zebflow 0.11.14

//! Pipeline domain model — the complete type system for graph execution, node contracts, and
//! runtime context.
//!
//! # Overview
//!
//! A **pipeline** is a directed graph of **nodes** connected by **edges**.
//! Each edge links one node's output pin to another node's input pin.  A node runs
//! once, when every edge into it has delivered a **payload** (a `serde_json::Value`) or
//! been skipped (`docs/contracts/node-conventions.md` §4). The only cycle allowed is the
//! edge from a `logic.retry`'s `retry` pin back to the node it retries; any other is
//! refused at activation.
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────────┐
//! │  PipelineGraph                                                  │
//! │                                                                 │
//! │  ┌───────────────────┐    edge: n0/out → n1/in                 │
//! │  │ PipelineNode (n0) │ ──────────────────────────────────────► │
//! │  │  kind: n.trigger  │                                         │
//! │  │  config: { path } │    ┌─────────────────────┐             │
//! │  └───────────────────┘    │ PipelineNode (n1)   │             │
//! │                           │  kind: postgres.query.run   │             │
//! │                           │  config: { query }  │             │
//! │                           └──────────┬──────────┘             │
//! │                                      │ edge: n1/out → n2/in   │
//! │                                      ▼                        │
//! │                           ┌─────────────────────┐             │
//! │                           │ PipelineNode (n2)   │             │
//! │                           │  kind: web.response.send│             │
//! │                           │  config: { tmpl }   │             │
//! │                           └─────────────────────┘             │
//! └─────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Three Separate Concerns
//!
//! This module contains three groups of types that must not be confused:
//!
//! | Group | Types | Purpose |
//! |-------|-------|---------|
//! | **Graph** | [`PipelineGraph`], [`PipelineNode`], [`PipelineEdge`] | Persisted pipeline definition — what gets saved to `*.zf.json` and loaded at runtime |
//! | **Definition** | [`NodeDefinition`], [`NodeScriptBridge`], [`NodeAiToolDefinition`] | Static contract describing what a node kind *is* — its schemas, pins, and capabilities |
//! | **Runtime** | [`PipelineContext`], [`PipelineOutput`], [`PipelineError`], [`ExecuteOptions`] | Values produced and consumed during a single pipeline execution run |
//!
//! # Node Definition vs Node Instance
//!
//! **[`NodeDefinition`]** is the *kind-level* contract — it is the same for every instance of
//! `postgres.query.run` across all pipelines. It is declared once per node kind via a `definition()`
//! function in each node module (e.g. `pg_query::definition()`), registered into a catalog,
//! and served at `/docs/node` via [`NodeContractDocument`].
//!
//! **[`PipelineNode`]** is a *graph-level instance* — a specific occurrence of a node kind
//! inside one pipeline, identified by a unique `id`, carrying its own `config` blob.  Many
//! `PipelineNode` instances can share the same `kind` string but have different configs.
//!
//! ```text
//!                  NodeDefinition (kind = "postgres.query.run")
//!                  ├── config_schema: { credential_id: required, query: required, ... }
//!                  ├── input_schema:  { type: object }
//!                  ├── output_schema: { query: { rows, columns, row_count, … } }
//!                  ├── input_pins:    ["in"]
//!                  └── output_pins:  ["out"]
//!
//!                           ↑ validates config against config_schema at register time
//!
//!   PipelineNode { id: "n1", kind: "postgres.query.run", config: { credential_id: "pg-main", query: "SELECT ..." } }
//!   PipelineNode { id: "n4", kind: "postgres.query.run", config: { credential_id: "pg-replica", query: "SELECT ..." } }
//! ```
//!
//! # Config Schema
//!
//! [`NodeDefinition::config_schema`] is the single source of truth for what static
//! configuration a node kind accepts.  It is a JSON Schema object derived from each node's
//! typed `Config` struct via `schemars::schema_for!` (available through the `rmcp::schemars`
//! re-export).
//!
//! Adding `#[derive(JsonSchema)]` to a node's `Config` struct and calling
//! `serde_json::to_value(schemars::schema_for!(Config))` in its `definition()` function
//! provides:
//!
//! 1. **Validation at register time** — `config` blobs on incoming [`PipelineNode`]s can be
//!    checked against the schema before a pipeline is saved to disk.  Bad config fails fast
//!    rather than at execution time.
//! 2. **UI dialog auto-generation** — the pipeline editor reads `config_schema.properties`
//!    to render the correct field types, labels, and required markers without hardcoding them
//!    in TypeScript.
//! 3. **DSL flag derivation** — the DSL parser can map `--flag` names to `config` JSON keys
//!    by reading schema property names, eliminating hardcoded mapping tables.
//! 4. **API docs** — `config_schema` is included in [`NodeContractItem`] and surfaced at
//!    `/docs/node` so external tooling can discover node capabilities programmatically.
//!
//! # Pin System
//!
//! Nodes communicate through named **pins**.  An output pin on one node connects to an input
//! pin on another via a [`PipelineEdge`].  The flowing value between pins is a
//! `serde_json::Value` (the **payload**).
//!
//! ```text
//!   NodeA { output_pins: ["out", "error"] }
//!       │
//!       │  PipelineEdge { from_node: "A", from_pin: "out", to_node: "B", to_pin: "in" }
//!       ▼
//!   NodeB { input_pins: ["in"] }
//! ```
//!
//! Pin names are declared in [`NodeDefinition`] (the kind-level contract) and instantiated in
//! [`PipelineNode`] (the graph instance).  Trigger nodes have no input pins.  Terminal nodes
//! need not have output pins — their payload becomes the HTTP response.
//!
//! # Authoring a New Node
//!
//! Every node module must expose:
//!
//! 1. A `pub fn definition() -> NodeDefinition` that populates all contract fields.
//! 2. A `pub struct Config` with `#[derive(Serialize, Deserialize, JsonSchema)]` documenting
//!    accepted config fields via doc comments.
//! 3. A `pub struct Node` that implements [`crate::pipeline::nodes::NodeHandler`].
//! 4. Register the `definition()` call in the family list,
//!    `pipeline/nodes/basic/<family>/mod.rs → definitions()`, which
//!    `basic/mod.rs → builtin_node_definitions()` reads.
//!
//! ```rust,ignore
//! // basic/<family>/my_node.rs
//! use schemars::JsonSchema;
//! use serde::{Deserialize, Serialize};
//!
//! #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
//! pub struct Config {
//!     /// Required: the thing the node needs.
//!     pub my_field: String,
//!     /// Optional override.
//!     #[serde(default)]
//!     pub optional_field: Option<String>,
//! }
//!
//! pub fn definition() -> NodeDefinition {
//!     NodeDefinition {
//!         kind: "example.thing.run".to_string(),
//!         title: "My Node".to_string(),
//!         description: "Does the thing.".to_string(),
//!         config_schema: serde_json::to_value(schemars::schema_for!(Config)).unwrap_or_default(),
//!         input_schema: serde_json::json!({ "type": "object" }),
//!         output_schema: serde_json::json!({ "type": "object", "properties": { "result": { "type": "string" } } }),
//!         input_pins: vec!["in".to_string()],
//!         output_pins: vec!["out".to_string()],
//!         ..Default::default()
//!     }
//! }
//! ```
//!
//! # Contract Document
//!
//! At startup, [`crate::pipeline::nodes::basic::builtin_node_definitions()`] returns all
//! [`NodeDefinition`]s.  The platform converts them to [`NodeContractItem`]s via
//! `From<NodeDefinition>` and assembles a [`NodeContractDocument`] served at `/docs/node`.
//! This endpoint is the programmatic API for everything that needs to know about node
//! capabilities: UI, DSL tooling, LLM assistant, external integrations.
//!
//! # Payload Flow
//!
//! During execution, the engine feeds an initial payload (from the HTTP request) into the
//! trigger node.  Each node receives an **input payload** (the output of the upstream node
//! that fired its input pin) and produces an **output payload** forwarded downstream.
//! [`NodeDefinition::input_schema`] and [`NodeDefinition::output_schema`] document these
//! payload shapes — they describe the *flowing data*, not the static config.
//!
//! The trigger node is special: it receives the envelope the ingress built and answers it
//! under one key, its source (`webhook: { body, query, … }`).  Downstream nodes add their
//! answers until a terminal node (e.g. `web.response.send`) produces the final HTTP response.

use std::fmt::{Display, Formatter};

use serde::{Deserialize, Serialize};
use serde_json::Value;

/// What a node kind can reach beyond the pipeline it runs in.
///
/// This is the vocabulary the package safety review speaks, and it is an enum
/// rather than a string because nobody types it. A native node declares its own
/// set here in Rust, so a misspelling is a build failure. A bundle-provided node
/// declares nothing at all: a composite's set is derived as the union of the
/// nodes its function pipelines compose, and a WASM node's is empty because
/// `wasm_host` instantiates every module with an empty import array and this
/// build links `wasmtime` without WASI -- the module gets linear memory and
/// arithmetic and no host function whatever. Declaration only becomes necessary
/// on the day the host grants imports, and on that day the import list would be
/// the declaration rather than a claim about one.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord, Hash)]
#[serde(rename_all = "snake_case")]
pub enum NodeCapability {
    /// Opens an outbound connection to a host the pipeline names.
    Network,
    /// Reads or writes files the project owns.
    Filesystem,
    /// Reads or writes a store the project configured.
    Database,
    /// Reads a secret out of the credential service.
    Credential,
    /// Executes code or a program it was handed.
    Process,
}

impl NodeCapability {
    /// Every capability, in declaration order.
    pub const ALL: &'static [Self] = &[
        Self::Network,
        Self::Filesystem,
        Self::Database,
        Self::Credential,
        Self::Process,
    ];

    /// Stable wire name, matching the serde representation.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Network => "network",
            Self::Filesystem => "filesystem",
            Self::Database => "database",
            Self::Credential => "credential",
            Self::Process => "process",
        }
    }
}

impl Display for NodeCapability {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.as_str())
    }
}

/// How a DSL flag value is interpreted by the parser.
///
/// The universal transformation rule `--flag-name` → `flag_name` (replace `-` with `_`)
/// applies to ALL flags automatically.  `DslFlagKind` controls only how the *value* is parsed.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(rename_all = "snake_case")]
pub enum DslFlagKind {
    /// Single string value consumed from the next token.
    ///
    /// `--template-path pages/blog-home` → `config["template_path"] = "pages/blog-home"`
    #[default]
    Scalar,
    /// A repeatable role (`node-conventions.md` §2): each occurrence is one
    /// value, a comma inside it included, so `--case a,b` is one case.
    /// `--from "posts.parquet as posts" --from "authors.csv as authors"`
    /// → `config["from"] = ["posts.parquet as posts","authors.csv as authors"]`
    RepeatedList,
    /// Presence flag — no value consumed; sets config key to `true`.
    ///
    /// `--batch` → `config["batch"] = true`
    Bool,
    /// Repeated `--flag key=value` occurrences collected into a JSON object.
    ///
    /// Each occurrence adds one entry. Split is on the **first** `=` only so values
    /// may themselves contain `=`.  Follows Docker/kubectl convention.
    ///
    /// `--claim player_id=$.player_id --claim roles=$.roles`
    /// → `config["claims"] = {"player_id": "$.player_id", "roles": "$.roles"}`
    KeyValuePairs,
    /// Repeated schema-field declarations collected into a JSON Schema object.
    ///
    /// Each occurrence consumes a compact field spec plus an optional description token:
    /// `--input columns:string[]! "Source column names."`
    /// → `config["input_schema"] = {"type":"object","required":["columns"],"properties":{...}}`.
    SchemaField,
}

/// One DSL flag declaration owned by a [`NodeDefinition`].
///
/// Each node declares the DSL flags it accepts.  The parser uses these as the
/// **single source of truth** for flag→config mapping:
///
/// - **Parsing**: `config_key` is the authoritative destination key. Undeclared flags are a parse error.
/// - **Value coercion**: `kind` controls how the value is interpreted (scalar, list, bool).
/// - **Help text**: `pipeline --help web.response.send` lists all flags with descriptions.
/// - **Required validation**: flags with `required: true` must be present when registering.
/// - **LLM context**: surfaced in `/docs/node` so agents know exactly what flags to emit.
///
/// Every flag used in DSL MUST be declared here. No fallback, no auto-rule.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct DslFlag {
    /// CLI flag name including `--` prefix (e.g. `"--template-path"`).
    ///
    /// Convention: always `--kebab-case` matching the `snake_case` config key
    /// via the universal rule (replace `-` with `_`).
    pub flag: String,
    /// Config JSON key this flag writes to (e.g. `"template_path"`).
    ///
    /// Must match the corresponding field in the node's `Config` struct exactly.
    pub config_key: String,
    /// Human-readable description for help text and LLM context.
    ///
    /// Write this as a single sentence starting with a verb:
    /// "Path to the TSX page file relative to templates/."
    pub description: String,
    /// How the flag value is parsed.
    pub kind: DslFlagKind,
    /// Whether this flag must be present for the node to be valid.
    pub required: bool,
    /// What the value is (`node-conventions.md` §4): one of [`FLAG_VALUE_TYPES`].
    /// Empty while a node predates the 0.11 metadata; the save-time checks
    /// and the generated signature read it.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub value: String,
    /// The closed words a choice accepts (§3); empty for a free value.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub choices: Vec<String>,
    /// The most times a repeatable role may be given.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_repeat: Option<u32>,
    /// The value is a secret (a password, a key): a run record masks it in
    /// the node's recorded config, as `secret_paths` masks the answer.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub secret: bool,
}

/// The value types a flag may declare (`node-conventions.md` §4). `file` takes
/// any FileRef or store key; `file:<kind>` only that FileRef kind.
pub const FLAG_VALUE_TYPES: &[&str] = &[
    "text", "number", "duration", "size", "json", "expression", "ref", "file", "file:image",
    "file:video", "file:audio", "file:pdf", "file:archive", "file:geojson", "file:csv",
    "file:parquet", "file:json", "process",
];

// ── Engine-level common flags ─────────────────────────────────────────────────
//
// These are automatically injected into every node definition by
// `builtin_node_definitions()`.  They are consumed by the engine, not the node.

/// DSL flags available on **every** node kind, consumed by the engine.
pub fn engine_common_dsl_flags() -> Vec<DslFlag> {
    vec![
        DslFlag {
            flag: "--title".to_string(),
            config_key: "title".to_string(),
            description: "Custom display title for this node instance. \
                Overrides the default node kind title shown on the canvas."
                .to_string(),
            kind: DslFlagKind::Scalar,
            required: false,
            ..Default::default()
        },
        DslFlag {
            flag: "--timeout".to_string(),
            config_key: NODE_TIMEOUT_KEY.to_string(),
            description: "How long this node may run, e.g. 30s, 2m or 1h (1s to 1h). \
                Omitted: the project's node timeout."
                .to_string(),
            kind: DslFlagKind::Scalar,
            required: false,
            value: "duration".to_string(),
            ..Default::default()
        },
        DslFlag {
            flag: "--preview".to_string(),
            config_key: "preview.out".to_string(),
            description: "Render this node's output under its box on the canvas: \
                `<as>[:<path>]` where as is image|video|audio|pdf|json|text|table|html \
                and path is a dot path into the payload. `off` removes it. \
                Presentation only — never read by the engine."
                .to_string(),
            kind: DslFlagKind::Scalar,
            required: false,
            ..Default::default()
        },
        DslFlag {
            flag: "--preview-in".to_string(),
            config_key: "preview.in".to_string(),
            description: "Render this node's input under its box on the canvas: \
                `<as>[:<path>]`, same kinds and path rule as `--preview`. \
                `off` removes it. Presentation only — never read by the engine."
                .to_string(),
            kind: DslFlagKind::Scalar,
            required: false,
            ..Default::default()
        },
    ]
}

/// The config key `--timeout` writes: how long one node may run.
pub const NODE_TIMEOUT_KEY: &str = "timeout";
/// The shortest `--timeout` a node takes.
pub const MIN_NODE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(1);
/// The longest `--timeout` a node takes (`node-conventions.md` §7).
pub const MAX_NODE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3600);
/// A `--timeout` that is not a duration, or is outside 1s–1h.
pub const NODE_TIMEOUT_CONFIG_CODE: &str = "FW_NODE_TIMEOUT_CONFIG";

/// A node's `--timeout`, read from its resolved config: `None` when it is not
/// given (the project's node timeout applies). A value that is not a
/// duration, or is under 1s or over 1h, is refused — never clamped into
/// range, so a typo cannot quietly become a different limit.
pub fn node_timeout(config: &serde_json::Value) -> Result<Option<std::time::Duration>, PipelineError> {
    let refuse = |shown: &str| {
        PipelineError::new(
            NODE_TIMEOUT_CONFIG_CODE,
            format!("--timeout '{shown}' must be a duration from 1s to 1h, such as 30s, 2m or 1h"),
        )
    };
    let text = match config.get(NODE_TIMEOUT_KEY) {
        None | Some(serde_json::Value::Null) => return Ok(None),
        Some(serde_json::Value::String(text)) if text.trim().is_empty() => return Ok(None),
        Some(serde_json::Value::String(text)) => text.trim().to_string(),
        Some(other) => return Err(refuse(&other.to_string())),
    };
    let duration = crate::pipeline::nodes::shared::units::duration(&text, "--timeout", NODE_TIMEOUT_CONFIG_CODE)
        .map_err(|_| refuse(&text))?;
    if duration < MIN_NODE_TIMEOUT || duration > MAX_NODE_TIMEOUT {
        return Err(refuse(&text));
    }
    Ok(Some(duration))
}

/// UI fields for engine-level common config, injected into every node definition.
pub fn engine_common_fields() -> Vec<NodeFieldDef> {
    vec![NodeFieldDef {
        name: NODE_TIMEOUT_KEY.to_string(),
        label: "Timeout".to_string(),
        field_type: NodeFieldType::Text,
        help: Some(
            "How long this node may run, e.g. 30s, 2m or 1h (1s to 1h). Empty: the project's node timeout."
                .to_string(),
        ),
        placeholder: Some("30s".to_string()),
        ..Default::default()
    }]
}

// ── Node field definitions ────────────────────────────────────────────────────
//
// These types describe the **UI form fields** for each node kind.  Every node
// declares its own `fields: Vec<NodeFieldDef>` inside `definition()`.  The
// pipeline editor reads these from the `/docs/node` API response and renders a
// generic config form — no TypeScript changes needed when a new node is added.

/// What kind of UI widget a node field uses.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum NodeFieldType {
    Text,
    Textarea,
    CodeEditor,
    Select,
    Datalist,
    MethodButtons,
    CopyUrl,
    Checkbox,
    Section,
    /// Multi-select checkbox list. Options come from `data_source` (live) or `options` (static).
    /// Config value is stored as `Vec<String>`.
    MultiCheckbox,
    /// Repeating key-value pair rows: [key input][value input][×] with [+ Add] button.
    /// Config value is stored as a JSON object (`Map<String, String>`).
    KeyValuePairs,
    /// Specialized claims editor for `auth.token.create`.
    /// Same layout as `KeyValuePairs` but each row has a "Public" checkbox
    /// that adds/removes `:public` at the end of the claim's name.
    /// Claims marked public are exposed in the browser via `ctx.auth`;
    /// unmarked claims stay server-only.
    ClaimsPairs,
    /// Function parameter definition builder for `trigger.function`.
    /// Each row has: param name | type dropdown | description text | remove button.
    /// Config value stored as JSON object: `{"name": {"type": "string", "description": "..."}}`
    ParamsBuilder,
    /// Numeric input (`<input type="number">`).
    /// Config value is stored as a JSON number.
    /// Supports optional `min`, `max`, `step` extra config.
    Number,
    /// Binding editor for `secure_request` HTTP credential profiles.
    /// Shows the required variable slots declared by the selected credential and stores
    /// a JSON object mapping variable name → JS expression string.
    SecureRequestBindings,
    /// Specialized route editor for `logic.match`.
    /// Config value is normalized into `cases` plus `default` route metadata.
    MatchCases,
    /// Reusable source binding table.
    /// Each row has: source expression/path | alias | remove button.
    /// Config value is stored as an array of objects:
    /// `[{"source":"datasets/posts.csv","alias":"posts"}]`.
    SourceBindings,
}

impl Default for NodeFieldType {
    fn default() -> Self {
        Self::Text
    }
}

/// Where a select/datalist field's options come from at runtime.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum NodeFieldDataSource {
    /// All live project credentials regardless of kind.
    CredentialsAll,
    /// Live project Postgres credentials.
    CredentialsPostgres,
    /// Live project JWT credentials.
    CredentialsJwt,
    /// Live project page templates.
    TemplatesPages,
    /// Live project browser credentials (kind prefix: browser_).
    CredentialsBrowser,
    /// Live project OpenAI-family credentials (kind = "openai" or "openrouter").
    CredentialsOpenAi,
    /// Live `secure_request` credentials that declare request variable slots.
    CredentialsSecureRequest,
    /// All node kinds registered as AI tools (`ai_tool.registered = true` in NodeDefinition).
    /// Returns `{ tool_name, tool_description }` pairs for checklist rendering.
    AiTools,
    /// Active function pipelines (trigger_kind = "function") in the current project.
    /// Returns pipeline slugs for datalist rendering in `function.result.call`.
    FunctionPipelines,
    /// Roles registered in the JWT credential selected in the sibling `auth_credential` field.
    /// Reads `auth_roles` from that credential's list item. Used by `trigger.webhook`.
    CredentialJwtRoles,
    /// Credentials usable as HTTP auth: `secure_request` and `oauth2` kinds.
    /// Used by `http.response.fetch` credential_id field.
    CredentialsHttpAuth,
    /// Credentials usable as webhook auth: `jwt_signing_key`, `hmac`, and `api_key` kinds.
    /// Used by `trigger.webhook` auth_credential field.
    CredentialsWebhookAuth,
    /// Credentials filtered by a specific kind string (e.g. `"credentials:telegram_bot"`).
    /// Used by composite node packages that define custom credential types.
    CredentialsByKind(String),
}

impl serde::Serialize for NodeFieldDataSource {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        match self {
            Self::CredentialsAll => serializer.serialize_str("credentials_all"),
            Self::CredentialsPostgres => serializer.serialize_str("credentials_postgres"),
            Self::CredentialsJwt => serializer.serialize_str("credentials_jwt"),
            Self::TemplatesPages => serializer.serialize_str("templates_pages"),
            Self::CredentialsBrowser => serializer.serialize_str("credentials_browser"),
            Self::CredentialsOpenAi => serializer.serialize_str("credentials_open_ai"),
            Self::CredentialsSecureRequest => {
                serializer.serialize_str("credentials_secure_request")
            }
            Self::AiTools => serializer.serialize_str("ai_tools"),
            Self::FunctionPipelines => serializer.serialize_str("function_pipelines"),
            Self::CredentialJwtRoles => serializer.serialize_str("credential_jwt_roles"),
            Self::CredentialsHttpAuth => serializer.serialize_str("credentials_http_auth"),
            Self::CredentialsWebhookAuth => serializer.serialize_str("credentials_webhook_auth"),
            Self::CredentialsByKind(kind) => {
                serializer.serialize_str(&format!("credentials:{}", kind))
            }
        }
    }
}

impl<'de> serde::Deserialize<'de> for NodeFieldDataSource {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let s = String::deserialize(deserializer)?;
        match s.as_str() {
            "credentials_all" => Ok(Self::CredentialsAll),
            "credentials_postgres" => Ok(Self::CredentialsPostgres),
            "credentials_jwt" => Ok(Self::CredentialsJwt),
            "templates_pages" => Ok(Self::TemplatesPages),
            "credentials_browser" => Ok(Self::CredentialsBrowser),
            "credentials_open_ai" => Ok(Self::CredentialsOpenAi),
            "credentials_secure_request" => Ok(Self::CredentialsSecureRequest),
            "ai_tools" => Ok(Self::AiTools),
            "function_pipelines" => Ok(Self::FunctionPipelines),
            "credential_jwt_roles" => Ok(Self::CredentialJwtRoles),
            "credentials_http_auth" => Ok(Self::CredentialsHttpAuth),
            "credentials_webhook_auth" => Ok(Self::CredentialsWebhookAuth),
            other => {
                if let Some(kind) = other.strip_prefix("credentials:") {
                    Ok(Self::CredentialsByKind(kind.to_string()))
                } else {
                    Err(serde::de::Error::unknown_variant(
                        other,
                        &[
                            "credentials_all",
                            "credentials_postgres",
                            "credentials_jwt",
                            "templates_pages",
                            "credentials_browser",
                            "credentials_open_ai",
                            "credentials_secure_request",
                            "ai_tools",
                            "function_pipelines",
                            "credential_jwt_roles",
                            "credentials_http_auth",
                            "credentials_webhook_auth",
                            "credentials:<kind>",
                        ],
                    ))
                }
            }
        }
    }
}

/// One option in a `select`, `datalist`, or `method_buttons` field.
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct SelectOptionDef {
    pub value: String,
    pub label: String,
}

/// One item in a code-editor sidebar section.
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct SidebarItem {
    pub label: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub type_hint: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
}

/// A collapsible group of items shown in the code-editor sidebar.
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct SidebarSection {
    pub title: String,
    pub items: Vec<SidebarItem>,
}

/// Describes one form field in a node's config dialog.
///
/// Nodes declare `fields: Vec<NodeFieldDef>` in their `definition()`.
/// The pipeline editor reads this from `/docs/node` and renders the form
/// generically — no frontend changes are needed when a new node is added.
#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct NodeFieldDef {
    /// Config key (also used as the HTML field name).
    pub name: String,
    /// Human-readable label shown above the field.
    pub label: String,
    /// Widget type.
    #[serde(rename = "type", default)]
    pub field_type: NodeFieldType,
    /// Optional help text shown below the field.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub help: Option<String>,
    /// Input placeholder text.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub placeholder: Option<String>,
    /// When true the field is read-only (display only).
    #[serde(default)]
    pub readonly: bool,
    /// Textarea height hint (number of rows).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub rows: Option<u32>,
    /// Code-editor language (e.g. `"javascript"`, `"sql"`, `"json"`).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub language: Option<String>,
    /// Static options for `select`, `datalist`, and `method_buttons`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub options: Vec<SelectOptionDef>,
    /// Dynamic options resolved from live project data at render time.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub data_source: Option<NodeFieldDataSource>,
    /// Default value used when the config key is absent.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub default_value: Option<serde_json::Value>,
    /// Sidebar sections shown next to a `code_editor` field.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub sidebar: Vec<SidebarSection>,
    /// Grid span override: `"full"` (col-span-2) or `"half"` (col-span-1).
    /// Omit to use the default based on field type.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub span: Option<String>,
}

/// One item in a node's dialog layout tree.
///
/// `Field(name)` references a `NodeFieldDef` by its `name` key.
/// `Row` renders children horizontally (equal-width flex columns).
/// `Col` renders children vertically (nested stack inside a row cell).
///
/// Top-level `NodeDefinition::layout` is an implicit vertical stack.
/// Serializes as untagged: `"field_name"`, `{"row":[...]}`, `{"col":[...]}`.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(untagged)]
pub enum LayoutItem {
    Field(String),
    Row { row: Vec<LayoutItem> },
    Col { col: Vec<LayoutItem> },
}

// ── Pipeline graph ─────────────────────────────────────────────────────────────

/// A complete runtime pipeline graph ready for execution.
///
/// The permanent repository schema is owned by
/// [`crate::contracts::kinds::PipelineSpec`]. Contract decoding converts that
/// stable source model into this runtime model so engine refactoring cannot
/// silently change `.zf.json`.
///
/// # Execution entry point
///
/// The engine starts traversal from [`entry_nodes`](PipelineGraph::entry_nodes).  If the
/// list is empty, the engine falls back to nodes that have no incoming edges (i.e. nodes
/// with no input pins — trigger nodes).
///
/// # JSON shape
///
/// ```json
/// {
///   "apiVersion": "zebflow.com/v1",
///   "kind": "Pipeline",
///   "metadata": { "name": "blog-home" },
///   "spec": {
///     "id": "blog-home",
///     "nodes": [ ... ],
///     "edges": [ ... ]
///   }
/// }
/// ```
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PipelineGraph {
    /// Unique pipeline id — matches the filename stem of the `.zf.json` file.
    pub id: String,
    /// Optional human-readable description of what this pipeline does.
    /// Stored in `.zf.json` and indexed in the catalog for search.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub description: Option<String>,
    /// Optional graph-level metadata stored alongside the pipeline source.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metadata: Option<PipelineGraphMetadata>,
    /// Explicit entry node ids.  If empty the engine uses nodes with no incoming edges.
    #[serde(default)]
    pub entry_nodes: Vec<String>,
    /// All nodes in this graph. Source order is stable but does not define
    /// execution order. The engine follows directed edges and supports cycles.
    pub nodes: Vec<PipelineNode>,
    /// All directed edges connecting output pins to input pins.
    pub edges: Vec<PipelineEdge>,
    /// Free-standing annotations drawn on the canvas.
    ///
    /// Deliberately a sibling of `nodes` rather than a node kind: a note has no
    /// pins, no config, and never runs, so keeping it out of `nodes` means no
    /// executor learns to skip it and no edge can point at one. `nodes` stays
    /// exactly what it was — the things that execute.
    ///
    /// They exist so a pipeline that arrives from the hub can explain itself:
    /// which credential to create, which node to change, what the flow does.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub notes: Vec<PipelineNote>,
}

impl PipelineGraph {
    /// The nodes a run starts from.
    ///
    /// `kinds/pipeline/README.md` freezes the default for an omitted
    /// `spec.entry_nodes`: "Runtime derives roots from graph connectivity."
    /// A root is a node no edge points at, and **every** root starts — a graph
    /// with two independent triggers is two roots, not one.
    ///
    /// The fallback is for a graph where every node has an incoming edge: it
    /// has no root by connectivity. Such a graph is a cycle, which activation
    /// refuses (`node-conventions.md` §4); source order names a node anyway,
    /// so a caller asking about an unvalidated draft still gets one.
    pub fn entry_node_ids(&self) -> Vec<&str> {
        if !self.entry_nodes.is_empty() {
            return self.entry_nodes.iter().map(String::as_str).collect();
        }
        let targets: std::collections::HashSet<&str> = self
            .edges
            .iter()
            .map(|edge| edge.to_node.as_str())
            .collect();
        let roots: Vec<&str> = self
            .nodes
            .iter()
            .map(|node| node.id.as_str())
            .filter(|id| !targets.contains(id))
            .collect();
        if roots.is_empty() {
            return self
                .nodes
                .first()
                .map(|node| vec![node.id.as_str()])
                .unwrap_or_default();
        }
        roots
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(deny_unknown_fields)]
pub struct PipelineGraphMetadata {
    #[serde(default)]
    pub locked: bool,
    #[serde(default)]
    pub settings: PipelineGraphSettings,
}

#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(deny_unknown_fields)]
pub struct PipelineGraphSettings {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub invocation_retention: Option<PipelineInvocationRetention>,
    /// Per-field overrides of project trace capture defaults. These bounds affect
    /// recorded node previews only; downstream execution receives complete data.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub trace_capture: Option<crate::pipeline::trace_capture::TraceCaptureSettings>,
}

#[derive(Debug, Clone, Serialize, Deserialize, Default)]
#[serde(deny_unknown_fields)]
pub struct PipelineInvocationRetention {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_invocations: Option<u32>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_age_secs: Option<u64>,
}

/// One node instance inside a [`PipelineGraph`].
///
/// This is the *instance* — not the kind-level contract.  Many nodes with the same `kind`
/// can exist in the same graph or across different graphs, each with different `config`.
///
/// # Config vs Definition
///
/// `config` is a raw `serde_json::Value` blob.  The engine deserializes it into the node
/// kind's typed `Config` struct at build/compile time.  The shape is defined by
/// [`NodeDefinition::config_schema`].
///
/// # Pins
///
/// `input_pins` and `output_pins` are stored per instance to allow the graph serialization
/// format to be self-describing.  They must be consistent with what
/// [`NodeDefinition::input_pins`] and [`NodeDefinition::output_pins`] declare for the kind.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PipelineNode {
    /// Unique node id within this graph (e.g. `"n0"`, `"trigger"`, `"render_blog"`).
    pub id: String,
    /// Kind identifier — must match a registered node kind (e.g. `"web.response.send"`).
    pub kind: String,
    /// Input pin names for this instance.
    #[serde(default)]
    pub input_pins: Vec<String>,
    /// Output pin names for this instance.
    #[serde(default)]
    pub output_pins: Vec<String>,
    /// Node-specific configuration blob.  Shape is defined by the kind's
    /// [`NodeDefinition::config_schema`].  Validated at register time; deserialized into
    /// a typed `Config` struct at engine build time.
    #[serde(default)]
    pub config: Value,
}

/// One annotation on the canvas. Presentation only — never executed, never
/// reachable by an edge, and carried through read and write untouched.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct PipelineNote {
    /// Unique id within this graph, so an editor can move one note without
    /// rewriting the rest.
    pub id: String,
    /// The note's body. Markdown, rendered by the editor.
    #[serde(default)]
    pub text: String,
    /// Canvas position, in the same space node positions use.
    #[serde(default)]
    pub x: f64,
    #[serde(default)]
    pub y: f64,
    /// Drawn size. An editor may grow a note to fit its text, but the stored
    /// size is what reopens.
    #[serde(default)]
    pub width: f64,
    #[serde(default)]
    pub height: f64,
    /// A colour name the editor maps to its palette, e.g. `"amber"`. Free text
    /// rather than an enum: a note nobody can read is a cosmetic problem, and
    /// refusing to open a pipeline over an unknown colour is not.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub color: String,
}

/// A directed connection from one node's output pin to another node's input pin.
///
/// Together, all edges in a [`PipelineGraph`] form a directed graph. The engine
/// follows edges to determine which node fires next and what payload it receives.
/// A cycle is refused at activation, unless it is a `logic.retry` `retry`
/// edge back to the node it retries.
///
/// # Example
///
/// ```json
/// { "from_node": "trigger", "from_pin": "out", "to_node": "query", "to_pin": "in" }
/// ```
///
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PipelineEdge {
    /// Source node id.
    pub from_node: String,
    /// Output pin name on the source node (e.g. `"out"`, `"true"`, `"error"`).
    pub from_pin: String,
    /// Target node id.
    pub to_node: String,
    /// Input pin name on the target node (almost always `"in"`).
    pub to_pin: String,
}

/// Reserved shape for a script bridge that does not exist yet.
///
/// **Nothing dispatches from a script back into a node handler.** The `n` object a
/// script receives is built by `build_capabilities_expr` in the Deno sandbox pool and
/// holds pure `time` and `math` helpers only; the sandbox's two host ops return a
/// result and read a project-local file, and `Deno.core` is hidden from user code.
/// There is no path from script source to a `NodeHandler`.
///
/// The field is kept because [`NodeDefinition`] is a frozen `zebflow.com/v1` contract
/// whose serialized shape cannot lose a field without a new API version. Every
/// definition in this repository declares `None`, and a reader must not take that as
/// a restriction being enforced — there is nothing to restrict.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
#[serde(deny_unknown_fields)]
pub struct NodeScriptBridge {
    /// Function name a bridge would expose, were one built (e.g. `"postgres.query.run"`).
    pub name: String,
    /// Reserved. No runtime reads this, because no bridge dispatches.
    #[serde(default)]
    pub enabled: bool,
}

/// Registers this node kind as a callable AI tool for LLM-based agents.
///
/// When `registered = true`, the assistant can invoke this node's capability by name
/// during an agentic pipeline run (e.g. Zebtune calling `postgres.query.run` to answer a question).
/// `tool_input_schema` is a JSON Schema object the LLM uses to form valid tool arguments.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct NodeAiToolDefinition {
    /// Whether this node is exposed as an AI tool.
    #[serde(default)]
    pub registered: bool,
    /// Stable tool name/id used by the LLM (e.g. `"query_database"`).
    #[serde(default)]
    pub tool_name: String,
    /// Human-readable description fed to the LLM to explain what the tool does.
    #[serde(default)]
    pub tool_description: String,
    /// JSON Schema for the tool's input arguments as seen by the LLM.
    #[serde(default)]
    pub tool_input_schema: Value,
}

/// The canonical contract for one node kind.
///
/// **One `NodeDefinition` per node kind** — not per instance.  Every node module exposes a
/// `pub fn definition() -> NodeDefinition` and registers it via
/// `builtin_node_definitions()`.
///
/// # Three Schemas
///
/// A `NodeDefinition` carries three independent JSON Schema objects.  Understanding the
/// difference is critical:
///
/// | Field | Describes | Populated by |
/// |-------|-----------|--------------|
/// | `config_schema` | Static fields in the node's `Config` struct | `schemars::schema_for!(Config)` |
/// | `input_schema`  | Payload shape this node *consumes* from upstream | Hand-written in `definition()` |
/// | `output_schema` | Payload shape this node *produces* downstream   | Hand-written in `definition()` |
///
/// `config_schema` is the authoritative source for what can be put in
/// [`PipelineNode::config`].  Downstream consumers (UI dialog renderer, DSL parser, register
/// validator) all read this schema rather than hardcoding field knowledge.
///
/// `input_schema` and `output_schema` document the *flowing payload* — the `serde_json::Value`
/// passed between nodes at runtime.  They are informational for now (not enforced at runtime)
/// but drive UI tooltips and LLM context.
///
/// # Pin Declarations
///
/// `input_pins` and `output_pins` list the pin names this kind exposes.  Trigger nodes have
/// empty `input_pins`.  Logic nodes like `logic.match` have dynamic `output_pins` (empty
/// here, populated per instance in the graph).
///
/// # Usage Dimensions
///
/// A node kind can be usable in two ways simultaneously:
/// 1. As a **graph node** — always true for any registered kind.
/// 2. As an **AI tool** — invokable by LLM agents (see [`NodeAiToolDefinition`]).
///
/// A third dimension, a script bridge, is declared by [`NodeScriptBridge`] and
/// implemented by nothing; read it before believing the declaration.
///
/// The [`NodeUsageMatrix`] in [`NodeContractItem`] carries all three declarations.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct NodeDefinition {
    /// Stable kind id — must be unique across all registered nodes.
    /// `family.noun.verb` (`node-conventions.md` §1, e.g. `postgres.query.run`, `web.response.send`).
    pub kind: String,
    /// Short display title for UI catalogs and tooltips (e.g. `"Postgres Query"`).
    pub title: String,
    /// Full behavior description for docs, UI dialogs, and LLM context.
    pub description: String,
    /// JSON Schema for the node's static `Config` struct.
    ///
    /// **Populate this via `schemars::schema_for!(Config)`** (available as
    /// `rmcp::schemars::schema_for!`).  Adding `#[derive(JsonSchema)]` to the `Config`
    /// struct and annotating fields with doc comments produces a schema with `required`,
    /// `properties`, `description`, and `type` automatically.
    ///
    /// Consumers:
    /// - **Validation**: `register_pipeline` checks each [`PipelineNode::config`] against
    ///   this schema before saving — fail fast before execution.
    /// - **UI**: The pipeline editor reads `properties` to render form fields without any
    ///   TypeScript-side hardcoding.
    /// - **DSL**: The parser maps `--flag` names to config keys using property names.
    /// - **API**: Exposed in [`NodeContractItem`] at `/docs/node`.
    ///
    /// Defaults to `Value::Null` (no schema declared yet) — treated as "accept anything".
    #[serde(default)]
    pub config_schema: Value,
    /// JSON Schema for the payload this node *reads* from the upstream output.
    ///
    /// Informational — not enforced at runtime.  Documents what fields the node expects
    /// to find in `input.payload` so pipeline authors know what upstream nodes must produce.
    #[serde(default)]
    pub input_schema: Value,
    /// JSON Schema for the payload this node *writes* to its output pins.
    ///
    /// Informational — not enforced at runtime.  Documents what fields downstream nodes
    /// can rely on after this node runs.
    #[serde(default)]
    pub output_schema: Value,
    /// Concrete examples used by generated documentation and tool callers.
    ///
    /// Examples are documentation, not fixtures executed during every pipeline run.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub examples: Vec<NodeExample>,
    /// Stable failure codes and their meaning.
    ///
    /// Runtime errors may include more context, but documented codes must keep the
    /// meaning declared here for the lifetime of this contract version.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub failure_semantics: Vec<NodeFailureSemantic>,
    /// Input pin names declared by this kind.  Empty for trigger nodes (no upstream).
    /// Almost always `["in"]` for processing nodes.
    #[serde(default)]
    pub input_pins: Vec<String>,
    /// Output pin names declared by this kind.  Common values: `["out"]`, `["out", "error"]`,
    /// `["true", "false"]`.  Empty for dynamic-pin nodes like `logic.match` — pins are
    /// defined per instance in the graph.
    #[serde(default)]
    pub output_pins: Vec<String>,
    /// What this kind can reach beyond the pipeline, declared once here.
    ///
    /// The package safety review reads this rather than guessing from the kind
    /// string, so a node that touches nothing outside its payload carries an
    /// empty set and says so. See [`NodeCapability`] for why no authoring
    /// surface accepts one of these as text.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub capabilities: Vec<NodeCapability>,
    /// Reserved, and honoured by nothing. See [`NodeScriptBridge`] for why: no code
    /// path reaches a node handler from inside a script, so this declares neither a
    /// permission nor a restriction. Every definition here declares `false`.
    #[serde(default)]
    pub script_available: bool,
    /// Reserved companion to `script_available`; both are declared together or not at
    /// all, which is all the contract validator checks.
    #[serde(default)]
    pub script_bridge: Option<NodeScriptBridge>,
    /// AI tool registration.  Only meaningful when `ai_tool.registered = true`.
    #[serde(default)]
    pub ai_tool: NodeAiToolDefinition,
    /// DSL flag declarations for this node kind.
    ///
    /// Each entry documents one `--flag-name` → `config_key` mapping.  The parser applies
    /// the universal rule (`-` → `_`) for any flag not listed here — declarations exist
    /// for help text, required-flag validation, and LLM context surfaced at `/docs/node`.
    ///
    /// # Convention
    ///
    /// Declare every user-facing flag.  Internal/platform-injected fields (e.g. `markup`,
    /// `route`) must NOT appear here — they are not settable by users or agents.
    #[serde(default)]
    pub dsl_flags: Vec<DslFlag>,
    /// UI field definitions for the node config dialog.
    ///
    /// Declares the form fields shown when a user clicks "Edit" on this node in the
    /// pipeline editor.  The editor reads these from `/docs/node` and renders a generic
    /// form — no TypeScript changes needed when a new node is added or modified.
    ///
    /// Every node prepends a `title` text field automatically.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub fields: Vec<NodeFieldDef>,
    /// Hierarchical layout tree for the config dialog.
    /// Empty = fall back to flat 2-column grid using `fields` order.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub layout: Vec<LayoutItem>,
    /// UI category path for grouping in the pipeline editor catalog.
    /// Dot-separated: `root.subcategory` (e.g. `"data.postgres"`, `"files.fs"`).
    /// First segment = root category button; second = subcategory header.
    /// Positions in this node's output that hold a secret regardless of value —
    /// `kinds/invocation-record` rule 3, by position.
    ///
    /// Rule 3 by *value* covers anything that came out of the credential store.
    /// This covers what it cannot: a secret the platform has never seen,
    /// arriving in someone else's response under a name no list holds. An OAuth
    /// token endpoint answering `access_token` is the motivating case, and a
    /// real authorization code reached a trace in full under the name `code`.
    ///
    /// JSON Pointer, plus `*` matching one object member or array element:
    ///
    ///   `/access_token`        one field
    ///   `/items/*/token`       that field in every element
    ///   `/users/*`             every value of a dynamically keyed map
    ///
    /// Masks the record only. What the next node receives is untouched —
    /// a level shapes the record, never the run.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub secret_paths: Vec<String>,
    /// The config key a bare first token fills in the DSL: `input.text prompt`
    /// is `input.text --name prompt`. The key must also be a declared flag, so
    /// the flag form always works and `graph_to_dsl` renders the bare form.
    /// `None` for every kind that has no positional.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub positional: Option<String>,
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub ui_category: String,
    /// Human-readable label for the leaf subcategory.
    /// If empty, the frontend title-cases the subcategory slug.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub ui_category_label: String,
    /// One profile per `--provider` word of a swappable task
    /// (`node-conventions.md` §11): what that provider takes beyond the
    /// kind's shared roles. Empty for a kind without `--provider`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub profiles: Vec<ProviderProfile>,
}

/// What one provider of a swappable task takes (`node-conventions.md` §11).
///
/// A flag named in some profile's `roles` is provider-specific: only the
/// providers listing it take it. Every other flag of the kind is shared.
/// A profile is added or extended, never changed.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct ProviderProfile {
    /// One word of the kind's `--provider` choices.
    pub provider: String,
    /// The credential kinds a `--credential` may name for this provider.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub credential_kinds: Vec<String>,
    /// The `--model` ids this provider takes; empty takes any.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub models: Vec<String>,
    /// The provider-specific roles it takes, by flag (`--image`).
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
    pub roles: std::collections::BTreeMap<String, ProfileRole>,
    /// A narrower word list for a choice flag of the kind, by flag.
    #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")]
    pub choices: std::collections::BTreeMap<String, Vec<String>>,
    /// The closed, typed `--option key=value` keys it takes.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub options: Vec<ProfileOption>,
}

/// How a provider takes one of the kind's provider-specific roles.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct ProfileRole {
    /// The provider needs it.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub required: bool,
    /// The most values a repeatable role takes from this provider.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_repeat: Option<u32>,
}

/// One `--option key=value` setting a provider takes.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct ProfileOption {
    pub key: String,
    /// What the value is: one of [`FLAG_VALUE_TYPES`].
    pub value: String,
    /// The closed words it takes; empty for a free value of its type.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub choices: Vec<String>,
    pub description: String,
}

/// One concrete node usage example.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct NodeExample {
    /// Short label shown in generated documentation.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub title: String,
    /// The node as it is written in a pipeline — one DSL segment, flags and
    /// body, without the leading `|`. This is the line a reader copies; the
    /// guard test parses it against the node's declared flags.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub dsl: String,
    /// Optional explanation of what the example demonstrates.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub description: String,
    /// Static node configuration for the example.
    #[serde(default, skip_serializing_if = "Value::is_null")]
    pub config: Value,
    /// Example upstream payload.
    #[serde(default, skip_serializing_if = "Value::is_null")]
    pub input: Value,
    /// Expected output payload.
    #[serde(default, skip_serializing_if = "Value::is_null")]
    pub output: Value,
}

impl NodeExample {
    /// The DSL line a reader copies — one node segment, no leading `|`.
    pub fn dsl(title: &str, dsl: &str) -> Self {
        Self { title: title.to_string(), dsl: dsl.to_string(), ..Default::default() }
    }

    /// The payload the node produces for this example.
    pub fn output(mut self, output: Value) -> Self {
        self.output = output;
        self
    }

    /// The payload the node received for this example.
    pub fn input(mut self, input: Value) -> Self {
        self.input = input;
        self
    }

    /// One sentence on what the example shows, when the DSL alone is not enough.
    pub fn note(mut self, description: &str) -> Self {
        self.description = description.to_string();
        self
    }
}

/// One documented runtime failure exposed by a node kind.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
#[serde(deny_unknown_fields)]
pub struct NodeFailureSemantic {
    /// Stable machine-readable error code.
    pub code: String,
    /// Human-readable explanation of the failure.
    pub description: String,
    /// Whether retrying unchanged input can reasonably succeed.
    #[serde(default)]
    pub retryable: bool,
    /// Optional actionable guidance for a retry or correction.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub retry_hint: String,
}

/// The unimplemented script bridge as reported in a [`NodeUsageMatrix`].
///
/// Every field is a restatement of the reserved [`NodeScriptBridge`] declaration, so
/// all three read `false`/empty for every node Zebflow ships. Nothing consumes this
/// view; it exists because `/docs/node` serializes the whole matrix.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct NodeScriptUsageContract {
    /// Mirror of `NodeDefinition::script_available`. Not a permission.
    pub available: bool,
    /// Function name a bridge would expose, were one built.
    #[serde(default)]
    pub bridge_name: String,
    /// Reserved. No bridge dispatches, so nothing is ever active.
    #[serde(default)]
    pub enabled: bool,
}

/// AI tool capability as exposed in a [`NodeUsageMatrix`].
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct NodeToolUsageContract {
    /// Whether this node is registered as an AI tool.
    pub registered: bool,
    /// Tool name/id.
    #[serde(default)]
    pub tool_name: String,
    /// Description fed to the LLM.
    #[serde(default)]
    pub tool_description: String,
    /// JSON Schema for tool arguments as seen by the LLM.
    #[serde(default)]
    pub tool_input_schema: Value,
}

/// Summary of where and how a node kind can be used.
///
/// A node is always usable as a pipeline graph node.  The matrix additionally records
/// whether it is callable from script and/or invokable by AI agents.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct NodeUsageMatrix {
    /// Always `true` — every registered kind is available as a graph node.
    pub pipeline_node: bool,
    /// Script bridge availability.
    pub script_usage: NodeScriptUsageContract,
    /// AI tool availability.
    pub tool_usage: NodeToolUsageContract,
}

/// One entry in the node contract document, ready for API serialization.
///
/// Produced from a [`NodeDefinition`] via `From<NodeDefinition> for NodeContractItem`.
/// Carries all three schemas plus the [`NodeUsageMatrix`].
///
/// Served inside [`NodeContractDocument`] at `GET /docs/node`.  External consumers
/// (pipeline editor UI, DSL tooling, LLM assistant, external integrations) use this
/// document to discover node capabilities without reading Rust source.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct NodeContractItem {
    /// Stable kind id.
    pub kind: String,
    /// Display title.
    pub title: String,
    /// Full behavior description.
    pub description: String,
    /// JSON Schema for the node's static config.  See [`NodeDefinition::config_schema`].
    #[serde(default)]
    pub config_schema: Value,
    /// JSON Schema for the consumed payload.
    #[serde(default)]
    pub input_schema: Value,
    /// JSON Schema for the produced payload.
    #[serde(default)]
    pub output_schema: Value,
    /// Concrete usage examples from the node definition.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub examples: Vec<NodeExample>,
    /// Stable documented failure codes from the node definition.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub failure_semantics: Vec<NodeFailureSemantic>,
    /// Declared input pin names.
    #[serde(default)]
    pub input_pins: Vec<String>,
    /// Declared output pin names.
    #[serde(default)]
    pub output_pins: Vec<String>,
    /// What this kind can reach beyond the pipeline.  Same as
    /// [`NodeDefinition::capabilities`].
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub capabilities: Vec<NodeCapability>,
    /// Where this node kind can be used.
    pub usage_matrix: NodeUsageMatrix,
    /// DSL flag declarations — same as [`NodeDefinition::dsl_flags`].
    ///
    /// Surfaced here so API consumers (UI, LLM, external tooling) can discover
    /// exactly what `--flags` a node accepts without reading Rust source.
    #[serde(default)]
    pub dsl_flags: Vec<DslFlag>,
    /// UI field definitions for the node config dialog.  Same as [`NodeDefinition::fields`].
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub fields: Vec<NodeFieldDef>,
    /// Hierarchical layout tree — passed through from [`NodeDefinition::layout`].
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub layout: Vec<LayoutItem>,
    /// UI category path (e.g. `"data.postgres"`). Same as [`NodeDefinition::ui_category`].
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub ui_category: String,
    /// Subcategory display label. Same as [`NodeDefinition::ui_category_label`].
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub ui_category_label: String,
    /// Unified icon URL for this node kind (resolved at API response time).
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub icon_url: String,
    /// Content hash of the icon SVG (first 8 hex chars of SHA-256) for cache-busting.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub icon_hash: String,
    /// Source of this node: `"native"`, `"composite"`, or `"wasm"`.
    #[serde(default = "default_node_source")]
    pub source: String,
    /// Node tier: `"official"` (native or platform-bundled) or `"community"` (installed per-project).
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub tier: String,
    /// Whether this node can actually run here.
    ///
    /// False when the definition came from an interface carried in `repo/nodes/`
    /// because the providing bundle is not installed. The node still renders
    /// with its real pins and fields; it just cannot execute.
    #[serde(default = "default_node_available")]
    pub available: bool,
    /// Credential requirements declared by the node package.
    ///
    /// Native nodes usually express credentials through `fields` and `dsl_flags`.
    /// Composite/WASM packages additionally declare credential kinds and placeholder
    /// mappings in their package manifest; surfacing that here lets agents and UIs
    /// see which project credential kind a composite node needs.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub credential_requirements: Vec<NodeCredentialRequirement>,
}

fn default_node_available() -> bool {
    true
}

fn default_node_source() -> String {
    "native".to_string()
}

/// Credential requirement metadata attached to a node contract.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct NodeCredentialRequirement {
    /// Required project credential kind, e.g. `"openai"` or `"telegram_bot"`.
    pub kind: String,
    /// Human-readable credential type title.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub title: String,
    /// Short credential description.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub description: String,
    /// Node config key that stores the selected credential id, usually `"credential_id"`.
    #[serde(default, skip_serializing_if = "String::is_empty")]
    pub config_key: String,
    /// Whether this credential is required for normal node execution.
    #[serde(default)]
    pub required: bool,
    /// Placeholder names populated from the credential into composite inner pipelines.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub placeholder_names: Vec<String>,
}

/// Root response envelope for `GET /docs/node`.
///
/// Contains all registered node contracts in a single document.  Clients should check
/// `ok = true` before processing `items`.  `schema_version` can be used to detect
/// breaking changes to the contract format.
///
/// # Assembling the document
///
/// ```rust,ignore
/// let items: Vec<NodeContractItem> = builtin_node_definitions()
///     .into_iter()
///     .map(NodeContractItem::from)
///     .collect();
/// let doc = NodeContractDocument {
///     ok: true,
///     schema_version: "1",
///     source: "builtin",
///     items,
/// };
/// ```
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
pub struct NodeContractDocument {
    /// `true` when the document was assembled successfully.
    pub ok: bool,
    /// Monotonic version of the contract schema format.  Increment on breaking changes.
    pub schema_version: &'static str,
    /// Source tag for traceability (e.g. `"builtin"`).
    pub source: &'static str,
    /// All node contract entries, sorted by kind.
    #[serde(default)]
    pub items: Vec<NodeContractItem>,
}

/// Converts a kind-level [`NodeDefinition`] into an API-facing [`NodeContractItem`].
///
/// This is the main path for building the `/docs/node` response.  It flattens the
/// `script_bridge` and `ai_tool` metadata into the flat [`NodeUsageMatrix`] structure.
impl From<NodeDefinition> for NodeContractItem {
    fn from(value: NodeDefinition) -> Self {
        let (bridge_name, bridge_enabled) = value
            .script_bridge
            .as_ref()
            .map(|bridge| (bridge.name.clone(), bridge.enabled))
            .unwrap_or_else(|| (String::new(), false));
        // A definition is implementation-neutral, so it cannot know how a node
        // is built. Bundle-provided nodes get their real source from the
        // package manifest when the API enriches this item.
        let source = default_node_source();
        let ui_category = value.ui_category;
        let ui_category_label = value.ui_category_label;
        Self {
            // A definition on its own is always runnable; the registry marks it
            // unavailable when it came from an interface with no bundle behind it.
            available: true,
            kind: value.kind,
            title: value.title,
            description: value.description,
            config_schema: value.config_schema,
            input_schema: value.input_schema,
            output_schema: value.output_schema,
            examples: value.examples,
            failure_semantics: value.failure_semantics,
            input_pins: value.input_pins,
            output_pins: value.output_pins,
            capabilities: value.capabilities,
            dsl_flags: value.dsl_flags,
            fields: value.fields,
            layout: value.layout,
            usage_matrix: NodeUsageMatrix {
                pipeline_node: true,
                script_usage: NodeScriptUsageContract {
                    available: value.script_available,
                    bridge_name,
                    enabled: bridge_enabled,
                },
                tool_usage: NodeToolUsageContract {
                    registered: value.ai_tool.registered,
                    tool_name: value.ai_tool.tool_name,
                    tool_description: value.ai_tool.tool_description,
                    tool_input_schema: value.ai_tool.tool_input_schema,
                },
            },
            ui_category,
            ui_category_label,
            icon_url: String::new(),
            icon_hash: String::new(),
            source,
            tier: String::new(),
            credential_requirements: Vec::new(),
        }
    }
}

/// Execution-scoped signal emitted by nodes or the engine during a pipeline run.
///
/// Signals are transport-neutral observable emissions — not part of the data flow.
/// Consumers (SSE adapter, WebSocket room, log sink, execution store) subscribe
/// independently via [`ExecutionBus`] and interpret signals according to their context.
///
/// Nodes emit signals in two ways:
/// - **Mid-execution**: call [`ExecutionBus::emit`] directly via `input.bus` (e.g.
///   `ai.text.generate` emitting thinking/tool_call steps in real-time).
/// - **Post-execution**: include a `__signal` key in the output payload. The engine
///   strips it before forwarding the payload downstream and routes it through the bus.
///   Supports string, object `{kind, message, data}`, or array of either.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Signal {
    /// Short machine-readable tag chosen by the emitter (e.g. `"thinking"`,
    /// `"progress"`, `"query"`, `"complete"`). No fixed vocabulary — nodes decide.
    pub kind: String,
    /// Human-readable description.
    pub message: String,
    /// Node that emitted the signal (set by the engine when routing `__signal`
    /// from output; set by the node itself when calling `bus.emit()` directly).
    #[serde(default)]
    pub node_id: String,
    /// Node kind (e.g. `"postgres.query.run"`).
    #[serde(default)]
    pub node_kind: String,
    /// Optional structured data attached to the signal.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub data: Option<serde_json::Value>,
    /// Timing string (e.g. `"142ms"`, `"00:00:03"`).
    #[serde(default)]
    pub at: String,
}

/// Backward-compatible alias — existing code that references `StepEvent` keeps compiling
/// during migration.  New code should use [`Signal`] directly.
pub type StepEvent = Signal;

/// Execution-scoped broadcast bus for [`Signal`] values.
///
/// Created by the execution caller (webhook handler, schedule runner, etc.) and passed
/// into `ExecuteOptions`. The engine threads it to every node. Multiple consumers can
/// subscribe independently. If no consumer subscribes, emitted signals are silently
/// dropped — zero runtime cost.
///
/// The bus lives for exactly one pipeline execution run and is dropped when the run ends.
#[derive(Debug, Clone)]
pub struct ExecutionBus {
    tx: tokio::sync::broadcast::Sender<Signal>,
}

impl ExecutionBus {
    /// Create a new bus with the given channel capacity.
    pub fn new(capacity: usize) -> Self {
        let (tx, _) = tokio::sync::broadcast::channel(capacity);
        Self { tx }
    }

    /// Emit a signal to all current subscribers.  If no subscribers exist the signal
    /// is silently dropped.
    pub fn emit(&self, signal: Signal) {
        let _ = self.tx.send(signal);
    }

    /// Subscribe a new independent receiver.
    pub fn subscribe(&self) -> tokio::sync::broadcast::Receiver<Signal> {
        self.tx.subscribe()
    }
}

/// Per-execution options passed to the pipeline engine.
///
/// Pass `Default::default()` for standard non-streaming execution.
#[derive(Debug, Clone, Default)]
pub struct ExecuteOptions {
    /// When set, the engine threads this bus to every node and routes `__signal`
    /// output keys through it.  Consumers (SSE, WS, log) subscribe independently.
    pub bus: Option<std::sync::Arc<ExecutionBus>>,
    /// When set, the first `web.response.send` of the run hands its envelope
    /// here the moment it runs (`node-conventions.md` §4: it answers the
    /// caller at once and later nodes keep running). The HTTP ingress waits
    /// on it beside the run.
    pub responder: Option<Responder>,
}

/// The one-shot hand-off of a run's response to the HTTP caller waiting on
/// it. Only the first envelope is sent; the receiver ends without one when
/// the run ends without a response.
#[derive(Clone, Default)]
pub struct Responder(
    std::sync::Arc<std::sync::Mutex<Option<tokio::sync::oneshot::Sender<Value>>>>,
);

impl Responder {
    pub fn channel() -> (Self, tokio::sync::oneshot::Receiver<Value>) {
        let (tx, rx) = tokio::sync::oneshot::channel();
        (Self(std::sync::Arc::new(std::sync::Mutex::new(Some(tx)))), rx)
    }

    /// Sends the envelope if nothing was sent before; `false` otherwise.
    pub fn send(&self, envelope: Value) -> bool {
        let sender = self.0.lock().unwrap_or_else(|e| e.into_inner()).take();
        sender.is_some_and(|tx| tx.send(envelope).is_ok())
    }
}

impl std::fmt::Debug for Responder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("Responder")
    }
}

/// Immutable execution context passed to every node during a pipeline run.
///
/// Contains the tenant/project scope and a unique request id for tracing.  The `input`
/// field carries the original trigger payload — nodes should read their flowing payload
/// from `NodeExecutionInput::payload`, not from here.
///
/// `route` carries the inbound HTTP request path for webhook-triggered pipelines.
/// It is injected into every node's `metadata` under the key `"route"` so that
/// downstream nodes can access the real request path without reading it from a
/// potentially-transformed payload.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineContext {
    /// Owner / tenant id (e.g. `"superadmin"`).
    pub owner: String,
    /// Project id (e.g. `"default"`).
    pub project: String,
    /// Pipeline id matching [`PipelineGraph::id`].
    pub pipeline: String,
    /// Unique id for this execution run — used in trace entries and logs.
    pub request_id: String,
    /// Inbound HTTP request path for webhook-triggered pipelines (e.g. `"/blog/my-post"`).
    ///
    /// Populated by the webhook handler from the matched request URI.  Empty string for
    /// non-webhook triggers (schedule, WS, manual).  Injected into node metadata as
    /// `"route"` so template nodes can build a correct [`crate::rwe::RenderContext`]
    /// without requiring a `route` field on the node config itself.
    #[serde(default)]
    pub route: String,
    /// The envelope that started this pipeline run. Passed into the entry
    /// node, which answers it under its source key (`webhook: { … }`).
    pub input: Value,
    /// Immutable trigger-event snapshot carried to every node in the run
    /// (`$trigger`).
    ///
    /// Set once at pipeline entry by the ingress; never mutated by nodes.
    /// For webhook triggers: the envelope plus `search`, `pathname` and the
    /// safe `headers`. For manual runs: the input with empty `params`,
    /// `query`, `auth`. For room events: the envelope with `auth`. `None`
    /// makes the engine use `input` itself, so `$trigger` is always the
    /// envelope.
    #[serde(default)]
    pub trigger: Option<Value>,
    /// Credential placeholder map for composite/WASM node execution.
    ///
    /// Maps placeholder names (e.g. `"BOT_TOKEN"`) to resolved credential values.
    /// Injected into every node's metadata under `"placeholder"` so that
    /// `$placeholder.BOT_TOKEN` resolves in `{{ expr }}` expressions.
    /// Only set for composite inner pipelines; `None` for regular pipelines.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub placeholder: Option<Value>,
}

/// Per-node execution record captured by the pipeline engine during a run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NodeTraceEntry {
    /// Node's `zfPipelineNodeId` (slug), e.g. `"fetch-user"`.
    pub node_id: String,
    /// Node kind, e.g. `"postgres.query.run"`.
    pub node_kind: String,
    /// Effective node config snapshot after expression resolution.
    ///
    /// Present when the engine can safely surface config-driven execution details in logs.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub config: Option<Value>,
    /// Wall-clock duration of this node in milliseconds.
    pub duration_ms: u64,
    /// Input payload delivered to this node.
    pub input: Value,
    /// Output payload produced by this node (`null` on error).
    pub output: Value,
    /// Set if this node threw a `PipelineError`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
    /// NodeIO status word: `ok`, `empty` (ran, deliberately emitted
    /// nothing), `skipped` (never ran: every edge into it was skipped — a
    /// branch not taken, `node-conventions.md` §4), `refused` (caller's
    /// fault), or `failed` (world's fault). Records written before the word
    /// existed read as `ok`, which is what they meant.
    ///
    /// Two engine words on top, since 2026-09-21. `retry`: a wait — a
    /// failure an `:error` edge handed to `logic.retry` (the node failed, the
    /// run did not; `error` and `error_code` still say why), or a
    /// `logic.retry` that sent a verdict round again (no error at all). The
    /// attempt is the count of `retry` entries for that node so far.
    /// `error_routed`: a failure an `:error` edge handed to anything else.
    #[serde(default = "default_trace_status")]
    pub status: String,
    /// The registered error code, when status is refused/failed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub error_code: Option<String>,
    /// A small JPEG of a **temporary** image FileRef this node's declared
    /// `--preview image` / `--preview-in image` would draw, so the canvas can
    /// show it after the run's files are deleted:
    /// `{ slot, mime, width, height, data_base64 }` (longest side 540 px,
    /// ≤ 64 KB) or `{ slot, snapshot_skipped }`. Beside the payload, never in
    /// it. See `trace_capture::preview_snapshot`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub preview_snapshot: Option<Value>,
}

fn default_trace_status() -> String {
    "ok".to_string()
}

/// Final output of a completed pipeline execution.
///
/// `value` is the payload produced by the last (terminal) node.  For `web.response.send` with
/// `--template` this contains `{ html, compiled_scripts, hydration_payload }`.  For all other terminal nodes
/// it is whatever the node returned.
///
/// `trace` is a flat ordered list of trace strings emitted by each node in execution order,
/// useful for debugging slow or failed runs.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineOutput {
    /// Final output payload from the terminal node.
    pub value: Value,
    /// What `web.response.send` answered the caller, when one ran: the
    /// envelope the HTTP layer turns into a response. Never part of `value`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub response: Option<Value>,
    /// Ordered trace entries from all nodes that executed during this run.
    pub trace: Vec<String>,
    /// Structured per-node trace (populated by BasicPipelineEngine).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub node_trace: Vec<NodeTraceEntry>,
}

impl PipelineOutput {
    /// A function pipeline's result — the one rule for every caller of a
    /// function (`function.result.call`, a composite node, an agent tool, a
    /// map layer): **its last node's answer**, the value the last node that
    /// answered (trace status `ok`) added under its own key (its noun, a
    /// trigger's source, `node-conventions.md` §6), read from the final
    /// payload. A last node with no key of its own (`logic.*`,
    /// `web.response.send`) answered nothing, so the result is then the final
    /// payload without the function trigger's `function` key. A function
    /// therefore ends in the node whose answer is its result, usually a
    /// `javascript.script.run` shaping it.
    pub fn function_result(&self) -> Value {
        let last_key = self
            .node_trace
            .iter()
            .rev()
            .find(|entry| entry.status == "ok")
            .and_then(|entry| crate::pipeline::nodes::answer_key(&entry.node_kind));
        if let Some(value) = last_key.and_then(|key| self.value.get(&key).cloned()) {
            return value;
        }
        let mut rest = self.value.clone();
        if let Some(map) = rest.as_object_mut() {
            map.remove("function");
        }
        rest
    }
}

/// A typed error produced at any point during pipeline execution.
///
/// `code` is a stable SCREAMING_SNAKE_CASE string prefixed by the node kind
/// (e.g. `"FW_NODE_POSTGRES_QUERY_RUN_CONFIG"`, `"FW_NODE_WEB_RENDER_COMPILE"`).  It is safe to
/// match on in tests and error handlers.  `message` is the human-readable detail.
///
/// `node_id` and `node_kind` are optionally populated by the engine at the node execution
/// boundary so error messages include the failing node identity.
///
/// # Stable error code conventions
///
/// | Prefix | Meaning |
/// |--------|---------|
/// | `FW_NODE_<KIND>_CONFIG` | Node config validation failed at build time |
/// | `FW_NODE_<KIND>_COMPILE` | Node compile/build phase failed |
/// | `FW_NODE_<KIND>_RUN` | Node execution failed at runtime |
/// | `FW_NODE_<KIND>_CREDENTIAL*` | Credential lookup or kind mismatch |
/// | `FW_NODE_<KIND>_INPUT_PIN` | Unexpected input pin name |
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PipelineError {
    /// Stable error code.  Never changes across versions for a given failure mode.
    pub code: &'static str,
    /// Human-readable error detail.
    pub message: String,
    /// Node instance id that produced this error (e.g. `"n1"`, `"query"`).
    /// Populated by the engine at the node execution boundary; `None` for config errors.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_id: Option<String>,
    /// Node kind that produced this error (e.g. `"postgres.query.run"`).
    /// Populated by the engine at the node execution boundary; `None` for config errors.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub node_kind: Option<String>,
    /// Partial per-node execution trace captured before the failure.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub node_trace: Vec<NodeTraceEntry>,
}

impl From<crate::zebfs::ZebFsError> for PipelineError {
    /// A store refusal keeps its own code (`ZEBFS_NOT_FOUND`), so a node's
    /// diagnostic names the store's reason.
    fn from(err: crate::zebfs::ZebFsError) -> Self {
        Self::new(err.code, err.message)
    }
}

impl PipelineError {
    /// Constructs a new error with a stable code and a descriptive message.
    pub fn new(code: &'static str, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
            node_id: None,
            node_kind: None,
            node_trace: vec![],
        }
    }

    /// Constructs an error attributed to a specific node instance.
    pub fn node(node_id: &str, node_kind: &str, code: &'static str, msg: &str) -> Self {
        Self {
            code,
            message: msg.to_string(),
            node_id: Some(node_id.to_string()),
            node_kind: Some(node_kind.to_string()),
            node_trace: vec![],
        }
    }
}

impl Display for PipelineError {
    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.code, self.message)
    }
}

impl std::error::Error for PipelineError {}

#[cfg(test)]
mod node_timeout_tests {
    use std::time::Duration;

    use serde_json::json;

    use super::{NODE_TIMEOUT_CONFIG_CODE, engine_common_dsl_flags, node_timeout};

    /// `--timeout` is a duration from 1s to 1h; anything else is refused,
    /// never clamped into range.
    #[test]
    fn the_engine_timeout_is_a_duration_from_one_second_to_one_hour() {
        assert_eq!(node_timeout(&json!({})).unwrap(), None);
        assert_eq!(node_timeout(&json!({ "timeout": "" })).unwrap(), None);
        assert_eq!(node_timeout(&json!({ "timeout": "30s" })).unwrap(), Some(Duration::from_secs(30)));
        assert_eq!(node_timeout(&json!({ "timeout": "1h" })).unwrap(), Some(Duration::from_secs(3600)));
        for bad in [json!("30"), json!(30), json!("500ms"), json!("2h"), json!("soon")] {
            let err = node_timeout(&json!({ "timeout": bad })).expect_err("refused");
            assert_eq!(err.code, NODE_TIMEOUT_CONFIG_CODE, "{bad}");
        }
        let flag = engine_common_dsl_flags().into_iter().find(|f| f.flag == "--timeout").expect("--timeout");
        assert_eq!((flag.config_key.as_str(), flag.value.as_str()), ("timeout", "duration"));
    }
}

#[cfg(test)]
mod function_result_tests {
    use serde_json::{Value, json};

    use super::PipelineOutput;

    fn output(value: Value, trace: &[(&str, &str)]) -> PipelineOutput {
        let node_trace = trace
            .iter()
            .map(|(kind, status)| {
                serde_json::from_value(json!({
                    "node_id": "n", "node_kind": kind, "duration_ms": 0,
                    "input": null, "output": null, "status": status
                }))
                .expect("trace entry")
            })
            .collect();
        PipelineOutput { value, response: None, trace: Vec::new(), node_trace }
    }

    /// The last node that answered decides; a control node at the end leaves
    /// the final payload without the trigger's `function`.
    #[test]
    fn a_functions_result_is_its_last_answer() {
        let value = json!({ "function": { "a": 1 }, "response": { "status": 200 }, "script": { "ok": true } });
        let ran = [("trigger.function", "ok"), ("http.response.fetch", "ok"), ("javascript.script.run", "ok")];
        assert_eq!(output(value.clone(), &ran).function_result(), json!({ "ok": true }));

        let skipped_last = [("trigger.function", "ok"), ("http.response.fetch", "ok"), ("javascript.script.run", "empty")];
        assert_eq!(output(value.clone(), &skipped_last).function_result(), json!({ "status": 200 }));

        let control_last = [("trigger.function", "ok"), ("javascript.script.run", "ok"), ("logic.if", "ok")];
        assert_eq!(
            output(value, &control_last).function_result(),
            json!({ "response": { "status": 200 }, "script": { "ok": true } })
        );
    }
}