noetl-server 2.8.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;

/// Worker event request.
///
/// R-1.2 PR-EE-2: aligned with the executor's `ExecutorEvent`
/// shape per the cross-repo event envelope reconciliation tracked
/// on noetl/ai-meta#30.  The legacy `name` field is renamed to
/// `event_type` with a `#[serde(alias = "name")]` so existing
/// worker / CLI clients sending `name` continue to deserialize
/// without breakage during the migration window.  The new
/// optional fields (`event_id`, `status`, `created_at`) align
/// with `noetl_executor::events::ExecutorEvent 0.3.1`; producers
/// that don't populate them get sensible server-side fallbacks
/// (DB-side snowflake_id() for `event_id`, name-derived `status`,
/// `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.
    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).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    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
}

/// 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.
    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.  R-1.2 PR-EE-2: prefer the
    // application-side snowflake from the request when present
    // (per `agents/rules/observability.md` Principle 3); fall
    // back to the server-side `noetl.snowflake_id()` function
    // when omitted.
    let event_id: i64 = match request.event_id.as_deref() {
        Some(raw) => raw.parse().map_err(|_| {
            AppError::Validation(format!("Invalid event_id: {raw}"))
        })?,
        None => generate_snowflake_id(&state).await?,
    };

    // 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.db)
    .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 on command.completed
    if request.event_type == "command.completed" && request.step.to_lowercase() != "end" {
        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);

    let row: Option<(i64, String, String, serde_json::Value, serde_json::Value)> =
        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(&state.db)
        .await?;

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

    let mut tx = state.db.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 = generate_snowflake_id_with_tx(&mut tx).await?;
    // 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?;
    let mut tx = state.db.begin().await?;
    let mut event_ids = Vec::with_capacity(request.events.len());

    for item in &request.events {
        // Batch path uses server-side snowflake (app-side event_id
        // per item isn't carried in BatchEventItem yet; left as a
        // follow-up if batch becomes the worker's primary path).
        let event_id = generate_snowflake_id_with_tx(&mut tx).await?;
        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
    // whose step is not the playbook's terminal `end` block.  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" && item.step.to_lowercase() != "end" {
            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.db)
        .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.db)
    .await?;

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

/// Generate a snowflake ID.
async fn generate_snowflake_id(state: &AppState) -> AppResult<i64> {
    let row: (i64,) = sqlx::query_as::<_, (i64,)>("SELECT noetl.snowflake_id()")
        .fetch_one(&state.db)
        .await?;

    Ok(row.0)
}

/// Generate a snowflake ID using an existing transaction.
async fn generate_snowflake_id_with_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
) -> AppResult<i64> {
    let row: (i64,) = sqlx::query_as::<_, (i64,)>("SELECT noetl.snowflake_id()")
        .fetch_one(&mut **tx)
        .await?;
    Ok(row.0)
}

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

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

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

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

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

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

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

    let playbook_yaml: String = sqlx::query_scalar(
        "SELECT content FROM noetl.catalog WHERE catalog_id = $1",
    )
    .bind(catalog_id)
    .fetch_one(&state.db)
    .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.
    let orchestrator = WorkflowOrchestrator::new();
    let result = orchestrator
        .evaluate(&events, &playbook, Some("command.completed"))
        .map_err(|e| AppError::Internal(format!("Orchestrator evaluate failed: {e}")))?;

    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 = generate_snowflake_id(state).await?;
        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.db)
        .await?;
    }

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

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

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

    // 6. Emit terminal playbook event when the orchestrator says so.
    if result.should_complete {
        let (event_type, status) = match &result.completion_status {
            Some(cs) if cs.status == "FAILED" => ("playbook.failed", "FAILED"),
            _ => ("playbook.completed", "COMPLETED"),
        };
        let event_id = generate_snowflake_id(state).await?;
        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.db)
        .await?;
        info!(
            execution_id,
            terminal_event = %event_type,
            "Orchestrator marked execution as terminal"
        );
    }

    Ok(commands_generated)
}

#[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());
    }

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