noetl-server 2.32.0

//! Event handling API handlers.
//!
//! Handles worker events and command retrieval endpoints.
//!
//! SECURITY: All event payloads are sanitized before storage to prevent
//! sensitive data (bearer tokens, passwords, API keys) from being persisted.

use axum::{
    extract::{Path, State},
    Json,
};
use serde::{Deserialize, Serialize};
use sqlx::Row;
use tracing::{debug, info, warn};

use crate::error::{AppError, AppResult};
use crate::sanitize::sanitize_sensitive_data;
use crate::state::AppState;

/// Deserialize a snowflake-id field that may arrive on the wire as
/// either a JSON string (the historical browser-facing shape) or a
/// JSON integer (the shape `noetl-events::ExecutorEvent` emits over
/// `.json(&event)`).  Both decode to `String` so the rest of the
/// handler is unchanged.
///
/// Why the lax decoder: the worker's canonical envelope types
/// `execution_id` / `event_id` as `i64`, the Rust server's request
/// shape kept them as `String` for browser JSON-number precision,
/// and the Python server (Pydantic v2 lax mode) coerced int→str
/// silently for over a year — so the drift only manifested once
/// Rust-on-both-ends went through the same path.  See
/// `noetl/ai-meta#55` for the surfacing in Phase F R5.
fn deserialize_string_or_i64<'de, D>(deserializer: D) -> std::result::Result<String, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::de::{self, Visitor};
    use std::fmt;

    struct StringOrI64;

    impl<'de> Visitor<'de> for StringOrI64 {
        type Value = String;

        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
            formatter.write_str("a string or signed/unsigned integer representing a snowflake id")
        }

        fn visit_str<E>(self, v: &str) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(v.to_string())
        }

        fn visit_string<E>(self, v: String) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(v)
        }

        fn visit_i64<E>(self, v: i64) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(v.to_string())
        }

        fn visit_u64<E>(self, v: u64) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(v.to_string())
        }
    }

    deserializer.deserialize_any(StringOrI64)
}

/// `Option<String>` variant of [`deserialize_string_or_i64`] for
/// optional id fields like `EventRequest.event_id`.  Accepts
/// missing / null / string / integer.
fn deserialize_optional_string_or_i64<'de, D>(
    deserializer: D,
) -> std::result::Result<Option<String>, D::Error>
where
    D: serde::Deserializer<'de>,
{
    use serde::de::{self, Visitor};
    use std::fmt;

    struct OptStringOrI64;

    impl<'de> Visitor<'de> for OptStringOrI64 {
        type Value = Option<String>;

        fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
            formatter.write_str(
                "null, a string, or a signed/unsigned integer representing an optional snowflake id",
            )
        }

        fn visit_none<E>(self) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(None)
        }

        fn visit_unit<E>(self) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(None)
        }

        fn visit_some<D2>(self, deserializer: D2) -> std::result::Result<Self::Value, D2::Error>
        where
            D2: serde::Deserializer<'de>,
        {
            deserialize_string_or_i64(deserializer).map(Some)
        }

        fn visit_str<E>(self, v: &str) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(Some(v.to_string()))
        }

        fn visit_string<E>(self, v: String) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(Some(v))
        }

        fn visit_i64<E>(self, v: i64) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(Some(v.to_string()))
        }

        fn visit_u64<E>(self, v: u64) -> std::result::Result<Self::Value, E>
        where
            E: de::Error,
        {
            Ok(Some(v.to_string()))
        }
    }

    deserializer.deserialize_any(OptStringOrI64)
}

/// Worker event request.
///
/// The shared subset of fields with the canonical
/// [`noetl_events::ExecutorEvent`] envelope (from the
/// `noetl-events` crate published off
/// [noetl/cli](https://github.com/noetl/cli)) — `execution_id`,
/// `step`, `event_type` (with `name` alias), `payload`/`context`,
/// `meta`, `worker_id`, `event_id`, `status`, `created_at` — is
/// wire-format compatible.  The wire-compat test
/// `wire_compat_round_trips_shared_subset_with_executor_event`
/// guards this property.  EE-4 (noetl/ai-meta#49) extracted the
/// shared envelope into the dedicated `noetl-events` crate and
/// added a direct dep on it here so the wire shape has a single
/// source of truth instead of being held in sync by hand-aligned
/// doc comments.
///
/// `EventRequest` keeps several server-only fields beyond the
/// canonical envelope: `result_kind`, `result_uri`, `event_ids`
/// (drive the constraint-compliant `{status, reference}` /
/// `{status, context}` result shape per noetl/server#29);
/// `actionable`, `informative` (control orchestrator dispatch +
/// log-only persistence).  Wire-encodes `execution_id` /
/// `event_id` as `String` for JSON-number precision in browser
/// clients, vs the envelope's `i64` — the `From` / `TryFrom`
/// impls below handle the conversion at the boundary.
///
/// Pre-EE-2 (`name`-field) worker / CLI clients keep working via
/// `#[serde(alias = "name")]`; producers that omit
/// `event_id` / `status` / `created_at` get sensible server-side
/// fallbacks (DB-side `snowflake_id()` for `event_id`, the
/// name-derived `status` returned by `event_status_from_name`,
/// `Utc::now()` for `created_at`).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventRequest {
    /// Execution ID.  Wire format is `String` (matches the Python
    /// `EventEmitRequest` and avoids JSON-number precision loss
    /// for large snowflakes in browser clients); parsed to `i64`
    /// before the DB write.
    ///
    /// The `deserialize_with` adapter accepts the worker's canonical
    /// `noetl-events::ExecutorEvent.execution_id: i64` wire shape
    /// as well, so Rust-on-both-ends doesn't fail at the boundary.
    /// See `noetl/ai-meta#55` for the drift this fixes.
    #[serde(deserialize_with = "deserialize_string_or_i64")]
    pub execution_id: String,
    /// Step name.
    pub step: String,
    /// Event type (e.g. `step.enter`, `call.done`, `step.exit`,
    /// `command.completed`).  R-1.2 PR-EE-2: renamed from `name`;
    /// the alias keeps pre-PR-EE clients working.
    #[serde(alias = "name")]
    pub event_type: String,
    /// Event payload/result data.
    ///
    /// R-1.2 PR-EE-2: `context` alias accepted so producers that
    /// send the executor's `ExecutorEvent.context` field
    /// deserialize cleanly into this `payload`.
    #[serde(default, alias = "context")]
    pub payload: serde_json::Value,
    /// Additional metadata.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub meta: Option<serde_json::Value>,
    /// Worker ID.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worker_id: Option<String>,
    /// Result kind: "data", "ref", or "refs".
    #[serde(default = "default_result_kind")]
    pub result_kind: String,
    /// Result URI for ref kind.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub result_uri: Option<String>,
    /// Event IDs for refs kind.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_ids: Option<Vec<i64>>,
    /// If true, server should take action.
    #[serde(default = "default_true")]
    pub actionable: bool,
    /// If true, event is for logging/observability.
    #[serde(default = "default_true")]
    pub informative: bool,
    /// Application-side snowflake ID for this event.  Per
    /// `agents/rules/observability.md` Principle 3, the emitting
    /// process generates this BEFORE the row hits the database so
    /// spans / metrics / cross-component correlation can use it
    /// immediately.  Wire format is `String` to avoid JSON-number
    /// precision loss; parsed to `i64` for the DB write.
    /// `None` falls back to the server-side `noetl.snowflake_id()`
    /// function (the existing default).
    ///
    /// Accepts both the `String` wire shape (browser clients) and
    /// the `i64` wire shape (worker's canonical envelope).  See
    /// `noetl/ai-meta#55`.
    #[serde(
        default,
        skip_serializing_if = "Option::is_none",
        deserialize_with = "deserialize_optional_string_or_i64"
    )]
    pub event_id: Option<String>,
    /// Lifecycle status (`STARTED` / `RUNNING` / `COMPLETED` /
    /// `FAILED`).  `None` falls back to name-based derivation in
    /// `event_status_from_name`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub status: Option<String>,
    /// Wall-clock when the event was produced.  `None` falls back
    /// to `chrono::Utc::now()`.  Stamping at emit time preserves
    /// per-component ordering across server-clock skew (matters
    /// when multiple workers emit in tight bursts).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub created_at: Option<chrono::DateTime<chrono::Utc>>,
}

fn default_result_kind() -> String {
    "data".to_string()
}

fn default_true() -> bool {
    true
}

/// Project the canonical `noetl_events::ExecutorEvent` envelope (the
/// shape every NoETL Rust producer emits through `EventSink`) into
/// the server's wire request shape.
///
/// Server-only fields (`result_kind`, `result_uri`, `event_ids`,
/// `actionable`, `informative`) get the same defaults the handler
/// applies when a producer omits them.  `execution_id` + `event_id`
/// flip to `String` because that's the wire format the server has
/// always exposed to browser clients (JSON-number precision).
impl From<noetl_events::ExecutorEvent> for EventRequest {
    fn from(ev: noetl_events::ExecutorEvent) -> Self {
        Self {
            execution_id: ev.execution_id.to_string(),
            step: ev.step,
            event_type: ev.event_type,
            payload: ev.context,
            meta: ev.meta,
            worker_id: ev.worker_id,
            result_kind: default_result_kind(),
            result_uri: None,
            event_ids: None,
            actionable: true,
            informative: true,
            event_id: ev.event_id.map(|id| id.to_string()),
            status: Some(ev.status),
            created_at: Some(ev.created_at),
        }
    }
}

/// Inverse of [`From<noetl_events::ExecutorEvent>`].  `TryFrom`
/// rather than `From` because the wire-shape `String` execution_id
/// and `String` event_id can fail to parse — the server returns 400
/// in that case in the actual handler.  Server-only fields
/// (`result_kind`, `result_uri`, `event_ids`, `actionable`,
/// `informative`) drop on the floor here — the canonical envelope
/// doesn't model them.  When `status` / `created_at` are absent on
/// the request, the conversion fills them with the same fallbacks
/// the handler uses (`event_status_from_name`, `Utc::now()`).
impl TryFrom<&EventRequest> for noetl_events::ExecutorEvent {
    type Error = anyhow::Error;

    fn try_from(req: &EventRequest) -> std::result::Result<Self, Self::Error> {
        let execution_id: i64 = req.execution_id.parse().map_err(|e| {
            anyhow::anyhow!(
                "execution_id {:?} not parseable as i64: {e}",
                req.execution_id
            )
        })?;
        let event_id = req
            .event_id
            .as_deref()
            .map(|s| s.parse::<i64>())
            .transpose()
            .map_err(|e| {
                anyhow::anyhow!("event_id {:?} not parseable as i64: {e}", req.event_id)
            })?;
        let status = req
            .status
            .clone()
            .unwrap_or_else(|| event_status_from_name(&req.event_type).to_string());
        let created_at = req.created_at.unwrap_or_else(chrono::Utc::now);
        Ok(Self {
            execution_id,
            event_type: req.event_type.clone(),
            step: req.step.clone(),
            status,
            created_at,
            context: req.payload.clone(),
            event_id,
            worker_id: req.worker_id.clone(),
            meta: req.meta.clone(),
        })
    }
}

/// Response for event handling.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventResponse {
    /// Status of the operation.
    pub status: String,
    /// Event ID that was created.
    pub event_id: i64,
    /// Number of commands generated.
    pub commands_generated: i32,
}

/// Request to claim a command atomically.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClaimRequest {
    /// Worker ID requesting the claim.
    pub worker_id: String,
}

/// Response for successful claim.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ClaimResponse {
    /// Status of the claim operation.
    pub status: String,
    /// Command event ID.
    pub event_id: i64,
    /// Execution ID.
    pub execution_id: i64,
    /// Node/step ID.
    pub node_id: String,
    /// Node/step name.
    pub node_name: String,
    /// Action/tool kind.
    pub action: String,
    /// Command context.
    pub context: serde_json::Value,
    /// Command metadata.
    pub meta: serde_json::Value,
}

/// A single batched worker event.
///
/// R-1.2 PR-EE-2: same `name` → `event_type` rename + serde alias
/// as `EventRequest`; same `context` alias for `payload`.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchEventItem {
    /// Step name.
    pub step: String,
    /// Event type.
    #[serde(alias = "name")]
    pub event_type: String,
    /// Event payload/result data.
    #[serde(default, alias = "context")]
    pub payload: serde_json::Value,
    /// If true, server should take action.
    #[serde(default)]
    pub actionable: bool,
    /// If true, event is for logging/observability.
    #[serde(default = "default_true")]
    pub informative: bool,
}

/// Request for batched event ingestion.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchEventRequest {
    /// Execution ID.  Accepts both string (browser clients) and
    /// integer (worker's `noetl-events::ExecutorEvent`) wire
    /// shapes.  See `noetl/ai-meta#55`.
    #[serde(deserialize_with = "deserialize_string_or_i64")]
    pub execution_id: String,
    /// Worker ID.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub worker_id: Option<String>,
    /// Events to persist.
    pub events: Vec<BatchEventItem>,
}

/// Response for batched event ingestion.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct BatchEventResponse {
    /// Status of the operation.
    pub status: String,
    /// Inserted event IDs.
    pub event_ids: Vec<i64>,
    /// Number of generated commands.
    pub commands_generated: i32,
}

/// Command details response.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CommandResponse {
    /// Execution ID.
    pub execution_id: i64,
    /// Node/step ID.
    pub node_id: String,
    /// Node/step name.
    pub node_name: String,
    /// Action/tool kind.
    pub action: String,
    /// Command context (tool config, args, etc.).
    pub context: serde_json::Value,
    /// Command metadata.
    pub meta: serde_json::Value,
}

/// Handle worker event.
///
/// POST /api/events
///
/// Worker reports completion with result (inline or ref).
/// Engine evaluates case/when/then and generates next commands.
///
/// Instrumented per
/// [`agents/rules/observability.md`](https://github.com/noetl/ai-meta/blob/main/agents/rules/observability.md)
/// Principle 1: a counter (`noetl_events_ingested_total{event_type,status}`)
/// and histogram (`noetl_event_ingest_duration_seconds{event_type}`) are
/// recorded on every dispatch.  See [`handle_event_inner`] for the body
/// of the handler.
pub async fn handle_event(
    state: State<AppState>,
    request: Json<EventRequest>,
) -> Result<Json<EventResponse>, AppError> {
    let event_type_for_metrics = request.0.event_type.clone();
    let started_at = std::time::Instant::now();

    let result = handle_event_inner(state, request).await;

    let status_label = if result.is_ok() { "ok" } else { "error" };
    let duration_seconds = started_at.elapsed().as_secs_f64();
    crate::metrics::record_event_ingest(
        &event_type_for_metrics,
        status_label,
        duration_seconds,
    );

    result
}

/// Inner body of [`handle_event`] — same logic, no instrumentation.
///
/// Split out so the wrapper can record metrics on both Ok and Err
/// paths without coupling the body to the recording call.
async fn handle_event_inner(
    State(state): State<AppState>,
    Json(request): Json<EventRequest>,
) -> Result<Json<EventResponse>, AppError> {
    debug!(
        "Event received: execution_id={}, step={}, event_type={}",
        request.execution_id, request.step, request.event_type
    );

    let execution_id: i64 = request
        .execution_id
        .parse()
        .map_err(|_| AppError::Validation("Invalid execution_id".to_string()))?;

    // Events that should NOT trigger engine processing
    let skip_engine_events = [
        "command.claimed",
        "command.started",
        "command.completed",
        "command.failed",
        "step.enter",
    ];

    // For command.claimed, check if already claimed
    if request.event_type == "command.claimed" {
        if let Some(command_id) = get_command_id(&request) {
            if check_already_claimed(&state, execution_id, &command_id, &request.worker_id).await? {
                // Already claimed by same worker - idempotent success
                return Ok(Json(EventResponse {
                    status: "ok".to_string(),
                    event_id: 0,
                    commands_generated: 0,
                }));
            }
        }
    }

    // Resolve status BEFORE building the result so the result
    // envelope can include it (the DB constraint
    // chk_event_result_shape requires every result row to carry
    // a string `status` key at the top level).
    //
    // R-1.2 PR-EE-2: prefer the application-supplied status when
    // present (so the worker can distinguish STARTED vs RUNNING
    // explicitly); fall back to name-based derivation when
    // omitted for pre-PR-EE clients.
    let derived_status: String = request
        .status
        .clone()
        .unwrap_or_else(|| event_status_from_name(&request.event_type).to_string());

    // Build result object based on kind, embedding the resolved
    // status.  See noetl/server#29 — previous shapes
    // (`{kind, data}`, etc.) violated `chk_event_result_shape`.
    let result_obj_raw = build_result_object(&request, &derived_status);
    // SECURITY: Sanitize result data to remove sensitive information (tokens, passwords, etc.)
    let result_obj = sanitize_sensitive_data(&result_obj_raw);

    // Resolve event_id.  Producers may stamp it client-side per
    // `agents/rules/observability.md` Principle 3 (worker, CLI in
    // distributed mode); when omitted, the server now stamps via
    // its own application-side `SnowflakeGenerator` instead of
    // the DB-side `noetl.snowflake_id()` function.  Phase F R1.5
    // (noetl/ai-meta#49) moved the fallback path here so the id
    // is available before the INSERT span opens and so per-shard
    // generation can be controlled via `NOETL_SERVER_MACHINE_ID`.
    let event_id: i64 = match request.event_id.as_deref() {
        Some(raw) => raw.parse().map_err(|_| {
            AppError::Validation(format!("Invalid event_id: {raw}"))
        })?,
        None => state.snowflake.generate()?,
    };

    // Get catalog_id from existing events
    let catalog_id = get_catalog_id(&state, execution_id).await?;

    // Build meta object with control flags
    let mut meta_obj = request.meta.clone().unwrap_or(serde_json::json!({}));
    if let serde_json::Value::Object(ref mut map) = meta_obj {
        map.insert(
            "actionable".to_string(),
            serde_json::json!(request.actionable),
        );
        map.insert(
            "informative".to_string(),
            serde_json::json!(request.informative),
        );
        if let Some(ref worker_id) = request.worker_id {
            map.insert("worker_id".to_string(), serde_json::json!(worker_id));
        }
    }
    // SECURITY: Sanitize meta data to remove sensitive information
    let meta_obj = sanitize_sensitive_data(&meta_obj);

    // `derived_status` is computed earlier (before the result
    // envelope build) so the constraint-required top-level
    // `status` key can be embedded; reused here as the `status`
    // column value.

    // Resolve created_at — prefer the application-supplied stamp
    // (avoids server-clock skew when ordering bursts).
    let created_at = request.created_at.unwrap_or_else(chrono::Utc::now);

    // Persist the event
    sqlx::query(
        r#"
        INSERT INTO noetl.event (
            event_id, execution_id, catalog_id, event_type,
            node_id, node_name, status, result, meta, created_at
        ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
        "#,
    )
    .bind(event_id)
    .bind(execution_id)
    .bind(catalog_id)
    .bind(&request.event_type)
    .bind(&request.step)
    .bind(&request.step)
    .bind(&derived_status)
    .bind(&result_obj)
    .bind(&meta_obj)
    .bind(created_at)
    .execute(state.pools.pool_for(execution_id))
    .await?;

    info!(
        "Event persisted: event_id={}, execution_id={}, event_type={}",
        event_id, execution_id, request.event_type
    );

    // Process through engine if applicable
    let commands_generated = if !skip_engine_events.contains(&request.event_type.as_str()) {
        // TODO: Implement engine event handling
        // This would call the orchestrator to evaluate next steps
        debug!("Would process through engine: event_type={}", request.event_type);
        0
    } else {
        debug!("Skipped engine for administrative event: {}", request.event_type);
        0
    };

    // Trigger orchestrator for workflow progression.
    //
    // `command.completed` advances the workflow to the next step;
    // `command.failed` checks whether the failure should terminate
    // the playbook (noetl/ai-meta#58 — without this trigger, failed
    // steps stalled the execution forever because the orchestrator
    // never got a chance to emit `playbook.failed`).
    //
    // The earlier `step != "end"` guard treated `end` as a sentinel
    // whose completion fired playbook.completed implicitly.  After
    // the noetl/ai-meta#54 orchestrator change (end is a real step
    // with its own `tool:` block), end's command.completed MUST
    // trigger the orchestrator — that's the pass where
    // `check_completion` sees end as done and emits
    // `playbook.completed`.  Without this trigger the playbook
    // stalled at `command.completed [end]` with no terminal event.
    let should_trigger_orchestrator =
        request.event_type == "command.completed" || request.event_type == "command.failed";
    if should_trigger_orchestrator {
        match trigger_orchestrator(&state, execution_id, event_id).await {
            Ok(cmds) => {
                info!(
                    "Orchestrator generated {} commands for execution {}",
                    cmds, execution_id
                );
            }
            Err(e) => {
                warn!("Orchestrator error: {}", e);
            }
        }
    }

    Ok(Json(EventResponse {
        status: "ok".to_string(),
        event_id,
        commands_generated,
    }))
}

/// Get command details from command.issued event.
///
/// GET /api/commands/{event_id}
///
/// Workers call this to fetch command config after NATS notification.
pub async fn get_command(
    State(state): State<AppState>,
    Path(event_id): Path<i64>,
) -> Result<Json<CommandResponse>, AppError> {
    debug!("Getting command for event_id={}", event_id);

    // Phase F R4-4: `GET /api/commands/{event_id}` is keyed by
    // event_id alone — execution_id isn't known until after the
    // lookup.  Use the cross-shard resolver: probe every shard,
    // first hit wins.  In single-pool fallback mode this is a
    // single probe against the one pool.
    let found = state
        .pools
        .find_first(|_shard_idx, pool| async move {
            sqlx::query_as::<_, (i64, String, String, serde_json::Value, serde_json::Value)>(
                r#"
                SELECT execution_id, node_name, node_type, context, meta
                FROM noetl.event
                WHERE event_id = $1 AND event_type = 'command.issued'
                "#,
            )
            .bind(event_id)
            .fetch_optional(&pool)
            .await
        })
        .await?;
    let row = found.map(|(_shard_idx, r)| r);

    match row {
        Some((execution_id, node_name, node_type, context, meta)) => Ok(Json(CommandResponse {
            execution_id,
            node_id: node_name.clone(),
            node_name,
            action: node_type,
            context,
            meta,
        })),
        None => Err(AppError::NotFound(format!(
            "command.issued event not found: {}",
            event_id
        ))),
    }
}

/// Atomically claim command and return command details.
///
/// POST /api/commands/{event_id}/claim
pub async fn claim_command(
    State(state): State<AppState>,
    Path(event_id): Path<i64>,
    Json(request): Json<ClaimRequest>,
) -> Result<Json<ClaimResponse>, AppError> {
    debug!(
        "Claim request received: event_id={}, worker_id={}",
        event_id, request.worker_id
    );

    // Phase F R4-4: resolve event_id -> execution_id via the
    // cross-shard probe, then open the tx on the per-execution
    // pool.  Two round trips in sharded mode (one probe + one tx
    // open) is the right trade-off vs. holding the tx open
    // across a fan-out scan — keeping shard-locality on the
    // claim transaction means the second SELECT (terminal-row
    // check) and any subsequent INSERTs all hit the same shard
    // and stay within the same tx scope.
    //
    // In single-pool fallback mode the resolver short-circuits
    // (one pool, one probe).
    let resolved_execution_id: Option<i64> = state
        .pools
        .find_first(|_shard_idx, pool| async move {
            sqlx::query_scalar::<_, i64>(
                r#"
                SELECT execution_id
                FROM noetl.event
                WHERE event_id = $1 AND event_type = 'command.issued'
                "#,
            )
            .bind(event_id)
            .fetch_optional(&pool)
            .await
        })
        .await?
        .map(|(_shard_idx, eid)| eid);

    let resolved_execution_id = resolved_execution_id.ok_or_else(|| {
        AppError::NotFound(format!("command.issued event not found: {}", event_id))
    })?;

    let mut tx = state
        .pools
        .pool_for(resolved_execution_id)
        .begin()
        .await?;

    let cmd_row = sqlx::query(
        r#"
        SELECT execution_id, catalog_id, node_name, node_type, context, meta
        FROM noetl.event
        WHERE event_id = $1 AND event_type = 'command.issued'
        "#,
    )
    .bind(event_id)
    .fetch_optional(&mut *tx)
    .await?;

    let Some(row) = cmd_row else {
        return Err(AppError::NotFound(format!(
            "command.issued event not found: {}",
            event_id
        )));
    };

    let execution_id: i64 = row.try_get("execution_id")?;
    let catalog_id: Option<i64> = row.try_get("catalog_id")?;
    let step: String = row.try_get("node_name")?;
    let tool_kind: String = row.try_get("node_type")?;
    let context: serde_json::Value = row
        .try_get("context")
        .unwrap_or_else(|_| serde_json::json!({}));
    let meta: serde_json::Value = row
        .try_get("meta")
        .unwrap_or_else(|_| serde_json::json!({}));
    let command_id = meta
        .get("command_id")
        .and_then(|v| v.as_str())
        .map(|s| s.to_string())
        .unwrap_or_else(|| format!("{}:{}:{}", execution_id, step, event_id));

    // If command already reached terminal state, skip re-claim.
    let terminal_row = sqlx::query(
        r#"
        SELECT event_type
        FROM noetl.event
        WHERE execution_id = $1
          AND event_type IN ('command.completed', 'command.failed')
          AND (meta->>'command_id' = $2 OR result->'data'->>'command_id' = $2)
        ORDER BY event_id DESC
        LIMIT 1
        "#,
    )
    .bind(execution_id)
    .bind(&command_id)
    .fetch_optional(&mut *tx)
    .await?;

    if terminal_row.is_some() {
        return Err(AppError::Conflict(
            "Command already reached terminal state".to_string(),
        ));
    }

    // If execution already cancelled, reject claim.
    let cancelled_row = sqlx::query(
        r#"
        SELECT 1
        FROM noetl.event
        WHERE execution_id = $1
          AND event_type = 'execution.cancelled'
        LIMIT 1
        "#,
    )
    .bind(execution_id)
    .fetch_optional(&mut *tx)
    .await?;

    if cancelled_row.is_some() {
        return Err(AppError::Conflict(
            "Execution has been cancelled".to_string(),
        ));
    }

    // Acquire advisory transaction lock by command id.
    let lock_row =
        sqlx::query("SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint) AS lock_acquired")
            .bind(&command_id)
            .fetch_one(&mut *tx)
            .await?;
    let lock_acquired: bool = lock_row.try_get("lock_acquired")?;
    if !lock_acquired {
        return Err(AppError::Conflict(
            "Command is being claimed by another worker".to_string(),
        ));
    }

    // Check if already claimed by another worker.
    let existing_claim = sqlx::query(
        r#"
        SELECT worker_id, meta
        FROM noetl.event
        WHERE execution_id = $1
          AND event_type = 'command.claimed'
          AND (meta->>'command_id' = $2 OR result->'data'->>'command_id' = $2)
        ORDER BY event_id DESC
        LIMIT 1
        "#,
    )
    .bind(execution_id)
    .bind(&command_id)
    .fetch_optional(&mut *tx)
    .await?;

    if let Some(existing) = existing_claim {
        let worker_id_db: Option<String> = existing.try_get("worker_id").ok();
        let worker_id_meta = existing
            .try_get::<serde_json::Value, _>("meta")
            .ok()
            .and_then(|value| {
                value
                    .get("worker_id")
                    .and_then(|v| v.as_str())
                    .map(|s| s.to_string())
            });
        let existing_worker = worker_id_db.or(worker_id_meta);

        if let Some(existing_worker_id) = existing_worker {
            if existing_worker_id != request.worker_id {
                return Err(AppError::Conflict(format!(
                    "Command already claimed by {}",
                    existing_worker_id
                )));
            }
            // Idempotent claim by same worker.
            tx.commit().await?;
            return Ok(Json(ClaimResponse {
                status: "ok".to_string(),
                event_id,
                execution_id,
                node_id: step.clone(),
                node_name: step,
                action: tool_kind,
                context,
                meta,
            }));
        }
    }

    let claim_event_id = state.snowflake.generate()?;
    // Constraint-compliant `{status, context}` envelope — see
    // noetl/server#29 for why `{kind, data}` was rejected.  The
    // explicit claim path was missed in v2.4.3 because the load
    // smoke only exercised handle_event; Phase D Round 2's
    // multi-step kind validation surfaced it.
    let claim_result = serde_json::json!({
        "status": "RUNNING",
        "context": {
            "command_id": command_id,
            "worker_id": request.worker_id,
        }
    });
    let claim_meta = serde_json::json!({
        "command_id": command_id,
        "worker_id": request.worker_id,
        "actionable": false,
        "informative": true,
    });

    sqlx::query(
        r#"
        INSERT INTO noetl.event (
            event_id, execution_id, catalog_id, event_type,
            node_id, node_name, status, result, meta, worker_id, created_at
        ) VALUES (
            $1, $2, $3, $4,
            $5, $6, $7, $8, $9, $10, $11
        )
        "#,
    )
    .bind(claim_event_id)
    .bind(execution_id)
    .bind(catalog_id)
    .bind("command.claimed")
    .bind(&step)
    .bind(&step)
    .bind("RUNNING")
    .bind(claim_result)
    .bind(claim_meta)
    .bind(&request.worker_id)
    .bind(chrono::Utc::now())
    .execute(&mut *tx)
    .await?;

    tx.commit().await?;

    Ok(Json(ClaimResponse {
        status: "ok".to_string(),
        event_id,
        execution_id,
        node_id: step.clone(),
        node_name: step,
        action: tool_kind,
        context,
        meta,
    }))
}

/// Handle batched worker events.
///
/// POST /api/events/batch
pub async fn handle_batch_events(
    State(state): State<AppState>,
    Json(request): Json<BatchEventRequest>,
) -> Result<Json<BatchEventResponse>, AppError> {
    if request.events.is_empty() {
        return Ok(Json(BatchEventResponse {
            status: "ok".to_string(),
            event_ids: Vec::new(),
            commands_generated: 0,
        }));
    }

    let execution_id: i64 = request
        .execution_id
        .parse()
        .map_err(|_| AppError::Validation("Invalid execution_id".to_string()))?;

    let catalog_id = get_catalog_id(&state, execution_id).await?;
    // Phase F R4-3: batch writes land on this execution's shard.
    let mut tx = state.pools.pool_for(execution_id).begin().await?;
    let mut event_ids = Vec::with_capacity(request.events.len());

    for item in &request.events {
        // Batch path uses the application-side snowflake
        // generator (Phase F R1.5 of noetl/ai-meta#49).  Per-item
        // app-side event_id isn't carried in BatchEventItem yet;
        // left as a follow-up if batch becomes the worker's
        // primary path.
        let event_id = state.snowflake.generate()?;
        let status = event_status_from_name(&item.event_type);

        // Constraint-compliant `{status, context}` envelope per
        // noetl/server#29.  Same shape rule as build_result_object
        // in handle_event_inner: `context` only when payload is
        // an object; otherwise emit `{status}` alone.
        let mut result_map = serde_json::Map::new();
        result_map.insert(
            "status".to_string(),
            serde_json::Value::String(status.to_string()),
        );
        if let serde_json::Value::Object(_) = item.payload {
            result_map.insert("context".to_string(), item.payload.clone());
        }
        let result_obj_raw = serde_json::Value::Object(result_map);
        let result_obj = sanitize_sensitive_data(&result_obj_raw);

        let mut meta_obj = serde_json::json!({
            "actionable": item.actionable,
            "informative": item.informative,
        });
        if let Some(worker_id) = &request.worker_id {
            if let serde_json::Value::Object(ref mut map) = meta_obj {
                map.insert("worker_id".to_string(), serde_json::json!(worker_id));
            }
        }
        let meta_obj = sanitize_sensitive_data(&meta_obj);

        sqlx::query(
            r#"
            INSERT INTO noetl.event (
                event_id, execution_id, catalog_id, event_type,
                node_id, node_name, status, result, meta, worker_id, created_at
            ) VALUES (
                $1, $2, $3, $4,
                $5, $6, $7, $8, $9, $10, $11
            )
            "#,
        )
        .bind(event_id)
        .bind(execution_id)
        .bind(catalog_id)
        .bind(&item.event_type)
        .bind(&item.step)
        .bind(&item.step)
        .bind(status)
        .bind(result_obj)
        .bind(meta_obj)
        .bind(&request.worker_id)
        .bind(chrono::Utc::now())
        .execute(&mut *tx)
        .await?;

        event_ids.push(event_id);
    }

    tx.commit().await?;

    // Trigger orchestrator for any command.completed in the batch,
    // including end (end is now a real dispatched step per
    // noetl/ai-meta#54 — its command.completed is the trigger that
    // makes check_completion emit playbook.completed).  Mirrors the
    // call site in `handle_event` above; runs once per qualifying
    // event so a batch with multiple completions can still advance
    // multi-step playbooks.  Errors are logged and swallowed so a
    // bad-state evaluation doesn't fail the whole batch ingest.
    for (idx, item) in request.events.iter().enumerate() {
        if item.event_type == "command.completed" {
            let trigger_event_id = event_ids[idx];
            match trigger_orchestrator(&state, execution_id, trigger_event_id).await {
                Ok(cmds) => {
                    info!(
                        "Orchestrator (batch) generated {} commands for execution {} step {}",
                        cmds, execution_id, item.step
                    );
                }
                Err(e) => {
                    warn!(
                        "Orchestrator error in batch for execution {} step {}: {}",
                        execution_id, item.step, e
                    );
                }
            }
        }
    }

    Ok(Json(BatchEventResponse {
        status: "ok".to_string(),
        event_ids,
        commands_generated: 0,
    }))
}

/// Extract command_id from request.
fn get_command_id(request: &EventRequest) -> Option<String> {
    // Try payload first
    if let Some(id) = request.payload.get("command_id").and_then(|v| v.as_str()) {
        return Some(id.to_string());
    }
    // Try meta
    if let Some(meta) = &request.meta {
        if let Some(id) = meta.get("command_id").and_then(|v| v.as_str()) {
            return Some(id.to_string());
        }
    }
    None
}

/// Check if command is already claimed.
async fn check_already_claimed(
    state: &AppState,
    execution_id: i64,
    command_id: &str,
    worker_id: &Option<String>,
) -> AppResult<bool> {
    let row: Option<(Option<String>, Option<serde_json::Value>)> =
        sqlx::query_as::<_, (Option<String>, Option<serde_json::Value>)>(
            r#"
            SELECT worker_id, meta FROM noetl.event
            WHERE execution_id = $1
              AND event_type = 'command.claimed'
              AND (meta->>'command_id' = $2 OR result->'data'->>'command_id' = $2)
            LIMIT 1
            "#,
        )
        .bind(execution_id)
        .bind(command_id)
        .fetch_optional(state.pools.pool_for(execution_id))
        .await?;

    if let Some((existing_worker, meta)) = row {
        let existing_worker_id = existing_worker.or_else(|| {
            meta.and_then(|m| {
                m.get("worker_id")
                    .and_then(|v| v.as_str())
                    .map(String::from)
            })
        });

        if let (Some(existing), Some(current)) = (&existing_worker_id, worker_id) {
            if existing != current {
                // Different worker - reject
                return Err(AppError::Conflict(format!(
                    "Command already claimed by {}",
                    existing
                )));
            }
            // Same worker - idempotent
            return Ok(true);
        }
    }

    Ok(false)
}

/// Build the `result` JSONB envelope for a `noetl.event` row.
///
/// Shape is constrained at the DB level by
/// `chk_event_result_shape`: top-level keys are limited to
/// `status` (required string), `reference` (optional object),
/// `context` (optional object).  Anything else fails the
/// constraint.  See noetl/server#29 for the history — the
/// previous `{kind, data}` / `{kind, store_tier, logical_uri}` /
/// `{kind, event_ids, total_parts}` envelopes all violated the
/// constraint and caused every POST /api/events that reached
/// the INSERT to 500.
///
/// Mapping:
/// - `result_kind = "ref"`  + `result_uri` set
///       → `{status, reference: {store_tier, logical_uri}}`
/// - `result_kind = "refs"` + `event_ids` set
///       → `{status, reference: {event_ids, total_parts}}`
/// - default (`"data"` or unknown):
///   - `payload` is a non-null object → `{status, context: <payload>}`
///   - `payload` is null/non-object   → `{status}`
fn build_result_object(request: &EventRequest, status: &str) -> serde_json::Value {
    let mut result = serde_json::Map::new();
    result.insert("status".to_string(), serde_json::Value::String(status.to_string()));

    match request.result_kind.as_str() {
        "ref" if request.result_uri.is_some() => {
            let uri = request.result_uri.as_ref().unwrap();
            let store_tier = if uri.starts_with("gs://") {
                "gcs"
            } else if uri.starts_with("s3://") {
                "s3"
            } else {
                "artifact"
            };
            result.insert(
                "reference".to_string(),
                serde_json::json!({
                    "store_tier": store_tier,
                    "logical_uri": uri,
                }),
            );
        }
        "refs" if request.event_ids.is_some() => {
            let event_ids = request.event_ids.as_ref().unwrap();
            result.insert(
                "reference".to_string(),
                serde_json::json!({
                    "event_ids": event_ids,
                    "total_parts": event_ids.len(),
                }),
            );
        }
        _ => {
            // Constraint requires `context` (when present) to be
            // an object.  Wire-format payload may be a primitive
            // for some legacy clients — skip the key entirely in
            // that case rather than corrupting the row.
            if let serde_json::Value::Object(_) = request.payload {
                result.insert("context".to_string(), request.payload.clone());
            }
        }
    }

    serde_json::Value::Object(result)
}

/// Get catalog_id from existing events.
async fn get_catalog_id(state: &AppState, execution_id: i64) -> AppResult<Option<i64>> {
    let row: Option<(i64,)> = sqlx::query_as::<_, (i64,)>(
        "SELECT catalog_id FROM noetl.event WHERE execution_id = $1 LIMIT 1",
    )
    .bind(execution_id)
    .fetch_optional(state.pools.pool_for(execution_id))
    .await?;

    Ok(row.map(|(id,)| id))
}

/// Map event name to status.
fn event_status_from_name(event_name: &str) -> &'static str {
    if event_name.contains("done")
        || event_name.contains("exit")
        || event_name.contains("completed")
    {
        "COMPLETED"
    } else if event_name.contains("error") || event_name.contains("failed") {
        "FAILED"
    } else {
        "RUNNING"
    }
}

/// Trigger orchestrator for workflow progression.
///
/// Phase D Round 2 of noetl/ai-meta#49 — wires the previously-
/// stubbed orchestrator to the real
/// [`crate::engine::WorkflowOrchestrator::evaluate`] pipeline.
///
/// Pipeline:
///
/// 1. Load all `noetl.event` rows for this execution (sorted by
///    `event_id` so [`WorkflowState::from_events`] reconstructs
///    state in the canonical order).
/// 2. Resolve `catalog_id` from one of the events, load the
///    playbook YAML, parse to [`Playbook`].
/// 3. Call `orchestrator.evaluate(&events, &playbook,
///    Some("command.completed"))`.
/// 4. For each [`EventToEmit`] returned (e.g. `step.enter` rows
///    for the next step), insert into `noetl.event`.
/// 5. For each [`engine::Command`] returned, look up the matching
///    [`Step`] in the playbook and call
///    [`crate::handlers::execute::persist_engine_command`] —
///    which inserts the `command.issued` event, the
///    `noetl.command` row, and publishes the NATS notification
///    (same code path the `/api/execute` first-command uses, so
///    the wire format stays consistent).
/// 6. If `result.should_complete` is set, emit a final
///    `playbook.completed` or `playbook.failed` event so
///    downstream consumers can observe terminal state.
///
/// `trigger_event_id` is the `event_id` of the event that
/// triggered this evaluation pass (usually the `command.completed`
/// row).  It's used as the `parent_event_id` for newly-inserted
/// events so the event log forms a proper causal chain.
async fn trigger_orchestrator(
    state: &AppState,
    execution_id: i64,
    trigger_event_id: i64,
) -> AppResult<i32> {
    use crate::engine::WorkflowOrchestrator;
    use sqlx::Row;

    debug!(
        execution_id,
        trigger_event_id, "trigger_orchestrator: loading events"
    );

    // 1. Load all events for this execution.
    //
    // `attempt` is stored inside `meta` JSONB (no dedicated column on
    // noetl.event today — same shape Python projector uses), so we
    // source it via `meta->>'attempt'` cast to int.
    let rows = sqlx::query(
        r#"
        SELECT event_id, execution_id, catalog_id,
               parent_event_id, parent_execution_id,
               event_type, node_id, node_name, node_type, status,
               context, meta, result, worker_id,
               NULLIF(meta->>'attempt', '')::int AS attempt,
               created_at
        FROM noetl.event
        WHERE execution_id = $1
        ORDER BY event_id ASC
        "#,
    )
    .bind(execution_id)
    .fetch_all(state.pools.pool_for(execution_id))
    .await?;

    let events: Vec<crate::db::models::Event> = rows
        .into_iter()
        .map(|r| crate::db::models::Event {
            id: r.try_get("event_id").unwrap_or(0),
            execution_id: r.try_get("execution_id").unwrap_or(0),
            catalog_id: r.try_get("catalog_id").unwrap_or(0),
            event_id: r.try_get("event_id").unwrap_or(0),
            parent_event_id: r.try_get("parent_event_id").ok(),
            parent_execution_id: r.try_get("parent_execution_id").ok(),
            event_type: r.try_get("event_type").unwrap_or_default(),
            node_id: r.try_get("node_id").ok(),
            node_name: r.try_get("node_name").ok(),
            node_type: r.try_get("node_type").ok(),
            status: r.try_get("status").unwrap_or_default(),
            context: r.try_get("context").ok(),
            meta: r.try_get("meta").ok(),
            result: r.try_get("result").ok(),
            worker_id: r.try_get("worker_id").ok(),
            attempt: r.try_get("attempt").ok(),
            created_at: r.try_get("created_at").unwrap_or_else(|_| chrono::Utc::now()),
        })
        .collect();

    if events.is_empty() {
        debug!(execution_id, "No events to evaluate — orchestrator exit early");
        return Ok(0);
    }

    // 2. Look up catalog_id + playbook content.
    let catalog_id = events
        .iter()
        .find_map(|e| {
            if e.catalog_id > 0 {
                Some(e.catalog_id)
            } else {
                None
            }
        })
        .ok_or_else(|| {
            AppError::Internal(format!(
                "No catalog_id found in events for execution {execution_id}"
            ))
        })?;

    // Phase F R4-3: noetl.catalog is a cluster-wide table.
    let playbook_yaml: String = sqlx::query_scalar(
        "SELECT content FROM noetl.catalog WHERE catalog_id = $1",
    )
    .bind(catalog_id)
    .fetch_one(state.pools.cluster())
    .await
    .map_err(|e| {
        AppError::Internal(format!(
            "Failed to load playbook for catalog_id {catalog_id}: {e}"
        ))
    })?;
    let playbook = crate::playbook::parser::parse_playbook(&playbook_yaml)?;

    // 3. Evaluate.
    //
    // Resolve the real trigger event's type from the loaded events
    // list instead of hard-coding `command.completed` — the same
    // path serves `command.failed` triggers (noetl/ai-meta#58) and
    // future trigger sources (e.g. `iterator_completed`,
    // `step.exit`).  Without this lookup, a failure trigger would
    // be misreported as a completion to the orchestrator, and the
    // failure-termination branch would never run.
    let trigger_event_type = events
        .iter()
        .find(|e| e.event_id == trigger_event_id)
        .map(|e| e.event_type.as_str())
        .unwrap_or("command.completed");
    let orchestrator = WorkflowOrchestrator::new();
    let result = match orchestrator.evaluate(&events, &playbook, Some(trigger_event_type)) {
        Ok(r) => r,
        Err(e) => {
            // A deterministic orchestrator evaluate failure (invalid
            // template in a step body, unknown step in a `next` arc,
            // malformed routing) fails identically on every retry.
            // Emitting only a WARN here strands the execution in RUNNING
            // forever — the next step is never issued and no terminal
            // event is written, so `/api/executions/{id}` reports RUNNING
            // indefinitely.  Instead write a terminal `playbook.failed`
            // event so the run resolves to FAILED with the error surfaced
            // to the client.
            //
            // noetl/ai-meta#54 (e2e regression sweep): `test_vars_template_access`
            // hung after `set_variables` because an invalid `{{ ctx.* }}`
            // template in a downstream step's `code` body tripped minijinja
            // inside `evaluate`, and the WARN-only path left no terminal
            // event.  This mirrors the noetl/ai-meta#58 `command.failed`
            // stall class — a deterministic failure must still produce a
            // terminal event.
            let msg = format!("Orchestrator evaluate failed: {e}");
            warn!(
                execution_id,
                error = %msg,
                "Orchestrator evaluate error is deterministic — terminating execution as FAILED"
            );
            emit_playbook_failed(state, execution_id, catalog_id, trigger_event_id, &msg).await?;
            return Ok(0);
        }
    };

    info!(
        execution_id,
        trigger_event_id,
        new_commands = result.commands.len(),
        new_events = result.events_to_emit.len(),
        should_complete = result.should_complete,
        "Orchestrator evaluate complete"
    );

    // 4. Emit pure events (step.enter etc.) before issuing new
    //    commands so the causal chain is correct.
    for emit in &result.events_to_emit {
        let event_id = state.snowflake.generate()?;
        let event_status = if emit.status.is_empty() {
            "STARTED".to_string()
        } else {
            emit.status.clone()
        };

        // Compose the constraint-compliant {status, context} result
        // envelope when context is present, else {status} alone.
        let result_obj = match &emit.context {
            Some(serde_json::Value::Object(_)) => serde_json::json!({
                "status": event_status,
                "context": emit.context.clone().unwrap(),
            }),
            _ => serde_json::json!({"status": event_status}),
        };

        sqlx::query(
            r#"
            INSERT INTO noetl.event (
                event_id, execution_id, catalog_id, event_type,
                node_id, node_name, status, result, meta, created_at, parent_event_id
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
            "#,
        )
        .bind(event_id)
        .bind(execution_id)
        .bind(catalog_id)
        .bind(&emit.event_type)
        .bind(emit.node_name.as_deref())
        .bind(emit.node_name.as_deref())
        .bind(&event_status)
        .bind(&result_obj)
        .bind(serde_json::json!({"emitted_by": "orchestrator"}))
        .bind(chrono::Utc::now())
        .bind(trigger_event_id)
        .execute(state.pools.pool_for(execution_id))
        .await?;
    }

    // 5. Issue new commands via the shared persist + publish helper.
    let mut commands_generated = 0i32;
    for command in &result.commands {
        let step = playbook.get_step(&command.step_name).ok_or_else(|| {
            AppError::Internal(format!(
                "Orchestrator returned command for unknown step '{}'",
                command.step_name
            ))
        })?;

        let render_context: std::collections::HashMap<String, serde_json::Value> =
            command.context.clone().unwrap_or_default();

        crate::handlers::execute::persist_engine_command(
            state,
            execution_id,
            catalog_id,
            trigger_event_id,
            step,
            command,
            &render_context,
            &playbook,
        )
        .await?;
        commands_generated += 1;
    }

    // 6. Emit terminal playbook event when the orchestrator says so.
    if result.should_complete {
        let (event_type, status) = match &result.completion_status {
            Some(cs) if cs.status == "FAILED" => ("playbook.failed", "FAILED"),
            _ => ("playbook.completed", "COMPLETED"),
        };
        let event_id = state.snowflake.generate()?;
        let terminal_meta = serde_json::to_value(&result.completion_status).unwrap_or_default();
        sqlx::query(
            r#"
            INSERT INTO noetl.event (
                event_id, execution_id, catalog_id, event_type,
                node_id, node_name, status, result, meta, created_at, parent_event_id
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
            "#,
        )
        .bind(event_id)
        .bind(execution_id)
        .bind(catalog_id)
        .bind(event_type)
        .bind("playbook")
        .bind("playbook")
        .bind(status)
        .bind(serde_json::json!({"status": status}))
        .bind(terminal_meta)
        .bind(chrono::Utc::now())
        .bind(trigger_event_id)
        .execute(state.pools.pool_for(execution_id))
        .await?;
        info!(
            execution_id,
            terminal_event = %event_type,
            "Orchestrator marked execution as terminal"
        );
    }

    Ok(commands_generated)
}

/// Emit a terminal `playbook.failed` event for an execution that hit a
/// deterministic, non-retryable orchestrator error during evaluate.
///
/// Without this an evaluate failure (invalid template in a step body,
/// unknown step in a `next` arc) leaves only a WARN in the server log
/// and strands the execution in RUNNING forever — no terminal event is
/// ever written.  The list-status aggregation in `list_executions`
/// maps `playbook.failed` -> FAILED, so writing this event resolves the
/// run and surfaces `error` to API readers.  Parented on the trigger
/// event so the causal chain stays intact.
async fn emit_playbook_failed(
    state: &AppState,
    execution_id: i64,
    catalog_id: i64,
    trigger_event_id: i64,
    error: &str,
) -> AppResult<()> {
    let event_id = state.snowflake.generate()?;
    sqlx::query(
        r#"
        INSERT INTO noetl.event (
            event_id, execution_id, catalog_id, event_type,
            node_id, node_name, status, result, meta, created_at, parent_event_id
        ) VALUES ($1, $2, $3, 'playbook.failed', 'playbook', 'playbook', 'FAILED', $4, $5, $6, $7)
        "#,
    )
    .bind(event_id)
    .bind(execution_id)
    .bind(catalog_id)
    .bind(serde_json::json!({"status": "FAILED", "context": {"error": error}}))
    .bind(serde_json::json!({
        "emitted_by": "orchestrator",
        "reason": "evaluate_error",
        "error": error,
    }))
    .bind(chrono::Utc::now())
    .bind(trigger_event_id)
    .execute(state.pools.pool_for(execution_id))
    .await?;
    info!(
        execution_id,
        terminal_event = "playbook.failed",
        "Orchestrator evaluate error → execution terminated as FAILED"
    );
    Ok(())
}

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

    fn test_request_skeleton() -> EventRequest {
        EventRequest {
            execution_id: "123".to_string(),
            step: "step1".to_string(),
            event_type: "step.exit".to_string(),
            payload: serde_json::json!({}),
            meta: None,
            worker_id: None,
            result_kind: "data".to_string(),
            result_uri: None,
            event_ids: None,
            actionable: true,
            informative: true,
            event_id: None,
            status: None,
            created_at: None,
        }
    }

    #[test]
    fn test_event_request_defaults() {
        // New canonical field name `event_type`.
        let json = r#"{"execution_id": "123", "step": "step1", "event_type": "step.enter"}"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();

        assert_eq!(request.event_type, "step.enter");
        assert_eq!(request.result_kind, "data");
        assert!(request.actionable);
        assert!(request.informative);
        assert!(request.event_id.is_none());
        assert!(request.status.is_none());
        assert!(request.created_at.is_none());
    }

    #[test]
    fn test_legacy_name_alias_deserializes_into_event_type() {
        // R-1.2 PR-EE-2 back-compat: pre-PR-EE worker / CLI
        // clients send `name` instead of `event_type`.  The
        // alias means they deserialize cleanly without a server
        // restart.
        let json = r#"{"execution_id": "123", "step": "step1", "name": "step.exit"}"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.event_type, "step.exit");
    }

    #[test]
    fn test_context_alias_deserializes_into_payload() {
        // Executor producers send the field as `context`; pre-PR
        // clients send `payload`.  Both deserialize into the same
        // field on the server side.
        let json = r#"{
            "execution_id": "123",
            "step": "step1",
            "event_type": "step.exit",
            "context": {"result": 42}
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.payload["result"], 42);
    }

    #[test]
    fn test_new_optional_fields_accept_executor_event_shape() {
        // Wire format matching noetl-executor 0.3.1 ExecutorEvent:
        // event_id (snowflake as String), status, created_at, plus
        // the `context` alias.
        let json = r#"{
            "execution_id": "478775660589088776",
            "event_type": "command.completed",
            "step": "fetch_calendar",
            "status": "COMPLETED",
            "created_at": "2026-05-31T03:14:15Z",
            "context": {"items": 42},
            "event_id": "478775660589088777",
            "worker_id": "worker-prod-7",
            "meta": {"attempts": 2}
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.event_type, "command.completed");
        assert_eq!(request.event_id.as_deref(), Some("478775660589088777"));
        assert_eq!(request.status.as_deref(), Some("COMPLETED"));
        assert_eq!(request.worker_id.as_deref(), Some("worker-prod-7"));
        assert!(request.created_at.is_some());
    }

    #[test]
    fn test_event_request_accepts_integer_execution_id() {
        // noetl/ai-meta#55 — the Rust worker emits
        // `noetl-events::ExecutorEvent` whose `execution_id` is
        // `i64`, so the wire shape is a JSON integer.  Without the
        // `deserialize_string_or_i64` adapter, strict serde rejects
        // it with "invalid type: integer, expected a string", and
        // every Rust-on-both-ends event emission fails.
        let json = r#"{
            "execution_id": 321079436235509760,
            "step": "start",
            "event_type": "step.start"
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.execution_id, "321079436235509760");
        assert_eq!(request.step, "start");
    }

    #[test]
    fn test_event_request_accepts_string_execution_id() {
        // Legacy browser-client wire shape (snowflake as JSON
        // string).  Confirms the lax decoder didn't regress the
        // documented `String` wire format that EE-2 / EE-4 kept
        // for browser JSON-number precision.
        let json = r#"{
            "execution_id": "321079436235509760",
            "step": "start",
            "event_type": "step.start"
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.execution_id, "321079436235509760");
    }

    #[test]
    fn test_event_request_accepts_integer_event_id() {
        // event_id is optional + `Option<String>`; the worker emits
        // `Option<i64>` from `noetl-events`.  Same drift as
        // execution_id, same fix.
        let json = r#"{
            "execution_id": "1",
            "step": "s",
            "event_type": "e",
            "event_id": 478775660589088777
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.event_id.as_deref(), Some("478775660589088777"));
    }

    #[test]
    fn test_event_request_event_id_null_is_none() {
        // Explicit null should land as None.  Optional decoder
        // sanity check.
        let json = r#"{
            "execution_id": "1",
            "step": "s",
            "event_type": "e",
            "event_id": null
        }"#;
        let request: EventRequest = serde_json::from_str(json).unwrap();
        assert!(request.event_id.is_none());
    }

    #[test]
    fn test_batch_event_request_accepts_integer_execution_id() {
        // BatchEventRequest is the second worker→server inbound
        // type with execution_id on it (the worker uses it for
        // batched event emission).  Same drift as the per-event
        // shape; same lax decoder.
        let json = r#"{
            "execution_id": 321079436235509760,
            "worker_id": "worker-rust-pool-0",
            "events": [
                {
                    "step": "start",
                    "event_type": "step.start"
                }
            ]
        }"#;
        let request: BatchEventRequest = serde_json::from_str(json).unwrap();
        assert_eq!(request.execution_id, "321079436235509760");
        assert_eq!(request.events.len(), 1);
    }

    #[test]
    fn test_event_request_rejects_garbage_execution_id() {
        // The lax decoder should still reject obviously-bogus
        // shapes (arrays, objects, floats) — only `string` and
        // `integer` are valid for a snowflake id field.  Floats
        // are rejected because the precision loss makes them
        // ambiguous (a Pythonish `12345.0` decodes via visit_f64
        // which we don't implement).
        let bogus_shapes = [
            r#"{"execution_id": [1,2,3], "step": "s", "event_type": "e"}"#,
            r#"{"execution_id": {"id": 1}, "step": "s", "event_type": "e"}"#,
        ];
        for json in bogus_shapes {
            let result: std::result::Result<EventRequest, _> = serde_json::from_str(json);
            assert!(result.is_err(), "Expected reject for {}", json);
        }
    }

    // build_result_object — constraint-compliant shape per
    // noetl/server#29.  The DB constraint allows only
    // `status` (required string), `reference` (optional object),
    // `context` (optional object); nothing else.

    #[test]
    fn test_build_result_object_data() {
        let request = EventRequest {
            payload: serde_json::json!({"output": "success"}),
            ..test_request_skeleton()
        };

        let result = build_result_object(&request, "COMPLETED");
        assert_eq!(result["status"], "COMPLETED");
        assert_eq!(result["context"]["output"], "success");
        // No disallowed top-level keys.
        assert!(result.get("kind").is_none());
        assert!(result.get("data").is_none());
        assert!(result.get("reference").is_none());
    }

    #[test]
    fn test_build_result_object_data_with_null_payload_omits_context() {
        let request = EventRequest {
            payload: serde_json::Value::Null,
            ..test_request_skeleton()
        };
        let result = build_result_object(&request, "STARTED");
        assert_eq!(result["status"], "STARTED");
        assert!(result.get("context").is_none(),
            "context must not be set when payload is non-object: {result}");
    }

    #[test]
    fn test_build_result_object_data_with_primitive_payload_omits_context() {
        // Constraint: when present, context must be an object.
        // Wrap-it-or-omit-it — we chose omit.
        let request = EventRequest {
            payload: serde_json::json!("just a string"),
            ..test_request_skeleton()
        };
        let result = build_result_object(&request, "RUNNING");
        assert!(result.get("context").is_none());
    }

    #[test]
    fn test_build_result_object_ref() {
        let request = EventRequest {
            result_kind: "ref".to_string(),
            result_uri: Some("gs://bucket/path/to/result".to_string()),
            ..test_request_skeleton()
        };

        let result = build_result_object(&request, "COMPLETED");
        assert_eq!(result["status"], "COMPLETED");
        let reference = &result["reference"];
        assert_eq!(reference["store_tier"], "gcs");
        assert_eq!(reference["logical_uri"], "gs://bucket/path/to/result");
        // No disallowed top-level keys.
        assert!(result.get("kind").is_none());
        assert!(result.get("store_tier").is_none());
        assert!(result.get("logical_uri").is_none());
    }

    #[test]
    fn test_build_result_object_refs() {
        let request = EventRequest {
            result_kind: "refs".to_string(),
            event_ids: Some(vec![100, 200, 300]),
            ..test_request_skeleton()
        };

        let result = build_result_object(&request, "COMPLETED");
        assert_eq!(result["status"], "COMPLETED");
        let reference = &result["reference"];
        assert_eq!(reference["event_ids"][0], 100);
        assert_eq!(reference["total_parts"], 3);
        assert!(result.get("event_ids").is_none(),
            "event_ids should be nested under reference, not top-level");
    }

    #[test]
    fn test_build_result_object_constraint_top_level_keys_only() {
        // The DB constraint allows ONLY {status, reference, context}
        // at the top level.  Walk all output shapes and assert none
        // emit anything else.
        let allowed: std::collections::HashSet<&str> =
            ["status", "reference", "context"].iter().copied().collect();

        let cases: Vec<(&str, EventRequest)> = vec![
            (
                "data with object payload",
                EventRequest {
                    payload: serde_json::json!({"k": "v"}),
                    ..test_request_skeleton()
                },
            ),
            (
                "data with null payload",
                EventRequest {
                    payload: serde_json::Value::Null,
                    ..test_request_skeleton()
                },
            ),
            (
                "ref",
                EventRequest {
                    result_kind: "ref".to_string(),
                    result_uri: Some("gs://foo".to_string()),
                    ..test_request_skeleton()
                },
            ),
            (
                "refs",
                EventRequest {
                    result_kind: "refs".to_string(),
                    event_ids: Some(vec![1, 2]),
                    ..test_request_skeleton()
                },
            ),
        ];
        for (label, req) in cases {
            let r = build_result_object(&req, "OK");
            let obj = r.as_object().expect("result must be object");
            for k in obj.keys() {
                assert!(
                    allowed.contains(k.as_str()),
                    "[{label}] disallowed top-level key: {k} (full result: {r})"
                );
            }
            assert_eq!(r["status"], "OK", "[{label}] status must be present");
        }
    }

    #[test]
    fn test_batch_event_item_legacy_name_alias() {
        let json = r#"{"step": "s", "name": "call.done", "payload": {}}"#;
        let item: BatchEventItem = serde_json::from_str(json).unwrap();
        assert_eq!(item.event_type, "call.done");
    }

    #[test]
    fn test_event_response_serialization() {
        let response = EventResponse {
            status: "ok".to_string(),
            event_id: 12345,
            commands_generated: 2,
        };

        let json = serde_json::to_string(&response).unwrap();
        assert!(json.contains("ok"));
        assert!(json.contains("12345"));
    }

    #[test]
    fn test_command_response_serialization() {
        let response = CommandResponse {
            execution_id: 12345,
            node_id: "step1".to_string(),
            node_name: "step1".to_string(),
            action: "python".to_string(),
            context: serde_json::json!({"tool_config": {}}),
            meta: serde_json::json!({"attempt": 1}),
        };

        let json = serde_json::to_string(&response).unwrap();
        assert!(json.contains("step1"));
        assert!(json.contains("python"));
    }

    // ---- EE-4 (noetl/ai-meta#49) wire-compat with noetl-events --------
    //
    // The server's `EventRequest` and the canonical
    // `noetl_events::ExecutorEvent` share a subset of fields that
    // make up the wire envelope every NoETL Rust producer emits.
    // The four tests below pin the round-trip semantics so a future
    // change to either type that breaks compat fails the build
    // here instead of in a kind-validation cycle.

    #[test]
    fn ee4_executor_event_converts_into_event_request() {
        let executor_event = noetl_events::ExecutorEvent {
            execution_id: 478775660589088776,
            event_type: "command.completed".to_string(),
            step: "fetch_calendar".to_string(),
            status: "COMPLETED".to_string(),
            created_at: chrono::DateTime::parse_from_rfc3339("2026-05-31T03:14:15Z")
                .unwrap()
                .with_timezone(&chrono::Utc),
            context: serde_json::json!({"items": 42}),
            event_id: Some(478775660589088777),
            worker_id: Some("worker-prod-7".to_string()),
            meta: Some(serde_json::json!({"attempts": 2})),
        };
        let req: EventRequest = executor_event.clone().into();
        // String wire format for browser precision.
        assert_eq!(req.execution_id, "478775660589088776");
        assert_eq!(req.event_id.as_deref(), Some("478775660589088777"));
        // Shared subset round-trips field-for-field.
        assert_eq!(req.event_type, executor_event.event_type);
        assert_eq!(req.step, executor_event.step);
        assert_eq!(req.status.as_deref(), Some(executor_event.status.as_str()));
        assert_eq!(req.created_at, Some(executor_event.created_at));
        assert_eq!(req.payload, executor_event.context);
        assert_eq!(req.worker_id, executor_event.worker_id);
        assert_eq!(req.meta, executor_event.meta);
        // Server-only fields take handler defaults.
        assert_eq!(req.result_kind, "data");
        assert!(req.result_uri.is_none());
        assert!(req.event_ids.is_none());
        assert!(req.actionable);
        assert!(req.informative);
    }

    #[test]
    fn ee4_event_request_converts_into_executor_event() {
        let req = EventRequest {
            execution_id: "478775660589088776".to_string(),
            step: "fetch_calendar".to_string(),
            event_type: "command.completed".to_string(),
            payload: serde_json::json!({"items": 42}),
            meta: Some(serde_json::json!({"attempts": 2})),
            worker_id: Some("worker-prod-7".to_string()),
            result_kind: "data".to_string(),
            result_uri: None,
            event_ids: None,
            actionable: true,
            informative: true,
            event_id: Some("478775660589088777".to_string()),
            status: Some("COMPLETED".to_string()),
            created_at: Some(
                chrono::DateTime::parse_from_rfc3339("2026-05-31T03:14:15Z")
                    .unwrap()
                    .with_timezone(&chrono::Utc),
            ),
        };
        let ev: noetl_events::ExecutorEvent =
            (&req).try_into().expect("convert with explicit fields");
        assert_eq!(ev.execution_id, 478775660589088776_i64);
        assert_eq!(ev.event_id, Some(478775660589088777_i64));
        assert_eq!(ev.status, "COMPLETED");
        assert_eq!(ev.created_at, req.created_at.unwrap());
        assert_eq!(ev.context, req.payload);
        assert_eq!(ev.worker_id, req.worker_id);
        assert_eq!(ev.meta, req.meta);
    }

    #[test]
    fn ee4_try_from_event_request_fills_defaults_for_missing_status_and_created_at() {
        // Producers that don't stamp `status` / `created_at` are
        // valid on the wire; the conversion must apply the same
        // fallbacks the handler uses, so callers building an
        // `ExecutorEvent` for downstream emit don't see surprises.
        let mut req = test_request_skeleton();
        req.event_type = "command.completed".to_string();
        req.status = None;
        req.created_at = None;
        let ev: noetl_events::ExecutorEvent =
            (&req).try_into().expect("convert with defaults");
        assert_eq!(ev.status, "COMPLETED"); // name-derived fallback
        // created_at falls back to now(); just assert it's non-zero
        // and recent enough to be sane.
        let age = chrono::Utc::now() - ev.created_at;
        assert!(age.num_seconds() >= 0 && age.num_seconds() < 60);
    }

    #[test]
    fn ee4_try_from_event_request_rejects_non_numeric_execution_id() {
        // The wire shape is "stringified i64".  Anything else is a
        // bug at the producer; the conversion surfaces it instead of
        // silently dropping the event into the log with execution_id=0.
        let req = EventRequest {
            execution_id: "not-a-number".to_string(),
            ..test_request_skeleton()
        };
        let err = noetl_events::ExecutorEvent::try_from(&req).unwrap_err();
        assert!(
            err.to_string().contains("execution_id"),
            "error should mention the field name: {err}"
        );
    }
}