a3s_code_core/

dynamic_workflow.rs

//! A3S Flow-backed dynamic workflow runtime.
//!
//! `DynamicWorkflowRuntime` lets hosts run a sandboxed PTC script as an A3S
//! Flow runtime. Flow owns durable replay and step lifecycle; A3S Code's
//! existing `program` tool remains the sandbox and tool-call boundary.

use crate::execution_identity::{
    ExecutionIdentityV1, DYNAMIC_WORKFLOW_CLAIM_IDENTITY_DOMAIN_V1,
    DYNAMIC_WORKFLOW_CONTINUATION_IDENTITY_DOMAIN_V1, DYNAMIC_WORKFLOW_INPUT_IDENTITY_DOMAIN_V1,
    FLOW_STEP_IDENTITY_DOMAIN_V1,
};
use crate::llm::{ModelGenerationAdmission, ModelGenerationConcurrency};
use crate::task_scheduler::{
    TaskLease, TaskPriority as SchedulerTaskPriority, TaskScheduler, TaskSchedulerQuota,
};
use crate::tools::{
    registry_tool_invoker, Tool, ToolContext, ToolInvoker, ToolOutput, ToolRegistry, ToolResult,
};
use crate::{
    agent::AgentEvent,
    flow_graph::{
        FileFlowDecisionLedger, FlowDecisionClaimOutcome, FlowDecisionClaimState,
        FlowDecisionLedger, FlowGraphObserver, MemoryFlowDecisionLedger,
    },
    planning::{Complexity, ExecutionPlan, Task, TaskStatus},
};
use a3s_flow::{
    CancellationRequest, FanoutFlowEventObserver, FlowEngine, FlowEvent, FlowEventEnvelope,
    FlowEventObserver, FlowEventStore, FlowRuntime, InMemoryEventStore, LocalFileEventStore,
    RuntimeBuildCompatibility, RuntimeBuildId, RuntimeCommand, StepInvocation, StepStatus,
    WorkflowInvocation, WorkflowRunSnapshot, WorkflowRunStatus, WorkflowSpec,
};
use anyhow::{Context, Result};
use async_trait::async_trait;
use chrono::Utc;
use serde::{Deserialize, Serialize};
use serde_json::{json, Map, Value};
use std::collections::{BTreeMap, BTreeSet};
use std::fs::OpenOptions;
use std::future::Future;
use std::num::NonZeroUsize;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, Weak};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::{broadcast, Mutex, OwnedSemaphorePermit, Semaphore};
use tokio_util::sync::CancellationToken;

const DYNAMIC_WORKFLOW_TOOL: &str = "dynamic_workflow";
const GENERATE_OBJECT_TOOL: &str = "generate_object";
const PROGRAM_TOOL: &str = "program";
const TASK_TOOL: &str = "task";
const PARALLEL_TASK_TOOL: &str = "parallel_task";
const MAX_INLINE_RETRY_RESUMES: usize = 8;
const MAX_INLINE_RETRY_DELAY: Duration = Duration::from_secs(5);
const DEFAULT_MAX_CONCURRENT_STEPS: usize = 4;
const MAX_MAX_CONCURRENT_STEPS: usize = 32;
const MAX_FLOW_STEP_ID_BYTES: usize = 256;
const MAX_FLOW_STEP_INPUT_BYTES: usize = 64 * 1024;
const MAX_DYNAMIC_WORKFLOW_INPUT_BYTES: usize = 128 * 1024;
const DEFAULT_DYNAMIC_WORKFLOW_LEASE_MS: u64 = 30_000;
const MAX_DYNAMIC_WORKFLOW_SETTLE: Duration = Duration::from_secs(5);
const MAX_DYNAMIC_WORKFLOW_CONTROL_REASON_BYTES: usize = 4 * 1024;
const DYNAMIC_WORKFLOW_LEASE_RELATIVE_PATH: &str = "leases";

fn dynamic_workflow_scheduler_quota_limit(limits: &DynamicWorkflowScriptLimits) -> usize {
    limits
        .max_concurrent_steps
        .unwrap_or(DEFAULT_MAX_CONCURRENT_STEPS)
        .clamp(1, MAX_MAX_CONCURRENT_STEPS)
}

fn provider_quota_for_context(
    context: &ToolContext,
) -> Option<crate::task_scheduler::TaskSchedulerQuota> {
    if context
        .model_generation_admission()
        .is_some_and(|admission| admission.has_scheduler_quota())
    {
        // The session-owned generation gate already reserves this exact pool
        // through the shared scheduler. Holding it again on the enclosing Flow
        // step would make a single-flight provider recursively wait on itself.
        return None;
    }
    let client = context.llm_client()?;
    let pool = client.model_generation_pool()?;
    crate::task_scheduler::TaskSchedulerQuota::new(
        pool.identity.clone(),
        pool.max_concurrency().get(),
    )
    .ok()
}

/// Runtime build used to pin newly-created dynamic workflow runs.
///
/// Deployments that need a stronger revision identity can provide an explicit
/// [`RuntimeBuildCompatibility`] through [`DynamicWorkflowTool::with_runtime_build_compatibility`].
pub const DYNAMIC_WORKFLOW_RUNTIME_BUILD_ID: &str =
    concat!("a3s-code-core-", env!("CARGO_PKG_VERSION"));

/// Legacy marker used only while deriving an identity for an unpinned history
/// created before dynamic workflow runtime-build fencing was enabled.
const LEGACY_UNPINNED_RUNTIME_BUILD_ID: &str = "<unpinned>";

/// Project-relative directory used for durable dynamic workflow history.
pub const DYNAMIC_WORKFLOW_STORE_RELATIVE_PATH: &str = ".a3s/workflow";

/// Cross-process serialized adapter for a local Flow event journal.
///
/// `a3s-flow::LocalFileEventStore` deliberately serializes writers only
/// inside one process. Dynamic workflows can be resumed or controlled by a
/// replacement process, so Code adds one small lock-file boundary around the
/// same append-only journal. The adapter does not project or cache workflow
/// state: Flow remains the sole event authority and its optimistic sequence
/// checks still decide whether an append is accepted.
#[derive(Debug, Clone)]
pub struct CrossProcessFlowEventStore {
    root: PathBuf,
    inner: LocalFileEventStore,
    process_lock: Arc<Mutex<()>>,
}

impl CrossProcessFlowEventStore {
    /// Create a cross-process event store rooted at `root`.
    pub fn new(root: impl Into<PathBuf>) -> Self {
        let root = root.into();
        Self {
            inner: LocalFileEventStore::new(root.clone()),
            root,
            process_lock: Arc::new(Mutex::new(())),
        }
    }

    /// Return the directory containing the journal files.
    pub fn root(&self) -> &Path {
        &self.root
    }

    async fn acquire_file_lock(&self) -> a3s_flow::Result<std::fs::File> {
        tokio::fs::create_dir_all(&self.root).await?;
        let lock_path = self.root.join(".flow-events.lock");
        match tokio::fs::symlink_metadata(&lock_path).await {
            Ok(metadata) if metadata.file_type().is_symlink() => {
                return Err(a3s_flow::FlowError::Store(format!(
                    "refusing to use symlinked event-store lock {}",
                    lock_path.display()
                )))
            }
            Ok(metadata) if !metadata.is_file() => {
                return Err(a3s_flow::FlowError::Store(format!(
                    "event-store lock {} exists but is not a file",
                    lock_path.display()
                )))
            }
            Ok(_) => {}
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
            Err(error) => return Err(a3s_flow::FlowError::Io(error)),
        }
        tokio::task::spawn_blocking(move || {
            use fs2::FileExt;
            let file = OpenOptions::new()
                .create(true)
                .truncate(false)
                .read(true)
                .write(true)
                .open(&lock_path)
                .map_err(a3s_flow::FlowError::Io)?;
            file.lock_exclusive().map_err(a3s_flow::FlowError::Io)?;
            Ok(file)
        })
        .await
        .map_err(|error| {
            a3s_flow::FlowError::Store(format!("event-store lock task failed: {error}"))
        })?
    }
}

#[async_trait]
impl FlowEventStore for CrossProcessFlowEventStore {
    async fn append(&self, run_id: &str, event: FlowEvent) -> a3s_flow::Result<FlowEventEnvelope> {
        let _process_guard = self.process_lock.lock().await;
        let _file_guard = self.acquire_file_lock().await?;
        self.inner.append(run_id, event).await
    }

    async fn append_if_sequence(
        &self,
        run_id: &str,
        expected_sequence: u64,
        event: FlowEvent,
    ) -> a3s_flow::Result<FlowEventEnvelope> {
        let _process_guard = self.process_lock.lock().await;
        let _file_guard = self.acquire_file_lock().await?;
        self.inner
            .append_if_sequence(run_id, expected_sequence, event)
            .await
    }

    async fn list(&self, run_id: &str) -> a3s_flow::Result<Vec<FlowEventEnvelope>> {
        let _process_guard = self.process_lock.lock().await;
        let _file_guard = self.acquire_file_lock().await?;
        self.inner.list(run_id).await
    }

    async fn list_run_ids(&self) -> a3s_flow::Result<Vec<String>> {
        let _process_guard = self.process_lock.lock().await;
        let _file_guard = self.acquire_file_lock().await?;
        self.inner.list_run_ids().await
    }
}

/// Resolve the durable dynamic workflow history directory for a local workspace.
pub fn dynamic_workflow_store_path(workspace_root: impl AsRef<Path>) -> PathBuf {
    workspace_root
        .as_ref()
        .join(DYNAMIC_WORKFLOW_STORE_RELATIVE_PATH)
}

/// Recover one completed step output from the exact durable workflow run.
///
/// Recovery is bound to the requested run ID and the original input query. It
/// never acts as a cross-run query cache and never promotes an incomplete step.
pub async fn recover_dynamic_workflow_step_output(
    workspace_root: impl AsRef<Path>,
    run_id: &str,
    expected_query: &str,
    step_id: &str,
) -> Result<Option<Value>> {
    if !safe_workflow_run_id(run_id) || expected_query.is_empty() || step_id.is_empty() {
        return Ok(None);
    }
    let workspace_root = workspace_root.as_ref();
    let store_root = dynamic_workflow_store_path(workspace_root);
    let log_path = store_root.join(format!("{run_id}.jsonl"));
    match tokio::fs::symlink_metadata(&log_path).await {
        Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
            return Ok(None)
        }
        Ok(_) => {}
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => {
            return Err(error).with_context(|| {
                format!("inspect dynamic workflow history {}", log_path.display())
            })
        }
    }
    validate_dynamic_workflow_directory(&workspace_root.join(".a3s"), ".a3s").await?;
    validate_dynamic_workflow_directory(&store_root, ".a3s/workflow").await?;
    validate_dynamic_workflow_log(&log_path).await?;

    let events = LocalFileEventStore::new(store_root).list(run_id).await?;
    let input_matches = events.iter().any(|envelope| {
        matches!(
            &envelope.event,
            FlowEvent::RunCreated { input, .. }
                if input.get("query").and_then(Value::as_str) == Some(expected_query)
        )
    });
    if !input_matches {
        return Ok(None);
    }
    Ok(events
        .iter()
        .rev()
        .find_map(|envelope| match &envelope.event {
            FlowEvent::StepCompleted {
                step_id: completed_step_id,
                output,
            } if completed_step_id == step_id => Some(output.clone()),
            _ => None,
        }))
}

/// Limits forwarded to the underlying PTC `program` tool.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct DynamicWorkflowScriptLimits {
    #[serde(skip_serializing_if = "Option::is_none")]
    pub timeout_ms: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_tool_calls: Option<usize>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub max_output_bytes: Option<usize>,
    /// Maximum independently session-bound model generations active at once.
    /// This orchestration limit is not forwarded to the PTC program sandbox.
    #[serde(default, skip_serializing)]
    pub max_concurrent_generations: Option<usize>,
    /// Maximum Flow step bodies that may execute concurrently for one run.
    /// This is an orchestration boundary and is never forwarded to QuickJS.
    #[serde(default, skip_serializing)]
    pub max_concurrent_steps: Option<usize>,
}

/// Point-in-time admission counters for one dynamic workflow runtime.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct DynamicWorkflowAdmissionStats {
    /// Effective per-run step concurrency limit.
    pub max_concurrent_steps: usize,
    /// Number of step bodies admitted since this runtime was created.
    pub admitted_steps: usize,
    /// Number of step bodies currently holding a local permit.
    pub active_steps: usize,
    /// Highest observed active step count.
    pub peak_active_steps: usize,
}

/// Bounded cumulative worker-claim diagnostics for one dynamic-workflow tool
/// instance.  The counters retain no run ids, labels, source, input, output,
/// or owner tokens; the durable Flow ledger remains the authority for the
/// per-run attempt number and lease state.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct DynamicWorkflowHealthSnapshot {
    /// Number of worker-claim attempts observed by this process.
    pub claim_attempts: u64,
    /// Number of claims that acquired a worker lease, including takeovers.
    pub claims: u64,
    /// Number of claims whose durable attempt number was greater than one.
    pub takeovers: u64,
    /// Number of attempts that found an already-completed claim.
    pub already_completed: u64,
    /// Number of attempts rejected because another worker is live.
    pub busy: u64,
    /// Number of identity/hash conflicts.
    pub conflicts: u64,
    /// Successful lease heartbeats.
    pub renewals: u64,
    /// Lease renewals or terminal operations fenced by a lost lease.
    pub lease_lost: u64,
    /// Successful terminal claim completions.
    pub completions: u64,
    /// Successful releases of non-terminal claims.
    pub releases: u64,
    /// Parent/control cancellation observations.
    pub cancellations: u64,
    /// Ledger or execution-boundary failures.
    pub failures: u64,
    /// Claims currently waiting on a ledger operation.
    pub in_flight: u64,
}

#[derive(Default)]
struct DynamicWorkflowMetrics {
    claim_attempts: AtomicU64,
    claims: AtomicU64,
    takeovers: AtomicU64,
    already_completed: AtomicU64,
    busy: AtomicU64,
    conflicts: AtomicU64,
    renewals: AtomicU64,
    lease_lost: AtomicU64,
    completions: AtomicU64,
    releases: AtomicU64,
    cancellations: AtomicU64,
    failures: AtomicU64,
    in_flight: AtomicU64,
}

impl DynamicWorkflowMetrics {
    fn snapshot(&self) -> DynamicWorkflowHealthSnapshot {
        DynamicWorkflowHealthSnapshot {
            claim_attempts: self.claim_attempts.load(Ordering::Relaxed),
            claims: self.claims.load(Ordering::Relaxed),
            takeovers: self.takeovers.load(Ordering::Relaxed),
            already_completed: self.already_completed.load(Ordering::Relaxed),
            busy: self.busy.load(Ordering::Relaxed),
            conflicts: self.conflicts.load(Ordering::Relaxed),
            renewals: self.renewals.load(Ordering::Relaxed),
            lease_lost: self.lease_lost.load(Ordering::Relaxed),
            completions: self.completions.load(Ordering::Relaxed),
            releases: self.releases.load(Ordering::Relaxed),
            cancellations: self.cancellations.load(Ordering::Relaxed),
            failures: self.failures.load(Ordering::Relaxed),
            in_flight: self.in_flight.load(Ordering::Relaxed),
        }
    }

    fn increment(counter: &AtomicU64) {
        counter
            .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |value| {
                Some(value.saturating_add(1))
            })
            .ok();
    }
}

/// Host-facing aggregate view over dynamic-workflow claim counters and the
/// optional agent-wide scheduler.  It is a read-only composition of local
/// metrics; it does not introduce a second workflow or event authority.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct DynamicWorkflowControlDiagnostics {
    pub workflow: DynamicWorkflowHealthSnapshot,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scheduler: Option<crate::task_scheduler::TaskSchedulerHealthSnapshot>,
    /// Live owner-quota occupancy for this workflow when direct Flow steps
    /// use the shared scheduler. The projection is absent when the runtime is
    /// session-bound or has no scheduler.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub scheduler_quota: Option<crate::task_scheduler::TaskSchedulerQuotaSnapshot>,
}

/// Bounded, host-facing projection of one dynamic workflow continuation.
///
/// The projection deliberately omits source, input, step arguments, outputs,
/// and worker owner tokens. Flow history remains available through the
/// explicit inspection APIs when a trusted host needs the full record.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
pub struct DynamicWorkflowControlSnapshot {
    /// Stable durable run identifier.
    pub run_id: String,
    /// Current materialized Flow status.
    pub status: WorkflowRunStatus,
    /// Last durable event sequence observed for the run.
    pub last_sequence: u64,
    /// Number of durable step definitions in the run.
    pub step_count: usize,
    /// Number of steps with a committed output.
    pub completed_steps: usize,
    /// Number of steps that remain actionable or in flight.
    pub open_steps: usize,
    /// Whether Flow has recorded a cleanup-aware cancellation request.
    pub cancellation_requested: bool,
    /// Digest-only identity reconstructed from immutable continuation facts.
    pub continuation_identity: ExecutionIdentityV1,
    /// Digest-only identity of the projected immutable step plan.
    pub plan_identity: ExecutionIdentityV1,
    /// Runtime build pinned by the durable run, or `None` for a legacy
    /// unpinned history.
    pub runtime_build_id: Option<String>,
    /// Redacted worker-lease state observed by the control operation.
    pub worker_lease: FlowDecisionClaimState,
}

/// A host-owned control handle for one dynamic workflow run.
///
/// The handle keeps the exact source, input, registry, and runtime policy
/// needed to replay the selected run. Mutating operations first acquire the
/// same worker lease used by [`DynamicWorkflowTool`], then ask A3S Flow to
/// append/drive its authoritative events. A live worker therefore remains the
/// sole executor; a controller retries after the redacted lease state reports
/// that the claim is busy.
#[must_use = "a dynamic workflow control handle should be used or explicitly dropped"]
#[derive(Clone)]
pub struct DynamicWorkflowControl {
    registry: Arc<ToolRegistry>,
    context: ToolContext,
    flow_event_store: Option<Arc<dyn FlowEventStore>>,
    run_id: String,
    source: Arc<str>,
    input: Value,
    allowed_tools: Vec<String>,
    limits: DynamicWorkflowScriptLimits,
    graph_observer: Option<FlowGraphObserver>,
    task_scheduler: Option<Arc<TaskScheduler>>,
    admit_steps_globally: bool,
    runtime_build_compatibility: Option<RuntimeBuildCompatibility>,
    continuation_lease_ledger: Option<Arc<dyn FlowDecisionLedger>>,
    continuation_lease_ms: u64,
    memory_continuation_lease_ledger: Arc<MemoryFlowDecisionLedger>,
    metrics: Arc<DynamicWorkflowMetrics>,
}

#[derive(Clone)]
struct DynamicStepAdmission {
    semaphore: Arc<Semaphore>,
    max_concurrent_steps: usize,
    admitted_steps: Arc<AtomicUsize>,
    active_steps: Arc<AtomicUsize>,
    peak_active_steps: Arc<AtomicUsize>,
}

impl DynamicStepAdmission {
    fn new(max_concurrent_steps: usize) -> Self {
        Self {
            semaphore: Arc::new(Semaphore::new(max_concurrent_steps)),
            max_concurrent_steps,
            admitted_steps: Arc::new(AtomicUsize::new(0)),
            active_steps: Arc::new(AtomicUsize::new(0)),
            peak_active_steps: Arc::new(AtomicUsize::new(0)),
        }
    }

    async fn acquire(
        &self,
        identity: ExecutionIdentityV1,
        cancellation: &tokio_util::sync::CancellationToken,
    ) -> a3s_flow::Result<DynamicStepLease> {
        let permit = tokio::select! {
            biased;
            _ = cancellation.cancelled() => {
                return Err(a3s_flow::FlowError::Runtime(
                    "dynamic workflow step admission cancelled".to_string(),
                ));
            }
            permit = Arc::clone(&self.semaphore).acquire_owned() => {
                permit.map_err(|_| a3s_flow::FlowError::Runtime(
                    "dynamic workflow step admission is closed".to_string(),
                ))?
            }
        };
        if cancellation.is_cancelled() {
            drop(permit);
            return Err(a3s_flow::FlowError::Runtime(
                "dynamic workflow step admission cancelled".to_string(),
            ));
        }

        let active = self.active_steps.fetch_add(1, Ordering::AcqRel) + 1;
        self.admitted_steps.fetch_add(1, Ordering::Relaxed);
        let mut observed = self.peak_active_steps.load(Ordering::Acquire);
        while active > observed {
            match self.peak_active_steps.compare_exchange(
                observed,
                active,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => break,
                Err(current) => observed = current,
            }
        }
        tracing::trace!(
            execution_identity = identity.key(),
            active_steps = active,
            max_concurrent_steps = self.max_concurrent_steps,
            "dynamic workflow step admitted"
        );
        Ok(DynamicStepLease {
            _permit: permit,
            identity,
            active_steps: Arc::clone(&self.active_steps),
            task_lease: None,
        })
    }

    fn stats(&self) -> DynamicWorkflowAdmissionStats {
        DynamicWorkflowAdmissionStats {
            max_concurrent_steps: self.max_concurrent_steps,
            admitted_steps: self.admitted_steps.load(Ordering::Acquire),
            active_steps: self.active_steps.load(Ordering::Acquire),
            peak_active_steps: self.peak_active_steps.load(Ordering::Acquire),
        }
    }
}

struct DynamicStepLease {
    _permit: OwnedSemaphorePermit,
    identity: ExecutionIdentityV1,
    active_steps: Arc<AtomicUsize>,
    task_lease: Option<TaskLease>,
}

impl DynamicStepLease {
    fn with_task_lease(mut self, task_lease: TaskLease) -> Self {
        self.task_lease = Some(task_lease);
        self
    }
}

impl Drop for DynamicStepLease {
    fn drop(&mut self) {
        let active = self.active_steps.fetch_sub(1, Ordering::AcqRel) - 1;
        tracing::trace!(
            execution_identity = self.identity.key(),
            active_steps = active,
            "dynamic workflow step admission released"
        );
    }
}

/// Runs A3S Flow workflow and step invocations through a sandboxed PTC script.
#[derive(Clone)]
pub struct DynamicWorkflowRuntime {
    invoker: Arc<dyn ToolInvoker>,
    context: ToolContext,
    source: Arc<str>,
    allowed_tools: Vec<String>,
    limits: DynamicWorkflowScriptLimits,
    parallel_generation_admission: Option<ModelGenerationAdmission>,
    step_admission: DynamicStepAdmission,
    /// Optional global scheduler for runtimes created outside an AgentSession.
    /// Session-bound callers already hold the enclosing scheduler lease and
    /// should leave this unset to avoid nested single-slot deadlocks.
    task_scheduler: Option<Arc<TaskScheduler>>,
    admit_steps_globally: bool,
    /// Optional owner quota for direct script-backed Flow steps. The quota is
    /// enforced by the same scheduler actor as global capacity.
    scheduler_quota: Option<crate::task_scheduler::TaskSchedulerQuota>,
    /// Provider/model capacity projected into the same scheduler actor as
    /// workflow-step admission.
    provider_quota: Option<crate::task_scheduler::TaskSchedulerQuota>,
    continuation_lease: Option<Arc<DynamicWorkflowLease>>,
}

impl DynamicWorkflowRuntime {
    pub fn new(
        registry: Arc<ToolRegistry>,
        context: ToolContext,
        source: impl Into<String>,
    ) -> Self {
        let provider_quota = provider_quota_for_context(&context);
        let allowed_tools = default_allowed_tools(&registry);
        // Session/agent callers install the governed gateway in ToolContext.
        // The raw registry adapter is retained only for explicit low-level
        // callers that construct this public runtime outside an AgentSession.
        let invoker = context
            .tool_invoker()
            .unwrap_or_else(|| registry_tool_invoker(registry));
        Self {
            invoker,
            context,
            source: Arc::from(source.into()),
            allowed_tools,
            limits: DynamicWorkflowScriptLimits::default(),
            parallel_generation_admission: None,
            step_admission: DynamicStepAdmission::new(DEFAULT_MAX_CONCURRENT_STEPS),
            task_scheduler: None,
            admit_steps_globally: false,
            scheduler_quota: None,
            provider_quota,
            continuation_lease: None,
        }
    }

    pub fn with_allowed_tools(mut self, allowed_tools: impl IntoIterator<Item = String>) -> Self {
        self.allowed_tools = sanitize_allowed_tools(allowed_tools);
        self
    }

    pub fn with_limits(mut self, limits: DynamicWorkflowScriptLimits) -> Self {
        let generation_concurrency = limits.max_concurrent_generations.unwrap_or(1).clamp(1, 4);
        self.parallel_generation_admission = NonZeroUsize::new(generation_concurrency)
            .filter(|maximum| maximum.get() > 1)
            .map(|maximum| {
                let admission =
                    ModelGenerationAdmission::new(ModelGenerationConcurrency::bounded(maximum));
                match self.context.model_generation_admission() {
                    Some(session_admission) => match admission.clone().with_scheduler_quota_from(
                        &session_admission,
                        "dynamic-workflow-model-generation",
                    ) {
                        Ok(admission) => admission,
                        Err(error) => {
                            tracing::warn!(
                                %error,
                                "failed to project session provider quota into dynamic workflow admission"
                            );
                            admission
                        }
                    },
                    None => admission,
                }
            });
        let step_concurrency = limits
            .max_concurrent_steps
            .unwrap_or(DEFAULT_MAX_CONCURRENT_STEPS)
            .clamp(1, MAX_MAX_CONCURRENT_STEPS);
        self.step_admission = DynamicStepAdmission::new(step_concurrency);
        self.limits = limits;
        self
    }

    /// Attach the agent-wide scheduler when this runtime is used as a
    /// standalone host adapter.
    ///
    /// `admit_steps_globally = false` is the correct setting for a runtime
    /// invoked from an existing AgentSession operation: the parent operation
    /// already owns the global lease and Flow steps use the local bounded gate.
    pub fn with_task_scheduler(
        mut self,
        scheduler: Arc<TaskScheduler>,
        admit_steps_globally: bool,
    ) -> Self {
        self.task_scheduler = Some(scheduler);
        self.admit_steps_globally = admit_steps_globally;
        self
    }

    /// Bind a digest-only owner quota to scheduler-backed Flow admissions.
    ///
    /// This does not create another queue or workflow authority; it only tells
    /// the existing scheduler how many slots this runtime may own at once.
    pub fn with_task_scheduler_quota(
        mut self,
        quota: crate::task_scheduler::TaskSchedulerQuota,
    ) -> Self {
        self.scheduler_quota = Some(quota);
        self
    }

    fn with_continuation_lease(mut self, lease: Arc<DynamicWorkflowLease>) -> Self {
        self.continuation_lease = Some(lease);
        self
    }

    /// Return local Flow-step admission counters for diagnostics and hosts.
    pub fn admission_stats(&self) -> DynamicWorkflowAdmissionStats {
        self.step_admission.stats()
    }

    /// Return the live owner-quota projection for this runtime, when it is
    /// layered on an agent-wide scheduler. The projection contains only the
    /// digest identity and current occupancy; Flow history and worker leases
    /// remain independent authorities.
    pub async fn scheduler_quota_snapshot(
        &self,
    ) -> std::result::Result<
        Option<crate::task_scheduler::TaskSchedulerQuotaSnapshot>,
        crate::task_scheduler::TaskSchedulerError,
    > {
        match (&self.task_scheduler, &self.scheduler_quota) {
            (Some(scheduler), Some(quota)) => scheduler.quota_snapshot(quota).await.map(Some),
            _ => Ok(None),
        }
    }

    async fn admit_step(&self, invocation: &StepInvocation) -> a3s_flow::Result<DynamicStepLease> {
        self.ensure_continuation_lease().await?;
        let identity = dynamic_workflow_step_identity(
            &invocation.run_id,
            &invocation.step_id,
            &invocation.step_name,
            &invocation.input,
        )
        .map_err(|error| a3s_flow::FlowError::Runtime(error.to_string()))?;
        let mut lease = self
            .step_admission
            .acquire(identity.clone(), &self.context.cancellation_token())
            .await?;

        // Host task fan-out has its own child-run scheduler boundary. Holding
        // a second global lease here would deadlock a max_active=1 scheduler
        // while the child waits for capacity, so only direct script-backed
        // steps opt into the standalone global admission.
        if self.admit_steps_globally
            && !matches!(
                invocation.step_name.as_str(),
                TASK_TOOL | PARALLEL_TASK_TOOL
            )
        {
            let Some(scheduler) = self.task_scheduler.as_ref() else {
                return Err(a3s_flow::FlowError::Runtime(
                    "global Flow-step admission requires a configured task scheduler".to_string(),
                ));
            };
            let label = format!(
                "flow:{}:{}:{}",
                invocation.run_id, invocation.step_id, invocation.step_name
            );
            let mut quotas = Vec::with_capacity(2);
            if let Some(quota) = self.scheduler_quota.as_ref() {
                quotas.push(quota.clone());
            }
            if let Some(quota) = self.provider_quota.as_ref() {
                if !quotas
                    .iter()
                    .any(|candidate| candidate.identity == quota.identity)
                {
                    quotas.push(quota.clone());
                }
            }
            let task_lease = if quotas.is_empty() {
                scheduler
                    .acquire_with_identity(
                        SchedulerTaskPriority::Foreground,
                        label,
                        Some(identity),
                        &self.context.cancellation_token(),
                    )
                    .await
            } else {
                scheduler
                    .acquire_with_quotas(
                        SchedulerTaskPriority::Foreground,
                        label,
                        &quotas,
                        Some(identity),
                        &self.context.cancellation_token(),
                    )
                    .await
            }
            .map_err(|error| a3s_flow::FlowError::Runtime(error.to_string()))?;
            lease = lease.with_task_lease(task_lease);
        }
        Ok(lease)
    }

    async fn ensure_continuation_lease(&self) -> a3s_flow::Result<()> {
        let Some(lease) = self.continuation_lease.as_ref() else {
            return Ok(());
        };
        if lease
            .renew()
            .await
            .map_err(|error| a3s_flow::FlowError::Runtime(error.to_string()))?
        {
            Ok(())
        } else {
            Err(a3s_flow::FlowError::Runtime(
                "dynamic workflow worker lease is no longer owned before admission".to_string(),
            ))
        }
    }

    async fn run_script(
        &self,
        payload: Value,
        context: &ToolContext,
    ) -> a3s_flow::Result<ToolResult> {
        let mut args = json!({
            "type": "script",
            "language": "javascript",
            "source": self.source.as_ref(),
            "inputs": payload,
            "allowed_tools": self.allowed_tools,
        });
        if let Some(object) = args.as_object_mut() {
            if let Ok(Value::Object(limits)) = serde_json::to_value(&self.limits) {
                if !limits.is_empty() {
                    object.insert("limits".to_string(), Value::Object(limits));
                }
            }
        }

        let result = self
            .invoker
            .invoke(
                crate::tools::ToolInvocation::runtime_internal(PROGRAM_TOOL, args),
                context,
            )
            .await;
        if result.exit_code != 0 {
            return Err(a3s_flow::FlowError::Runtime(result.output));
        }
        Ok(result)
    }

    async fn context_for_step(
        &self,
        run_id: &str,
        step_id: &str,
        step_name: &str,
    ) -> a3s_flow::Result<ToolContext> {
        if step_name != GENERATE_OBJECT_TOOL {
            return Ok(self.context.clone().with_run_id(run_id.to_string()));
        }
        if let (Some(admission), Some(client)) = (
            self.parallel_generation_admission.as_ref(),
            self.context.llm_client(),
        ) {
            let fork_id = format!("{run_id}:{step_id}");
            if let Some(forked_client) = client.fork_for_session(&fork_id) {
                let permit = admission
                    .acquire(&self.context.cancellation_token())
                    .await
                    .map_err(|error| {
                        a3s_flow::FlowError::Runtime(format!(
                            "parallel model-generation admission failed before workflow step: {error}"
                        ))
                    })?;
                return self
                    .context
                    .clone()
                    .with_run_id(run_id.to_string())
                    .with_llm_client(forked_client)
                    .with_model_generation_permit(admission.clone(), Arc::new(permit))
                    .map_err(|error| {
                        a3s_flow::FlowError::Runtime(format!(
                            "bind parallel model-generation admission to workflow step: {error}"
                        ))
                    });
            }
        }
        let Some(admission) = self.context.model_generation_admission() else {
            return Ok(self.context.clone().with_run_id(run_id.to_string()));
        };
        let permit = admission
            .acquire(&self.context.cancellation_token())
            .await
            .map_err(|error| {
                a3s_flow::FlowError::Runtime(format!(
                    "model-generation admission failed before workflow step: {error}"
                ))
            })?;
        self.context
            .clone()
            .with_run_id(run_id.to_string())
            .with_model_generation_permit(admission, Arc::new(permit))
            .map_err(|error| {
                a3s_flow::FlowError::Runtime(format!(
                    "bind model-generation admission to workflow step: {error}"
                ))
            })
    }

    async fn run_tool_step(
        &self,
        run_id: &str,
        tool_name: &str,
        args: Value,
    ) -> a3s_flow::Result<Value> {
        let context = self.context.clone().with_run_id(run_id.to_string());
        let result = self
            .invoker
            .invoke(
                self.context
                    .nested_tool_invocation(tool_name.to_string(), args),
                &context,
            )
            .await;
        if result.exit_code != 0 {
            return Err(a3s_flow::FlowError::Runtime(result.output));
        }
        Ok(json!({
            "tool": result.name,
            "output": result.output,
            "exit_code": result.exit_code,
            "metadata": result.metadata,
        }))
    }
}

#[async_trait]
impl FlowRuntime for DynamicWorkflowRuntime {
    async fn run_workflow(
        &self,
        invocation: WorkflowInvocation,
    ) -> a3s_flow::Result<RuntimeCommand> {
        self.ensure_continuation_lease().await?;
        let payload = invocation_payload("workflow", &invocation.run_id, &invocation.history)
            .with("input", invocation.input);
        let context = self
            .context
            .clone()
            .with_run_id(invocation.run_id.to_string());
        let result = self.run_script(payload.into_value(), &context).await?;
        serde_json::from_value(script_result(&result)?).map_err(a3s_flow::FlowError::from)
    }

    async fn run_step(&self, invocation: StepInvocation) -> a3s_flow::Result<Value> {
        let _admission = self.admit_step(&invocation).await?;
        if matches!(
            invocation.step_name.as_str(),
            TASK_TOOL | PARALLEL_TASK_TOOL
        ) {
            return self
                .run_tool_step(&invocation.run_id, &invocation.step_name, invocation.input)
                .await;
        }

        let context = self
            .context_for_step(
                &invocation.run_id,
                &invocation.step_id,
                &invocation.step_name,
            )
            .await?;
        let payload = invocation_payload("step", &invocation.run_id, &invocation.history)
            .with("step_id", invocation.step_id)
            .with("step_name", invocation.step_name)
            .with("input", invocation.input);
        let result = self.run_script(payload.into_value(), &context).await?;
        script_result(&result)
    }
}

/// Derive the stable identity for one dynamic Flow step admission.
///
/// The returned value is digest-only. The step input is included in the
/// derivation so retries with different arguments cannot share a lease, but no
/// input bytes are retained in the identity or emitted by the scheduler.
pub fn dynamic_workflow_step_identity(
    run_id: &str,
    step_id: &str,
    step_name: &str,
    input: &Value,
) -> Result<ExecutionIdentityV1, crate::execution_identity::ExecutionIdentityError> {
    for (field, value) in [
        ("run_id", run_id),
        ("step_id", step_id),
        ("step_name", step_name),
    ] {
        if value.is_empty()
            || value.len() > MAX_FLOW_STEP_ID_BYTES
            || value.contains('\0')
            || value.lines().count() != 1
        {
            return Err(
                crate::execution_identity::ExecutionIdentityError::InvalidClaimField(field),
            );
        }
    }
    let encoded_input = serde_json::to_vec(input).map_err(|error| {
        crate::execution_identity::ExecutionIdentityError::Serialization(error.to_string())
    })?;
    if encoded_input.len() > MAX_FLOW_STEP_INPUT_BYTES {
        return Err(
            crate::execution_identity::ExecutionIdentityError::Serialization(format!(
                "dynamic Flow step input exceeds {} bytes",
                MAX_FLOW_STEP_INPUT_BYTES
            )),
        );
    }
    ExecutionIdentityV1::derive(
        FLOW_STEP_IDENTITY_DOMAIN_V1,
        &json!({
            "run_id": run_id,
            "step_id": step_id,
            "step_name": step_name,
            "input": input,
        }),
    )
}

fn dynamic_workflow_input_identity(
    input: &Value,
) -> Result<ExecutionIdentityV1, crate::execution_identity::ExecutionIdentityError> {
    let encoded = serde_json::to_vec(input).map_err(|error| {
        crate::execution_identity::ExecutionIdentityError::Serialization(error.to_string())
    })?;
    if encoded.len() > MAX_DYNAMIC_WORKFLOW_INPUT_BYTES {
        return Err(
            crate::execution_identity::ExecutionIdentityError::Serialization(format!(
                "dynamic workflow input exceeds {} bytes",
                MAX_DYNAMIC_WORKFLOW_INPUT_BYTES
            )),
        );
    }
    ExecutionIdentityV1::derive(DYNAMIC_WORKFLOW_INPUT_IDENTITY_DOMAIN_V1, input)
}

/// Derive the stable, digest-only claim identity for one dynamic workflow.
///
/// Unlike the continuation identity, this identity intentionally excludes the
/// evolving plan and step history. It therefore remains constant while a
/// worker replays, retries, or takes over the same durable run. The complete
/// continuation identity is still validated first, so malformed or
/// mixed-generation history can never acquire a worker lease.
pub fn dynamic_workflow_claim_identity(
    run_id: &str,
    source: &str,
    input: &Value,
    runtime_build_id: &str,
    history: &[FlowEventEnvelope],
) -> std::result::Result<ExecutionIdentityV1, crate::execution_identity::ExecutionIdentityError> {
    dynamic_workflow_continuation_identity(run_id, source, input, runtime_build_id, history)?;
    let input_identity = dynamic_workflow_input_identity(input)?;
    let effective_runtime_build_id = history
        .iter()
        .find_map(|envelope| match &envelope.event {
            FlowEvent::RunCreated { spec, .. } => Some(
                spec.runtime_build_id
                    .as_ref()
                    .map(ToString::to_string)
                    .unwrap_or_else(|| LEGACY_UNPINNED_RUNTIME_BUILD_ID.to_string()),
            ),
            _ => None,
        })
        .unwrap_or_else(|| runtime_build_id.to_string());
    ExecutionIdentityV1::derive(
        DYNAMIC_WORKFLOW_CLAIM_IDENTITY_DOMAIN_V1,
        &json!({
            "run_id": run_id,
            "source_hash": source_hash(source),
            "input_identity": input_identity.digest,
            "runtime_build_id": effective_runtime_build_id,
        }),
    )
}

/// Reconstruct the immutable identity of a dynamic workflow continuation.
///
/// The Flow journal already persists the run definition and every step
/// definition. This adapter binds those facts to the current runtime build,
/// source, and initial input without introducing a second journal. Progress,
/// retries, event sequence, and step outputs are intentionally excluded, so a
/// restart observes the same continuation identity before and after replay.
/// A malformed or mixed-generation history is rejected before a step body can
/// be admitted.
pub fn dynamic_workflow_continuation_identity(
    run_id: &str,
    source: &str,
    input: &Value,
    runtime_build_id: &str,
    history: &[FlowEventEnvelope],
) -> std::result::Result<ExecutionIdentityV1, crate::execution_identity::ExecutionIdentityError> {
    if !safe_workflow_run_id(run_id) {
        return Err(crate::execution_identity::ExecutionIdentityError::InvalidClaimField("run_id"));
    }
    if source.is_empty() {
        return Err(crate::execution_identity::ExecutionIdentityError::InvalidClaimField("source"));
    }
    RuntimeBuildId::new(runtime_build_id.to_string()).map_err(|error| {
        crate::execution_identity::ExecutionIdentityError::Serialization(format!(
            "invalid dynamic workflow runtime build id: {error}"
        ))
    })?;
    let input_identity = dynamic_workflow_input_identity(input)?;
    let expected_source_hash = source_hash(source);
    let expected_spec = WorkflowSpec::rust_embedded(
        "a3s-code.dynamic-workflow",
        expected_source_hash.as_str(),
        "ptc",
        "run",
    );
    let mut saw_run_created = false;
    let mut persisted_runtime_build: Option<String> = None;
    let mut step_identities = BTreeMap::<String, String>::new();
    let mut terminal_seen = false;

    for (index, envelope) in history.iter().enumerate() {
        if envelope.run_id != run_id {
            return Err(
                crate::execution_identity::ExecutionIdentityError::InvalidClaimField("run_id"),
            );
        }
        if envelope.sequence != index as u64 + 1 {
            return Err(
                crate::execution_identity::ExecutionIdentityError::InvalidClaimField("sequence"),
            );
        }
        if terminal_seen {
            return Err(
                crate::execution_identity::ExecutionIdentityError::InvalidClaimField("terminal"),
            );
        }
        if index == 0 && !matches!(&envelope.event, FlowEvent::RunCreated { .. }) {
            return Err(
                crate::execution_identity::ExecutionIdentityError::InvalidClaimField("run_created"),
            );
        }

        match &envelope.event {
            FlowEvent::RunCreated {
                spec,
                input: persisted_input,
            } => {
                if saw_run_created {
                    return Err(
                        crate::execution_identity::ExecutionIdentityError::InvalidClaimField(
                            "run_created",
                        ),
                    );
                }
                saw_run_created = true;
                if spec.name != expected_spec.name
                    || spec.runtime != expected_spec.runtime
                    || !spec.patch_markers.is_empty()
                    || !spec.signal_names.is_empty()
                {
                    return Err(
                        crate::execution_identity::ExecutionIdentityError::InvalidClaimField(
                            "workflow_spec",
                        ),
                    );
                }
                if spec.version != expected_source_hash {
                    return Err(
                        crate::execution_identity::ExecutionIdentityError::InvalidClaimField(
                            "source",
                        ),
                    );
                }
                let persisted_input_identity = dynamic_workflow_input_identity(persisted_input)?;
                if persisted_input_identity != input_identity {
                    return Err(
                        crate::execution_identity::ExecutionIdentityError::InvalidClaimField(
                            "input",
                        ),
                    );
                }
                if let Some(build_id) = &spec.runtime_build_id {
                    RuntimeBuildId::new(build_id.as_str().to_string()).map_err(|error| {
                        crate::execution_identity::ExecutionIdentityError::Serialization(format!(
                            "invalid persisted dynamic workflow runtime build id: {error}"
                        ))
                    })?;
                    persisted_runtime_build = Some(build_id.as_str().to_string());
                } else {
                    persisted_runtime_build = Some(LEGACY_UNPINNED_RUNTIME_BUILD_ID.to_string());
                }
            }
            FlowEvent::StepCreated {
                step_id,
                step_name,
                input,
                retry,
            } => {
                let admission_identity =
                    dynamic_workflow_step_identity(run_id, step_id, step_name, input)?;
                // Retry behavior is part of Flow's immutable step definition,
                // even though it is not needed for the scheduler admission
                // lease. Bind it here so a conflicting duplicate cannot hide
                // behind the digest-only admission identity.
                let identity = ExecutionIdentityV1::derive(
                    FLOW_STEP_IDENTITY_DOMAIN_V1,
                    &json!({
                        "admission": admission_identity.digest,
                        "retry": retry,
                    }),
                )?;
                if step_identities
                    .insert(step_id.clone(), identity.digest)
                    .is_some()
                {
                    return Err(
                        crate::execution_identity::ExecutionIdentityError::InvalidClaimField(
                            "step_definition",
                        ),
                    );
                }
            }
            _ => {}
        }

        terminal_seen = matches!(
            &envelope.event,
            FlowEvent::RunCompleted { .. }
                | FlowEvent::RunFailed { .. }
                | FlowEvent::RunCancelled { .. }
                | FlowEvent::RunTimedOut { .. }
                | FlowEvent::RunRetryExhausted { .. }
                | FlowEvent::RunHostShutdown { .. }
                | FlowEvent::RunContinuedAsNew { .. }
        );
    }

    if !history.is_empty() && !saw_run_created {
        return Err(
            crate::execution_identity::ExecutionIdentityError::InvalidClaimField("run_created"),
        );
    }
    let effective_runtime_build = persisted_runtime_build
        .as_deref()
        .unwrap_or(runtime_build_id);
    if effective_runtime_build != LEGACY_UNPINNED_RUNTIME_BUILD_ID {
        RuntimeBuildId::new(effective_runtime_build.to_string()).map_err(|error| {
            crate::execution_identity::ExecutionIdentityError::Serialization(format!(
                "invalid effective dynamic workflow runtime build id: {error}"
            ))
        })?;
    }
    let plan = dynamic_workflow_execution_plan(history);
    let plan_identity = plan.definition_identity()?;
    ExecutionIdentityV1::derive(
        DYNAMIC_WORKFLOW_CONTINUATION_IDENTITY_DOMAIN_V1,
        &json!({
            "run_id": run_id,
            "source_hash": expected_source_hash,
            "input_identity": input_identity.digest,
            "runtime_build_id": effective_runtime_build,
            "plan_identity": plan_identity.digest,
            "step_identities": step_identities,
        }),
    )
}

/// Project the complete durable Flow history into Code's canonical plan model.
///
/// Flow remains the execution authority; this is a read-only adapter used by
/// progress events and metadata. Replaying the full history (rather than only
/// observing newly appended events) makes resumed runs expose the same plan as
/// fresh runs.
pub fn dynamic_workflow_execution_plan(history: &[FlowEventEnvelope]) -> ExecutionPlan {
    let mut plan = ExecutionPlan::new("dynamic workflow", Complexity::Medium);
    for envelope in history {
        match &envelope.event {
            FlowEvent::StepCreated {
                step_id,
                step_name,
                input,
                ..
            } => {
                let task = Task::new(
                    step_id.clone(),
                    workflow_step_description(step_id, step_name, Some(input)),
                )
                .with_tool(step_name.clone());
                plan.upsert_step(task);
            }
            FlowEvent::StepStarted { step_id, .. } => {
                plan.mark_status(step_id, TaskStatus::InProgress);
            }
            FlowEvent::StepRetrying { step_id, .. } => {
                plan.mark_status(step_id, TaskStatus::InProgress);
            }
            FlowEvent::StepCompleted { step_id, .. } => {
                plan.mark_status(step_id, TaskStatus::Completed);
            }
            FlowEvent::StepFailed { step_id, .. } => {
                plan.mark_status(step_id, TaskStatus::Failed);
            }
            FlowEvent::RunFailed { .. }
            | FlowEvent::RunTimedOut { .. }
            | FlowEvent::RunRetryExhausted { .. } => {
                for task in &mut plan.steps {
                    if task.status.is_active() {
                        task.status = TaskStatus::Failed;
                    }
                }
            }
            FlowEvent::RunCancelled { .. } | FlowEvent::RunHostShutdown { .. } => {
                for task in &mut plan.steps {
                    if task.status.is_active() {
                        task.status = TaskStatus::Cancelled;
                    }
                }
            }
            _ => {}
        }
    }
    plan
}

struct WorkflowProgressState {
    plan: ExecutionPlan,
}

impl WorkflowProgressState {
    fn new(plan: ExecutionPlan) -> Self {
        Self { plan }
    }

    fn upsert_step(
        &mut self,
        step_id: &str,
        step_name: &str,
        input: Option<&Value>,
        status: TaskStatus,
    ) {
        let content = workflow_step_description(step_id, step_name, input);
        self.plan.upsert_step(
            Task::new(step_id.to_string(), content)
                .with_tool(step_name)
                .with_status(status),
        );
    }

    fn mark_status(&mut self, step_id: &str, status: TaskStatus) {
        self.plan.mark_status(step_id, status);
    }

    fn step_position(&self, step_id: &str) -> (usize, usize) {
        let total = self.plan.steps.len().max(1);
        let number = self
            .plan
            .steps
            .iter()
            .position(|task| task.id == step_id)
            .map(|idx| idx + 1)
            .unwrap_or(total);
        (number, total)
    }

    fn step_description(&self, step_id: &str) -> String {
        self.plan
            .steps
            .iter()
            .find(|task| task.id == step_id)
            .map(|task| task.content.clone())
            .unwrap_or_else(|| step_id.to_string())
    }

    fn tasks(&self) -> &[Task] {
        &self.plan.steps
    }

    fn snapshot(&self) -> ExecutionPlan {
        self.plan.clone()
    }
}

struct AgentEventFlowObserver {
    tx: broadcast::Sender<AgentEvent>,
    session_id: String,
    state: Mutex<WorkflowProgressState>,
}

impl AgentEventFlowObserver {
    fn new(tx: broadcast::Sender<AgentEvent>, session_id: String, plan: ExecutionPlan) -> Self {
        Self {
            tx,
            session_id,
            state: Mutex::new(WorkflowProgressState::new(plan)),
        }
    }

    fn emit_task_update(&self, tasks: &[Task]) {
        let _ = self.tx.send(AgentEvent::TaskUpdated {
            session_id: self.session_id.clone(),
            tasks: tasks.to_vec(),
        });
    }
}

#[async_trait]
impl FlowEventObserver for AgentEventFlowObserver {
    async fn observe(&self, envelope: FlowEventEnvelope) {
        match envelope.event {
            FlowEvent::RunStarted => {
                let _ = self.tx.send(AgentEvent::PlanningStart {
                    prompt: "dynamic_workflow".to_string(),
                });
                let state = self.state.lock().await;
                if !state.plan.steps.is_empty() {
                    let plan = state.snapshot();
                    let _ = self.tx.send(AgentEvent::PlanningEnd {
                        estimated_steps: plan.steps.len(),
                        plan,
                    });
                }
            }
            FlowEvent::StepCreated {
                step_id,
                step_name,
                input,
                ..
            } => {
                let mut state = self.state.lock().await;
                state.upsert_step(&step_id, &step_name, Some(&input), TaskStatus::Pending);
                self.emit_task_update(state.tasks());
                let plan = state.snapshot();
                let _ = self.tx.send(AgentEvent::PlanningEnd {
                    estimated_steps: plan.steps.len(),
                    plan,
                });
            }
            FlowEvent::StepStarted { step_id, .. } => {
                let mut state = self.state.lock().await;
                state.mark_status(&step_id, TaskStatus::InProgress);
                self.emit_task_update(state.tasks());
                let (step_number, total_steps) = state.step_position(&step_id);
                let _ = self.tx.send(AgentEvent::StepStart {
                    description: state.step_description(&step_id),
                    step_id,
                    step_number,
                    total_steps,
                });
            }
            FlowEvent::StepCompleted { step_id, .. } => {
                let mut state = self.state.lock().await;
                state.mark_status(&step_id, TaskStatus::Completed);
                self.emit_task_update(state.tasks());
                let (step_number, total_steps) = state.step_position(&step_id);
                let _ = self.tx.send(AgentEvent::StepEnd {
                    step_id,
                    status: TaskStatus::Completed,
                    step_number,
                    total_steps,
                });
            }
            FlowEvent::StepRetrying { step_id, .. } => {
                let mut state = self.state.lock().await;
                state.mark_status(&step_id, TaskStatus::InProgress);
                self.emit_task_update(state.tasks());
            }
            FlowEvent::StepFailed { step_id, .. } => {
                let mut state = self.state.lock().await;
                state.mark_status(&step_id, TaskStatus::Failed);
                self.emit_task_update(state.tasks());
                let (step_number, total_steps) = state.step_position(&step_id);
                let _ = self.tx.send(AgentEvent::StepEnd {
                    step_id,
                    status: TaskStatus::Failed,
                    step_number,
                    total_steps,
                });
            }
            FlowEvent::RunFailed { .. }
            | FlowEvent::RunTimedOut { .. }
            | FlowEvent::RunRetryExhausted { .. } => {
                let mut state = self.state.lock().await;
                for task in &mut state.plan.steps {
                    if task.status.is_active() {
                        task.status = TaskStatus::Failed;
                    }
                }
                self.emit_task_update(state.tasks());
            }
            FlowEvent::RunCancelled { .. } | FlowEvent::RunHostShutdown { .. } => {
                let mut state = self.state.lock().await;
                for task in &mut state.plan.steps {
                    if task.status.is_active() {
                        task.status = TaskStatus::Cancelled;
                    }
                }
                self.emit_task_update(state.tasks());
            }
            _ => {}
        }
    }
}

fn workflow_step_description(step_id: &str, step_name: &str, input: Option<&Value>) -> String {
    if matches!(step_name, TASK_TOOL | PARALLEL_TASK_TOOL) {
        let count = input
            .and_then(|value| value.get("tasks"))
            .and_then(Value::as_array)
            .map(Vec::len)
            .unwrap_or(0);
        if count > 0 {
            return bounded_workflow_description(&format!(
                "Fan out {count} parallel subagent task(s)"
            ));
        }
    }

    let description = input
        .and_then(|value| value.get("description").or_else(|| value.get("title")))
        .and_then(Value::as_str)
        .map(str::to_string)
        .unwrap_or_else(|| {
            if step_name == step_id {
                step_id.to_string()
            } else {
                format!("{step_name}: {step_id}")
            }
        });
    bounded_workflow_description(&description)
}

fn bounded_workflow_description(value: &str) -> String {
    const MAX_DESCRIPTION_BYTES: usize = 512;
    if value.len() <= MAX_DESCRIPTION_BYTES {
        return value.to_string();
    }
    let mut end = MAX_DESCRIPTION_BYTES.saturating_sub("…".len());
    while end > 0 && !value.is_char_boundary(end) {
        end -= 1;
    }
    format!("{}…", &value[..end])
}

#[derive(Clone)]
struct DynamicWorkflowLease {
    ledger: Arc<dyn FlowDecisionLedger>,
    decision_id: String,
    request_hash: String,
    identity: ExecutionIdentityV1,
    owner_id: String,
    lease_ms: u64,
    attempt: u32,
    metrics: Arc<DynamicWorkflowMetrics>,
}

impl DynamicWorkflowLease {
    async fn renew(&self) -> Result<bool> {
        let result = self
            .ledger
            .renew_with_identity(
                &self.decision_id,
                &self.request_hash,
                &self.identity,
                &self.owner_id,
                dynamic_workflow_now_ms(),
                self.lease_ms,
            )
            .await;
        match &result {
            Ok(true) => DynamicWorkflowMetrics::increment(&self.metrics.renewals),
            Ok(false) => DynamicWorkflowMetrics::increment(&self.metrics.lease_lost),
            Err(_) => {
                DynamicWorkflowMetrics::increment(&self.metrics.lease_lost);
                DynamicWorkflowMetrics::increment(&self.metrics.failures);
            }
        }
        result
    }

    async fn complete(&self) -> Result<()> {
        let result = self
            .ledger
            .complete_with_identity(
                &self.decision_id,
                &self.request_hash,
                &self.identity,
                &self.owner_id,
                dynamic_workflow_now_ms(),
            )
            .await;
        match &result {
            Ok(()) => DynamicWorkflowMetrics::increment(&self.metrics.completions),
            Err(_) => DynamicWorkflowMetrics::increment(&self.metrics.failures),
        }
        result
    }

    async fn release(&self) -> Result<()> {
        let result = self
            .ledger
            .release_with_identity(
                &self.decision_id,
                &self.request_hash,
                &self.identity,
                &self.owner_id,
            )
            .await;
        match &result {
            Ok(()) => DynamicWorkflowMetrics::increment(&self.metrics.releases),
            Err(_) => DynamicWorkflowMetrics::increment(&self.metrics.failures),
        }
        result
    }
}

enum DynamicWorkflowLeaseClaim {
    Owned(DynamicWorkflowLease),
    AlreadyCompleted,
}

fn dynamic_workflow_lease_key(identity: &ExecutionIdentityV1) -> (String, String) {
    (
        format!("dynamic-workflow:{}", identity.digest),
        identity.digest.clone(),
    )
}

async fn claim_dynamic_workflow_lease(
    ledger: Arc<dyn FlowDecisionLedger>,
    identity: ExecutionIdentityV1,
    lease_ms: u64,
    metrics: Arc<DynamicWorkflowMetrics>,
) -> Result<DynamicWorkflowLeaseClaim> {
    DynamicWorkflowMetrics::increment(&metrics.claim_attempts);
    let mut in_flight = DynamicWorkflowClaimInFlight::new(Arc::clone(&metrics));
    identity
        .validate()
        .map_err(|error| anyhow::anyhow!(error))?;
    let (decision_id, request_hash) = dynamic_workflow_lease_key(&identity);
    let owner_id = format!("dynamic-workflow-worker-{}", uuid::Uuid::new_v4());
    let outcome = ledger
        .claim_with_identity(
            &decision_id,
            &request_hash,
            &identity,
            &owner_id,
            dynamic_workflow_now_ms(),
            lease_ms,
        )
        .await
        .context("admit dynamic workflow worker lease")?;
    let result = match outcome {
        FlowDecisionClaimOutcome::Claimed { attempt } => {
            DynamicWorkflowMetrics::increment(&metrics.claims);
            if attempt > 1 {
                DynamicWorkflowMetrics::increment(&metrics.takeovers);
            }
            Ok(DynamicWorkflowLeaseClaim::Owned(DynamicWorkflowLease {
                ledger,
                decision_id,
                request_hash,
                identity,
                owner_id,
                lease_ms,
                attempt,
                metrics,
            }))
        }
        FlowDecisionClaimOutcome::Completed => {
            DynamicWorkflowMetrics::increment(&metrics.already_completed);
            Ok(DynamicWorkflowLeaseClaim::AlreadyCompleted)
        }
        FlowDecisionClaimOutcome::Busy {
            lease_expires_at_ms,
        } => {
            DynamicWorkflowMetrics::increment(&metrics.busy);
            anyhow::bail!("dynamic workflow worker lease is busy until {lease_expires_at_ms}")
        }
        FlowDecisionClaimOutcome::Conflict => {
            DynamicWorkflowMetrics::increment(&metrics.conflicts);
            anyhow::bail!("dynamic workflow worker lease identity conflicts with its claim")
        }
    };
    in_flight.finish();
    result
}

struct DynamicWorkflowClaimInFlight {
    metrics: Arc<DynamicWorkflowMetrics>,
    finished: bool,
}

impl DynamicWorkflowClaimInFlight {
    fn new(metrics: Arc<DynamicWorkflowMetrics>) -> Self {
        DynamicWorkflowMetrics::increment(&metrics.in_flight);
        Self {
            metrics,
            finished: false,
        }
    }

    fn finish(&mut self) {
        if !self.finished {
            self.metrics.in_flight.fetch_sub(1, Ordering::Relaxed);
            self.finished = true;
        }
    }
}

impl Drop for DynamicWorkflowClaimInFlight {
    fn drop(&mut self) {
        if !self.finished {
            self.finish();
            DynamicWorkflowMetrics::increment(&self.metrics.failures);
        }
    }
}

fn dynamic_workflow_now_ms() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
        .min(u128::from(u64::MAX)) as u64
}

fn is_terminal_workflow_event(event: &FlowEvent) -> bool {
    matches!(
        event,
        FlowEvent::RunCompleted { .. }
            | FlowEvent::RunFailed { .. }
            | FlowEvent::RunCancelled { .. }
            | FlowEvent::RunTimedOut { .. }
            | FlowEvent::RunRetryExhausted { .. }
            | FlowEvent::RunHostShutdown { .. }
            | FlowEvent::RunContinuedAsNew { .. }
    )
}

fn history_is_terminal(history: &[FlowEventEnvelope]) -> bool {
    history
        .last()
        .is_some_and(|envelope| is_terminal_workflow_event(&envelope.event))
}

/// Model-visible tool that executes a dynamic workflow through A3S Flow.
pub struct DynamicWorkflowTool {
    registry: DynamicWorkflowRegistry,
    flow_event_store: Option<Arc<dyn FlowEventStore>>,
    graph_observer: Option<FlowGraphObserver>,
    task_scheduler: Option<Arc<TaskScheduler>>,
    admit_steps_globally: bool,
    runtime_build_compatibility: Option<RuntimeBuildCompatibility>,
    continuation_lease_ledger: Option<Arc<dyn FlowDecisionLedger>>,
    continuation_lease_ms: u64,
    memory_continuation_lease_ledger: Arc<MemoryFlowDecisionLedger>,
    metrics: Arc<DynamicWorkflowMetrics>,
}

enum DynamicWorkflowRegistry {
    Standalone(Arc<ToolRegistry>),
    RegistryBound(Weak<ToolRegistry>),
}

impl DynamicWorkflowRegistry {
    fn resolve(&self) -> Option<Arc<ToolRegistry>> {
        match self {
            Self::Standalone(registry) => Some(Arc::clone(registry)),
            Self::RegistryBound(registry) => registry.upgrade(),
        }
    }
}

impl DynamicWorkflowTool {
    pub fn new(registry: Arc<ToolRegistry>) -> Self {
        Self {
            registry: DynamicWorkflowRegistry::Standalone(registry),
            flow_event_store: None,
            graph_observer: None,
            task_scheduler: None,
            admit_steps_globally: false,
            runtime_build_compatibility: None,
            continuation_lease_ledger: None,
            continuation_lease_ms: DEFAULT_DYNAMIC_WORKFLOW_LEASE_MS,
            memory_continuation_lease_ledger: Arc::new(MemoryFlowDecisionLedger::new()),
            metrics: Arc::new(DynamicWorkflowMetrics::default()),
        }
    }

    fn new_registry_bound(registry: Arc<ToolRegistry>) -> Self {
        Self {
            registry: DynamicWorkflowRegistry::RegistryBound(Arc::downgrade(&registry)),
            flow_event_store: None,
            graph_observer: None,
            task_scheduler: None,
            admit_steps_globally: false,
            runtime_build_compatibility: None,
            continuation_lease_ledger: None,
            continuation_lease_ms: DEFAULT_DYNAMIC_WORKFLOW_LEASE_MS,
            memory_continuation_lease_ledger: Arc::new(MemoryFlowDecisionLedger::new()),
            metrics: Arc::new(DynamicWorkflowMetrics::default()),
        }
    }

    /// Return bounded cumulative worker-claim diagnostics for this tool
    /// instance. Durable per-run attempt state remains in the Flow ledger.
    pub fn health(&self) -> DynamicWorkflowHealthSnapshot {
        self.metrics.snapshot()
    }

    /// Project committed Flow events into an optional reactive state graph.
    /// A3S Flow remains the workflow execution source of truth.
    pub fn with_graph_observer(mut self, observer: FlowGraphObserver) -> Self {
        self.graph_observer = Some(observer);
        self
    }

    /// Use a host-owned Flow event store for workflow history and control.
    ///
    /// This is the extension point for remote or database-backed hosts. The
    /// store must provide its own durable append/sequence contract; Code does
    /// not mirror its events into a second journal or cache. When omitted,
    /// local workspaces use [`CrossProcessFlowEventStore`] and non-local
    /// contexts use a process-local in-memory store for compatibility.
    pub fn with_flow_event_store(mut self, store: Arc<dyn FlowEventStore>) -> Self {
        self.flow_event_store = Some(store);
        self
    }

    /// Configure optional global admission for direct script-backed Flow
    /// steps. Session-bound registrations should leave this disabled because
    /// the enclosing session operation already owns the global lease.
    pub fn with_task_scheduler(
        mut self,
        scheduler: Arc<TaskScheduler>,
        admit_steps_globally: bool,
    ) -> Self {
        self.task_scheduler = Some(scheduler);
        self.admit_steps_globally = admit_steps_globally;
        self
    }

    /// Fence new and resumed runs to an explicit runtime-build compatibility
    /// set. By default the tool pins new runs to
    /// [`DYNAMIC_WORKFLOW_RUNTIME_BUILD_ID`] and temporarily accepts legacy
    /// unpinned histories; hosts that can replay older builds should add them
    /// to this compatibility set explicitly.
    pub fn with_runtime_build_compatibility(
        mut self,
        compatibility: RuntimeBuildCompatibility,
    ) -> Self {
        self.runtime_build_compatibility = Some(compatibility);
        self
    }

    /// Use a caller-owned durable lease ledger for worker admission.
    ///
    /// The ledger stores only claim metadata and digests; A3S Flow's event
    /// store remains the workflow source of truth. Hosts that run multiple
    /// workers should provide a shared ledger (for example a
    /// [`FileFlowDecisionLedger`]) and choose a lease long enough for one
    /// heartbeat interval. Without an explicit ledger, local workspaces use a
    /// sidecar file ledger and remote/in-memory contexts use a tool-scoped
    /// memory ledger.
    pub fn with_continuation_lease_ledger(
        mut self,
        ledger: Arc<dyn FlowDecisionLedger>,
        lease_ms: u64,
    ) -> Self {
        self.continuation_lease_ledger = Some(ledger);
        self.continuation_lease_ms = lease_ms.max(1);
        self
    }

    /// Change the default worker lease while retaining automatic ledger
    /// selection.
    pub fn with_continuation_lease_ms(mut self, lease_ms: u64) -> Self {
        self.continuation_lease_ms = lease_ms.max(1);
        self
    }

    /// Bind a host control handle to one durable run and its replay inputs.
    ///
    /// The handle is intentionally separate from the model-visible tool call:
    /// a host must provide the exact source and initial input before it can
    /// inspect, drive, or cancel a run. No operation is performed until a
    /// method on the returned handle is awaited.
    pub fn control(
        &self,
        run_id: impl Into<String>,
        source: impl Into<String>,
        input: Value,
        ctx: &ToolContext,
    ) -> Result<DynamicWorkflowControl> {
        let registry = self
            .registry
            .resolve()
            .ok_or_else(|| anyhow::anyhow!("tool registry is closed"))?;
        let run_id = run_id.into();
        if !safe_workflow_run_id(&run_id) {
            anyhow::bail!(
                "dynamic workflow run_id must contain only ASCII letters, numbers, '-' or '_'"
            );
        }
        let source = source.into();
        if source.is_empty() {
            anyhow::bail!("dynamic workflow source must not be empty");
        }
        dynamic_workflow_input_identity(&input)
            .map_err(|error| anyhow::anyhow!("invalid dynamic workflow input: {error}"))?;
        let allowed_tools = default_allowed_tools(&registry);
        Ok(DynamicWorkflowControl {
            registry,
            context: ctx.clone(),
            flow_event_store: self.flow_event_store.clone(),
            run_id,
            source: Arc::from(source),
            input,
            allowed_tools,
            limits: DynamicWorkflowScriptLimits::default(),
            graph_observer: self.graph_observer.clone(),
            task_scheduler: self.task_scheduler.clone(),
            admit_steps_globally: self.admit_steps_globally,
            runtime_build_compatibility: self.runtime_build_compatibility.clone(),
            continuation_lease_ledger: self.continuation_lease_ledger.clone(),
            continuation_lease_ms: self.continuation_lease_ms,
            memory_continuation_lease_ledger: Arc::clone(&self.memory_continuation_lease_ledger),
            metrics: Arc::clone(&self.metrics),
        })
    }

    async fn continuation_lease_ledger_for_context(
        &self,
        ctx: &ToolContext,
    ) -> Result<Arc<dyn FlowDecisionLedger>> {
        dynamic_workflow_lease_ledger_for_context(
            self.continuation_lease_ledger.as_ref(),
            &self.memory_continuation_lease_ledger,
            ctx,
        )
        .await
    }
}

impl DynamicWorkflowControl {
    /// Return the immutable run id bound to this control handle.
    pub fn run_id(&self) -> &str {
        &self.run_id
    }

    /// Return bounded cumulative worker-claim diagnostics shared by this
    /// tool and all controls derived from it.
    pub fn health(&self) -> DynamicWorkflowHealthSnapshot {
        self.metrics.snapshot()
    }

    /// Compose workflow claim counters with the optional agent-wide scheduler
    /// health snapshot. This is a read-only diagnostic view; Flow history and
    /// the worker lease remain the only authorities for workflow state.
    pub async fn diagnostics(&self) -> Result<DynamicWorkflowControlDiagnostics> {
        let scheduler = match self.task_scheduler.as_ref() {
            Some(scheduler) => Some(scheduler.health().await.map_err(|error| {
                anyhow::anyhow!("read dynamic workflow scheduler health: {error}")
            })?),
            None => None,
        };
        let scheduler_quota = if self.admit_steps_globally {
            if let (Some(scheduler), Some(quota)) = (
                self.task_scheduler.as_ref(),
                // Diagnostics are valid before a control handle has started
                // its run. `allow_missing` keeps this read-only projection
                // from turning an observation into an implicit start/lookup
                // precondition while still deriving the same stable claim
                // identity that the first drive will use.
                self.prepare(true).await?.scheduler_quota,
            ) {
                Some(scheduler.quota_snapshot(&quota).await.map_err(|error| {
                    anyhow::anyhow!("read dynamic workflow scheduler quota: {error}")
                })?)
            } else {
                None
            }
        } else {
            None
        };
        Ok(DynamicWorkflowControlDiagnostics {
            workflow: self.metrics.snapshot(),
            scheduler,
            scheduler_quota,
        })
    }

    /// Replace the set of tools available to replayed script steps.
    pub fn with_allowed_tools(mut self, allowed_tools: impl IntoIterator<Item = String>) -> Self {
        self.allowed_tools = sanitize_allowed_tools(allowed_tools);
        self
    }

    /// Replace the bounded script and orchestration limits used during replay.
    pub fn with_limits(mut self, limits: DynamicWorkflowScriptLimits) -> Self {
        self.limits = limits;
        self
    }

    /// Project control-driven Flow events into a graph observer.
    pub fn with_graph_observer(mut self, observer: FlowGraphObserver) -> Self {
        self.graph_observer = Some(observer);
        self
    }

    /// Configure optional global admission for direct script-backed steps.
    pub fn with_task_scheduler(
        mut self,
        scheduler: Arc<TaskScheduler>,
        admit_steps_globally: bool,
    ) -> Self {
        self.task_scheduler = Some(scheduler);
        self.admit_steps_globally = admit_steps_globally;
        self
    }

    /// Set the runtime-build compatibility policy used by control replay.
    pub fn with_runtime_build_compatibility(
        mut self,
        compatibility: RuntimeBuildCompatibility,
    ) -> Self {
        self.runtime_build_compatibility = Some(compatibility);
        self
    }

    /// Use a caller-owned worker lease ledger for control operations.
    pub fn with_continuation_lease_ledger(
        mut self,
        ledger: Arc<dyn FlowDecisionLedger>,
        lease_ms: u64,
    ) -> Self {
        self.continuation_lease_ledger = Some(ledger);
        self.continuation_lease_ms = lease_ms.max(1);
        self
    }

    /// Change the worker lease while retaining automatic ledger selection.
    pub fn with_continuation_lease_ms(mut self, lease_ms: u64) -> Self {
        self.continuation_lease_ms = lease_ms.max(1);
        self
    }
}

async fn dynamic_workflow_lease_ledger_for_context(
    explicit: Option<&Arc<dyn FlowDecisionLedger>>,
    memory: &Arc<MemoryFlowDecisionLedger>,
    ctx: &ToolContext,
) -> Result<Arc<dyn FlowDecisionLedger>> {
    if let Some(ledger) = explicit {
        return Ok(Arc::clone(ledger));
    }
    let Some(root) = ctx.workspace_services.local_root() else {
        let ledger: Arc<dyn FlowDecisionLedger> = memory.clone();
        return Ok(ledger);
    };
    let workflow_root = dynamic_workflow_store_path(root);
    validate_dynamic_workflow_directory(&root.join(".a3s"), ".a3s").await?;
    validate_dynamic_workflow_directory(&workflow_root, ".a3s/workflow").await?;
    let lease_root = workflow_root.join(DYNAMIC_WORKFLOW_LEASE_RELATIVE_PATH);
    validate_dynamic_workflow_directory(&lease_root, ".a3s/workflow/leases").await?;
    Ok(Arc::new(FileFlowDecisionLedger::new(lease_root)))
}

#[async_trait]
impl Tool for DynamicWorkflowTool {
    fn name(&self) -> &str {
        DYNAMIC_WORKFLOW_TOOL
    }

    fn description(&self) -> &str {
        "Run a local dynamic workflow with A3S Flow. The workflow source is a sandboxed JavaScript PTC script that may call allowed ctx tools; A3S Flow records workflow and step history."
    }

    fn parameters(&self) -> Value {
        json!({
            "type": "object",
            "additionalProperties": false,
            "properties": {
                "source": {
                    "type": "string",
                    "description": "JavaScript PTC source defining async function run(ctx, inputs). For inputs.kind='workflow', return a Flow command: {type:'complete', output}, {type:'fail', error}, {type:'schedule_step', step_id, step_name, input, retry?}, or {type:'schedule_steps', steps:[...]}. For inputs.kind='step', return the step JSON output. A scheduled step with step_name='task' bypasses QuickJS and calls the host task tool directly with input as its arguments; legacy parallel_task steps remain readable."
                },
                "input": {
                    "type": "object",
                    "description": "Initial workflow input."
                },
                "run_id": {
                    "type": "string",
                    "description": "Optional durable run id. Reusing it with the same source and input is idempotent."
                },
                "allowed_tools": {
                    "type": "array",
                    "description": "Tool names the workflow script may call through ctx. Defaults to all registered tools except program, dynamic_workflow, and the legacy parallel_task alias. Direct task fan-out is blocked inside QuickJS; schedule a host task step instead. Login-registered tools such as runtime are allowed when present.",
                    "items": { "type": "string" }
                },
                "limits": {
                    "type": "object",
                    "additionalProperties": false,
                    "properties": {
                        "timeoutMs": { "type": "integer", "minimum": 1 },
                        "maxToolCalls": { "type": "integer", "minimum": 1 },
                        "maxOutputBytes": { "type": "integer", "minimum": 1 },
                        "maxConcurrentGenerations": {
                            "type": "integer",
                            "minimum": 1,
                            "maximum": 4,
                            "description": "Optional bounded fan-out for independently session-bound generate_object steps. Providers without session forking remain single-flight."
                        },
                        "maxConcurrentSteps": {
                            "type": "integer",
                            "minimum": 1,
                            "maximum": 32,
                            "description": "Optional per-workflow limit for concurrently executing Flow step bodies."
                        }
                    }
                }
            },
            "required": ["source"]
        })
    }

    async fn execute(&self, args: &Value, ctx: &ToolContext) -> Result<ToolOutput> {
        let Some(registry) = self.registry.resolve() else {
            return Ok(ToolOutput::error("Tool registry is closed"));
        };
        let Some(source) = args.get("source").and_then(Value::as_str) else {
            return Ok(ToolOutput::error("dynamic_workflow requires source"));
        };
        let input = args.get("input").cloned().unwrap_or_else(|| json!({}));
        let allowed_tools = args
            .get("allowed_tools")
            .and_then(Value::as_array)
            .map(|items| {
                items
                    .iter()
                    .filter_map(Value::as_str)
                    .map(ToString::to_string)
                    .collect::<Vec<_>>()
            })
            .unwrap_or_else(|| default_allowed_tools(&registry));
        let limits: DynamicWorkflowScriptLimits = args
            .get("limits")
            .cloned()
            .and_then(|value| serde_json::from_value(value).ok())
            .unwrap_or_default();

        let (runtime_build_id, runtime_build_compatibility) =
            match dynamic_workflow_runtime_configuration(self.runtime_build_compatibility.as_ref())
            {
                Ok(configuration) => configuration,
                Err(error) => {
                    return Ok(ToolOutput::error(format!(
                        "invalid dynamic workflow runtime build identity: {error}"
                    )))
                }
            };
        let source_hash = source_hash(source);
        let base_spec = WorkflowSpec::rust_embedded(
            "a3s-code.dynamic-workflow",
            source_hash.as_str(),
            "ptc",
            "run",
        );

        if ctx.is_cancelled() {
            return Ok(ToolOutput::error(
                "dynamic_workflow was cancelled before admission",
            ));
        }
        let requested_run_id = args.get("run_id").and_then(Value::as_str);
        if requested_run_id.is_some_and(|run_id| !safe_workflow_run_id(run_id)) {
            return Ok(ToolOutput::error(
                "dynamic_workflow run_id must contain only ASCII letters, numbers, '-' or '_'",
            ));
        }
        let run_id = requested_run_id
            .map(ToString::to_string)
            .unwrap_or_else(|| uuid::Uuid::new_v4().to_string());
        let store = match self.flow_event_store.clone() {
            Some(store) => Ok(store),
            None => flow_store_for_context(ctx, Some(&run_id)).await,
        };
        let store = match store {
            Ok(store) => store,
            Err(error) => return Ok(ToolOutput::error(error.to_string())),
        };
        let prior_history = match store.list(&run_id).await {
            Ok(history) => history,
            Err(a3s_flow::FlowError::RunNotFound(_)) => Vec::new(),
            Err(error) => return Ok(ToolOutput::error(error.to_string())),
        };
        // Preserve the immutable build pin already persisted in a resumed
        // history. A new run is pinned to this worker's current build; a
        // legacy run remains intentionally unpinned during migration. Keeping
        // the exact persisted spec is required because Flow treats the whole
        // workflow definition as the idempotent start contract.
        let (spec, effective_runtime_build_id) = prior_history
            .iter()
            .find_map(|envelope| match &envelope.event {
                FlowEvent::RunCreated { spec, .. } => Some(spec.clone()),
                _ => None,
            })
            .map(|persisted_spec| {
                let runtime_build_id = persisted_spec
                    .runtime_build_id
                    .as_ref()
                    .map(ToString::to_string);
                (persisted_spec, runtime_build_id)
            })
            .unwrap_or_else(|| {
                let spec = base_spec.with_runtime_build(runtime_build_id.clone());
                (spec, Some(runtime_build_id.to_string()))
            });
        let claim_identity = match dynamic_workflow_claim_identity(
            &run_id,
            source,
            &input,
            runtime_build_id.as_str(),
            &prior_history,
        ) {
            Ok(identity) => identity,
            Err(error) => {
                return Ok(ToolOutput::error(format!(
                    "dynamic workflow continuation identity rejected: {error}"
                )))
            }
        };
        let lease_ledger = match self.continuation_lease_ledger_for_context(ctx).await {
            Ok(ledger) => ledger,
            Err(error) => return Ok(ToolOutput::error(error.to_string())),
        };
        let lease_claim = match claim_dynamic_workflow_lease(
            lease_ledger,
            claim_identity.clone(),
            self.continuation_lease_ms,
            Arc::clone(&self.metrics),
        )
        .await
        {
            Ok(claim) => claim,
            Err(error) => return Ok(ToolOutput::error(error.to_string())),
        };
        let (lease, mut lease_state) = match lease_claim {
            DynamicWorkflowLeaseClaim::Owned(lease) => (Some(lease), "claimed"),
            DynamicWorkflowLeaseClaim::AlreadyCompleted => {
                if !history_is_terminal(&prior_history) {
                    return Ok(ToolOutput::error(
                        "dynamic workflow worker claim is completed but its durable run is not terminal",
                    ));
                }
                (None, "already_completed")
            }
        };
        let parent_cancellation = ctx.cancellation_token();
        let child_cancellation = parent_cancellation.child_token();
        let workflow_context = ctx
            .clone()
            .with_run_id(run_id.clone())
            .with_cancellation(child_cancellation.clone());
        let scheduler_quota = if self.admit_steps_globally && self.task_scheduler.is_some() {
            Some(
                TaskSchedulerQuota::new(
                    lease
                        .as_ref()
                        .map(|lease| lease.identity.clone())
                        .unwrap_or_else(|| claim_identity.clone()),
                    dynamic_workflow_scheduler_quota_limit(&limits),
                )
                .map_err(|error| {
                    anyhow::anyhow!("build dynamic workflow scheduler quota: {error}")
                })?,
            )
        } else {
            None
        };
        let mut runtime = DynamicWorkflowRuntime::new(registry, workflow_context.clone(), source)
            .with_allowed_tools(allowed_tools)
            .with_limits(limits);
        if let Some(scheduler) = &self.task_scheduler {
            runtime = runtime.with_task_scheduler(Arc::clone(scheduler), self.admit_steps_globally);
            if let Some(quota) = scheduler_quota {
                runtime = runtime.with_task_scheduler_quota(quota);
            }
        }
        if let Some(lease) = lease.as_ref() {
            runtime = runtime.with_continuation_lease(Arc::new(lease.clone()));
        }
        let runtime = Arc::new(runtime);
        let runtime_for_metadata = Arc::clone(&runtime);
        let initial_plan = if prior_history.is_empty() {
            ExecutionPlan::new("dynamic workflow", Complexity::Medium)
        } else {
            dynamic_workflow_execution_plan(&prior_history)
        };
        let mut observers: Vec<Arc<dyn FlowEventObserver>> = Vec::new();
        if let Some(tx) = ctx.agent_event_tx.clone() {
            observers.push(Arc::new(AgentEventFlowObserver::new(
                tx,
                ctx.session_id.clone().unwrap_or_default(),
                initial_plan,
            )));
        }
        if let Some(observer) = &self.graph_observer {
            observers.push(Arc::new(observer.clone()));
        }
        let mut engine_builder = FlowEngine::builder(runtime)
            .with_store(store)
            .with_runtime_build_compatibility(runtime_build_compatibility);
        if !observers.is_empty() {
            engine_builder = engine_builder
                .with_observer(Arc::new(FanoutFlowEventObserver::from_observers(observers)));
        }
        let engine = engine_builder.build();
        let input_for_identity = input.clone();

        let execution_result = if let Some(lease) = lease.as_ref() {
            drive_dynamic_workflow_with_lease(
                &engine,
                &run_id,
                spec,
                input,
                DynamicWorkflowDriveContext {
                    source,
                    input_for_identity: &input_for_identity,
                    runtime_build_id: runtime_build_id.as_str(),
                    workflow_context: &workflow_context,
                    parent_cancellation,
                    child_cancellation,
                },
                lease,
            )
            .await
        } else {
            let started_run_id = match engine.start_with_id(&run_id, spec, input).await {
                Ok(run_id) => run_id,
                Err(error) => return Ok(ToolOutput::error(error.to_string())),
            };
            if started_run_id != run_id {
                return Ok(ToolOutput::error(format!(
                    "dynamic workflow engine returned unexpected run id `{started_run_id}` for `{run_id}`"
                )));
            }
            let snapshot = match drive_inline_retries(&engine, &run_id, &workflow_context).await {
                Ok(snapshot) => snapshot,
                Err(error) => return Ok(ToolOutput::error(error.to_string())),
            };
            collect_dynamic_workflow_execution(
                &engine,
                &run_id,
                snapshot,
                source,
                &input_for_identity,
                runtime_build_id.as_str(),
            )
            .await
        };
        let execution = match execution_result {
            Ok(execution) => execution,
            Err(err) => return Ok(ToolOutput::error(err.to_string())),
        };
        let DynamicWorkflowExecution {
            snapshot,
            history,
            continuation_identity,
        } = execution;
        if lease.is_some() {
            lease_state = if snapshot.status.is_terminal() {
                "completed"
            } else {
                "released"
            };
        }

        let output = match &snapshot.output {
            Some(output) => {
                serde_json::to_string_pretty(output).unwrap_or_else(|_| output.to_string())
            }
            None => snapshot
                .error
                .clone()
                .unwrap_or_else(|| format!("workflow status: {:?}", snapshot.status)),
        };

        let status = snapshot.status;
        let plan = dynamic_workflow_execution_plan(&history);
        let plan_identity = match plan.definition_identity() {
            Ok(identity) => identity,
            Err(error) => return Ok(ToolOutput::error(error.to_string())),
        };
        let metadata = json!({
            "dynamic_workflow": {
                "run_id": run_id,
                "status": format!("{:?}", snapshot.status),
                "last_sequence": snapshot.last_sequence,
                "source_hash": source_hash,
                "runtime_build_id": effective_runtime_build_id,
                "snapshot": snapshot,
                "history": history,
                "plan": plan,
                "plan_identity": plan_identity,
                "continuation_identity": continuation_identity,
                "worker_lease": {
                    "state": lease_state,
                    "attempt": lease.as_ref().map(|lease| lease.attempt),
                },
                "admission": runtime_for_metadata.admission_stats(),
            }
        });
        let output = match status {
            WorkflowRunStatus::Completed => ToolOutput::success(output),
            WorkflowRunStatus::Failed | WorkflowRunStatus::Cancelled => ToolOutput::error(output),
            _ => ToolOutput::error(format!(
                "dynamic_workflow ended without a terminal result: {status:?}; {output}"
            )),
        };

        Ok(output.with_metadata(metadata))
    }
}

struct DynamicWorkflowControlPreparation {
    store: Arc<dyn FlowEventStore>,
    prior_history: Vec<FlowEventEnvelope>,
    spec: WorkflowSpec,
    effective_runtime_build_id: Option<String>,
    runtime_build_id: RuntimeBuildId,
    runtime_build_compatibility: RuntimeBuildCompatibility,
    claim_identity: ExecutionIdentityV1,
    scheduler_quota: Option<TaskSchedulerQuota>,
    lease_ledger: Arc<dyn FlowDecisionLedger>,
}

fn dynamic_workflow_runtime_configuration(
    configured: Option<&RuntimeBuildCompatibility>,
) -> Result<(RuntimeBuildId, RuntimeBuildCompatibility)> {
    let runtime_build_id =
        match configured {
            Some(compatibility) => compatibility.current_build_id().clone(),
            None => RuntimeBuildId::new(DYNAMIC_WORKFLOW_RUNTIME_BUILD_ID.to_string()).map_err(
                |error| anyhow::anyhow!("invalid dynamic workflow runtime build identity: {error}"),
            )?,
        };
    let compatibility = configured.cloned().unwrap_or_else(|| {
        RuntimeBuildCompatibility::new(runtime_build_id.clone()).accept_unpinned()
    });
    Ok((runtime_build_id, compatibility))
}

impl DynamicWorkflowControl {
    async fn prepare(&self, allow_missing: bool) -> Result<DynamicWorkflowControlPreparation> {
        let (runtime_build_id, runtime_build_compatibility) =
            dynamic_workflow_runtime_configuration(self.runtime_build_compatibility.as_ref())?;
        let source_hash = source_hash(self.source.as_ref());
        let base_spec = WorkflowSpec::rust_embedded(
            "a3s-code.dynamic-workflow",
            source_hash.as_str(),
            "ptc",
            "run",
        );
        let store = match self.flow_event_store.clone() {
            Some(store) => store,
            None => flow_store_for_context(&self.context, Some(&self.run_id)).await?,
        };
        let prior_history = match store.list(&self.run_id).await {
            Ok(history) => history,
            Err(a3s_flow::FlowError::RunNotFound(_)) if allow_missing => Vec::new(),
            Err(error) => return Err(error.into()),
        };
        if !allow_missing && prior_history.is_empty() {
            return Err(a3s_flow::FlowError::RunNotFound(self.run_id.clone()).into());
        }
        let (spec, effective_runtime_build_id) = prior_history
            .iter()
            .find_map(|envelope| match &envelope.event {
                FlowEvent::RunCreated { spec, .. } => Some(spec.clone()),
                _ => None,
            })
            .map(|persisted_spec| {
                let runtime_build_id = persisted_spec
                    .runtime_build_id
                    .as_ref()
                    .map(ToString::to_string);
                (persisted_spec, runtime_build_id)
            })
            .unwrap_or_else(|| {
                let spec = base_spec.with_runtime_build(runtime_build_id.clone());
                (spec, Some(runtime_build_id.to_string()))
            });
        let claim_identity = dynamic_workflow_claim_identity(
            &self.run_id,
            self.source.as_ref(),
            &self.input,
            runtime_build_id.as_str(),
            &prior_history,
        )
        .map_err(|error| {
            anyhow::anyhow!("dynamic workflow continuation identity rejected: {error}")
        })?;
        let scheduler_quota = if self.admit_steps_globally && self.task_scheduler.is_some() {
            Some(
                TaskSchedulerQuota::new(
                    claim_identity.clone(),
                    dynamic_workflow_scheduler_quota_limit(&self.limits),
                )
                .map_err(|error| {
                    anyhow::anyhow!("build dynamic workflow scheduler quota: {error}")
                })?,
            )
        } else {
            None
        };
        let lease_ledger = dynamic_workflow_lease_ledger_for_context(
            self.continuation_lease_ledger.as_ref(),
            &self.memory_continuation_lease_ledger,
            &self.context,
        )
        .await?;
        Ok(DynamicWorkflowControlPreparation {
            store,
            prior_history,
            spec,
            effective_runtime_build_id,
            runtime_build_id,
            runtime_build_compatibility,
            claim_identity,
            scheduler_quota,
            lease_ledger,
        })
    }

    fn build_engine(
        &self,
        preparation: &DynamicWorkflowControlPreparation,
        workflow_context: ToolContext,
        lease: Option<&DynamicWorkflowLease>,
    ) -> FlowEngine {
        let mut runtime = DynamicWorkflowRuntime::new(
            Arc::clone(&self.registry),
            workflow_context,
            self.source.as_ref(),
        )
        .with_allowed_tools(self.allowed_tools.clone())
        .with_limits(self.limits.clone());
        if let Some(scheduler) = &self.task_scheduler {
            runtime = runtime.with_task_scheduler(Arc::clone(scheduler), self.admit_steps_globally);
            if let Some(quota) = preparation.scheduler_quota.clone() {
                runtime = runtime.with_task_scheduler_quota(quota);
            }
        }
        if let Some(lease) = lease {
            runtime = runtime.with_continuation_lease(Arc::new(lease.clone()));
        }
        let runtime = Arc::new(runtime);
        let initial_plan = if preparation.prior_history.is_empty() {
            ExecutionPlan::new("dynamic workflow", Complexity::Medium)
        } else {
            dynamic_workflow_execution_plan(&preparation.prior_history)
        };
        let mut observers: Vec<Arc<dyn FlowEventObserver>> = Vec::new();
        if let Some(tx) = self.context.agent_event_tx.clone() {
            observers.push(Arc::new(AgentEventFlowObserver::new(
                tx,
                self.context.session_id.clone().unwrap_or_default(),
                initial_plan,
            )));
        }
        if let Some(observer) = &self.graph_observer {
            observers.push(Arc::new(observer.clone()));
        }
        let mut builder = FlowEngine::builder(runtime)
            .with_store(Arc::clone(&preparation.store))
            .with_runtime_build_compatibility(preparation.runtime_build_compatibility.clone());
        if !observers.is_empty() {
            builder =
                builder.with_observer(Arc::new(FanoutFlowEventObserver::from_observers(observers)));
        }
        builder.build()
    }

    async fn read_state(
        &self,
        preparation: &DynamicWorkflowControlPreparation,
        engine: &FlowEngine,
    ) -> Result<(WorkflowRunSnapshot, Vec<FlowEventEnvelope>)> {
        for _ in 0..3 {
            // Flow exposes history and its projection as separate reads. Read
            // the journal on both sides of the projection so a concurrent
            // append cannot be mistaken for an atomic snapshot; retry a
            // bounded number of times when the sequences disagree.
            let before_history = engine.history(&self.run_id).await?;
            let snapshot = engine.snapshot(&self.run_id).await?;
            let history = engine.history(&self.run_id).await?;
            let before_sequence = before_history
                .last()
                .map(|event| event.sequence)
                .unwrap_or(0);
            let last_sequence = history.last().map(|event| event.sequence).unwrap_or(0);
            if before_sequence == last_sequence && snapshot.last_sequence == last_sequence {
                return Ok((snapshot, history));
            }
        }
        let expected = preparation
            .prior_history
            .last()
            .map(|event| event.sequence)
            .unwrap_or(0);
        anyhow::bail!(
            "dynamic workflow state changed repeatedly while inspecting run `{}` (initial sequence {expected})",
            self.run_id
        )
    }

    async fn summarize(
        &self,
        preparation: &DynamicWorkflowControlPreparation,
        snapshot: WorkflowRunSnapshot,
        history: Vec<FlowEventEnvelope>,
    ) -> Result<DynamicWorkflowControlSnapshot> {
        let continuation_identity = dynamic_workflow_continuation_identity(
            &self.run_id,
            self.source.as_ref(),
            &self.input,
            preparation.runtime_build_id.as_str(),
            &history,
        )
        .map_err(|error| {
            anyhow::anyhow!(
                "dynamic workflow continuation identity rejected during inspection: {error}"
            )
        })?;
        let plan = dynamic_workflow_execution_plan(&history);
        let plan_identity = plan.definition_identity()?;
        let (decision_id, request_hash) = dynamic_workflow_lease_key(&preparation.claim_identity);
        let worker_lease = preparation
            .lease_ledger
            .inspect_with_identity(&decision_id, &request_hash, &preparation.claim_identity)
            .await?;
        let completed_steps = snapshot
            .steps
            .values()
            .filter(|step| step.status == StepStatus::Completed)
            .count();
        let open_steps = snapshot
            .steps
            .values()
            .filter(|step| {
                !matches!(
                    step.status,
                    StepStatus::Completed | StepStatus::Failed | StepStatus::Cancelled
                )
            })
            .count();
        Ok(DynamicWorkflowControlSnapshot {
            run_id: self.run_id.clone(),
            status: snapshot.status,
            last_sequence: snapshot.last_sequence,
            step_count: snapshot.steps.len(),
            completed_steps,
            open_steps,
            cancellation_requested: snapshot.cancellation.is_some(),
            continuation_identity,
            plan_identity,
            runtime_build_id: preparation.effective_runtime_build_id.clone(),
            worker_lease,
        })
    }

    async fn claim(
        &self,
        preparation: &DynamicWorkflowControlPreparation,
    ) -> Result<DynamicWorkflowLeaseClaim> {
        claim_dynamic_workflow_lease(
            Arc::clone(&preparation.lease_ledger),
            preparation.claim_identity.clone(),
            self.continuation_lease_ms,
            Arc::clone(&self.metrics),
        )
        .await
    }

    /// Inspect the durable run and return a bounded, digest-only projection.
    pub async fn inspect(&self) -> Result<DynamicWorkflowControlSnapshot> {
        let preparation = self.prepare(false).await?;
        let engine = self.build_engine(&preparation, self.context.clone(), None);
        let (snapshot, history) = self.read_state(&preparation, &engine).await?;
        self.summarize(&preparation, snapshot, history).await
    }

    /// Read the complete durable Flow history for trusted host diagnostics.
    ///
    /// Unlike [`Self::inspect`], this method intentionally returns persisted
    /// input and step output values. Callers should apply their own redaction
    /// policy before exposing the result outside a trusted control plane.
    pub async fn history(&self) -> Result<Vec<FlowEventEnvelope>> {
        let preparation = self.prepare(false).await?;
        let history = preparation.store.list(&self.run_id).await?;
        dynamic_workflow_continuation_identity(
            &self.run_id,
            self.source.as_ref(),
            &self.input,
            preparation.runtime_build_id.as_str(),
            &history,
        )
        .map_err(|error| anyhow::anyhow!("dynamic workflow history rejected: {error}"))?;
        Ok(history)
    }

    /// Request cleanup-aware durable cancellation and settle the worker lease.
    ///
    /// A live worker keeps ownership and causes this method to return a busy
    /// error; the caller can retry after the lease expires or the worker
    /// releases it. This preserves one executor for side-effecting steps.
    pub async fn request_cancellation(
        &self,
        reason: Option<String>,
    ) -> Result<DynamicWorkflowControlSnapshot> {
        validate_dynamic_workflow_control_reason(reason.as_deref())?;
        if self.context.is_cancelled() {
            anyhow::bail!("dynamic workflow control was cancelled before admission");
        }
        let preparation = self.prepare(false).await?;
        let claim = self.claim(&preparation).await?;
        let lease = match claim {
            DynamicWorkflowLeaseClaim::Owned(lease) => lease,
            DynamicWorkflowLeaseClaim::AlreadyCompleted => {
                let engine = self.build_engine(&preparation, self.context.clone(), None);
                let (snapshot, history) = self.read_state(&preparation, &engine).await?;
                if !snapshot.status.is_terminal() {
                    anyhow::bail!(
                        "dynamic workflow worker claim is completed but its durable run is not terminal"
                    );
                }
                return self.summarize(&preparation, snapshot, history).await;
            }
        };
        let parent_cancellation = self.context.cancellation_token();
        let child_cancellation = parent_cancellation.child_token();
        let workflow_context = self
            .context
            .clone()
            .with_cancellation(child_cancellation.clone());
        let engine = self.build_engine(&preparation, workflow_context.clone(), Some(&lease));
        let future = engine.request_cancellation(&self.run_id, CancellationRequest::new(reason));
        let execution = drive_dynamic_workflow_control_future(
            &engine,
            &self.run_id,
            DynamicWorkflowDriveContext {
                source: self.source.as_ref(),
                input_for_identity: &self.input,
                runtime_build_id: preparation.runtime_build_id.as_str(),
                workflow_context: &workflow_context,
                parent_cancellation,
                child_cancellation,
            },
            &lease,
            future,
        )
        .await?;
        self.summarize(&preparation, execution.snapshot, execution.history)
            .await
    }

    /// Immediately terminate the run through Flow's durable cancellation
    /// transition while still fencing the worker claim.
    pub async fn force_cancel(
        &self,
        reason: Option<String>,
    ) -> Result<DynamicWorkflowControlSnapshot> {
        validate_dynamic_workflow_control_reason(reason.as_deref())?;
        if self.context.is_cancelled() {
            anyhow::bail!("dynamic workflow control was cancelled before admission");
        }
        let preparation = self.prepare(false).await?;
        let claim = self.claim(&preparation).await?;
        let lease = match claim {
            DynamicWorkflowLeaseClaim::Owned(lease) => lease,
            DynamicWorkflowLeaseClaim::AlreadyCompleted => {
                let engine = self.build_engine(&preparation, self.context.clone(), None);
                let (snapshot, history) = self.read_state(&preparation, &engine).await?;
                if !snapshot.status.is_terminal() {
                    anyhow::bail!(
                        "dynamic workflow worker claim is completed but its durable run is not terminal"
                    );
                }
                return self.summarize(&preparation, snapshot, history).await;
            }
        };
        let parent_cancellation = self.context.cancellation_token();
        let child_cancellation = parent_cancellation.child_token();
        let workflow_context = self
            .context
            .clone()
            .with_cancellation(child_cancellation.clone());
        let engine = self.build_engine(&preparation, workflow_context.clone(), Some(&lease));
        let future = async {
            engine.force_cancel(&self.run_id, reason).await?;
            engine.snapshot(&self.run_id).await
        };
        let execution = drive_dynamic_workflow_control_future(
            &engine,
            &self.run_id,
            DynamicWorkflowDriveContext {
                source: self.source.as_ref(),
                input_for_identity: &self.input,
                runtime_build_id: preparation.runtime_build_id.as_str(),
                workflow_context: &workflow_context,
                parent_cancellation,
                child_cancellation,
            },
            &lease,
            future,
        )
        .await?;
        self.summarize(&preparation, execution.snapshot, execution.history)
            .await
    }

    /// Resume or start the bound run under the same worker fencing contract as
    /// the model-visible dynamic workflow tool.
    pub async fn drive(&self) -> Result<DynamicWorkflowControlSnapshot> {
        if self.context.is_cancelled() {
            anyhow::bail!("dynamic workflow control was cancelled before admission");
        }
        let preparation = self.prepare(true).await?;
        let claim = self.claim(&preparation).await?;
        let lease = match claim {
            DynamicWorkflowLeaseClaim::Owned(lease) => lease,
            DynamicWorkflowLeaseClaim::AlreadyCompleted => {
                let engine = self.build_engine(&preparation, self.context.clone(), None);
                let (snapshot, history) = self.read_state(&preparation, &engine).await?;
                if !snapshot.status.is_terminal() {
                    anyhow::bail!(
                        "dynamic workflow worker claim is completed but its durable run is not terminal"
                    );
                }
                return self.summarize(&preparation, snapshot, history).await;
            }
        };
        let parent_cancellation = self.context.cancellation_token();
        let child_cancellation = parent_cancellation.child_token();
        let workflow_context = self
            .context
            .clone()
            .with_cancellation(child_cancellation.clone());
        let engine = self.build_engine(&preparation, workflow_context.clone(), Some(&lease));
        let execution = drive_dynamic_workflow_with_lease(
            &engine,
            &self.run_id,
            preparation.spec.clone(),
            self.input.clone(),
            DynamicWorkflowDriveContext {
                source: self.source.as_ref(),
                input_for_identity: &self.input,
                runtime_build_id: preparation.runtime_build_id.as_str(),
                workflow_context: &workflow_context,
                parent_cancellation,
                child_cancellation,
            },
            &lease,
        )
        .await?;
        self.summarize(&preparation, execution.snapshot, execution.history)
            .await
    }
}

struct DynamicWorkflowExecution {
    snapshot: WorkflowRunSnapshot,
    history: Vec<FlowEventEnvelope>,
    continuation_identity: ExecutionIdentityV1,
}

struct DynamicWorkflowDriveContext<'a> {
    source: &'a str,
    input_for_identity: &'a Value,
    runtime_build_id: &'a str,
    workflow_context: &'a ToolContext,
    parent_cancellation: CancellationToken,
    child_cancellation: CancellationToken,
}

async fn collect_dynamic_workflow_execution(
    engine: &FlowEngine,
    run_id: &str,
    snapshot: WorkflowRunSnapshot,
    source: &str,
    input: &Value,
    runtime_build_id: &str,
) -> Result<DynamicWorkflowExecution> {
    let history = engine.history(run_id).await?;
    let continuation_identity =
        dynamic_workflow_continuation_identity(run_id, source, input, runtime_build_id, &history)
            .map_err(|error| {
            anyhow::anyhow!("dynamic workflow continuation identity rejected after replay: {error}")
        })?;
    Ok(DynamicWorkflowExecution {
        snapshot,
        history,
        continuation_identity,
    })
}

fn validate_dynamic_workflow_control_reason(reason: Option<&str>) -> Result<()> {
    let Some(reason) = reason else {
        return Ok(());
    };
    if reason.len() > MAX_DYNAMIC_WORKFLOW_CONTROL_REASON_BYTES {
        anyhow::bail!(
            "dynamic workflow cancellation reason exceeds {} bytes",
            MAX_DYNAMIC_WORKFLOW_CONTROL_REASON_BYTES
        );
    }
    if reason.contains('\0') {
        anyhow::bail!("dynamic workflow cancellation reason contains a NUL byte");
    }
    Ok(())
}

async fn drive_dynamic_workflow_control_future<F>(
    engine: &FlowEngine,
    run_id: &str,
    drive_context: DynamicWorkflowDriveContext<'_>,
    lease: &DynamicWorkflowLease,
    future: F,
) -> Result<DynamicWorkflowExecution>
where
    F: Future<Output = a3s_flow::Result<WorkflowRunSnapshot>>,
{
    let DynamicWorkflowDriveContext {
        source,
        input_for_identity,
        runtime_build_id,
        workflow_context: _workflow_context,
        parent_cancellation,
        child_cancellation,
    } = drive_context;
    tokio::pin!(future);

    async fn collect_result(
        engine: &FlowEngine,
        run_id: &str,
        result: a3s_flow::Result<WorkflowRunSnapshot>,
        source: &str,
        input: &Value,
        runtime_build_id: &str,
    ) -> Result<DynamicWorkflowExecution> {
        let snapshot = result?;
        collect_dynamic_workflow_execution(
            engine,
            run_id,
            snapshot,
            source,
            input,
            runtime_build_id,
        )
        .await
    }

    let heartbeat_period = Duration::from_millis((lease.lease_ms / 3).max(1));
    let first_heartbeat = tokio::time::Instant::now() + heartbeat_period;
    let mut heartbeat = tokio::time::interval_at(first_heartbeat, heartbeat_period);
    loop {
        tokio::select! {
            biased;
            result = &mut future => {
                let result = collect_result(
                    engine,
                    run_id,
                    result,
                    source,
                    input_for_identity,
                    runtime_build_id,
                ).await;
                return settle_dynamic_workflow_lease(lease, result).await;
            }
            _ = parent_cancellation.cancelled() => {
                DynamicWorkflowMetrics::increment(&lease.metrics.cancellations);
                child_cancellation.cancel();
                match tokio::time::timeout(MAX_DYNAMIC_WORKFLOW_SETTLE, &mut future).await {
                    Ok(result) => {
                        let result = collect_result(
                            engine,
                            run_id,
                            result,
                            source,
                            input_for_identity,
                            runtime_build_id,
                        ).await;
                        match settle_dynamic_workflow_lease(lease, result).await {
                            Ok(execution) if execution.snapshot.status.is_terminal() => {
                                return Ok(execution)
                            }
                            Ok(_) => anyhow::bail!("dynamic workflow control cancelled by its parent"),
                            Err(error) => return Err(error).context("settle dynamic workflow control cancellation"),
                        }
                    }
                    Err(_) => anyhow::bail!(
                        "dynamic workflow control cancellation did not settle within {} seconds; worker lease remains fenced until expiry",
                        MAX_DYNAMIC_WORKFLOW_SETTLE.as_secs()
                    ),
                }
            }
            _ = heartbeat.tick() => {
                match lease.renew().await {
                    Ok(true) => {}
                    Ok(false) => {
                        child_cancellation.cancel();
                        if tokio::time::timeout(MAX_DYNAMIC_WORKFLOW_SETTLE, &mut future)
                            .await
                            .is_ok()
                        {
                            let _ = lease.release().await;
                        }
                        anyhow::bail!("dynamic workflow control worker lease was lost before completion");
                    }
                    Err(error) => {
                        child_cancellation.cancel();
                        if tokio::time::timeout(MAX_DYNAMIC_WORKFLOW_SETTLE, &mut future)
                            .await
                            .is_ok()
                        {
                            let _ = lease.release().await;
                        }
                        return Err(error).context("renew dynamic workflow control worker lease");
                    }
                }
            }
        }
    }
}

async fn settle_dynamic_workflow_lease(
    lease: &DynamicWorkflowLease,
    result: Result<DynamicWorkflowExecution>,
) -> Result<DynamicWorkflowExecution> {
    match result {
        Ok(execution) if execution.snapshot.status.is_terminal() => {
            lease
                .complete()
                .await
                .context("complete dynamic workflow worker lease")?;
            Ok(execution)
        }
        Ok(execution) => {
            lease
                .release()
                .await
                .context("release suspended dynamic workflow worker lease")?;
            Ok(execution)
        }
        Err(error) => {
            if let Err(release_error) = lease.release().await {
                tracing::warn!(
                    error = %release_error,
                    "failed to release dynamic workflow worker lease after execution error"
                );
            }
            Err(error)
        }
    }
}

async fn drive_dynamic_workflow_with_lease(
    engine: &FlowEngine,
    run_id: &str,
    spec: WorkflowSpec,
    input: Value,
    drive_context: DynamicWorkflowDriveContext<'_>,
    lease: &DynamicWorkflowLease,
) -> Result<DynamicWorkflowExecution> {
    let DynamicWorkflowDriveContext {
        source,
        input_for_identity,
        runtime_build_id,
        workflow_context,
        parent_cancellation,
        child_cancellation,
    } = drive_context;
    let execution = async {
        let started_run_id = engine.start_with_id(run_id, spec, input).await?;
        if started_run_id != run_id {
            anyhow::bail!(
                "dynamic workflow engine returned unexpected run id `{started_run_id}` for `{run_id}`"
            );
        }
        let snapshot = drive_inline_retries(engine, run_id, workflow_context).await?;
        collect_dynamic_workflow_execution(
            engine,
            run_id,
            snapshot,
            source,
            input_for_identity,
            runtime_build_id,
        )
        .await
    };
    tokio::pin!(execution);

    let heartbeat_period = Duration::from_millis((lease.lease_ms / 3).max(1));
    let first_heartbeat = tokio::time::Instant::now() + heartbeat_period;
    let mut heartbeat = tokio::time::interval_at(first_heartbeat, heartbeat_period);
    loop {
        tokio::select! {
            biased;
            result = &mut execution => {
                return settle_dynamic_workflow_lease(lease, result).await;
            }
            _ = parent_cancellation.cancelled() => {
                DynamicWorkflowMetrics::increment(&lease.metrics.cancellations);
                child_cancellation.cancel();
                match tokio::time::timeout(MAX_DYNAMIC_WORKFLOW_SETTLE, &mut execution).await {
                    Ok(result) => {
                        match settle_dynamic_workflow_lease(lease, result).await {
                            Ok(execution) if execution.snapshot.status.is_terminal() => {
                                return Ok(execution)
                            }
                            Ok(_) => anyhow::bail!("dynamic workflow cancelled by its parent"),
                            Err(error) => return Err(error).context("settle dynamic workflow cancellation"),
                        }
                    }
                    Err(_) => anyhow::bail!(
                        "dynamic workflow cancellation did not settle within {} seconds; worker lease remains fenced until expiry",
                        MAX_DYNAMIC_WORKFLOW_SETTLE.as_secs()
                    ),
                }
            }
            _ = heartbeat.tick() => {
                match lease.renew().await {
                    Ok(true) => {}
                    Ok(false) => {
                        child_cancellation.cancel();
                        if tokio::time::timeout(MAX_DYNAMIC_WORKFLOW_SETTLE, &mut execution)
                            .await
                            .is_ok()
                        {
                            let _ = lease.release().await;
                        }
                        anyhow::bail!("dynamic workflow worker lease was lost before execution completed");
                    }
                    Err(error) => {
                        child_cancellation.cancel();
                        if tokio::time::timeout(MAX_DYNAMIC_WORKFLOW_SETTLE, &mut execution)
                            .await
                            .is_ok()
                        {
                            let _ = lease.release().await;
                        }
                        return Err(error).context("renew dynamic workflow worker lease");
                    }
                }
            }
        }
    }
}

/// Drive short, persisted step retries inside the originating tool call.
///
/// A3S Flow deliberately suspends at a delayed retry boundary. Interactive
/// waits and hooks must remain suspended for an external host, but a bounded
/// retry delay is ordinary fault recovery: returning it as a terminal tool
/// error forces every caller to reimplement the scheduler and previously made
/// DeepResearch abandon its event-sourced run. Retry attempts and their delay
/// remain authoritative in the Flow journal; this helper only waits for the
/// due time and asks the engine to replay the same run.
async fn drive_inline_retries(
    engine: &FlowEngine,
    run_id: &str,
    ctx: &ToolContext,
) -> Result<WorkflowRunSnapshot> {
    for _ in 0..MAX_INLINE_RETRY_RESUMES {
        let snapshot = engine.snapshot(run_id).await?;
        if snapshot.status.is_terminal() {
            return Ok(snapshot);
        }
        let Some(retry_after) = snapshot
            .steps
            .values()
            .filter(|step| step.status == StepStatus::Pending)
            .filter_map(|step| step.retry_after)
            .min()
        else {
            return Ok(snapshot);
        };
        let delay = retry_after
            .signed_duration_since(Utc::now())
            .to_std()
            .unwrap_or_default();
        if delay > MAX_INLINE_RETRY_DELAY {
            return Ok(snapshot);
        }
        let cancellation = ctx.cancellation_token();
        tokio::select! {
            biased;
            _ = cancellation.cancelled() => {
                anyhow::bail!("dynamic_workflow cancelled while waiting for a scheduled retry");
            }
            _ = tokio::time::sleep(delay) => {}
        }
        engine.drive(run_id).await?;
    }
    engine.snapshot(run_id).await.map_err(Into::into)
}

pub fn register_dynamic_workflow(registry: &Arc<ToolRegistry>) {
    registry.register(Arc::new(DynamicWorkflowTool::new_registry_bound(
        Arc::clone(registry),
    )));
}

/// Register a dynamic workflow Tool against a host-owned Flow event store.
///
/// The registry-bound Tool keeps only a weak registry reference, so adding a
/// durable store does not create a registry/tool ownership cycle. Use this for
/// database-backed or remote adapters that must be shared by model-visible
/// execution and [`DynamicWorkflowTool::control`].
pub fn register_dynamic_workflow_with_event_store(
    registry: &Arc<ToolRegistry>,
    store: Arc<dyn FlowEventStore>,
) {
    registry.register(Arc::new(
        DynamicWorkflowTool::new_registry_bound(Arc::clone(registry)).with_flow_event_store(store),
    ));
}

/// Register a dynamic workflow tool with an explicit scheduler policy.
///
/// This is intended for hosts that construct a Flow runtime outside the
/// normal AgentSession operation boundary. AgentSession registrations should
/// use [`register_dynamic_workflow`] so the enclosing run lease remains the
/// single global admission boundary.
pub fn register_dynamic_workflow_with_scheduler(
    registry: &Arc<ToolRegistry>,
    scheduler: Arc<TaskScheduler>,
    admit_steps_globally: bool,
) {
    registry.register(Arc::new(
        DynamicWorkflowTool::new_registry_bound(Arc::clone(registry))
            .with_task_scheduler(scheduler, admit_steps_globally),
    ));
}

async fn flow_store_for_context(
    ctx: &ToolContext,
    requested_run_id: Option<&str>,
) -> Result<Arc<dyn FlowEventStore>> {
    match ctx.workspace_services.local_root() {
        Some(root) => {
            let store = dynamic_workflow_store_path(root);
            validate_dynamic_workflow_directory(&root.join(".a3s"), ".a3s").await?;
            validate_dynamic_workflow_directory(&store, ".a3s/workflow").await?;
            if let Some(run_id) = requested_run_id.filter(|run_id| safe_workflow_run_id(run_id)) {
                validate_dynamic_workflow_log(&store.join(format!("{run_id}.jsonl"))).await?;
            }
            Ok(Arc::new(CrossProcessFlowEventStore::new(store)))
        }
        None => Ok(Arc::new(InMemoryEventStore::new())),
    }
}

async fn validate_dynamic_workflow_directory(path: &Path, label: &str) -> Result<()> {
    match tokio::fs::symlink_metadata(path).await {
        Ok(metadata) if metadata.file_type().is_symlink() => {
            anyhow::bail!("refusing to use symlinked dynamic workflow directory {label}")
        }
        Ok(metadata) if !metadata.is_dir() => {
            anyhow::bail!("dynamic workflow path {label} exists but is not a directory")
        }
        Ok(_) => Ok(()),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error).with_context(|| format!("inspect dynamic workflow path {label}")),
    }
}

async fn validate_dynamic_workflow_log(path: &Path) -> Result<()> {
    match tokio::fs::symlink_metadata(path).await {
        Ok(metadata) if metadata.file_type().is_symlink() => anyhow::bail!(
            "refusing to read or append symlinked dynamic workflow history {}",
            path.display()
        ),
        Ok(metadata) if !metadata.is_file() => anyhow::bail!(
            "dynamic workflow history path {} exists but is not a file",
            path.display()
        ),
        Ok(_) => Ok(()),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(error) => Err(error)
            .with_context(|| format!("inspect dynamic workflow history {}", path.display())),
    }
}

fn safe_workflow_run_id(run_id: &str) -> bool {
    !run_id.is_empty()
        && run_id
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
}

struct PayloadBuilder {
    value: Map<String, Value>,
}

impl PayloadBuilder {
    fn with(mut self, key: &str, value: impl Serialize) -> Self {
        self.value.insert(
            key.to_string(),
            serde_json::to_value(value).unwrap_or(Value::Null),
        );
        self
    }

    fn into_value(self) -> Value {
        Value::Object(self.value)
    }
}

fn invocation_payload(kind: &str, run_id: &str, history: &[FlowEventEnvelope]) -> PayloadBuilder {
    let mut value = Map::new();
    value.insert("kind".to_string(), json!(kind));
    value.insert("run_id".to_string(), json!(run_id));
    value.insert("history".to_string(), json!(history));
    value.insert("step_outputs".to_string(), completed_step_outputs(history));
    value.insert("step_failures".to_string(), failed_step_outputs(history));
    PayloadBuilder { value }
}

fn completed_step_outputs(history: &[FlowEventEnvelope]) -> Value {
    let mut outputs = Map::new();
    for envelope in history {
        if let FlowEvent::StepCompleted { step_id, output } = &envelope.event {
            outputs.insert(step_id.clone(), output.clone());
        }
    }
    Value::Object(outputs)
}

fn failed_step_outputs(history: &[FlowEventEnvelope]) -> Value {
    let mut outputs = Map::new();
    for envelope in history {
        if let FlowEvent::StepFailed {
            step_id,
            attempt,
            error,
        } = &envelope.event
        {
            outputs.insert(
                step_id.clone(),
                json!({
                    "attempt": attempt,
                    "error": error,
                }),
            );
        }
    }
    Value::Object(outputs)
}

fn script_result(result: &ToolResult) -> a3s_flow::Result<Value> {
    result
        .metadata
        .as_ref()
        .and_then(|metadata| metadata.get("script_result"))
        .cloned()
        .ok_or_else(|| {
            a3s_flow::FlowError::Runtime(
                "PTC program result did not include script_result metadata".to_string(),
            )
        })
}

fn default_allowed_tools(registry: &ToolRegistry) -> Vec<String> {
    sanitize_allowed_tools(registry.list())
}

fn sanitize_allowed_tools(items: impl IntoIterator<Item = String>) -> Vec<String> {
    let mut tools = items.into_iter().collect::<BTreeSet<_>>();
    tools.remove(PROGRAM_TOOL);
    tools.remove(DYNAMIC_WORKFLOW_TOOL);
    tools.remove(PARALLEL_TASK_TOOL);
    tools.into_iter().collect()
}

fn source_hash(source: &str) -> String {
    sha256::digest(source.as_bytes())
}

#[cfg(test)]
#[path = "dynamic_workflow/tests.rs"]
mod tests;