noetl-server 3.13.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.
/// Resolve any `{status, reference: {ref: "noetl://..."}}` results in `events`
/// back to the inline `{status, context: <data>}` shape the orchestrator reads.
///
/// Results-by-reference: the worker stages an over-budget tool result in the
/// result store and emits only a small reference on the event (keeping the
/// event log lean).  The orchestrator needs the data to evaluate `output.*`
/// guards / `set:` and to dispatch cursor bodies, so it rehydrates the
/// reference here — only for the events it's about to apply, not the whole log.
/// Inline results (no `reference`) and unresolvable refs are left untouched.
/// Resolve over-budget result references inline before the orchestrator
/// applies the events.
///
/// The worker stages results larger than its inline budget in the durable
/// result store and emits a `{data: {_ref}}` placeholder plus a sibling
/// `reference` block instead of the data (see
/// `repos/worker/src/executor/command.rs` `build_reference_or_inline`).  The
/// orchestrator's state machine (`extract_user_data`, the cursor drive,
/// `build_context`) reads the actual data out of the event — a placeholder
/// makes a cursor claim look like it returned zero rows, which stalls the
/// loop.  So before `from_events` / `evaluate_state` see the events, we swap
/// each placeholder for the resolved data.
///
/// The reference sits at one of two paths depending on how the `call.done`
/// envelope wrapped the tool result. Nested (the standard envelope) puts the
/// tool result under `context.result`, so the reference is at
/// `result.context.result.reference` and the placeholder at
/// `result.context.result.context.data._ref`. Top-level (an un-wrapped tool
/// result) puts the reference at `result.reference` and the placeholder at
/// `result.context.data._ref`.
///
/// In both cases `reference.ref` is the `noetl://` URI and the sibling
/// `context` is replaced with the resolved store payload (`{data: {...}}`),
/// then the `reference` block is dropped so the result reads like an inline
/// one.
async fn hydrate_result_references(
    events: &mut [crate::db::models::Event],
    result_store: &crate::services::result_store::ResultStoreService,
    keep_refs: bool,
) {
    use crate::services::result_store::parse_noetl_ref;

    enum Shape {
        /// reference at `result.context.result.reference`
        Nested,
        /// reference at `result.reference`
        TopLevel,
    }

    let mut hydrated = 0usize;
    let mut kept = 0usize;
    for ev in events.iter_mut() {
        let Some(result) = ev.result.as_mut() else {
            continue;
        };
        // Locate the reference URI + remember which envelope shape carried it.
        let (shape, ref_uri) = if let Some(u) = result
            .pointer("/context/result/reference/ref")
            .and_then(|v| v.as_str())
        {
            (Shape::Nested, u.to_string())
        } else if let Some(u) = result.pointer("/reference/ref").and_then(|v| v.as_str()) {
            (Shape::TopLevel, u.to_string())
        } else {
            continue;
        };

        // References-in-state (noetl/ai-meta#101 phase 2): when gated on AND the
        // reference carries an `extracted` predicate block (phase 1), keep the
        // reference and surface `extracted` as the readable `context.data` —
        // instead of resolving the full payload inline.  `extract_user_data`
        // then reads `extracted` for `when:`/`set:` evaluation; the reference
        // block stays on the event so `build_context` carries it and the worker
        // resolves the bulk payload at render time.  No `extracted` (pre-phase-1
        // refs) falls through to the resolve path below for back-compat.
        if keep_refs {
            let extracted = match shape {
                Shape::Nested => result
                    .pointer("/context/result/reference/extracted")
                    .cloned(),
                Shape::TopLevel => result.pointer("/reference/extracted").cloned(),
            };
            if let Some(extracted) = extracted {
                let data = serde_json::json!({ "data": extracted });
                match shape {
                    Shape::Nested => {
                        if let Some(inner) = result
                            .get_mut("context")
                            .and_then(|c| c.get_mut("result"))
                            .and_then(|r| r.as_object_mut())
                        {
                            inner.insert("context".to_string(), data);
                            kept += 1;
                        }
                    }
                    Shape::TopLevel => {
                        if let Some(obj) = result.as_object_mut() {
                            obj.insert("context".to_string(), data);
                            kept += 1;
                        }
                    }
                }
                continue; // keep the reference; do not resolve
            }
        }

        let parsed = match parse_noetl_ref(&ref_uri) {
            Ok(p) => p,
            Err(e) => {
                warn!(execution_id = ev.execution_id, ref_uri, %e, "unparseable result reference; left as-is");
                continue;
            }
        };
        let resolved = match result_store.resolve(&parsed).await {
            Ok(Some(data)) => data,
            Ok(None) => {
                warn!(execution_id = ev.execution_id, ref_uri, "result reference not found in store; left as-is");
                continue;
            }
            Err(e) => {
                warn!(execution_id = ev.execution_id, ref_uri, %e, "result reference resolution failed; left as-is");
                continue;
            }
        };

        // Splice the resolved payload over the `{data: {_ref}}` placeholder
        // and drop the `reference` block so the event reads as inline.
        match shape {
            Shape::Nested => {
                if let Some(inner) = result
                    .get_mut("context")
                    .and_then(|c| c.get_mut("result"))
                    .and_then(|r| r.as_object_mut())
                {
                    inner.insert("context".to_string(), resolved);
                    inner.remove("reference");
                    hydrated += 1;
                }
            }
            Shape::TopLevel => {
                if let Some(obj) = result.as_object_mut() {
                    obj.insert("context".to_string(), resolved);
                    obj.remove("reference");
                    hydrated += 1;
                }
            }
        }
    }
    if hydrated > 0 || kept > 0 {
        debug!(
            hydrated,
            kept, "hydrate_result_references: resolved / kept-by-reference over-budget results"
        );
    }
}

/// Columns the orchestrator reads from `noetl.event`, in the order
/// [`parse_event_rows`] expects.
const ORCH_EVENT_COLS: &str = 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 "#;

/// Map rows selected via [`ORCH_EVENT_COLS`] into `Event`s.  `attempt` lives in
/// `meta` JSONB (no dedicated column), sourced via the `NULLIF(...)::int` alias.
fn parse_event_rows(rows: Vec<sqlx::postgres::PgRow>) -> Vec<crate::db::models::Event> {
    use sqlx::Row;
    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()
}

/// Outcome of a bounded state rebuild (noetl/ai-meta#101 block b).  The caller
/// sets `applied_count = total` (the live event count) after a rebuild: a
/// rebuild folds in every event after the snapshot, so the state reflects all
/// events; any straggler the window still missed is caught by the next
/// trigger's count mismatch.
struct RebuildResult {
    state: crate::engine::state::WorkflowState,
    last_event_id: i64,
    snapshot_version: i64,
    routing_meta: Option<serde_json::Value>,
}

/// Margin (seconds) the snapshot rebuild re-scans behind the snapshot's write
/// time.  A straggler that lands below the snapshot `version` *after* the
/// snapshot is written carries a recent-ish `created_at` (the column is the
/// event's emit time, not its snowflake id), so loading events with
/// `created_at > snapshot.updated_at - MARGIN` catches it.  Covers worker
/// clock skew + emit-to-insert latency; re-applying the overlap is safe because
/// cursor counters are gated by the snapshot's `cursor_issued`/`cursor_completed`
/// id-sets.  Bounded small: machine-id interleaving makes a straggler's
/// `created_at` only milliseconds off; the margin only needs to cover
/// emit-to-insert latency + worker clock skew (NTP keeps this sub-second).
const REBUILD_STRAGGLER_MARGIN_SECS: i64 = 30;

/// Rebuild orchestrator state with a **bounded** event load: from the latest
/// `projection_snapshot` + the events after it (by id, plus a `created_at`
/// margin window to catch below-watermark stragglers), or — when no snapshot
/// exists yet — the full (still-small) early log.  This replaced the unbounded
/// full-log replay that OOM'd the server at scale: a straggler below the
/// incremental watermark used to trigger a reload of the entire (growing) event
/// log on nearly every completion under high concurrency.
/// Resolve over-budget cursor claim references into inline `claim_rows` before
/// the drive runs (references-in-state, noetl/ai-meta#101 phase 2).  A referenced
/// claim has `claim_ref` set + empty `claim_rows` (the sync `apply_event` can't
/// resolve); this async pass fills the rows so the cursor fans out instead of
/// wrongly draining.  No-op unless the flag kept a claim reference.
async fn resolve_cursor_claim_refs(
    ws: &mut crate::engine::state::WorkflowState,
    result_store: &crate::services::result_store::ResultStoreService,
) {
    use crate::services::result_store::parse_noetl_ref;
    for step in ws.steps.values_mut() {
        if !step.is_cursor {
            continue;
        }
        for frame in step.cursor_frames.values_mut() {
            if !frame.claim_rows.is_empty() {
                continue;
            }
            let Some(uri) = frame.claim_ref.clone() else {
                continue;
            };
            let parsed = match parse_noetl_ref(&uri) {
                Ok(p) => p,
                Err(e) => {
                    warn!(uri, %e, "cursor claim reference unparseable; left as drain");
                    continue;
                }
            };
            match result_store.resolve(&parsed).await {
                Ok(Some(data)) => match extract_claim_rows(&data) {
                    Some(rows) => {
                        frame.claim_rows = rows;
                        frame.claim_ref = None;
                    }
                    None => warn!(uri, "cursor claim reference resolved but no rows array found"),
                },
                Ok(None) => warn!(uri, "cursor claim reference not found in store"),
                Err(e) => warn!(uri, %e, "cursor claim reference resolve failed"),
            }
        }
    }
}

/// Pull the claimed rows array out of a resolved claim payload, trying the
/// common shapes the store may hold for a tool result.
fn extract_claim_rows(data: &serde_json::Value) -> Option<Vec<serde_json::Value>> {
    for ptr in ["/rows", "/data/rows", "/result/rows", "/context/data/rows"] {
        if let Some(arr) = data.pointer(ptr).and_then(|v| v.as_array()) {
            return Some(arr.clone());
        }
    }
    crate::engine::state::extract_user_data(data)
        .and_then(|d| d.get("rows").and_then(|r| r.as_array()).cloned())
}

async fn rebuild_state(
    pool: &crate::db::DbPool,
    result_store: &crate::services::result_store::ResultStoreService,
    execution_id: i64,
    keep_refs: bool,
) -> AppResult<RebuildResult> {
    use crate::services::orch_snapshot;
    match orch_snapshot::load_latest(pool, execution_id).await? {
        Some(snap) => {
            // Events after the snapshot: newer by id, OR (straggler) below the
            // version watermark but emitted within the margin of the snapshot.
            let margin_floor =
                snap.updated_at - chrono::Duration::seconds(REBUILD_STRAGGLER_MARGIN_SECS);
            let mut events_since = parse_event_rows(
                sqlx::query(&format!(
                    "{ORCH_EVENT_COLS} WHERE execution_id = $1 \
                       AND (event_id > $2 OR created_at > $3) ORDER BY event_id ASC"
                ))
                .bind(execution_id)
                .bind(snap.version)
                .bind(margin_floor)
                .fetch_all(pool)
                .await?,
            );
            hydrate_result_references(&mut events_since, result_store, keep_refs).await;
            let mut ws = snap.state;
            for e in &events_since {
                ws.apply_event(e);
            }
            // The window includes everything after the snapshot, so the highest
            // loaded id is the current head.
            let last_event_id = events_since
                .iter()
                .map(|e| e.event_id)
                .max()
                .unwrap_or(snap.version);
            Ok(RebuildResult {
                state: ws,
                last_event_id,
                snapshot_version: snap.version,
                routing_meta: snap.routing_meta,
            })
        }
        None => {
            let mut all_events = parse_event_rows(
                sqlx::query(&format!(
                    "{ORCH_EVENT_COLS} WHERE execution_id = $1 ORDER BY event_id ASC"
                ))
                .bind(execution_id)
                .fetch_all(pool)
                .await?,
            );
            hydrate_result_references(&mut all_events, result_store, keep_refs).await;
            let state = crate::engine::state::WorkflowState::from_events(&all_events)
                .ok_or_else(|| AppError::Validation("No events found for execution".to_string()))?;
            let last_event_id = all_events.last().map(|e| e.event_id).unwrap_or(0);
            let routing_meta = all_events
                .iter()
                .find(|e| e.event_type == "playbook_started")
                .and_then(|e| e.meta.clone());
            Ok(RebuildResult {
                state,
                last_event_id,
                snapshot_version: 0,
                routing_meta,
            })
        }
    }
}

async fn trigger_orchestrator(
    state: &AppState,
    execution_id: i64,
    trigger_event_id: i64,
) -> AppResult<i32> {
    trigger_orchestrator_inner(state, execution_id, trigger_event_id, false).await
}

/// Background reconcile poller (noetl/ai-meta#101 block b — small-tier
/// resilience).  Guarantees the orchestrator never permanently stalls.
///
/// The hot path advances state incrementally and only does the (O(events))
/// consistency `COUNT` on a throttle, so a non-triggering straggler that lands
/// in a throttle gap can leave a cursor unable to fan out — and once the cursor
/// stops, no further `command.completed` events arrive, so there is no trigger
/// to retry on.  On a constrained backend (e.g. a small Cloud SQL tier behind
/// PgBouncer, where DB latency widens the gap) this is exactly the deadlock that
/// stopped a 10×1000 run.
///
/// This task periodically force-reconciles every cached (non-terminal)
/// execution: a fresh `COUNT` + bounded rebuild folds in any missed straggler
/// and re-evaluates, so processing always resumes — slow under backpressure is
/// fine; stopping is not.  Cost is bounded: one cheap rebuild per active
/// execution per tick.
pub fn spawn_orchestrator_reconciler(state: AppState) {
    tokio::spawn(async move {
        const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(8);
        loop {
            tokio::time::sleep(RECONCILE_INTERVAL).await;
            for execution_id in state.orch_cache.active_executions() {
                // `i64::MAX` as the trigger id keeps the immediate-straggler
                // shortcut from firing on an already-applied event.
                match trigger_orchestrator_inner(&state, execution_id, i64::MAX, true).await {
                    Ok(n) if n > 0 => {
                        info!(execution_id, commands = n, "reconcile poller advanced a stuck execution")
                    }
                    Ok(_) => {}
                    Err(e) => {
                        warn!(execution_id, %e, "reconcile poller: orchestrator trigger failed")
                    }
                }
            }
        }
    });
}

/// Orchestrator trigger.  `force_count` is set by the background reconcile
/// poller ([`spawn_orchestrator_reconciler`]): it forces a fresh consistency
/// `COUNT` (bypassing the throttle) and proceeds to evaluate even when no new
/// events arrived, so a stuck execution — one whose cursor missed a
/// non-triggering straggler and therefore stopped emitting events, leaving no
/// trigger to retry on — is reconciled and advanced.  The poller passes
/// `trigger_event_id = i64::MAX` so the immediate-straggler shortcut is skipped.
async fn trigger_orchestrator_inner(
    state: &AppState,
    execution_id: i64,
    trigger_event_id: i64,
    force_count: bool,
) -> AppResult<i32> {
    use crate::engine::WorkflowOrchestrator;

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

    // 1. Incremental state with bounded rebuild (noetl/ai-meta#100, #101 block b).
    //
    // The cached `WorkflowState` advances by applying only events newer than
    // `last_event_id` behind a per-execution lock (serialising one execution's
    // triggers).  When the cache is cold, or a straggler is detected (an event
    // below the incremental watermark — workers' snowflake ids from different
    // machines in the same millisecond interleave), it rebuilds from the latest
    // `projection_snapshot` + events-since (bounded), NOT the whole event log:
    // the unbounded replay spiked memory and OOM'd the server at scale.  See
    // `rebuild_state` + `services::orch_snapshot`.

    let pool = state.pools.pool_for(execution_id);
    // Results-by-reference: an over-budget tool result is stored in the result
    // store and the event carries `{status, reference}` (no inline data).  The
    // orchestrator resolves those references back to inline data before reading
    // them so referenced results are transparent to state/templates.
    let result_store = crate::services::result_store::ResultStoreService::new(
        pool.clone(),
        state.snowflake.clone(),
    );

    // Per-execution lock: serialises this execution's concurrent completion
    // triggers so only one advances the cached state at a time.
    let cache_slot = state.orch_cache.entry(execution_id);
    let mut cache = cache_slot.lock().await;

    // The consistency `COUNT(*)` over this execution's event partition is
    // O(events) — ≈27ms at 60k events — and only detects a *non-triggering*
    // straggler (an event type that doesn't fire the orchestrator, inserted
    // below the high-water mark).  Running it on every trigger throttles the
    // whole orchestrator on a large log.  So throttle it: a fresh `total` gates
    // the mismatch→rebuild + snapshot paths; between checks the incremental
    // apply + the immediate straggler handling below carry correctness.  Cold
    // cache always counts (it needs `total` to seed `applied_count`).
    const COUNT_THROTTLE: std::time::Duration = std::time::Duration::from_millis(1000);
    let now = std::time::Instant::now();
    let do_count = force_count
        || cache.state.is_none()
        || cache
            .last_count_check
            .is_none_or(|t| now.duration_since(t) >= COUNT_THROTTLE);
    let total: Option<i64> = if do_count {
        cache.last_count_check = Some(now);
        Some(
            sqlx::query_scalar("SELECT COUNT(*) FROM noetl.event WHERE execution_id = $1")
                .bind(execution_id)
                .fetch_one(pool)
                .await?,
        )
    } else {
        None
    };

    // Warm a cold cache from the latest snapshot + events-since (bounded), or
    // the full (still-small) early log when no snapshot exists yet.
    let mut did_rebuild = false;
    if cache.state.is_none() {
        let r = rebuild_state(pool, &result_store, execution_id, state.config.refs_in_state).await?;
        cache.state = Some(r.state);
        cache.last_event_id = r.last_event_id;
        cache.applied_count = total.unwrap_or(0);
        cache.snapshot_version = r.snapshot_version;
        cache.routing_meta = r.routing_meta;
        did_rebuild = true;
    }

    // Immediate straggler: the event that fired this trigger is at/below the
    // watermark (a late insert below the high-water mark — interleaved snowflake
    // ids).  It is not in the `> watermark` load below, so apply it directly.
    // `command.completed`/`failed` are the only types that trigger, so this
    // catches the cursor-relevant stragglers (body completions) with no COUNT
    // and no rebuild.  Idempotent: cursor counters are gated by the
    // `cursor_issued`/`cursor_completed` id-sets.
    let mut straggler_applied = false;
    if !did_rebuild && trigger_event_id <= cache.last_event_id {
        let mut strag = parse_event_rows(
            sqlx::query(&format!(
                "{ORCH_EVENT_COLS} WHERE execution_id = $1 AND event_id = $2"
            ))
            .bind(execution_id)
            .bind(trigger_event_id)
            .fetch_all(pool)
            .await?,
        );
        hydrate_result_references(&mut strag, &result_store, state.config.refs_in_state).await;
        if let Some(ws) = cache.state.as_mut() {
            for e in &strag {
                ws.apply_event(e);
            }
        }
        straggler_applied = !strag.is_empty();
        // Count the straggler so the next consistency check matches and does NOT
        // fire an (expensive) bounded rebuild for an event we already applied.
        // The straggler is below the watermark, so it is never in the
        // `> watermark` load — no double count.  A rare double-trigger overcount
        // self-heals: the next check mismatches once and the rebuild resets
        // `applied_count = total`.
        cache.applied_count += strag.len() as i64;
    }

    // Events newer than what's applied.
    let mut new_events: Vec<crate::db::models::Event> = parse_event_rows(
        sqlx::query(&format!(
            "{ORCH_EVENT_COLS} WHERE execution_id = $1 AND event_id > $2 ORDER BY event_id ASC"
        ))
        .bind(execution_id)
        .bind(cache.last_event_id)
        .fetch_all(pool)
        .await?,
    );
    hydrate_result_references(&mut new_events, &result_store, state.config.refs_in_state).await;

    // Trigger type + drain timestamp.  The trigger event is normally among the
    // new events; after a cold rebuild / straggler apply that already folded it
    // in, fall back to a default type (and a None drain timestamp).
    let trigger_event_type = new_events
        .iter()
        .find(|e| e.event_id == trigger_event_id)
        .map(|e| e.event_type.clone())
        .unwrap_or_else(|| "command.completed".to_string());
    let latest_ts = new_events.last().map(|e| e.created_at);

    // Consistency.  With a fresh `total`, applying the new events should account
    // for ALL events; a shortfall means a straggler is below the watermark — and
    // crucially it may be a *non-triggering* one (the cursor claim's `call.done`
    // carrying the row batch), which would otherwise leave the cursor unable to
    // fan out and stop emitting events entirely.  A bounded rebuild re-scans the
    // recent window and folds it in.  This check runs whether or not there are
    // new events, so the reconcile poller (no new events, fresh count) still
    // catches a stuck execution.  Without a fresh `total` (throttled), trust the
    // incremental apply.
    let mismatch = matches!(total, Some(t) if cache.applied_count + new_events.len() as i64 != t);
    if mismatch {
        let r = rebuild_state(pool, &result_store, execution_id, state.config.refs_in_state).await?;
        cache.state = Some(r.state);
        cache.last_event_id = r.last_event_id;
        cache.applied_count = total.unwrap_or(cache.applied_count);
        cache.snapshot_version = r.snapshot_version;
        cache.routing_meta = r.routing_meta;
        did_rebuild = true;
    } else if !new_events.is_empty() {
        let ws = cache.state.as_mut().unwrap();
        for e in &new_events {
            ws.apply_event(e);
        }
        cache.applied_count += new_events.len() as i64;
        cache.last_event_id = new_events
            .last()
            .map(|e| e.event_id)
            .unwrap_or(cache.last_event_id);
    }

    // Nothing changed and this is not a forced reconcile → exit early.  A forced
    // reconcile still evaluates (to drive a cursor that may now advance).
    if new_events.is_empty() && !did_rebuild && !straggler_applied && !force_count {
        debug!(execution_id, "No new events to evaluate — orchestrator exit early");
        return Ok(0);
    }

    // Persist a snapshot once enough events have accrued since the last one,
    // gated on a *fresh* `total` confirming the state is fully consistent (no
    // outstanding straggler) — so the snapshot is a clean base for rebuilds.
    // Best-effort: a snapshot failure must not fail the trigger.
    const SNAPSHOT_INTERVAL_EVENTS: i64 = 500;
    if total == Some(cache.applied_count)
        && cache.last_event_id - cache.snapshot_version >= SNAPSHOT_INTERVAL_EVENTS
    {
        let version = cache.last_event_id;
        let applied = cache.applied_count;
        let routing = cache.routing_meta.clone();
        let saved = if let Some(ws) = cache.state.as_ref() {
            crate::services::orch_snapshot::save(
                pool,
                execution_id,
                version,
                applied,
                routing.as_ref(),
                ws,
            )
            .await
        } else {
            Ok(())
        };
        match saved {
            Ok(()) => cache.snapshot_version = version,
            Err(e) => {
                warn!(execution_id, %e, "orch_snapshot.save failed; continuing without snapshot")
            }
        }
    }

    // 2. Look up catalog_id (from the cached state) + playbook content.
    let catalog_id = cache.state.as_ref().map(|s| s.catalog_id).unwrap_or(0);
    if catalog_id == 0 {
        return Err(AppError::Internal(format!(
            "No catalog_id found 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 against the cached state.
    let orchestrator = WorkflowOrchestrator::new();
    let ws = cache.state.as_mut().unwrap();
    // References-in-state (noetl/ai-meta#101 phase 2): resolve any over-budget
    // cursor claim references into inline rows before the drive reads them — a
    // referenced claim has `claim_ref` set + empty `claim_rows`, and without
    // this the cursor would see 0 rows and wrongly DRAIN.  No-op unless the flag
    // kept a claim reference (flag-off claims are resolved inline by hydrate).
    resolve_cursor_claim_refs(ws, &result_store).await;
    let result = match orchestrator.evaluate_state(
        ws,
        latest_ts,
        &playbook,
        Some(trigger_event_type.as_str()),
    ) {
        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?;
    }

    // Recover the per-execution routing (dedicated pool segment + W3C trace)
    // the `/api/execute` caller supplied, persisted on the `playbook_started`
    // event meta, so every follow-up command lands on the same segment and
    // carries the same trace context (noetl/ai-meta#90 Phase 2).
    let routing = cache
        .routing_meta
        .as_ref()
        .map(crate::handlers::execute::CommandRouting::from_started_meta)
        .unwrap_or_default();

    // 5. Issue new commands — batched (noetl/ai-meta#102 step 1).  A cursor
    //    fan-out's N body commands now persist as two multi-row INSERTs (all
    //    `command.issued` events, then all `noetl.command` rows) instead of ~2N
    //    individual round-trips through PgBouncer to Cloud SQL — the write-path
    //    bottleneck on a small tier.  NATS publishes still loop (in-cluster).
    let commands_generated = crate::handlers::execute::persist_engine_commands_batch(
        state,
        execution_id,
        catalog_id,
        trigger_event_id,
        &result.commands,
        &playbook,
        &routing,
    )
    .await?;

    // 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"
        );
        // Free the cached state for a terminal execution (noetl/ai-meta#100).
        // Drop the guard first so the slot's Arc refcount can fall to zero.
        drop(cache);
        state.orch_cache.evict(execution_id);
    }

    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}"
        );
    }
}