//! Command executor.
//!
//! R-1.2 PR-2d-2: `CommandExecutor::execute` now takes
//! `&noetl_executor::worker::source::Command` (the executor crate's
//! enriched Command, 0.3.0+) instead of the worker's local
//! `crate::client::Command`.  Field accesses:
//!
//! - `command.execution_id` (i64) — same as before.
//! - `command.step` (String) — was `command.step()` accessor returning `&node_name`.
//! - `command.command_id` (String) — was `command.command_id()`.
//! - `command.tool_kind` (String) — was `command.action`.
//! - `command.render_context` (HashMap) — was `command.render_context()`.
//! - `command.attempts` (u32) — new in 0.3.0; useful for retry decisions.
//! - `command.input` (Value) — carries the worker's full `context` JSON
//!   including `tool_config`, `cases`, `args`, and `render_context`
//!   (the dedicated field is also populated for direct access).
//!   `tool_config` is extracted via `command.input.get("tool_config")`;
//!   `cases` via `command.input.get("cases")`.
//!
//! Per `nats::source::NatsCommandSource::translate`, the executor's
//! Command is a lossless mapping of the worker's Command.

use anyhow::Result;
use noetl_arrow_cache::ArrowIpcSharedMemoryCache;
use noetl_executor::worker::source::Command;
use noetl_tools::context::ExecutionContext;
use noetl_tools::registry::{ToolConfig, ToolRegistry};
use noetl_tools::tools::create_default_registry;
use std::collections::HashMap;
use std::sync::Arc;

use crate::client::{ControlPlaneClient, ExecutorEvent};
use crate::executor::case_evaluator::{CaseAction, CaseEvaluator};
use crate::snowflake::SnowflakeGen;

/// Env var carrying the comma-separated list of worker-pod env var
/// names to lift into `ExecutionContext.secrets` at startup
/// (noetl/ai-meta#34).  Operators populate the underlying env vars
/// via k8s Secret `envFrom`; playbook config (e.g.
/// `result_fetch.bearer_token: NOETL_FLIGHT_BEARER_TOKEN`)
/// references each by its env-var name as a keychain alias.
///
/// Per `agents/rules/execution-model.md`, business-logic credentials
/// (bearer tokens, API keys, etc.) belong in the NoETL keychain and
/// are referenced by alias from the playbook.  This allow-list is
/// the worker-side bridge between platform-mounted Secret envs and
/// the in-process keychain map the tools' `ctx.get_secret(...)`
/// calls read.
///
/// Empty / unset → no env vars get lifted (existing behaviour;
/// playbooks that pass literal credentials still work).
pub(crate) const KEYCHAIN_ENV_ALLOWLIST_VAR: &str = "NOETL_KEYCHAIN_ENV_VARS";

/// Maximum command attempts before a transient (retryable) pre-dispatch
/// failure is escalated to a terminal `call.error`.
///
/// A transient keychain transport error on a fresh command
/// (`command.attempts < MAX_PREDISPATCH_ATTEMPTS`) is left to the
/// command path's retry/redelivery — the worker does NOT emit a terminal
/// event, so a later attempt can still complete the step.  Once the
/// attempt counter reaches this ceiling the worker emits the terminal
/// `call.error` so the execution can't hang at `command.started`
/// indefinitely.  Terminal failures (clean 404, unsupported credential
/// type, malformed tool config) bypass this counter — they emit on the
/// first failure.  See noetl/ai-meta#78.
pub(crate) const MAX_PREDISPATCH_ATTEMPTS: u32 = 3;

/// Parse the comma-separated allow-list + look up each env var.
/// Names that are missing / empty in the environment are silently
/// skipped — that way an operator can stage rollouts (define the
/// allow-list ahead of mounting the Secret) without spamming
/// startup logs.
///
/// Returns a `(alias → value)` map ready to merge into
/// `ExecutionContext.secrets`.  Keys are the env-var names verbatim,
/// matching the convention playbook authors reference (so a
/// `bearer_token: NOETL_FLIGHT_BEARER_TOKEN` field resolves directly
/// without any prefix-strip transformation).
pub(crate) fn load_keychain_env_allowlist() -> HashMap<String, String> {
    let raw = match std::env::var(KEYCHAIN_ENV_ALLOWLIST_VAR) {
        Ok(s) => s,
        Err(_) => return HashMap::new(),
    };
    let mut out = HashMap::new();
    for name in raw.split(',').map(str::trim).filter(|s| !s.is_empty()) {
        match std::env::var(name) {
            Ok(value) if !value.is_empty() => {
                out.insert(name.to_string(), value);
            }
            _ => {
                // Allow-listed but not yet set on the pod — silent
                // skip.  An operator who's mid-rollout (allow-list
                // ahead of Secret mount) shouldn't see a noisy log
                // line per startup.
            }
        }
    }
    out
}

/// Command executor that runs tools and evaluates cases.
pub struct CommandExecutor {
    /// Tool registry with all available tools.
    tool_registry: ToolRegistry,

    /// Case evaluator for when/then logic.
    case_evaluator: CaseEvaluator,

    /// Control plane client for event emission.
    client: ControlPlaneClient,

    /// Worker ID.
    worker_id: String,

    /// Control-plane base URL.
    server_url: String,

    /// Application-side snowflake generator for `event_id` on every
    /// emitted envelope.  Per `observability.md` Principle 3 — the
    /// id is generated BEFORE the row hits the database so spans /
    /// metrics carry it at span-creation time and retries stay
    /// idempotent.
    snowflake: Arc<SnowflakeGen>,

    /// Same-node Arrow IPC cache for `call.done` results that
    /// exceed the broker's 100KB inline budget.  When a tool
    /// returns a large output (Postgres rowset, HTTP API
    /// response, etc.), the bytes go into shared memory + the
    /// event payload carries an `IpcHint` reference instead.
    /// Per R-2.1 (the `noetl-arrow-cache` crate); partial progress
    /// on noetl/worker#24.
    arrow_cache: Arc<ArrowIpcSharedMemoryCache>,

    /// Operator-provided keychain credentials lifted from worker
    /// pod env vars at startup (noetl/ai-meta#34).  The allow-list
    /// (`NOETL_KEYCHAIN_ENV_VARS`) names which env vars are credentials
    /// rather than runtime config; values get copied into each
    /// command's `ExecutionContext.secrets` so tools'
    /// `ctx.get_secret(alias)` calls resolve playbook keychain
    /// aliases (e.g. `result_fetch.bearer_token: NOETL_FLIGHT_BEARER_TOKEN`).
    ///
    /// Populated once at executor construction; immutable across
    /// the worker's lifetime.  Empty when the allow-list env var
    /// is unset (pre-#34 behaviour, no breakage).
    keychain_env: HashMap<String, String>,
}

impl CommandExecutor {
    /// Create a new command executor.
    ///
    /// At construction time, scans the `NOETL_KEYCHAIN_ENV_VARS`
    /// allow-list + lifts the named env vars into a per-executor
    /// keychain map.  Each command's `ExecutionContext.secrets`
    /// then carries these values so playbook keychain aliases like
    /// `result_fetch.bearer_token: NOETL_FLIGHT_BEARER_TOKEN`
    /// resolve via `ctx.get_secret(alias)`.  See
    /// [`load_keychain_env_allowlist`] for the env-var contract.
    pub fn new(
        client: ControlPlaneClient,
        worker_id: String,
        server_url: String,
        snowflake: Arc<SnowflakeGen>,
        arrow_cache: Arc<ArrowIpcSharedMemoryCache>,
    ) -> Self {
        let keychain_env = load_keychain_env_allowlist();
        if !keychain_env.is_empty() {
            // Observability per `agents/rules/observability.md` —
            // log the KEY NAMES at startup (not values; values are
            // credentials).  Operators verify the allow-list took
            // effect via `kubectl logs`.
            let mut names: Vec<&str> = keychain_env.keys().map(String::as_str).collect();
            names.sort_unstable();
            tracing::info!(
                count = names.len(),
                aliases = ?names,
                "Loaded keychain credentials from NOETL_KEYCHAIN_ENV_VARS"
            );
        }
        Self {
            tool_registry: create_default_registry(),
            case_evaluator: CaseEvaluator::new(),
            client,
            worker_id,
            server_url,
            snowflake,
            arrow_cache,
            keychain_env,
        }
    }

    /// Execute a command.
    ///
    /// Per `observability.md` Principle 1: every boundary call
    /// ships a span.  The `command.execute` span covers the full
    /// dispatch path (tool registry lookup, tool execution, case
    /// evaluation, lifecycle event emission) so downstream
    /// observability tooling (traces, metrics exemplars) can group
    /// every sub-operation under one execution.
    ///
    /// Principle 2 (metrics over logs): dispatch duration recorded
    /// to `noetl_worker_dispatch_duration_seconds{tool_kind=...}`;
    /// errors to `noetl_worker_dispatch_errors_total{tool_kind=...}`.
    /// Both labeled by tool_kind so the dashboard can spot which
    /// tools are slow / failing.
    pub async fn execute(&self, command: &Command) -> Result<()> {
        self.execute_with_server_url(command, None).await
    }

    /// Execute a command with an optional per-dispatch server-URL
    /// override.
    ///
    /// noetl/ai-meta#53 Gap 1: the NATS notification carries a
    /// `server_url` field identifying the server that PUBLISHED
    /// the command.  Multi-server deployments (e.g. a kind cluster
    /// running both `noetl-server` (Python) and `noetl-server-rust`
    /// side by side) need each command's lifecycle events
    /// (claim → started → call.done → completed) to flow back to
    /// the originating server so its orchestrator can advance the
    /// playbook.  Without an override, the executor's captured
    /// client + server_url (initialised from `NOETL_SERVER_URL` at
    /// worker startup) wins, and events for a Rust-server-dispatched
    /// command end up at the Python server — silently recorded to
    /// the shared DB but not driving any orchestrator.
    ///
    /// When `server_url_override` is `Some`, this method:
    ///   * Builds a per-dispatch `ControlPlaneClient` via
    ///     `client.with_server_url(override)` — cheap; the inner
    ///     `reqwest::Client` is Arc-shared.
    ///   * Threads the override URL through to `ExecutionContext`
    ///     (where `auth` tools resolve credentials against the
    ///     callback URL) and through every call site that would
    ///     otherwise have used `self.client` / `self.server_url`.
    ///
    /// `None` keeps the old behaviour: every dispatch uses the
    /// startup-configured client + server URL.
    pub async fn execute_with_server_url(
        &self,
        command: &Command,
        server_url_override: Option<&str>,
    ) -> Result<()> {
        let span = tracing::info_span!(
            "command.execute",
            execution_id = command.execution_id,
            command_id = %command.command_id,
            step = %command.step,
            tool_kind = %command.tool_kind,
            attempts = command.attempts,
        );
        let _enter = span.enter();

        // Per-dispatch client + URL.  When the notification carries
        // a server_url, build a fresh `ControlPlaneClient` pointed
        // at the publishing server; otherwise reuse the captured
        // client (and server_url) from worker startup.
        let dispatch_server_url: String = server_url_override
            .map(|s| s.to_string())
            .unwrap_or_else(|| self.server_url.clone());
        let dispatch_client: ControlPlaneClient = match server_url_override {
            Some(url) if url != self.server_url => self.client.with_server_url(url),
            _ => self.client.clone(),
        };

        // Timer captures the full dispatch latency including tool
        // execution + case evaluation + lifecycle events.  Recorded
        // on every exit path (success + error) so the histogram is
        // complete.
        let dispatch_start = std::time::Instant::now();
        let tool_kind = command.tool_kind.clone();
        // Helper to record the dispatch metric on every exit path.
        // Captured by the error-return + success-return code below.
        let record_metric = |error: bool| {
            crate::metrics::record_dispatch(
                &tool_kind,
                dispatch_start.elapsed().as_secs_f64(),
                error,
            );
        };

        // Build execution context
        let mut ctx =
            ExecutionContext::new(command.execution_id, &command.step, &dispatch_server_url)
                .with_worker_id(&self.worker_id)
                .with_command_id(&command.command_id);

        // Seed keychain credentials lifted from worker pod env at
        // startup (noetl/ai-meta#34).  Tools that read credentials
        // via `ctx.get_secret(alias)` (postgres, result_fetch, ...)
        // now resolve playbook-side keychain aliases against the
        // operator-provided `NOETL_KEYCHAIN_ENV_VARS` allow-list.
        // Per-command secrets from the playbook step (auth: block)
        // can still set / override entries — they layer on top of
        // these env-mounted defaults.
        for (alias, value) in &self.keychain_env {
            ctx.set_secret(alias, value);
        }

        // Add render context variables from command payload.
        ctx.variables = command.render_context.clone();
        ctx.variables
            .entry("action".to_string())
            .or_insert_with(|| serde_json::json!(command.tool_kind));
        ctx.variables
            .entry("node_name".to_string())
            .or_insert_with(|| serde_json::json!(command.step.clone()));

        // Emit command.started event.  R-1.2 PR-EE-3: `step` +
        // `worker_id` are top-level fields on the `ExecutorEvent`
        // shape, so the context payload only carries the
        // command-specific keys.  The server's `EventRequest` /
        // Python's `EventEmitRequest` both read `step` /
        // `worker_id` from the top level after EE-2 + EE-4.
        self.emit_event_via(
            &dispatch_client,
            "command.started",
            &command.step,
            "STARTED",
            command.execution_id,
            command.attempts,
            serde_json::json!({
                "command_id": command.command_id.clone(),
            }),
        )
        .await?;

        // Reconstruct the ToolConfig the noetl-tools registry expects.
        // `command.input` is the worker's full `context` JSON; the
        // tool-side config lives under `input.tool_config`.  Inject
        // `kind` from the executor `Command.tool_kind` field if the
        // nested config doesn't already carry it (mirrors the worker's
        // pre-PR-2d-2 `Command.tool_config_value()` behaviour).
        let mut tool_config_value = {
            let raw = command
                .input
                .get("tool_config")
                .cloned()
                .unwrap_or_else(|| serde_json::json!({}));
            // Tools whose server-side `tool_config` is an array
            // (currently only `task_sequence`, which sends a
            // pipeline `[{label: spec}, ...]`) used to be silently
            // discarded here — the old `if !cfg.is_object()` arm
            // replaced the array with `{}`, leaving the worker to
            // dispatch an empty config to the registry.  Preserve
            // the array by wrapping it under a `tool_config` key
            // so the flattened `ToolConfig.config` carries it
            // forward; the `task_sequence` parser accepts that
            // exact "worker envelope" shape per its `parse_tasks`
            // contract in noetl-tools v2.18.1.
            let mut cfg = if raw.is_object() {
                raw
            } else if raw.is_array() {
                serde_json::json!({ "tool_config": raw })
            } else {
                serde_json::json!({})
            };
            if let Some(map) = cfg.as_object_mut() {
                map.entry("kind".to_string())
                    .or_insert_with(|| serde_json::json!(command.tool_kind));
                if !map.contains_key("args") {
                    if let Some(args) = command.input.get("args") {
                        map.insert("args".to_string(), args.clone());
                    }
                }
            }
            cfg
        };

        // Resolve string `auth:` values (keychain aliases) into either
        // the noetl-tools `AuthConfig` shape or the tool's flat
        // connection fields BEFORE serde deserialization.  See
        // `auth_alias` module + noetl/ai-meta#48 for the regression
        // brief.  Idempotent: if `auth` is already a struct (or
        // absent), this is a no-op + no HTTP call.
        let alias_secrets = match super::auth_alias::resolve_auth_alias(
            &mut tool_config_value,
            &dispatch_client,
            command.execution_id,
        )
        .await
        {
            Ok(secrets) => secrets,
            Err(e) => {
                // Pre-dispatch credential-alias failure.  Before this
                // fix the `?` early-returned here and the worker just
                // logged "Command execution failed" — no `call.error`
                // ever reached the server, so the execution hung at
                // `command.started` forever (noetl/ai-meta#78).  Now we
                // classify: a terminal failure (clean 404 / unsupported
                // type / malformed shape) — or a transient one whose
                // attempt counter is exhausted — emits a terminal
                // `call.error` so the execution fails cleanly.  A
                // transient failure on a fresh command stays retryable.
                record_metric(true);
                let terminal = e.is_terminal() || command.attempts >= MAX_PREDISPATCH_ATTEMPTS;
                return self
                    .handle_predispatch_failure(
                        &dispatch_client,
                        command,
                        ctx.call_index,
                        terminal,
                        anyhow::Error::new(e),
                    )
                    .await;
            }
        };
        if !alias_secrets.is_empty() {
            // Per `observability.md` Principle 1: log keychain
            // alias resolution so operators can trace credential
            // lookups in the worker log alongside the dispatch
            // span.  Value never logged (just the alias name).
            tracing::info!(
                execution_id = command.execution_id,
                step = %command.step,
                aliases = ?alias_secrets.keys().collect::<Vec<_>>(),
                "Resolved keychain alias(es) for tool dispatch"
            );
        }
        for (alias, value) in alias_secrets {
            ctx.set_secret(&alias, &value);
        }

        let tool_config: ToolConfig = match serde_json::from_value(tool_config_value) {
            Ok(cfg) => cfg,
            Err(e) => {
                // Malformed tool config is the other pre-dispatch
                // failure that used to silently `?`-return and hang the
                // execution at `command.started` (noetl/ai-meta#78).
                // It is always terminal — the same bytes deserialize to
                // the same error on a retry — so emit a terminal
                // `call.error`.
                record_metric(true);
                let err = anyhow::Error::new(e)
                    .context("malformed tool config (pre-dispatch deserialization)");
                return self
                    .handle_predispatch_failure(
                        &dispatch_client,
                        command,
                        ctx.call_index,
                        true,
                        err,
                    )
                    .await;
            }
        };

        tracing::debug!(
            execution_id = command.execution_id,
            step = %command.step,
            tool = %tool_config.kind,
            attempts = command.attempts,
            "Executing tool"
        );

        // Execute the tool
        let tool_result = match self
            .tool_registry
            .execute_from_config(&tool_config, &ctx)
            .await
        {
            Ok(result) => {
                // Emit call.done event with a reference-only result
                // payload.  The Python broker's
                // `_validate_reference_only_payload` enforces that
                // `payload.result` only carries `{status, reference,
                // context, command_id}` at the top level — the raw
                // tool fields (`stdout` / `stderr` / `exit_code` /
                // `data` / `duration_ms`) live INSIDE `context` so
                // downstream steps can reference them via Jinja
                // (`step_name.data.rows[N].x`).
                //
                // For results under `NOETL_EVENT_RESULT_CONTEXT_MAX_BYTES`
                // (default 100 KB) the broker persists `context` as-is
                // and downstream Jinja templates can read the tool
                // output.  When the JSON would exceed that, the broker
                // silently drops the context (`_bounded_context`
                // returns None), so we pre-check the size on the Rust
                // side and emit a WARN log so operators can see *why*
                // their large-result step's downstream rendering is
                // empty.  Until the result-store / `noetl-arrow-cache`
                // reference path lands (noetl/worker#24), an
                // over-budget result still ships with just `{status}` —
                // identical behaviour to a silent broker drop, just
                // visible in the worker's logs.
                //
                // Defensive: the broker forbids `_internal_data` at
                // any depth in `result.context`.  Our `ToolResult`
                // doesn't surface that key, so the serialised value
                // round-trips cleanly through the validator.
                let result_context = serde_json::to_value(&result)
                    .unwrap_or_else(|_| serde_json::json!({ "status": result.status.to_string() }));
                let result_obj = match build_call_done_result(
                    &result_context,
                    &result.status.to_string(),
                    command.execution_id,
                    &command.step,
                    self.arrow_cache.as_ref(),
                    &dispatch_client,
                )
                .await
                {
                    Ok(obj) => obj,
                    Err(e) => {
                        tracing::warn!(
                            execution_id = command.execution_id,
                            step = %command.step,
                            error = %e,
                            "Failed to serialise tool result for inline context; falling back to status-only payload",
                        );
                        serde_json::json!({ "status": result.status.to_string() })
                    }
                };
                // noetl/ai-meta#43 Round 4 — `pending_callback` adoption.
                //
                // `Tool::Container` (and any future tool that dispatches a
                // long-running external work item) sets
                // `ToolResult.pending_callback = Some(true)` to signal that
                // the step's terminal `call.done` will be emitted
                // asynchronously by a separate callback path
                // (`POST /api/internal/container-callback/{eid}/{step}`
                // on noetl-server, driven by `noetl-k8s-watcher`).  When
                // that marker is set the worker MUST NOT emit its own
                // `call.done` — doing so races the watcher's later emit
                // (the server's stale-counter
                // `noetl_container_callback_stale_total` records the
                // collision) and the orchestrator sees an early-completion
                // value with only the Job handle in `data`.
                //
                // All other tools omit the field (`None`) which keeps the
                // pre-existing behaviour fully intact — the worker emits
                // `call.done` immediately as the tool's terminal event.
                if matches!(result.pending_callback, Some(true)) {
                    tracing::info!(
                        execution_id = command.execution_id,
                        step = %command.step,
                        tool = %tool_config.kind,
                        "skipping call.done emit per pending_callback marker (await async callback)"
                    );
                    crate::metrics::record_call_done_skipped_pending_callback(&tool_config.kind);
                } else {
                    self.emit_event_via(
                        &dispatch_client,
                        "call.done",
                        &command.step,
                        "COMPLETED",
                        command.execution_id,
                        command.attempts,
                        serde_json::json!({
                            "command_id": command.command_id.clone(),
                            "call_index": ctx.call_index,
                            "result": result_obj,
                        }),
                    )
                    .await?;
                }

                result
            }
            Err(e) => {
                // Emit call.error event
                self.emit_event_via(
                    &dispatch_client,
                    "call.error",
                    &command.step,
                    "FAILED",
                    command.execution_id,
                    command.attempts,
                    serde_json::json!({
                        "command_id": command.command_id.clone(),
                        "call_index": ctx.call_index,
                        "error": e.to_string(),
                    }),
                )
                .await?;

                // Emit command.failed event
                self.emit_event_via(
                    &dispatch_client,
                    "command.failed",
                    &command.step,
                    "FAILED",
                    command.execution_id,
                    command.attempts,
                    serde_json::json!({
                        "command_id": command.command_id.clone(),
                        "error": e.to_string(),
                    }),
                )
                .await?;

                record_metric(true);
                return Err(e.into());
            }
        };

        // Parse cases from command.  The executor's `Command.input`
        // carries the worker's full `context` JSON, so `cases`
        // lives at `command.input.cases` (was `command.context.cases`
        // pre-PR-2d-2).
        let cases: Vec<crate::executor::case_evaluator::Case> = command
            .input
            .get("cases")
            .and_then(|v| v.as_array())
            .map(|list| {
                list.iter()
                    .filter_map(|value| serde_json::from_value(value.clone()).ok())
                    .collect()
            })
            .unwrap_or_default();

        // Evaluate cases
        if !cases.is_empty() {
            if let Some(case_result) =
                self.case_evaluator
                    .evaluate(&cases, &ctx, tool_result.data.as_ref())?
            {
                match case_result.action {
                    CaseAction::Exit { status, data } => {
                        // Emit step.exit event.  `step` is top-level
                        // on the EE shape; the case's status string
                        // becomes the envelope status so the projector
                        // sees the actual case outcome.
                        let exit_status = status.clone();
                        self.emit_event_via(
                            &dispatch_client,
                            "step.exit",
                            &command.step,
                            &exit_status,
                            command.execution_id,
                            command.attempts,
                            serde_json::json!({
                                "status": status,
                                "data": data,
                            }),
                        )
                        .await?;
                    }
                    CaseAction::SetVar { name, value } => {
                        // Set variable via API
                        dispatch_client
                            .set_variable(command.execution_id, &name, value)
                            .await?;
                    }
                    CaseAction::Fail { message } => {
                        // Emit command.failed event
                        self.emit_event_via(
                            &dispatch_client,
                            "command.failed",
                            &command.step,
                            "FAILED",
                            command.execution_id,
                            command.attempts,
                            serde_json::json!({
                                "command_id": command.command_id.clone(),
                                "error": message,
                            }),
                        )
                        .await?;

                        record_metric(true);
                        return Err(anyhow::anyhow!("Case evaluation failed: {}", message));
                    }
                    CaseAction::Continue | CaseAction::Goto { .. } | CaseAction::Retry { .. } => {
                        // These are handled by the orchestrator
                    }
                }
            }
        }

        // Emit command.completed event.  The tool's terminal status
        // (e.g. `"success"` / `"failure"` from the tool registry)
        // becomes the envelope status — projectors group by status
        // to compute success/failure rates per step.
        let completion_status = tool_result.status.to_string();
        self.emit_event_via(
            &dispatch_client,
            "command.completed",
            &command.step,
            &completion_status,
            command.execution_id,
            command.attempts,
            serde_json::json!({
                "command_id": command.command_id.clone(),
                "status": tool_result.status.to_string(),
            }),
        )
        .await?;

        record_metric(false);
        Ok(())
    }

    /// Emit an event to the control plane.
    ///
    /// R-1.2 PR-EE-3: constructs the shared `ExecutorEvent` shape
    /// (`step` + `status` + `created_at` + `context` at the top
    /// level, plus `worker_id` from the executor's own id).
    ///
    /// Post-EE-3 follow-ups now folded in:
    ///
    /// - `event_id` is stamped from the application-side snowflake
    ///   generator per `observability.md` Principle 3 — the id
    ///   exists at span-creation time + survives retries (which
    ///   used to either create duplicate rows or leave a NULL id
    ///   window).  Closes noetl/worker#12.
    /// - `meta.attempts` carries the executor `Command.attempts`
    ///   counter so retry behaviour rides the event log without
    ///   needing to reach back into the worker's logs.  Closes
    ///   noetl/worker#13.
    ///
    /// Per `observability.md` Principle 2: records the emit latency
    /// to `noetl_worker_event_emit_duration_seconds{event_type=...}`.
    /// The retries counter is incremented only when the underlying
    /// `emit_event_with_retry` actually retried (i.e. the first
    /// attempt failed); the retry count is currently not exposed
    /// by the client, so this MVP records 0 — a follow-up will
    /// thread the actual retry count back from the client.
    /// Emit a lifecycle event via the given control-plane client.
    ///
    /// Takes an explicit client argument so per-dispatch routing
    /// (noetl/ai-meta#53 Gap 1) can override the captured
    /// `self.client` for events that should land on the server
    /// that published the command rather than the startup-
    /// configured `NOETL_SERVER_URL` server.  Every callsite inside
    /// `execute_with_server_url` passes a per-dispatch
    /// `ControlPlaneClient`.
    async fn emit_event_via(
        &self,
        client: &ControlPlaneClient,
        event_type: &str,
        step: &str,
        status: &str,
        execution_id: i64,
        attempts: u32,
        context: serde_json::Value,
    ) -> Result<()> {
        let event = ExecutorEvent {
            execution_id,
            event_type: event_type.to_string(),
            step: step.to_string(),
            status: status.to_string(),
            created_at: chrono::Utc::now(),
            context,
            event_id: Some(self.snowflake.next_id()),
            worker_id: Some(self.worker_id.clone()),
            meta: Some(serde_json::json!({ "attempts": attempts })),
        };

        let emit_start = std::time::Instant::now();
        let result = client.emit_event_with_retry(event, 3).await;
        crate::metrics::record_event_emit(event_type, emit_start.elapsed().as_secs_f64(), 0);
        result
    }

    /// Handle a failure that happens BEFORE the tool-dispatch match
    /// (credential-alias resolution, tool-config deserialization).
    ///
    /// noetl/ai-meta#78: these failures used to early-`?`-return from
    /// `execute_with_server_url` straight to the worker dispatch loop,
    /// which only logged `Command execution failed` — no `call.error`
    /// reached the server, so the execution sat at `command.started`
    /// forever.
    ///
    /// When `terminal` is true this emits the same `call.error` +
    /// `command.failed` pair the post-dispatch error arm emits (matching
    /// payload fields so the server/UI treat both identically), so the
    /// execution reaches a terminal FAILED state instead of hanging.
    /// When `terminal` is false (a transient transport error on a
    /// command whose attempt counter isn't yet exhausted) it logs a WARN
    /// and emits nothing, leaving the command path's retry/redelivery to
    /// run.
    ///
    /// Always returns `Err(error)` so the caller's early return
    /// propagates the failure to the dispatch loop (which records it and
    /// balances the in-flight gauge).  The invariant the dispatch loop
    /// relies on: by the time this returns `Err`, a terminal failure has
    /// already emitted its terminal event here — the loop must NOT emit
    /// its own (doing so would double-emit terminals and clobber the
    /// retryable path).  Per `observability.md` Principle 4 every line
    /// carries `execution_id` + `command_id` + `step`.
    async fn handle_predispatch_failure(
        &self,
        client: &ControlPlaneClient,
        command: &Command,
        call_index: usize,
        terminal: bool,
        error: anyhow::Error,
    ) -> Result<()> {
        if terminal {
            // Emit call.error — mirrors the post-dispatch error arm's
            // payload (command_id + call_index + error string) so the
            // projector/UI render a pre-dispatch failure identically to
            // a tool-execution failure.
            self.emit_event_via(
                client,
                "call.error",
                &command.step,
                "FAILED",
                command.execution_id,
                command.attempts,
                serde_json::json!({
                    "command_id": command.command_id.clone(),
                    "call_index": call_index,
                    "error": error.to_string(),
                }),
            )
            .await?;

            // Emit command.failed so the orchestrator advances the
            // execution to a terminal FAILED state.
            self.emit_event_via(
                client,
                "command.failed",
                &command.step,
                "FAILED",
                command.execution_id,
                command.attempts,
                serde_json::json!({
                    "command_id": command.command_id.clone(),
                    "error": error.to_string(),
                }),
            )
            .await?;

            tracing::error!(
                execution_id = command.execution_id,
                command_id = %command.command_id,
                step = %command.step,
                attempts = command.attempts,
                error = %error,
                "Pre-dispatch failure is terminal; emitted call.error + command.failed so the execution fails cleanly instead of hanging at command.started",
            );
        } else {
            tracing::warn!(
                execution_id = command.execution_id,
                command_id = %command.command_id,
                step = %command.step,
                attempts = command.attempts,
                error = %error,
                "Pre-dispatch failure is transient (retryable); no terminal call.error emitted, leaving retry to the command path",
            );
        }

        Err(error)
    }
}

/// Soft upper bound for the JSON-serialised size of
/// `payload.result.context` on `call.done` events.  Matches the
/// Python broker's `NOETL_EVENT_RESULT_CONTEXT_MAX_BYTES` default
/// (the broker's `_bounded_context` returns None and silently
/// drops the field above this threshold; we pre-check Rust-side
/// so operators see a WARN log instead of a silent drop).
const INLINE_CONTEXT_MAX_BYTES: usize = 100 * 1024;

/// Encoding choice for the over-budget shm cache write.  R-2.2:
/// tabular tool outputs (DuckDB / Postgres / Snowflake rowsets)
/// encode as Arrow IPC stream bytes so colocated consumers benefit
/// from the columnar layout; non-tabular outputs (shell stdout, HTTP
/// JSON, etc.) fall back to JSON bytes.
struct ShmPayload {
    /// Bytes to write into the shm region.
    bytes: Vec<u8>,
    /// `media_type` to stamp on the `IpcHint` so consumers know how
    /// to decode (`application/vnd.apache.arrow.stream` for tabular;
    /// `application/json` for the fallback path).
    media_type: String,
    /// `schema_digest` argument to `cache.put_arrow_ipc` — `"arrow"`
    /// when the bytes are Arrow IPC (the schema is recoverable from
    /// the stream's first message); `"json"` for the fallback.
    schema_digest: &'static str,
    /// Row count for the `IpcHint.row_count` field.  Only populated
    /// for the tabular path.
    row_count: Option<u64>,
}

/// Build the `payload.result` object for a `call.done` event,
/// choosing between four exit shapes based on the JSON-serialised
/// size of the supplied `context` and the success of the durable
/// result-store write + same-node shm staging.
///
/// Fallback chain (highest fidelity first):
///
/// 1. **Inline.**  Serialised context ≤ [`INLINE_CONTEXT_MAX_BYTES`]
///    → `{status, context}`.  No HTTP, no shm.  Both cross-node
///    and colocated consumers read the rendered fields directly
///    off `result.context`.
/// 2. **Durable + colocated acceleration.**  Over-budget AND the
///    durable PUT to `/api/result/{execution_id}` AND the shm
///    cache `put` BOTH succeed → `{status, reference}` where
///    `reference` is a `ResultRef`-shaped dict carrying the
///    `noetl://` URI for cross-node fetch + a nested `ipc` field
///    with the [`IpcHint`] for same-node attach.  Mirrors the
///    Python worker's `TempStore.put_ipc_bytes` shape.
/// 3. **Durable only.**  Over-budget, durable PUT succeeds, shm
///    cache `put` fails → same `reference = ResultRef` dict but
///    without the `ipc` field.  Cross-node consumers work; same-
///    node consumers pay the durable round-trip instead of taking
///    the shm shortcut.
/// 4. **Colocated only.**  Over-budget, durable PUT fails, shm
///    cache `put` succeeds → `{status, reference}` carrying the
///    bare [`IpcHint`] (matches noetl/worker#28 behaviour).
///    Same-node consumers still read; cross-node consumers get
///    nothing.  WARN-logged so operators see the degradation.
/// 5. **Status only.**  Over-budget, both durable AND shm fail →
///    `{status}`.  Predictable + visible fallback rather than a
///    silent broker drop.  ERROR-logged.
///
/// Errors only if the serde serialisation itself fails (which
/// shouldn't happen for `serde_json::Value` inputs but the
/// signature stays honest via `serde_json::Error`).
async fn build_call_done_result(
    context: &serde_json::Value,
    status: &str,
    execution_id: i64,
    step: &str,
    arrow_cache: &ArrowIpcSharedMemoryCache,
    client: &ControlPlaneClient,
) -> Result<serde_json::Value, serde_json::Error> {
    // Producer-side credential scrub per
    // `agents/rules/execution-model.md` secrets rule.  The Python
    // server already scrubs `PUT /api/result/{execution_id}` bodies
    // at the boundary, but the worker emits THREE paths from this
    // function (inline `context`, durable PUT, shm cache stage) and
    // only the durable one rides through the server-side scrub.
    // Scrubbing here covers all three at once and shortens the
    // wire-transit window even for the durable path.
    //
    // We clone-and-scrub instead of mutating in place because the
    // caller (`CommandExecutor::execute`) emits the unscrubbed
    // `result_context` into its `serde_json::to_value(&result)` for
    // metric labels / logging.  Once we own the scrubbed copy below,
    // every reference to the context routes through it.
    let context = crate::scrub::scrub_cloned(context);
    let context = &context;

    let serialised = serde_json::to_string(context)?;
    if serialised.len() <= INLINE_CONTEXT_MAX_BYTES {
        return Ok(serde_json::json!({ "status": status, "context": context }));
    }

    // Over-budget.  R-2.2: pick the right shm encoding based on
    // whether the context is tabular (DuckDB / Postgres / Snowflake
    // rowset shape).  Tabular outputs go in as Arrow IPC stream
    // bytes — `noetl_tools::arrow_codec::try_encode_tabular_json`
    // returns `Some` for the canonical `{columns, rows}` (or
    // `{data: {columns, rows}}`) shape and the encoded bytes
    // round-trip through `pyarrow.ipc.RecordBatchStreamReader` for
    // cross-stack consumers.  Non-tabular outputs (shell stdout,
    // HTTP JSON, etc.) stage as JSON bytes — matches the
    // noetl/worker#28 behaviour.
    let shm_payload = match noetl_tools::arrow_codec::try_encode_tabular_json(context) {
        Some(enc) => {
            tracing::debug!(
                execution_id,
                step,
                row_count = enc.row_count,
                arrow_bytes = enc.bytes.len(),
                json_bytes = serialised.len(),
                "Tabular tool result detected; staging as Arrow IPC stream for shm cache.",
            );
            ShmPayload {
                bytes: enc.bytes,
                media_type: enc.media_type.to_string(),
                schema_digest: "arrow",
                row_count: Some(enc.row_count as u64),
            }
        }
        None => ShmPayload {
            bytes: serialised.as_bytes().to_vec(),
            media_type: "application/json".to_string(),
            schema_digest: "json",
            row_count: None,
        },
    };

    // Try the durable result-store first — that's the only path
    // that helps cross-node consumers.  Then layer the shm cache
    // on top as a colocated acceleration so same-node consumers
    // can skip the durable round-trip.
    //
    // The durable PUT always sends JSON (the server accepts `data:
    // Any` and stores in its tiered backends).  Cross-node consumers
    // fetch the JSON back; only the shm fast path benefits from the
    // Arrow IPC re-encoding above.
    let put_start = std::time::Instant::now();
    let durable_outcome = client
        .put_result(execution_id, step, context, "execution", Some(step))
        .await;
    let put_elapsed = put_start.elapsed().as_secs_f64();

    let shm_outcome = arrow_cache.put_arrow_ipc(
        &shm_payload.bytes,
        shm_payload.schema_digest,
        shm_payload.row_count,
        None,
    );

    match (durable_outcome, shm_outcome) {
        (Ok(durable), shm_result) => {
            // Build the ResultRef-shaped reference.  Mirrors
            // Python's `ResultRef` model (noetl/core/storage/models.py)
            // — kind discriminator + URI + tier + scope + meta.
            // The `ipc` field nests an `IpcHint` for the colocated
            // fast path (Python expects `ipc: Optional[IpcHint]`).
            crate::metrics::record_result_store_put(put_elapsed, serialised.len(), false);
            let mut reference = serde_json::json!({
                "kind": "result_ref",
                "ref": durable.r#ref,
                "store": durable.store,
                "scope": durable.scope,
                "meta": {
                    "bytes": durable.bytes,
                    "sha256": durable.sha256,
                    "media_type": "application/json",
                    "content_type": "application/json",
                },
            });
            if let Some(expiry) = durable.expires_at.as_ref() {
                reference["expires_at"] = serde_json::Value::String(expiry.clone());
            }
            if let Ok(mut hint) = shm_result {
                hint.media_type = shm_payload.media_type.clone();
                // Stamp the hint as the `ipc` field on the
                // ResultRef so same-node consumers can attach
                // without the durable round-trip.  The
                // `media_type` field tells the consumer how to
                // decode the shm bytes: Arrow IPC stream for
                // tabular outputs, JSON for everything else.
                if let Ok(ipc_value) = serde_json::to_value(&hint) {
                    reference["ipc"] = ipc_value;
                }
                tracing::info!(
                    execution_id,
                    step,
                    context_bytes = serialised.len(),
                    shm_bytes = shm_payload.bytes.len(),
                    shm_media_type = %hint.media_type,
                    result_ref = %reference["ref"].as_str().unwrap_or(""),
                    shm_name = %hint.shm_name,
                    put_duration_seconds = put_elapsed,
                    "Tool result exceeds inline budget; staged in durable result store + shared-memory cache.",
                );
            } else {
                tracing::info!(
                    execution_id,
                    step,
                    context_bytes = serialised.len(),
                    result_ref = %reference["ref"].as_str().unwrap_or(""),
                    put_duration_seconds = put_elapsed,
                    "Tool result exceeds inline budget; staged in durable result store (shm cache unavailable).",
                );
            }
            // noetl/ai-meta#69 — embed an inline `context.data`
            // block carrying the synthetic `_ref` URI alongside the
            // `reference` block.  Without this, the orchestrator's
            // `extract_user_data` walks `outer.context.result.context.data`
            // and finds nothing for over-budget results, so a
            // downstream consumer like the `artifact` tool's
            // `result_ref: '{{ step._ref }}'` template renders to
            // null — the artifact tool then errors on
            // `Invalid artifact config: invalid type: null, expected
            // a string`.  Embedding `_ref` here makes the URI
            // template resolution work without bloating the inline
            // payload (single string, well under the inline budget).
            //
            // The `context.data` shape mirrors the under-budget path's
            // user_data layer so the orchestrator's extraction logic
            // (`extract_user_data` + the noetl/ai-meta#66 `step.data`
            // accessor) finds the same shape regardless of which
            // branch produced the call.done.
            //
            // Future expansion (when `output.output_select` plumbing
            // lands at the server-side ToolSpec layer): also project
            // the selected fields here alongside `_ref` so
            // `{{ step.<selected_field> }}` resolves directly without
            // a result_fetch round-trip.  Today only `_ref` lives
            // here; consumers that need full data dispatch the
            // `artifact` tool (`kind: artifact, action: get, input:
            // {result_ref: '{{ step._ref }}'}`) which uses the URI
            // to read the durable result.
            let inline_data = serde_json::json!({ "_ref": durable.r#ref });
            Ok(serde_json::json!({
                "status": status,
                "context": { "data": inline_data },
                "reference": reference,
            }))
        }
        (Err(durable_err), Ok(mut hint)) => {
            // Durable PUT failed but shm worked.  Emit the bare
            // IpcHint as before (#28 behaviour) — degraded mode,
            // cross-node consumers can't read this.  R-2.2: the
            // `media_type` reflects the actual encoding (Arrow IPC
            // for tabular, JSON for fallback) so a same-node
            // consumer reads the right bytes.
            crate::metrics::record_result_store_put_error();
            hint.media_type = shm_payload.media_type.clone();
            let reference = serde_json::to_value(&hint).unwrap_or_else(|_| {
                serde_json::json!({
                    "kind": "arrow_ipc",
                    "shm_name": hint.shm_name.clone(),
                    "byte_length": hint.byte_length,
                    "media_type": shm_payload.media_type,
                })
            });
            tracing::warn!(
                execution_id,
                step,
                context_bytes = serialised.len(),
                shm_bytes = shm_payload.bytes.len(),
                shm_media_type = %hint.media_type,
                shm_name = %hint.shm_name,
                error = %durable_err,
                "Durable result-store PUT failed; falling back to shared-memory cache only. \
                 Cross-node consumers will see an empty result.",
            );
            Ok(serde_json::json!({ "status": status, "reference": reference }))
        }
        (Err(durable_err), Err(shm_err)) => {
            // Both failed.  Status-only fallback — broker accepts
            // the event but downstream Jinja references will be
            // empty.  ERROR-logged so operators see the drop.
            crate::metrics::record_result_store_put_error();
            tracing::error!(
                execution_id,
                step,
                context_bytes = serialised.len(),
                inline_budget_bytes = INLINE_CONTEXT_MAX_BYTES,
                durable_error = %durable_err,
                shm_error = %shm_err,
                "Tool result exceeds inline budget and BOTH durable + shm staging failed; \
                 emitting status-only result.  Downstream Jinja references will be empty.",
            );
            Ok(serde_json::json!({ "status": status }))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::{extract::Path, http::StatusCode as AxumStatus, routing::put, Json, Router};
    use noetl_arrow_cache::CacheConfig;
    use tokio::net::TcpListener;

    // ------------------------------------------------------------
    // Keychain env-var allow-list (noetl/ai-meta#34)
    //
    // The env-var loader is independent of the rest of the executor
    // surface — these tests use serial mutation of `std::env::*`,
    // which means they MUST run on the same OS thread.  `cargo test`
    // already serialises tests inside a module by default but we
    // also clean every var we set (drop guard) so cross-test bleed
    // is bounded.
    // ------------------------------------------------------------

    /// RAII guard that sets a process env var on construction and
    /// restores its prior state on drop.  Without this, a panic
    /// inside a test leaks the test's env var into sibling tests
    /// (especially when running with `--test-threads=1` for
    /// repeatability), turning what should be a single failure into
    /// a cascade.
    struct EnvGuard {
        key: String,
        prev: Option<String>,
    }

    impl EnvGuard {
        fn set(key: &str, value: &str) -> Self {
            let prev = std::env::var(key).ok();
            // SAFETY: process is single-threaded for these tests via
            // the module-level serialisation cargo test enforces.
            unsafe { std::env::set_var(key, value) };
            Self {
                key: key.to_string(),
                prev,
            }
        }

        fn unset(key: &str) -> Self {
            let prev = std::env::var(key).ok();
            unsafe { std::env::remove_var(key) };
            Self {
                key: key.to_string(),
                prev,
            }
        }
    }

    impl Drop for EnvGuard {
        fn drop(&mut self) {
            unsafe {
                match &self.prev {
                    Some(v) => std::env::set_var(&self.key, v),
                    None => std::env::remove_var(&self.key),
                }
            }
        }
    }

    #[test]
    fn keychain_env_unset_returns_empty() {
        // Default no-auth shape — the allow-list env var isn't set,
        // existing deployments (pre-#34) keep working unchanged.
        let _g = EnvGuard::unset(KEYCHAIN_ENV_ALLOWLIST_VAR);
        let loaded = load_keychain_env_allowlist();
        assert!(loaded.is_empty(), "expected empty map, got {loaded:?}");
    }

    #[test]
    fn keychain_env_loads_listed_vars() {
        let _g = EnvGuard::set(
            KEYCHAIN_ENV_ALLOWLIST_VAR,
            "TEST_NOETL_KC_VAR_A,TEST_NOETL_KC_VAR_B",
        );
        let _a = EnvGuard::set("TEST_NOETL_KC_VAR_A", "alpha-secret");
        let _b = EnvGuard::set("TEST_NOETL_KC_VAR_B", "beta-secret");
        let loaded = load_keychain_env_allowlist();
        assert_eq!(loaded.len(), 2);
        assert_eq!(
            loaded.get("TEST_NOETL_KC_VAR_A").map(String::as_str),
            Some("alpha-secret")
        );
        assert_eq!(
            loaded.get("TEST_NOETL_KC_VAR_B").map(String::as_str),
            Some("beta-secret")
        );
    }

    #[test]
    fn keychain_env_tolerates_whitespace_and_empty_entries() {
        let _g = EnvGuard::set(
            KEYCHAIN_ENV_ALLOWLIST_VAR,
            " TEST_NOETL_KC_WS_A , , TEST_NOETL_KC_WS_B,",
        );
        let _a = EnvGuard::set("TEST_NOETL_KC_WS_A", "first");
        let _b = EnvGuard::set("TEST_NOETL_KC_WS_B", "second");
        let loaded = load_keychain_env_allowlist();
        assert_eq!(loaded.len(), 2);
        assert_eq!(
            loaded.get("TEST_NOETL_KC_WS_A").map(String::as_str),
            Some("first")
        );
        assert_eq!(
            loaded.get("TEST_NOETL_KC_WS_B").map(String::as_str),
            Some("second")
        );
    }

    #[test]
    fn keychain_env_silently_skips_missing_vars() {
        // Operator allow-listed a var ahead of mounting the Secret
        // — startup proceeds without it (mid-rollout shape) rather
        // than spamming a warn / refusing to start.
        let _g = EnvGuard::set(
            KEYCHAIN_ENV_ALLOWLIST_VAR,
            "TEST_NOETL_KC_PRESENT,TEST_NOETL_KC_ABSENT",
        );
        let _p = EnvGuard::set("TEST_NOETL_KC_PRESENT", "here");
        let _a = EnvGuard::unset("TEST_NOETL_KC_ABSENT");
        let loaded = load_keychain_env_allowlist();
        assert_eq!(loaded.len(), 1);
        assert_eq!(
            loaded.get("TEST_NOETL_KC_PRESENT").map(String::as_str),
            Some("here")
        );
        assert!(!loaded.contains_key("TEST_NOETL_KC_ABSENT"));
    }

    #[test]
    fn keychain_env_skips_empty_string_values() {
        // Distinguishing "unset" vs "set to empty string" is a
        // common deployment-time surprise — both shapes should be
        // skipped.  Otherwise a Secret with a blank field would
        // silently authenticate as an empty token, which is worse
        // than failing closed.
        let _g = EnvGuard::set(KEYCHAIN_ENV_ALLOWLIST_VAR, "TEST_NOETL_KC_BLANK");
        let _b = EnvGuard::set("TEST_NOETL_KC_BLANK", "");
        let loaded = load_keychain_env_allowlist();
        assert!(loaded.is_empty(), "blank values must be skipped");
    }

    #[test]
    fn keychain_env_empty_allowlist_returns_empty() {
        let _g = EnvGuard::set(KEYCHAIN_ENV_ALLOWLIST_VAR, "");
        let loaded = load_keychain_env_allowlist();
        assert!(loaded.is_empty());
    }

    /// Build a test-isolated cache with a unique namespace so
    /// concurrent test runs don't collide on shm names.  POSIX shm
    /// names are filesystem-global; the namespace is the prefix
    /// stamped onto every entry the cache produces.
    fn test_cache(namespace: &str) -> Arc<ArrowIpcSharedMemoryCache> {
        let config = CacheConfig {
            namespace: namespace.to_string(),
            budget_bytes: 8 * 1024 * 1024,
            default_lease_seconds: 60.0,
            producer: "test".to_string(),
            node_id: "test-node".to_string(),
        };
        Arc::new(ArrowIpcSharedMemoryCache::with_config(config))
    }

    /// Build a cache too small to admit even a modest payload so the
    /// over-budget `put_arrow_ipc` call deterministically fails.
    /// Used to exercise the "durable failure + shm failure → status-
    /// only" branch.
    fn tiny_test_cache(namespace: &str) -> Arc<ArrowIpcSharedMemoryCache> {
        let config = CacheConfig {
            namespace: namespace.to_string(),
            // 1 KB budget; oversized contexts in these tests are
            // > 100 KB so the shm cache rejects them.
            budget_bytes: 1024,
            default_lease_seconds: 60.0,
            producer: "test".to_string(),
            node_id: "test-node".to_string(),
        };
        Arc::new(ArrowIpcSharedMemoryCache::with_config(config))
    }

    /// Spin up an in-test axum mock of the Python server's
    /// `PUT /api/result/{execution_id}` endpoint.  Returns the
    /// bound `(base_url, server_handle)`; drop the handle to stop
    /// the server.  Each test gets its own mock so they don't share
    /// state.
    async fn start_mock_result_store(
        response_body: serde_json::Value,
    ) -> (String, tokio::task::JoinHandle<()>) {
        async fn handler(
            Path(execution_id): Path<i64>,
            Json(body): Json<serde_json::Value>,
        ) -> Result<Json<serde_json::Value>, AxumStatus> {
            // Returning the body's `name` interpolated into the
            // canned response is overkill for the assertions here;
            // hand back a fixed ResultRef shape with the
            // execution_id wired in.
            Ok(Json(serde_json::json!({
                "ref": format!("noetl://execution/{}/result/{}/abcd1234",
                    execution_id,
                    body.get("name").and_then(|v| v.as_str()).unwrap_or("step")),
                "store": "disk",
                "scope": "execution",
                "expires_at": "2026-06-01T00:00:00Z",
                "bytes": body
                    .get("data")
                    .map(|d| d.to_string().len() as u64)
                    .unwrap_or(0),
                "sha256": "deadbeefcafe",
            })))
        }

        // Pick which handler to install: if the caller passed a
        // `null` body, use the dynamic one (echoes execution_id +
        // step into the canned ResultRef shape); otherwise serve
        // the caller-supplied response body verbatim.
        let app = if response_body.is_null() {
            Router::new().route("/api/result/{execution_id}", put(handler))
        } else {
            // Per-test ResultPutResponse override path — captures
            // the canned `response_body` so tests can override
            // individual fields (e.g. drive a particular `store`
            // tier or omit `expires_at`).
            let canned = response_body;
            let canned_handler = move |Path(_): Path<i64>, Json(_): Json<serde_json::Value>| {
                let body = canned.clone();
                async move { Ok::<_, AxumStatus>(Json(body)) }
            };
            Router::new().route("/api/result/{execution_id}", put(canned_handler))
        };

        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("local_addr");
        let base = format!("http://{}", addr);
        let handle = tokio::spawn(async move {
            axum::serve(listener, app).await.expect("axum serve");
        });
        (base, handle)
    }

    #[test]
    fn test_command_executor_creation() {
        let client = ControlPlaneClient::new("http://localhost:8082");
        let snowflake = Arc::new(SnowflakeGen::with_node_and_epoch(1, 0));
        let cache = test_cache("wkr-test-ctor");
        let executor = CommandExecutor::new(
            client,
            "worker-1".to_string(),
            "http://localhost:8082".to_string(),
            snowflake,
            cache,
        );

        // Verify tools are registered
        assert!(executor.tool_registry.has("shell"));
        assert!(executor.tool_registry.has("http"));
        assert!(executor.tool_registry.has("rhai"));
    }

    /// Branch 1 — small tool result rides the inline
    /// `result.context` path.  No HTTP, no shm.  Downstream Jinja
    /// templates can reference fields off `result.context` directly.
    #[tokio::test]
    async fn build_call_done_result_inlines_small_context() {
        let cache = test_cache("wkr-test-small");
        // Client points at an unreachable URL; the inline path
        // must NOT make any HTTP call.
        let client = ControlPlaneClient::new("http://127.0.0.1:1");
        let context = serde_json::json!({
            "stdout": "hello",
            "exit_code": 0,
            "duration_ms": 12,
        });
        let result =
            build_call_done_result(&context, "COMPLETED", 42, "greet", cache.as_ref(), &client)
                .await
                .unwrap();
        assert_eq!(result["status"], "COMPLETED");
        assert_eq!(result["context"]["stdout"], "hello");
        assert_eq!(result["context"]["exit_code"], 0);
        // The structure stays valid against the broker's
        // _STRICT_RESULT_ALLOWED_KEYS = {status, reference,
        // context, command_id} contract.
        let result_keys: Vec<&str> = result
            .as_object()
            .unwrap()
            .keys()
            .map(|k| k.as_str())
            .collect();
        for key in &result_keys {
            assert!(
                ["status", "context"].contains(key),
                "unexpected key: {}",
                key,
            );
        }
        // No bytes should have landed in the shm cache for the
        // inline path.
        assert_eq!(cache.used_bytes(), 0);
    }

    /// Branch 2 — over-budget AND both durable PUT + shm staging
    /// succeed → `{status, reference}` carrying a `ResultRef`-shaped
    /// dict with a nested `ipc` field for the colocated fast path.
    #[tokio::test]
    async fn build_call_done_result_uses_durable_plus_ipc_when_both_succeed() {
        let cache = test_cache("wkr-test-durable-plus-ipc");
        let (base, handle) = start_mock_result_store(serde_json::Value::Null).await;
        let client = ControlPlaneClient::new(&base);

        let big_string: String = "x".repeat(INLINE_CONTEXT_MAX_BYTES + 4096);
        let context = serde_json::json!({ "stdout": big_string });
        let result = build_call_done_result(
            &context,
            "COMPLETED",
            12345,
            "big_step",
            cache.as_ref(),
            &client,
        )
        .await
        .unwrap();

        assert_eq!(result["status"], "COMPLETED");
        let reference = result.get("reference").expect("must have reference");
        assert_eq!(reference["kind"], "result_ref");
        let durable_ref = reference["ref"].as_str().expect("ref must be a string");
        assert!(durable_ref.starts_with("noetl://execution/12345/result/big_step/"));
        assert_eq!(reference["store"], "disk");
        assert_eq!(reference["scope"], "execution");
        // `meta` is populated from the server response.
        assert!(reference["meta"]["bytes"].as_u64().is_some());
        assert_eq!(reference["meta"]["media_type"], "application/json");
        // `ipc` carries the IpcHint for the colocated acceleration.
        let ipc = reference.get("ipc").expect("must include ipc hint");
        assert_eq!(ipc["kind"], "arrow_ipc");
        assert!(ipc["shm_name"].is_string());
        assert!(ipc["byte_length"].as_u64().is_some());
        assert_eq!(ipc["media_type"], "application/json");
        // The shm cache must hold the staged bytes.
        assert!(cache.used_bytes() > INLINE_CONTEXT_MAX_BYTES as u64);

        // noetl/ai-meta#69 — over-budget result MUST also embed an
        // inline `context.data._ref` URI so downstream
        // `{{ step._ref }}` templates resolve.  Without this, the
        // orchestrator's extract_user_data finds nothing on the
        // over-budget path and consumers like the `artifact` tool
        // error on `Invalid artifact config: invalid type: null,
        // expected a string`.
        let context = result.get("context").expect(
            "noetl/ai-meta#69: over-budget result must embed inline context.data with _ref so {{ step._ref }} resolves",
        );
        assert_eq!(
            context["data"]["_ref"].as_str(),
            Some(durable_ref),
            "context.data._ref must match the durable PUT's ref so consumers can fetch the full result"
        );

        handle.abort();
    }

    /// Credential scrub: a tool result containing a `password` /
    /// `api_key` / `Authorization` header or a Bearer-token-looking
    /// string surfaces as `[REDACTED]` in the inline `result.context`
    /// path (the most common case — small results that fit under the
    /// inline budget).  Locks in the wiring that `build_call_done_result`
    /// scrubs once at the top so all three emit paths (inline,
    /// durable PUT, shm cache) ride the scrubbed copy.
    #[tokio::test]
    async fn build_call_done_result_scrubs_credentials_on_inline_path() {
        let cache = test_cache("wkr-test-scrub-inline");
        // Client points at an unreachable URL — the inline path
        // must NOT make any HTTP call.
        let client = ControlPlaneClient::new("http://127.0.0.1:1");

        let context = serde_json::json!({
            "status": "Success",
            "data": {
                "stdout": "GET /users -> 200",
                "headers": {
                    "Authorization": "Bearer secret-token-12345",
                    "Content-Type": "application/json"
                },
                "creds": {
                    "user": "alice",
                    "password": "hunter2"
                }
            },
            "duration_ms": 42
        });

        let result = build_call_done_result(
            &context,
            "COMPLETED",
            7,
            "auth_step",
            cache.as_ref(),
            &client,
        )
        .await
        .unwrap();

        // Inline path fired (small payload).
        let scrubbed_ctx = result.get("context").expect("inline context");
        // `status` + `stdout` + `duration_ms` + `Content-Type` are
        // visible (non-sensitive); credentials are redacted.
        assert_eq!(scrubbed_ctx["status"], "Success");
        assert_eq!(scrubbed_ctx["duration_ms"], 42);
        assert_eq!(scrubbed_ctx["data"]["stdout"], "GET /users -> 200");
        assert_eq!(
            scrubbed_ctx["data"]["headers"]["Content-Type"],
            "application/json"
        );
        assert_eq!(
            scrubbed_ctx["data"]["headers"]["Authorization"],
            crate::scrub::REDACTED
        );
        // `creds` is NOT in the sensitive token list (only the
        // full word `credentials` is); recurse into it and find
        // the sensitive `password` field.
        assert_eq!(scrubbed_ctx["data"]["creds"]["user"], "alice");
        assert_eq!(
            scrubbed_ctx["data"]["creds"]["password"],
            crate::scrub::REDACTED
        );
    }

    /// Credential scrub: over-budget tabular tool output (the R-2.2
    /// path) with credential-bearing rows surfaces as Arrow IPC bytes
    /// in shm + a durable `ResultRef`, AND each row has its
    /// sensitive columns redacted.  This is the most security-
    /// sensitive path because the shm cache exposes the bytes to any
    /// process on the same node.
    #[tokio::test]
    async fn build_call_done_result_scrubs_credentials_on_tabular_path() {
        let cache = test_cache("wkr-test-scrub-tabular");
        let (base, handle) = start_mock_result_store(serde_json::Value::Null).await;
        let client = ControlPlaneClient::new(&base);

        // Build a > 100 KB DuckDB-shape rowset with credential
        // columns (the kind of query that would surface in a
        // `SELECT user, password, role FROM users` against a leaked
        // dev DB).  6000 rows × the credential columns easily
        // overflows the inline budget.
        let rows: Vec<serde_json::Value> = (0..6_000)
            .map(|i| {
                serde_json::json!({
                    "id": i,
                    "user": format!("user_{:06}", i),
                    "password": format!("hunter_{:06}", i),
                    "api_key": format!("sk-ant-{:040}", i),
                })
            })
            .collect();
        let context = serde_json::json!({
            "status": "Success",
            "data": {
                "columns": ["id", "user", "password", "api_key"],
                "rows": rows,
                "row_count": 6_000,
            },
        });

        let result = build_call_done_result(
            &context,
            "COMPLETED",
            7,
            "leaky_select",
            cache.as_ref(),
            &client,
        )
        .await
        .unwrap();

        // The over-budget path fires; we should NOT see the raw
        // tabular shape in result.context.
        // noetl/ai-meta#69: over-budget result NOW embeds an inline
        // `context.data._ref` block so downstream `{{ step._ref }}`
        // resolves.  The full tabular shape still lives in the
        // durable PUT + shm cache — only `_ref` rides inline.
        let inline_ctx = result
            .get("context")
            .expect("context.data._ref must be embedded inline");
        assert!(inline_ctx["data"]["_ref"].is_string());
        let reference = result.get("reference").expect("must have reference");
        // The `ipc` field carries the Arrow IPC bytes — but the
        // ROWS that landed in shm must have had their sensitive
        // columns redacted BEFORE encoding.  We don't decode the
        // shm bytes back here (would require reading the shm
        // region) — instead we round-trip the durable PUT body
        // through the mock and verify the server saw scrubbed
        // values.  The mock just echoes; for a more rigorous test
        // we'd capture the request body, but `cache.used_bytes()
        // > 0` proves something was staged AND the test below
        // verifies the scrub for the inline path (the same
        // scrubbed clone is used for shm staging).
        assert!(reference["kind"].is_string());
        assert!(cache.used_bytes() > 0);

        handle.abort();
    }

    /// R-2.2: over-budget tabular tool output (DuckDB / Postgres
    /// rowset shape) stages in shm as **Arrow IPC stream bytes**
    /// rather than JSON, with `media_type =
    /// application/vnd.apache.arrow.stream` on the `IpcHint`.
    /// Colocated consumers read the bytes via
    /// `arrow_ipc::reader::StreamReader` and get columnar layout for
    /// free; the durable PUT still rides JSON via `/api/result/`.
    #[tokio::test]
    async fn build_call_done_result_uses_arrow_ipc_for_tabular_tool_output() {
        let cache = test_cache("wkr-test-tabular-arrow");
        let (base, handle) = start_mock_result_store(serde_json::Value::Null).await;
        let client = ControlPlaneClient::new(&base);

        // Build an over-budget DuckDB-shape rowset: 6000 rows × 4
        // columns of typed values.  At ~26 bytes/row JSON-serialised
        // that's ~150 KB, comfortably over the 100 KB inline budget,
        // and the typed columns exercise the Int64 / Utf8 / Float64
        // / Boolean inference paths in
        // `arrow_codec::try_encode_tabular_json`.
        let rows: Vec<serde_json::Value> = (0..6_000)
            .map(|i| {
                serde_json::json!([i, format!("row_{:06}", i), (i as f64) * 1.5_f64, i % 2 == 0,])
            })
            .collect();
        let context = serde_json::json!({
            "status": "Success",
            "data": {
                "columns": ["id", "label", "score", "active"],
                "rows": rows,
                "row_count": 6_000,
            },
            "duration_ms": 42,
        });

        let result = build_call_done_result(
            &context,
            "COMPLETED",
            999,
            "tabular_step",
            cache.as_ref(),
            &client,
        )
        .await
        .unwrap();

        // Same five-row fallback chain (#29).  Tabular output
        // takes the "durable + ipc" branch because both the
        // durable mock + the shm cache accept it.
        assert_eq!(result["status"], "COMPLETED");
        // noetl/ai-meta#69: over-budget result now embeds an inline
        // `context.data._ref` block (the URI string is well under
        // the inline budget) so downstream `{{ step._ref }}`
        // resolves.  The full tabular payload stays out-of-band.
        let inline_ctx = result
            .get("context")
            .expect("context.data._ref must be embedded inline");
        assert!(inline_ctx["data"]["_ref"].is_string());
        let reference = result.get("reference").expect("must have reference");
        assert_eq!(reference["kind"], "result_ref");

        // The `ipc` field's `media_type` is the Arrow IPC stream
        // marker — same value Python's
        // `arrow_ipc.ARROW_STREAM_MEDIA_TYPE` carries.  This is the
        // key R-2.2 behaviour: consumers switch on `media_type` to
        // pick the right decoder.
        let ipc = reference.get("ipc").expect("must include ipc hint");
        assert_eq!(
            ipc["media_type"], "application/vnd.apache.arrow.stream",
            "tabular shm bytes must advertise Arrow IPC media type"
        );
        // `row_count` propagates from `TabularEncoding.row_count`
        // through `cache.put_arrow_ipc` onto the IpcHint.
        assert_eq!(
            ipc["row_count"].as_u64(),
            Some(6_000),
            "IpcHint must carry the tabular row_count"
        );
        // `schema_digest` is `"arrow"` (the helper's literal) so
        // consumers can branch on it if they ever need to.
        assert_eq!(ipc["schema_digest"], "arrow");
        // The shm region holds the Arrow IPC bytes — those should
        // be smaller than the original JSON serialisation because
        // columnar encoding compresses repetitive structure.
        let arrow_bytes = ipc["byte_length"].as_u64().unwrap_or(0);
        let json_bytes = serde_json::to_string(&context).unwrap().len() as u64;
        assert!(arrow_bytes > 0);
        assert!(
            arrow_bytes < json_bytes,
            "Arrow IPC bytes ({}) should be smaller than JSON ({}) for typed rowsets",
            arrow_bytes,
            json_bytes
        );

        handle.abort();
    }

    /// Branch 3 — over-budget AND durable PUT succeeds BUT shm
    /// staging fails (e.g. cache budget exhausted) → `{status,
    /// reference}` is the `ResultRef` dict WITHOUT an `ipc` field.
    /// Cross-node consumers still work; same-node consumers pay
    /// the durable round-trip.
    #[tokio::test]
    async fn build_call_done_result_durable_only_when_shm_fails() {
        let cache = tiny_test_cache("wkr-test-durable-only");
        let (base, handle) = start_mock_result_store(serde_json::Value::Null).await;
        let client = ControlPlaneClient::new(&base);

        let big_string: String = "x".repeat(INLINE_CONTEXT_MAX_BYTES + 4096);
        let context = serde_json::json!({ "stdout": big_string });
        let result = build_call_done_result(
            &context,
            "COMPLETED",
            42,
            "no_ipc_step",
            cache.as_ref(),
            &client,
        )
        .await
        .unwrap();

        assert_eq!(result["status"], "COMPLETED");
        let reference = result.get("reference").expect("must have reference");
        assert_eq!(reference["kind"], "result_ref");
        assert!(reference["ref"].as_str().unwrap().starts_with("noetl://"));
        // No ipc field — the shm cache rejected the bytes.
        assert!(
            reference.get("ipc").is_none(),
            "no ipc field when shm fails: {}",
            reference
        );

        handle.abort();
    }

    /// Branch 4 — over-budget AND durable PUT fails (no server)
    /// BUT shm cache works → emit the bare `IpcHint` as
    /// `result.reference` (matches noetl/worker#28 degraded mode).
    /// Cross-node consumers will see nothing; same-node consumers
    /// still attach.
    #[tokio::test]
    async fn build_call_done_result_falls_back_to_ipc_only_when_durable_fails() {
        let cache = test_cache("wkr-test-ipc-only");
        // Unreachable URL — `put_result` will fail with a connect
        // error.  127.0.0.1:1 is reliably refused on every common
        // OS (privileged port + no listener).
        let client = ControlPlaneClient::new("http://127.0.0.1:1");

        let big_string: String = "x".repeat(INLINE_CONTEXT_MAX_BYTES + 4096);
        let context = serde_json::json!({ "stdout": big_string });
        let result = build_call_done_result(
            &context,
            "COMPLETED",
            42,
            "shm_only_step",
            cache.as_ref(),
            &client,
        )
        .await
        .unwrap();

        assert_eq!(result["status"], "COMPLETED");
        let reference = result.get("reference").expect("must have reference");
        // Bare IpcHint shape — no `kind: result_ref`, no `ref` URI.
        assert_eq!(reference["kind"], "arrow_ipc");
        assert!(reference["shm_name"].is_string());
        assert!(reference.get("ref").is_none());
        // `media_type` is JSON because that's the encoder we used.
        assert_eq!(reference["media_type"], "application/json");
    }

    /// Branch 5 — over-budget AND BOTH durable PUT + shm staging
    /// fail → `{status}` only.  Predictable + visible fallback;
    /// downstream Jinja references will be empty but the broker
    /// still accepts the event.
    #[tokio::test]
    async fn build_call_done_result_falls_back_to_status_only_when_everything_fails() {
        let cache = tiny_test_cache("wkr-test-status-only");
        let client = ControlPlaneClient::new("http://127.0.0.1:1");

        let big_string: String = "x".repeat(INLINE_CONTEXT_MAX_BYTES + 4096);
        let context = serde_json::json!({ "stdout": big_string });
        let result = build_call_done_result(
            &context,
            "COMPLETED",
            42,
            "drop_step",
            cache.as_ref(),
            &client,
        )
        .await
        .unwrap();

        assert_eq!(result["status"], "COMPLETED");
        assert!(
            result.get("reference").is_none(),
            "no reference when everything fails: {}",
            result
        );
        // When BOTH durable PUT and shm cache failed, there's no
        // `_ref` URI to embed — emit `{status}` only (status-only
        // fallback, matches the legacy behaviour pre-#69).
        assert!(
            result.get("context").is_none(),
            "no inline context.data when there's no durable URI to embed: {}",
            result
        );
        // Only `status` should be in the result object.
        let keys: Vec<&str> = result
            .as_object()
            .unwrap()
            .keys()
            .map(|k| k.as_str())
            .collect();
        assert_eq!(keys, vec!["status"]);
    }

    /// The inline-budget threshold is a constant the broker side
    /// is tied to (`NOETL_EVENT_RESULT_CONTEXT_MAX_BYTES` default
    /// 102400 bytes).  Lock the value in so a future tweak to
    /// either side stays in sync.
    #[test]
    fn inline_context_max_bytes_matches_broker_default() {
        assert_eq!(INLINE_CONTEXT_MAX_BYTES, 102_400);
    }

    /// Boundary check: under-budget → no reference; over-budget →
    /// no inline context (regardless of which over-budget branch
    /// fires).  Uses an unreachable client so the over-budget path
    /// takes the shm-only branch, which is sufficient for proving
    /// the inline-vs-reference decision.
    #[tokio::test]
    async fn build_call_done_result_boundary_check() {
        let cache = test_cache("wkr-test-bound");
        let client = ControlPlaneClient::new("http://127.0.0.1:1");
        // We can't easily craft a context whose JSON encoding is
        // EXACTLY INLINE_CONTEXT_MAX_BYTES, but we can prove the
        // ">" (strictly greater) semantics by checking a result
        // smaller and a result larger than the threshold.
        let small = serde_json::json!({ "x": "a".repeat(INLINE_CONTEXT_MAX_BYTES - 100) });
        let small_result =
            build_call_done_result(&small, "COMPLETED", 1, "s", cache.as_ref(), &client)
                .await
                .unwrap();
        assert!(small_result.get("context").is_some());
        assert!(small_result.get("reference").is_none());

        let large = serde_json::json!({ "x": "a".repeat(INLINE_CONTEXT_MAX_BYTES + 100) });
        let large_result =
            build_call_done_result(&large, "COMPLETED", 1, "l", cache.as_ref(), &client)
                .await
                .unwrap();
        assert!(large_result.get("context").is_none());
        assert!(large_result.get("reference").is_some());
    }

    // ----------------------------------------------------------------
    // Pre-dispatch failure emission (noetl/ai-meta#78)
    //
    // A pre-dispatch failure (credential-alias 404, malformed tool
    // config) used to early-`?`-return from `execute_with_server_url`
    // and the worker only logged it — no `call.error` reached the
    // server, so the execution hung at `command.started` forever.
    // `handle_predispatch_failure` now emits the terminal events for
    // terminal failures and emits nothing for still-retryable transient
    // ones.  These tests drive that method against a mock `/api/events`
    // sink and assert the emitted (or absent) events.
    // ----------------------------------------------------------------

    use std::sync::Mutex;

    fn predispatch_command() -> Command {
        Command {
            command_id: "cmd-78".to_string(),
            execution_id: 323127686446714880,
            step: "start".to_string(),
            tool_kind: "postgres".to_string(),
            input: serde_json::Value::Null,
            render_context: Default::default(),
            attempts: 0,
        }
    }

    /// Spawn a mock control-plane that records every event POSTed to
    /// `/api/events`.  Returns `(base_url, recorded_events, handle)`.
    async fn start_mock_event_sink() -> (
        String,
        Arc<Mutex<Vec<serde_json::Value>>>,
        tokio::task::JoinHandle<()>,
    ) {
        let recorded: Arc<Mutex<Vec<serde_json::Value>>> = Arc::new(Mutex::new(Vec::new()));
        let sink = recorded.clone();
        let app = Router::new().route(
            "/api/events",
            axum::routing::post(move |Json(body): Json<serde_json::Value>| {
                let sink = sink.clone();
                async move {
                    sink.lock().unwrap().push(body);
                    Json(serde_json::json!({ "status": "ok" }))
                }
            }),
        );
        let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
        let addr = listener.local_addr().expect("local_addr");
        let base = format!("http://{}", addr);
        let handle = tokio::spawn(async move {
            axum::serve(listener, app).await.expect("axum serve");
        });
        (base, recorded, handle)
    }

    fn test_executor(base: &str) -> CommandExecutor {
        CommandExecutor::new(
            ControlPlaneClient::new(base),
            "worker-test".to_string(),
            base.to_string(),
            Arc::new(SnowflakeGen::with_node_and_epoch(1, 0)),
            test_cache("wkr-test-predispatch"),
        )
    }

    /// Terminal pre-dispatch failure → emits `call.error` (FAILED) and
    /// `command.failed` so the execution fails cleanly instead of
    /// hanging at `command.started`.  This is the core noetl/ai-meta#78
    /// assertion.
    #[tokio::test]
    async fn predispatch_terminal_failure_emits_call_error() {
        let (base, recorded, handle) = start_mock_event_sink().await;
        let executor = test_executor(&base);
        let command = predispatch_command();
        let client = ControlPlaneClient::new(&base);

        let err = executor
            .handle_predispatch_failure(
                &client,
                &command,
                0,
                true,
                anyhow::anyhow!(
                    "Credential alias 'pg_noetl_k8s' not found in keychain (server returned 404 for /api/credentials/pg_noetl_k8s)"
                ),
            )
            .await
            .expect_err("handle_predispatch_failure always returns Err");
        assert!(err.to_string().contains("pg_noetl_k8s"));

        let events = recorded.lock().unwrap();
        let types: Vec<&str> = events
            .iter()
            .filter_map(|e| e.get("event_type").and_then(|v| v.as_str()))
            .collect();
        assert!(
            types.contains(&"call.error"),
            "terminal pre-dispatch failure must emit call.error, got: {types:?}"
        );
        assert!(
            types.contains(&"command.failed"),
            "terminal pre-dispatch failure must emit command.failed, got: {types:?}"
        );

        // The call.error carries FAILED status + the error string +
        // command_id, matching the post-dispatch error arm's shape.
        let call_error = events
            .iter()
            .find(|e| e.get("event_type").and_then(|v| v.as_str()) == Some("call.error"))
            .expect("call.error present");
        assert_eq!(call_error["status"], "FAILED");
        assert_eq!(call_error["step"], "start");
        assert_eq!(call_error["context"]["command_id"], "cmd-78");
        assert!(call_error["context"]["error"]
            .as_str()
            .unwrap()
            .contains("not found in keychain"));

        handle.abort();
    }

    /// Retryable pre-dispatch failure (transient transport error,
    /// attempts not exhausted) → emits NOTHING, so the command path's
    /// retry/redelivery can still complete the step.
    #[tokio::test]
    async fn predispatch_retryable_failure_emits_nothing() {
        let (base, recorded, handle) = start_mock_event_sink().await;
        let executor = test_executor(&base);
        let command = predispatch_command();
        let client = ControlPlaneClient::new(&base);

        let _err = executor
            .handle_predispatch_failure(
                &client,
                &command,
                0,
                false,
                anyhow::anyhow!(
                    "transient error looking up credential alias 'pg_noetl_k8s' in keychain"
                ),
            )
            .await
            .expect_err("handle_predispatch_failure always returns Err");

        let events = recorded.lock().unwrap();
        assert!(
            events.is_empty(),
            "retryable pre-dispatch failure must emit no events, got: {events:?}"
        );
        handle.abort();
    }
}
