//! Declarative reconcile engine for a single app.
//!
//! Takes a desired-state [`Bundle`] (an app's scripts, routes, triggers,
//! and the names of its secrets) and either diffs it against the live
//! database ([`ApplyService::plan`], read-only) or reconciles it in one
//! transaction ([`ApplyService::apply`]).
//!
//! Identity keys match the DB UNIQUE constraints so the diff is stable:
//! scripts by `lower(name)`, routes by the `(method, host, path)` tuple,
//! triggers by their per-kind semantic tuple, secrets by name.
//!
//! **Trigger-identity limitation:** a trigger's identity *is* its semantic
//! definition, so the diff only ever Creates or Deletes (a change is
//! delete-old + create-new). For queue the identity is `queue|{queue_name}`
//! (script-independent — it encodes the one-consumer-per-queue invariant)
//! and for email it is `email|{script}`. Consequence: rebinding a queue to
//! a different script, changing its visibility timeout, or rotating an email
//! trigger's secret/reference diffs as a **NoOp and is not applied**. To
//! change those, recreate the trigger (drop it from the manifest, apply
//! `--prune`, then re-add it).
//!
//! Other known limitations vs. the interactive API: a script **rename**
//! diffs as delete+create, so under `--prune` it loses the old script's id,
//! version counter, and execution logs; an `endpoint`→`module` kind flip is
//! not guarded against *live* (un-pruned) routes/triggers still bound to the
//! script; and cross-pattern route conflict-vs-live is not checked (only
//! identical tuples collide). See `validate_bundle`.

use std::collections::{BTreeSet, HashMap, HashSet};
use std::sync::Arc;

use picloud_orchestrator_core::routing::{pattern, RouteTable};
use picloud_shared::{
    AdminUserId, AppId, DispatchMode, DocsEventOp, FilesEventOp, GroupId, HostKind, KvEventOp,
    MasterKey, PathKind, Route, Script, ScriptId, ScriptKind, ScriptOwner, ScriptSandbox,
    ScriptValidator, TriggerId,
};
use serde::{Deserialize, Serialize};
use sqlx::PgPool;

use crate::app_domain_repo::AppDomainRepository;
use crate::app_repo::AppRepository;
use crate::authz::AuthzRepo;
use crate::repo::{
    delete_script_tx, insert_script_tx, update_script_tx, NewScript, ScriptPatch, ScriptRepository,
    ScriptRepositoryError,
};
use crate::route_repo::{delete_route_tx, insert_route_tx, NewRoute, RouteRepository};
use crate::sandbox::SandboxCeiling;
use crate::secrets_repo::SecretsRepo;
use crate::trigger_config::TriggerConfig;
use crate::trigger_repo::{
    delete_trigger_tx, insert_email_trigger_tx, insert_trigger_tx, Trigger, TriggerDetails,
    TriggerDispatchMode, TriggerRepo, TriggerRepoError,
};
use crate::vars_repo::{VarOwner, VarsRepo};

/// Reserved path prefixes that user routes may not claim (mirrors the
/// route-create rejection — see CLAUDE.md "Reserved path prefixes").
const RESERVED_PATH_PREFIXES: &[&str] = &["/api/", "/admin/"];
const RESERVED_PATH_EXACT: &[&str] = &["/healthz", "/version"];

// ----------------------------------------------------------------------------
// Wire types — desired state (request) and the computed plan (response)
// ----------------------------------------------------------------------------

/// Desired state for one app, sent by `pic plan` / `pic apply`.
#[derive(Debug, Clone, Deserialize)]
pub struct Bundle {
    #[serde(default)]
    pub scripts: Vec<BundleScript>,
    #[serde(default)]
    pub routes: Vec<BundleRoute>,
    #[serde(default)]
    pub triggers: Vec<BundleTrigger>,
    /// Declared secret *names* only; values are pushed out-of-band.
    #[serde(default)]
    pub secrets: Vec<String>,
    /// Declared app-owned config `vars` — key → JSON value. Unlike secrets,
    /// var values are non-secret and live inline in the manifest. Written at
    /// app scope `*` (an app *is* an environment, §2 — per-env scoping is a
    /// group concern, deferred to the nested-manifest model). Tombstones are
    /// not expressible here; use `pic vars set --tombstone`.
    #[serde(default)]
    pub vars: std::collections::BTreeMap<String, serde_json::Value>,
    /// Declared extension-point *names* (§5.5) — module names this node offers
    /// for descendants to provide/override. Name-only, like `secrets`; the
    /// optional default body is a co-located `kind = module` script.
    #[serde(default)]
    pub extension_points: Vec<String>,
    /// Declared shared group collections (§11.6) — `(name, kind)` markers this
    /// node offers as cross-app-shared (`kind` ∈ `kv`/`docs`). The data lives in
    /// the per-kind store (`group_kv_entries` / `group_docs`). Authored on
    /// `[group]` nodes only (the CLI rejects it on `[app]`).
    #[serde(default)]
    pub collections: Vec<CollectionSpec>,
}

/// One declared shared-collection marker on the wire: a name + its store kind.
#[derive(Debug, Clone, Deserialize)]
pub struct CollectionSpec {
    pub name: String,
    /// Storage kind; defaults to `kv` for back-compat with the name-only form.
    #[serde(default = "default_collection_kind")]
    pub kind: String,
}

fn default_collection_kind() -> String {
    "kv".to_string()
}

/// The shared-collection kinds the apply engine accepts.
const COLLECTION_KINDS: &[&str] = &["kv", "docs"];

#[derive(Debug, Clone, Deserialize)]
pub struct BundleScript {
    pub name: String,
    pub source: String,
    #[serde(default)]
    pub kind: ScriptKind,
    #[serde(default)]
    pub description: Option<String>,
    #[serde(default)]
    pub timeout_seconds: Option<i32>,
    #[serde(default)]
    pub memory_limit_mb: Option<i32>,
    #[serde(default)]
    pub sandbox: Option<ScriptSandbox>,
    /// Three-state lifecycle (§4.3); omitted ⇒ active. Declarative (not
    /// leave-as-is): a script absent the key is `enabled = true`.
    #[serde(default = "picloud_shared::default_true")]
    pub enabled: bool,
}

#[derive(Debug, Clone, Deserialize)]
pub struct BundleRoute {
    /// Name of the script this route binds to.
    pub script: String,
    #[serde(default)]
    pub method: Option<String>,
    pub host_kind: HostKind,
    #[serde(default)]
    pub host: String,
    #[serde(default)]
    pub host_param_name: Option<String>,
    pub path_kind: PathKind,
    pub path: String,
    #[serde(default)]
    pub dispatch_mode: DispatchMode,
    /// Three-state lifecycle (§4.3); omitted ⇒ active.
    #[serde(default = "picloud_shared::default_true")]
    pub enabled: bool,
}

/// Desired trigger, tagged by kind on the wire (`{ "kind": "cron", … }`).
#[derive(Debug, Clone, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum BundleTrigger {
    Kv {
        script: String,
        collection_glob: String,
        #[serde(default)]
        ops: Vec<KvEventOp>,
        #[serde(default)]
        dispatch_mode: Option<TriggerDispatchMode>,
        #[serde(default)]
        retry_max_attempts: Option<u32>,
    },
    Docs {
        script: String,
        collection_glob: String,
        #[serde(default)]
        ops: Vec<DocsEventOp>,
        #[serde(default)]
        dispatch_mode: Option<TriggerDispatchMode>,
        #[serde(default)]
        retry_max_attempts: Option<u32>,
    },
    Files {
        script: String,
        collection_glob: String,
        #[serde(default)]
        ops: Vec<FilesEventOp>,
        #[serde(default)]
        dispatch_mode: Option<TriggerDispatchMode>,
        #[serde(default)]
        retry_max_attempts: Option<u32>,
    },
    Cron {
        script: String,
        schedule: String,
        #[serde(default = "default_timezone")]
        timezone: String,
        #[serde(default)]
        dispatch_mode: Option<TriggerDispatchMode>,
        #[serde(default)]
        retry_max_attempts: Option<u32>,
    },
    Pubsub {
        script: String,
        topic_pattern: String,
        #[serde(default)]
        dispatch_mode: Option<TriggerDispatchMode>,
        #[serde(default)]
        retry_max_attempts: Option<u32>,
    },
    Email {
        script: String,
        /// Name of the secret (set via `pic secret set`) holding the
        /// inbound HMAC value; resolved + sealed at apply time. Never the
        /// value itself.
        inbound_secret_ref: String,
        #[serde(default)]
        dispatch_mode: Option<TriggerDispatchMode>,
        #[serde(default)]
        retry_max_attempts: Option<u32>,
    },
    Queue {
        script: String,
        queue_name: String,
        #[serde(default)]
        visibility_timeout_secs: Option<u32>,
        #[serde(default)]
        dispatch_mode: Option<TriggerDispatchMode>,
        #[serde(default)]
        retry_max_attempts: Option<u32>,
    },
}

fn default_timezone() -> String {
    "UTC".to_string()
}

impl BundleTrigger {
    /// The script name this trigger fires.
    #[must_use]
    pub fn script(&self) -> &str {
        match self {
            Self::Kv { script, .. }
            | Self::Docs { script, .. }
            | Self::Files { script, .. }
            | Self::Cron { script, .. }
            | Self::Pubsub { script, .. }
            | Self::Email { script, .. }
            | Self::Queue { script, .. } => script,
        }
    }

    /// Whether this trigger is an email trigger, which resolves and decrypts
    /// a stored secret by reference at apply time (gating an extra capability).
    #[must_use]
    pub fn is_email(&self) -> bool {
        matches!(self, Self::Email { .. })
    }

    /// Stable semantic identity (the per-kind tuple), used for the diff.
    #[must_use]
    pub fn identity(&self) -> String {
        match self {
            Self::Kv {
                script,
                collection_glob,
                ops,
                ..
            } => format!(
                "kv|{script}|{collection_glob}|{}",
                sorted_csv(ops.iter().copied().map(KvEventOp::as_str))
            ),
            Self::Docs {
                script,
                collection_glob,
                ops,
                ..
            } => format!(
                "docs|{script}|{collection_glob}|{}",
                sorted_csv(ops.iter().copied().map(DocsEventOp::as_str))
            ),
            Self::Files {
                script,
                collection_glob,
                ops,
                ..
            } => format!(
                "files|{script}|{collection_glob}|{}",
                sorted_csv(ops.iter().copied().map(FilesEventOp::as_str))
            ),
            Self::Cron {
                script,
                schedule,
                timezone,
                ..
            } => format!("cron|{script}|{schedule}|{timezone}"),
            Self::Pubsub {
                script,
                topic_pattern,
                ..
            } => format!("pubsub|{script}|{topic_pattern}"),
            Self::Email { script, .. } => format!("email|{script}"),
            Self::Queue { queue_name, .. } => format!("queue|{queue_name}"),
        }
    }

    #[must_use]
    pub fn kind_str(&self) -> &'static str {
        match self {
            Self::Kv { .. } => "kv",
            Self::Docs { .. } => "docs",
            Self::Files { .. } => "files",
            Self::Cron { .. } => "cron",
            Self::Pubsub { .. } => "pubsub",
            Self::Email { .. } => "email",
            Self::Queue { .. } => "queue",
        }
    }
}

/// One change in the plan. `key` is the human-renderable identity.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct ResourceChange {
    pub op: Op,
    pub key: String,
    /// Optional one-line note (e.g. why an update, or "value must be set").
    #[serde(skip_serializing_if = "Option::is_none")]
    pub detail: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum Op {
    Create,
    Update,
    NoOp,
    Delete,
}

/// The computed diff, grouped by resource kind.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)]
pub struct Plan {
    pub scripts: Vec<ResourceChange>,
    pub routes: Vec<ResourceChange>,
    pub triggers: Vec<ResourceChange>,
    pub secrets: Vec<ResourceChange>,
    #[serde(default)]
    pub vars: Vec<ResourceChange>,
    #[serde(default)]
    pub extension_points: Vec<ResourceChange>,
    #[serde(default)]
    pub collections: Vec<ResourceChange>,
}

impl Plan {
    /// True when every resource is a no-op (nothing to apply).
    #[must_use]
    pub fn is_noop(&self) -> bool {
        self.scripts
            .iter()
            .chain(&self.routes)
            .chain(&self.triggers)
            .chain(&self.secrets)
            .chain(&self.vars)
            .chain(&self.extension_points)
            .chain(&self.collections)
            .all(|c| c.op == Op::NoOp)
    }
}

/// What `plan` returns: the diff plus a fingerprint of the live state it was
/// computed against. The token is flattened onto the plan JSON, so the wire
/// shape stays `{ scripts, routes, triggers, secrets, state_token }`.
#[derive(Debug, Clone, Serialize)]
pub struct PlanResult {
    #[serde(flatten)]
    pub plan: Plan,
    pub state_token: String,
}

// ----------------------------------------------------------------------------
// Tree apply (Phase 5): a whole project subtree applied in one transaction.
// ----------------------------------------------------------------------------

/// Whether a tree node is an app or a group.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum NodeKind {
    App,
    Group,
}

/// One node of a project tree: its kind, slug (must pre-exist on the server),
/// and its desired-state bundle (a group's bundle has no routes/triggers).
/// Request-only (deserialized from the CLI's tree bundle).
#[derive(Debug, Clone, Deserialize)]
pub struct TreeNode {
    pub kind: NodeKind,
    pub slug: String,
    pub bundle: Bundle,
}

/// A whole project subtree submitted to `apply_tree`.
#[derive(Debug, Clone, Default, Deserialize)]
pub struct TreeBundle {
    #[serde(default)]
    pub nodes: Vec<TreeNode>,
}

/// One node's diff within a tree plan.
#[derive(Debug, Clone, Serialize)]
pub struct NodePlan {
    pub kind: NodeKind,
    pub slug: String,
    #[serde(flatten)]
    pub plan: Plan,
}

/// What `plan_tree` returns: a per-node diff plus ONE combined bound-plan token
/// (fingerprint of every node's content + every in-scope group's structure
/// version, so a reparent or content edit between plan and apply is caught).
#[derive(Debug, Clone, Serialize)]
pub struct TreePlanResult {
    pub nodes: Vec<NodePlan>,
    pub state_token: String,
}

// ----------------------------------------------------------------------------
// Current state snapshot
// ----------------------------------------------------------------------------

/// The app's live state, loaded once for a diff.
#[derive(Debug, Default)]
pub struct CurrentState {
    pub scripts: Vec<Script>,
    pub routes: Vec<Route>,
    pub triggers: Vec<Trigger>,
    pub secret_names: Vec<String>,
    /// App's OWN `vars` at scope `*` (key, value), non-tombstone — the set
    /// the manifest reconciles. Inherited group vars and tombstones are
    /// excluded (the manifest manages only the app's own real values).
    pub vars: Vec<(String, serde_json::Value)>,
    /// Extension-point marker names declared directly at this node (§5.5).
    pub extension_point_names: Vec<String>,
    /// Shared group collections declared directly at this node (§11.6), as
    /// `(name, kind)` pairs.
    pub collections: Vec<(String, String)>,
}

/// One row of the read-only extension-point report (§5.5).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExtensionPointInfo {
    pub name: String,
    /// True when this node owns the marker (vs inheriting it from an ancestor).
    pub declared_here: bool,
    /// Resolved provider for an app (`app override` / `inherited default`), or
    /// `None` when unset (no provider — a plan error) or for a group node.
    pub provider: Option<String>,
}

/// One row of the read-only §11.6 shared-collection report.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CollectionInfo {
    pub name: String,
    pub kind: String,
}

// ----------------------------------------------------------------------------
// Errors
// ----------------------------------------------------------------------------

#[derive(Debug, thiserror::Error)]
pub enum ApplyError {
    #[error("app not found: {0}")]
    AppNotFound(String),
    #[error("invalid manifest: {0}")]
    Invalid(String),
    #[error(
        "live state changed since `pic plan` (someone edited this app's scripts, \
         routes, triggers, or secrets); re-run `pic plan` to review, then apply — \
         or `pic apply --force` to skip the check"
    )]
    StateMoved,
    #[error("forbidden")]
    Forbidden,
    #[error("authorization repo error: {0}")]
    AuthzRepo(String),
    #[error("backend: {0}")]
    Backend(String),
}

// ----------------------------------------------------------------------------
// Service
// ----------------------------------------------------------------------------

/// Reconcile engine. Holds trait-object repos for the read/diff path; the
/// transactional write path (`apply`) reuses the same handles plus `pool`.
#[derive(Clone)]
pub struct ApplyService {
    /// Pool for the transactional write path (`apply`).
    pub pool: PgPool,
    pub scripts: Arc<dyn ScriptRepository>,
    pub routes: Arc<dyn RouteRepository>,
    pub triggers: Arc<dyn TriggerRepo>,
    pub secrets: Arc<dyn SecretsRepo>,
    /// App-owned `vars` reconciliation (read/diff path; the transactional
    /// write goes through `set_app_var_tx` / `delete_app_var_tx`).
    pub vars: Arc<dyn VarsRepo>,
    pub apps: Arc<dyn AppRepository>,
    /// Group lookups (Phase 5): resolve a group slug→id for group/tree apply.
    pub groups: Arc<dyn crate::group_repo::GroupRepository>,
    /// Module resolver (Phase 4b): lexical, origin-aware module lookup used by
    /// the single-node dangling-import plan check (§5.5).
    pub modules: Arc<dyn picloud_shared::ModuleSource>,
    /// App domain claims — validates a route's host belongs to the app.
    pub domains: Arc<dyn AppDomainRepository>,
    pub authz: Arc<dyn AuthzRepo>,
    pub validator: Arc<dyn ScriptValidator>,
    pub sandbox_ceiling: SandboxCeiling,
    /// Default retry/dispatch settings filled when a bundle trigger omits them.
    pub trigger_config: TriggerConfig,
    /// Shared route snapshot, rebuilt once after a successful apply.
    pub route_table: Arc<RouteTable>,
    /// Master key — seals email triggers' resolved inbound secrets.
    pub master_key: MasterKey,
}

impl ApplyService {
    /// Compute the diff between `bundle` and app `app_id`'s live state.
    /// Read-only: validates the bundle, loads current state, diffs.
    ///
    /// # Errors
    /// `Invalid` for a malformed bundle; `Backend` for repo failures.
    pub async fn plan(&self, app_id: AppId, bundle: &Bundle) -> Result<PlanResult, ApplyError> {
        self.plan_owner(ApplyOwner::App(app_id), bundle).await
    }

    /// Owner-generic plan (Phase 5): diff `bundle` against an app OR group node.
    /// A group node has only scripts + vars — routes/triggers are rejected, and
    /// the app-only inherited-target + route-host validation is skipped.
    ///
    /// # Errors
    /// `Invalid` for a malformed bundle; `Backend` for repo failures.
    pub async fn plan_owner(
        &self,
        owner: ApplyOwner,
        bundle: &Bundle,
    ) -> Result<PlanResult, ApplyError> {
        let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
        self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
        self.check_imports_resolve(owner, bundle).await?;
        self.check_extension_points_provided(owner, bundle).await?;
        if let ApplyOwner::App(app_id) = owner {
            self.validate_route_hosts(app_id, bundle).await?;
        }
        let current = self.load_current(owner).await?;
        let current_names = self.resolve_current_target_names(&current).await?;
        validate_email_secrets_present(bundle, &current.secret_names)?;
        Ok(PlanResult {
            plan: compute_diff_with_names(&current, bundle, &current_names),
            // Fingerprint of the live state this plan was computed against, so
            // `apply` can refuse if the node changed underneath it (§4.2).
            state_token: state_token_with_names(&current, &current_names),
        })
    }

    /// Inherited group-script targets for an app node; a group node has no
    /// routes/triggers to bind, so there is nothing to inherit (empty map).
    async fn resolve_inherited_targets_for(
        &self,
        owner: ApplyOwner,
        bundle: &Bundle,
    ) -> Result<HashMap<String, ScriptId>, ApplyError> {
        match owner {
            ApplyOwner::App(app_id) => self.resolve_inherited_targets(app_id, bundle).await,
            ApplyOwner::Group(_) => Ok(HashMap::new()),
        }
    }

    /// `validate_bundle`, plus a group-node guard: a group owns only scripts +
    /// vars (+ declared secret names), so a `[group]` manifest carrying routes
    /// or triggers is a 422 — those are app concerns.
    fn validate_bundle_for(
        &self,
        owner: ApplyOwner,
        bundle: &Bundle,
        inherited_endpoints: &HashSet<String>,
    ) -> Result<(), ApplyError> {
        if matches!(owner, ApplyOwner::Group(_)) {
            if !bundle.routes.is_empty() {
                return Err(ApplyError::Invalid(
                    "a group manifest cannot declare routes — routes bind to an app".into(),
                ));
            }
            if !bundle.triggers.is_empty() {
                return Err(ApplyError::Invalid(
                    "a group manifest cannot declare triggers — triggers belong to an app".into(),
                ));
            }
        }
        // §11.6: shared collections are owned by GROUPS. Reject them on an app
        // node — the inverse of the group route/trigger guard above. The CLI
        // already prevents this (`ManifestApp` has no `collections` field), so
        // this is defense-in-depth against a hand-rolled wire `Bundle`; without
        // it the reconcile would insert an inert `app_id`-owned marker row that
        // no resolver ever reads.
        if matches!(owner, ApplyOwner::App(_)) && !bundle.collections.is_empty() {
            return Err(ApplyError::Invalid(
                "an app manifest cannot declare shared collections — \
                 they are owned by a group"
                    .into(),
            ));
        }
        self.validate_bundle(bundle, inherited_endpoints)
    }

    /// Reconcile app `app_id` to `bundle` in a single transaction. Creates
    /// and updates are always applied; deletions of resources absent from the
    /// bundle are executed only when `prune` is set (secrets are never pruned).
    ///
    /// # Errors
    /// `Invalid` for a bad bundle; `Backend` for repo/transaction failures.
    /// Reconcile ONE node's scripts/routes/triggers/vars inside an existing
    /// transaction (Phase 5). Returns the node's final `name -> script_id` map
    /// (current + created), which the tree apply records so a later app node
    /// can bind a route to a group script created earlier in the SAME tx. The
    /// single-node `apply_owner` and the multi-node tree apply both call this.
    #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
    async fn reconcile_node_tx(
        &self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        owner: ApplyOwner,
        bundle: &Bundle,
        plan: &Plan,
        current: &CurrentState,
        current_names: &HashMap<ScriptId, String>,
        inherited: &HashMap<String, ScriptId>,
        prune: bool,
        actor: AdminUserId,
        report: &mut ApplyReport,
    ) -> Result<HashMap<String, ScriptId>, ApplyError> {
        let bundle_scripts: HashMap<String, &BundleScript> = bundle
            .scripts
            .iter()
            .map(|s| (s.name.to_lowercase(), s))
            .collect();
        let current_scripts: HashMap<String, &Script> = current
            .scripts
            .iter()
            .map(|s| (s.name.to_lowercase(), s))
            .collect();
        // name -> id for resolving route/trigger targets (existing + created).
        let mut name_to_id: HashMap<String, ScriptId> = current
            .scripts
            .iter()
            .map(|s| (s.name.to_lowercase(), s.id))
            .collect();

        // 1. Scripts — create + update first so routes/triggers resolve.
        for ch in &plan.scripts {
            let key = ch.key.to_lowercase();
            match ch.op {
                Op::Create => {
                    let bs = bundle_scripts[&key];
                    let (script_app_id, script_group_id) = owner.script_owner();
                    let new = NewScript {
                        app_id: script_app_id,
                        group_id: script_group_id,
                        name: bs.name.clone(),
                        description: bs.description.clone(),
                        source: bs.source.clone(),
                        kind: bs.kind,
                        timeout_seconds: bs.timeout_seconds,
                        memory_limit_mb: bs.memory_limit_mb,
                        sandbox: bs.sandbox,
                        enabled: bs.enabled,
                        imports: self.script_imports(bs)?,
                    };
                    let created = insert_script_tx(&mut *tx, &new).await.map_err(map_repo)?;
                    name_to_id.insert(created.name.to_lowercase(), created.id);
                    report.scripts_created += 1;
                }
                Op::Update => {
                    let bs = bundle_scripts[&key];
                    let cur = current_scripts[&key];
                    let patch = ScriptPatch {
                        name: None,
                        // Sparse: `None` (omitted in the manifest) leaves the
                        // stored description untouched; `Some(text)` sets it.
                        // Mirrors `script_update_reason`'s leave-as-is rule.
                        description: bs.description.clone().map(Some),
                        source: Some(bs.source.clone()),
                        timeout_seconds: bs.timeout_seconds,
                        memory_limit_mb: bs.memory_limit_mb,
                        sandbox: bs.sandbox,
                        kind: Some(bs.kind),
                        // Declarative: always reconcile to the manifest's value.
                        enabled: Some(bs.enabled),
                        imports: Some(self.script_imports(bs)?),
                    };
                    update_script_tx(&mut *tx, cur.id, &patch)
                        .await
                        .map_err(map_repo)?;
                    report.scripts_updated += 1;
                }
                Op::NoOp | Op::Delete => {}
            }
        }

        // Phase 4: route/trigger targets the manifest doesn't define as app-own
        // scripts may bind to inherited group endpoints. `or_insert` keeps
        // app-own precedence (those names aren't in `inherited` by
        // construction, but be defensive about CoW).
        for (name, id) in inherited {
            name_to_id.entry(name.clone()).or_insert(*id);
        }

        // 2. Routes — identity is the (method, host, path) tuple; an
        // "update" (rebind / dispatch change) is delete + insert.
        let current_routes: HashMap<String, &Route> = current
            .routes
            .iter()
            .map(|r| {
                (
                    route_key(
                        r.method.as_deref(),
                        r.host_kind,
                        &r.host,
                        r.path_kind,
                        &r.path,
                    ),
                    r,
                )
            })
            .collect();
        let bundle_routes: HashMap<String, &BundleRoute> = bundle
            .routes
            .iter()
            .map(|r| {
                (
                    route_key(
                        r.method.as_deref(),
                        r.host_kind,
                        &r.host,
                        r.path_kind,
                        &r.path,
                    ),
                    r,
                )
            })
            .collect();
        // 2. Routes — delete before insert so a binding freed by an update's
        // delete-half or a prune removal can be reused by a create in the
        // SAME apply without tripping the per-app UNIQUE binding index.
        // Phase A — deletes: update replacements always; removals when pruning.
        for ch in &plan.routes {
            match ch.op {
                Op::Update => {
                    let cur = current_routes[&ch.key];
                    delete_route_tx(&mut *tx, cur.id).await.map_err(map_repo)?;
                }
                Op::Delete if prune => {
                    if let Some(cur) = current_routes.get(&ch.key) {
                        delete_route_tx(&mut *tx, cur.id).await.map_err(map_repo)?;
                        report.routes_deleted += 1;
                    }
                }
                _ => {}
            }
        }
        // Phase B — inserts: creates and update replacements.
        for ch in &plan.routes {
            let is_update = match ch.op {
                Op::Create => false,
                Op::Update => true,
                _ => continue,
            };
            let br = bundle_routes[&ch.key];
            let sid = resolve_script(&name_to_id, &br.script)?;
            // Routes only exist on app nodes (a group bundle has none, so this
            // loop is empty for a group — validated above).
            let app_id = owner.app_id().expect("routes only exist on app nodes");
            insert_bundle_route(&mut *tx, app_id, sid, br).await?;
            if is_update {
                report.routes_updated += 1;
            } else {
                report.routes_created += 1;
            }
        }

        // 3. Triggers — semantic identity captures the whole definition,
        // so the only op is Create (a change is delete-old + create-new;
        // the stale one is removed by prune). Additive: create only.
        let bundle_triggers: HashMap<String, &BundleTrigger> =
            bundle.triggers.iter().map(|t| (t.identity(), t)).collect();
        for ch in &plan.triggers {
            if ch.op != Op::Create {
                continue;
            }
            // Triggers only exist on app nodes (a group bundle has none).
            let app_id = owner.app_id().expect("triggers only exist on app nodes");
            let bt = bundle_triggers[&ch.key];
            let sid = resolve_script(&name_to_id, bt.script())?;
            if let BundleTrigger::Email {
                inbound_secret_ref, ..
            } = bt
            {
                // Resolve the referenced secret, decrypt it, and re-seal it
                // into the email trigger's detail row — the value never
                // travels in the manifest.
                let (ct, nonce) = self.resolve_and_seal(app_id, inbound_secret_ref).await?;
                insert_email_trigger_tx(&mut *tx, app_id, sid, actor, &ct, &nonce)
                    .await
                    .map_err(map_trig)?;
            } else {
                let (dispatch, retry_max, backoff, base) = self.trigger_settings(bt);
                let details = bundle_trigger_details(
                    bt,
                    self.trigger_config.queue_default_visibility_timeout_secs,
                );
                insert_trigger_tx(
                    &mut *tx, app_id, sid, actor, dispatch, retry_max, backoff, base, &details,
                )
                .await
                .map_err(map_trig)?;
            }
            report.triggers_created += 1;
        }

        // 3b. Vars — upsert every Create/Update at app scope `*`, in-tx so a
        // later failure rolls them back with everything else.
        for ch in &plan.vars {
            match ch.op {
                Op::Create => {
                    owner
                        .set_var_tx(&mut *tx, &ch.key, &bundle.vars[&ch.key])
                        .await?;
                    report.vars_created += 1;
                }
                Op::Update => {
                    owner
                        .set_var_tx(&mut *tx, &ch.key, &bundle.vars[&ch.key])
                        .await?;
                    report.vars_updated += 1;
                }
                Op::NoOp | Op::Delete => {}
            }
        }

        // 3c. Extension-point markers (§5.5) — insert each Create (idempotent).
        // Name-only identity, so there is no Update; deletes happen in prune.
        for ch in &plan.extension_points {
            if ch.op == Op::Create {
                crate::extension_point_repo::insert_extension_point_tx(
                    &mut *tx,
                    owner.as_script_owner(),
                    &ch.key,
                )
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
                report.extension_points_created += 1;
            }
        }

        // 3d. Shared group-collection markers (§11.6) — insert each Create
        // (idempotent). Name-only identity; deletes happen in prune. Pruning a
        // marker hides the store but does NOT drop its `group_kv_entries` data
        // (that dies only with the owning group).
        for ch in &plan.collections {
            if ch.op == Op::Create {
                let kind = ch.detail.as_deref().unwrap_or("kv");
                crate::group_collection_repo::insert_collection_tx(
                    &mut *tx,
                    owner.as_script_owner(),
                    &ch.key,
                    kind,
                )
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
                report.collections_created += 1;
            }
        }

        // 4. Prune (only with --prune): delete stale triggers, then scripts
        // (route removals already happened in the route delete-pass above).
        // Secret pruning is deliberately deferred (destructive + irreversible):
        // stale secrets are surfaced in the plan but never deleted here.
        if prune {
            // Include inherited group-script bindings (Phase 4) so a trigger
            // bound to a group script resolves its identity and is pruned when
            // dropped from the manifest — not silently orphaned.
            let id_to_name: HashMap<ScriptId, String> = current
                .scripts
                .iter()
                .map(|s| (s.id, s.name.clone()))
                .chain(current_names.iter().map(|(id, n)| (*id, n.clone())))
                .collect();
            let trigger_id_by_identity: HashMap<String, TriggerId> = current
                .triggers
                .iter()
                .filter_map(|t| current_trigger_identity(t, &id_to_name).map(|i| (i, t.id)))
                .collect();
            for ch in &plan.triggers {
                if ch.op == Op::Delete {
                    if let Some(id) = trigger_id_by_identity.get(&ch.key) {
                        delete_trigger_tx(&mut *tx, *id).await.map_err(map_trig)?;
                        report.triggers_deleted += 1;
                    }
                }
            }

            // Scripts owning a trigger the manifest can't represent (email,
            // dead_letter) must not be silently pruned: deleting the script
            // would cascade-destroy that trigger (and its sealed secret) via
            // the `triggers.script_id ON DELETE CASCADE` FK, defeating the
            // "never prune email triggers" guarantee. Refuse and point the
            // operator at the explicit removal path.
            let protected: HashSet<ScriptId> = current
                .triggers
                .iter()
                .filter(|t| {
                    matches!(
                        t.details,
                        TriggerDetails::Email { .. } | TriggerDetails::DeadLetter { .. }
                    )
                })
                .map(|t| t.script_id)
                .collect();
            let script_id_by_name: HashMap<String, ScriptId> = current
                .scripts
                .iter()
                .map(|s| (s.name.to_lowercase(), s.id))
                .collect();
            for ch in &plan.scripts {
                if ch.op == Op::Delete {
                    if let Some(id) = script_id_by_name.get(&ch.key.to_lowercase()) {
                        if protected.contains(id) {
                            return Err(ApplyError::Invalid(format!(
                                "cannot prune script `{}`: it still has an email or \
                                 dead-letter trigger that the manifest can't represent; \
                                 remove the trigger with `pic triggers rm` first",
                                ch.key
                            )));
                        }
                        delete_script_tx(&mut *tx, *id).await.map_err(map_repo)?;
                        report.scripts_deleted += 1;
                    }
                }
            }

            // Vars are prunable config (unlike secrets, which are never
            // pruned): a live app var the manifest dropped is deleted.
            for ch in &plan.vars {
                if ch.op == Op::Delete {
                    owner.delete_var_tx(&mut *tx, &ch.key).await?;
                    report.vars_deleted += 1;
                }
            }

            // Extension-point markers are prunable config too (§5.5).
            for ch in &plan.extension_points {
                if ch.op == Op::Delete {
                    crate::extension_point_repo::delete_extension_point_tx(
                        &mut *tx,
                        owner.as_script_owner(),
                        &ch.key,
                    )
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?;
                    report.extension_points_deleted += 1;
                }
            }

            // Shared group-collection markers are prunable config too (§11.6).
            // Only the marker is removed here — the data survives until the
            // owning group is deleted.
            for ch in &plan.collections {
                if ch.op == Op::Delete {
                    let kind = ch.detail.as_deref().unwrap_or("kv");
                    crate::group_collection_repo::delete_collection_tx(
                        &mut *tx,
                        owner.as_script_owner(),
                        &ch.key,
                        kind,
                    )
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?;
                    report.collections_deleted += 1;
                }
            }
        }
        Ok(name_to_id)
    }

    pub async fn apply(
        &self,
        app_id: AppId,
        bundle: &Bundle,
        prune: bool,
        actor: AdminUserId,
        expected_token: Option<&str>,
    ) -> Result<ApplyReport, ApplyError> {
        self.apply_owner(
            ApplyOwner::App(app_id),
            bundle,
            prune,
            actor,
            expected_token,
        )
        .await
    }

    /// Owner-generic apply (Phase 5): reconcile an app OR group node to `bundle`
    /// in a single transaction. A group node reconciles only its scripts + vars.
    ///
    /// # Errors
    /// `Invalid` for a bad bundle; `Backend` for repo/transaction failures.
    #[allow(clippy::too_many_lines)]
    pub async fn apply_owner(
        &self,
        owner: ApplyOwner,
        bundle: &Bundle,
        prune: bool,
        actor: AdminUserId,
        expected_token: Option<&str>,
    ) -> Result<ApplyReport, ApplyError> {
        let inherited = self.resolve_inherited_targets_for(owner, bundle).await?;
        self.validate_bundle_for(owner, bundle, &keys_set(&inherited))?;
        self.check_imports_resolve(owner, bundle).await?;
        self.check_extension_points_provided(owner, bundle).await?;
        if let ApplyOwner::App(app_id) = owner {
            self.validate_route_hosts(app_id, bundle).await?;
        }

        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?;
        // Serialize concurrent applies to the same app. NOTE: this lock
        // serializes apply-vs-apply only — `load_current` below reads on the
        // pool (a separate connection), so an *interactive* admin write
        // committing between the read and our writes can surface as a
        // unique-constraint conflict (clean tx rollback, no corruption) or a
        // stale no-op delete. Reading current state through `&mut *tx` is the
        // proper fix and is left as a follow-up. (The queue one-consumer
        // invariant is closed independently in `insert_trigger_tx`.)
        sqlx::query("SELECT pg_advisory_xact_lock($1)")
            .bind(apply_lock_key(owner))
            .execute(&mut *tx)
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?;

        let current = self.load_current(owner).await?;
        // Phase 4: names of group scripts the current routes/triggers bind to,
        // so the token / diff / prune resolve inherited bindings (see method).
        let current_names = self.resolve_current_target_names(&current).await?;
        // Bound-plan check (§4.2): if the caller passed the token from a prior
        // `pic plan`, refuse when the live state changed since. Computed from
        // the same `load_current` snapshot the diff uses, before any mutation
        // (so a stale apply rolls back nothing). NOTE: like the diff, this read
        // is on the pool, not `&mut *tx` (see the lock comment above), so it
        // catches drift between plan and apply but shares that same narrow
        // apply-vs-interactive-write window — it is not a substitute for the
        // tx-scoped read the follow-up will add.
        if let Some(expected) = expected_token {
            if state_token_with_names(&current, &current_names) != expected {
                return Err(ApplyError::StateMoved);
            }
        }
        // Surface a missing email-secret reference here so `plan` and `apply`
        // agree, rather than only failing deep in `resolve_and_seal` below.
        validate_email_secrets_present(bundle, &current.secret_names)?;
        let plan = compute_diff_with_names(&current, bundle, &current_names);
        let mut report = ApplyReport::default();
        // §4.7 warning: an enabled binding pointing at a disabled script is
        // deployed-but-unreachable. Not an error (it's valid desired state),
        // but surfaced so the operator isn't surprised by a silent 404.
        report.warnings.extend(disabled_target_warnings(bundle));
        // A group endpoint is reached by *inheritance* (a descendant app binds
        // it by name), never by its own route, so the "no route/trigger"
        // warning is noise for every group script — emit it for apps only.
        if matches!(owner, ApplyOwner::App(_)) {
            report
                .warnings
                .extend(unreachable_endpoint_warnings(bundle));
        }

        self.reconcile_node_tx(
            &mut tx,
            owner,
            bundle,
            &plan,
            &current,
            &current_names,
            &inherited,
            prune,
            actor,
            &mut report,
        )
        .await?;

        tx.commit()
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?;

        // Post-commit: rebuild the orchestrator's route snapshot once. A
        // failure here doesn't undo the committed DB write — the table
        // self-heals on the next route write or restart. A group node touches
        // no routes, so there's nothing to refresh.
        if matches!(owner, ApplyOwner::App(_)) {
            if let Err(e) = self.refresh_route_table().await {
                tracing::warn!(error = %e, "apply: route table refresh failed");
                report
                    .warnings
                    .push("route table refresh failed; it will self-heal".into());
            }
        }
        Ok(report)
    }

    // ------------------------------------------------------------------------
    // Tree apply (Phase 5): a project subtree reconciled in ONE transaction.
    // ------------------------------------------------------------------------

    /// Diff a whole project subtree, read-only. Returns a per-node plan + one
    /// combined bound-plan token.
    ///
    /// # Errors
    /// `Invalid` for a malformed/unknown bundle; `Backend` for repo failures.
    pub async fn plan_tree(&self, bundle: &TreeBundle) -> Result<TreePlanResult, ApplyError> {
        let (prepared, token) = self.prepare_tree(bundle).await?;
        let nodes = prepared
            .into_iter()
            .map(|p| NodePlan {
                kind: p.node.kind,
                slug: p.node.slug.clone(),
                plan: p.plan,
            })
            .collect();
        Ok(TreePlanResult {
            nodes,
            state_token: token,
        })
    }

    /// Reconcile a whole project subtree in a single transaction (all-or-
    /// nothing). Groups reconcile before apps, so an app route/trigger can bind
    /// a group script created earlier in the SAME transaction (resolved via the
    /// in-memory `group_script_index`, since uncommitted rows aren't visible to
    /// the pool-based inherited resolver).
    ///
    /// # Errors
    /// `Invalid` for a bad bundle; `StateMoved` if the bound token is stale;
    /// `Backend` for repo/transaction failures.
    pub async fn apply_tree(
        &self,
        bundle: &TreeBundle,
        prune: bool,
        actor: AdminUserId,
        expected_token: Option<&str>,
    ) -> Result<ApplyReport, ApplyError> {
        let (prepared, token) = self.prepare_tree(bundle).await?;
        if let Some(expected) = expected_token {
            if token != expected {
                return Err(ApplyError::StateMoved);
            }
        }

        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?;
        // Lock every node's apply key, in sorted order, so concurrent applies
        // touching any shared node serialize and the ordering can't deadlock.
        let mut lock_keys: Vec<i64> = prepared.iter().map(|p| apply_lock_key(p.owner)).collect();
        lock_keys.sort_unstable();
        lock_keys.dedup();
        for k in lock_keys {
            sqlx::query("SELECT pg_advisory_xact_lock($1)")
                .bind(k)
                .execute(&mut *tx)
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
        }

        let mut report = ApplyReport::default();
        let mut group_script_index: HashMap<GroupId, HashMap<String, ScriptId>> = HashMap::new();
        let mut any_app = false;

        // Phase A — groups first: reconcile each and record its post-create
        // `name -> id` so descendant app nodes can resolve inherited bindings.
        for p in &prepared {
            if let ApplyOwner::Group(gid) = p.owner {
                report
                    .warnings
                    .extend(disabled_target_warnings(&p.node.bundle));
                let name_to_id = self
                    .reconcile_node_tx(
                        &mut tx,
                        p.owner,
                        &p.node.bundle,
                        &p.plan,
                        &p.current,
                        &p.current_names,
                        &HashMap::new(),
                        prune,
                        actor,
                        &mut report,
                    )
                    .await?;
                group_script_index.insert(gid, name_to_id);
            }
        }

        // Phase B — apps: resolve inherited targets against in-tree groups (the
        // ids just created) and pre-existing ancestors, then reconcile.
        for p in &prepared {
            if let ApplyOwner::App(_) = p.owner {
                any_app = true;
                report
                    .warnings
                    .extend(disabled_target_warnings(&p.node.bundle));
                report
                    .warnings
                    .extend(unreachable_endpoint_warnings(&p.node.bundle));
                let inherited = self
                    .tree_inherited_for_app(&p.app_chain, &p.node.bundle, &group_script_index)
                    .await?;
                self.reconcile_node_tx(
                    &mut tx,
                    p.owner,
                    &p.node.bundle,
                    &p.plan,
                    &p.current,
                    &p.current_names,
                    &inherited,
                    prune,
                    actor,
                    &mut report,
                )
                .await?;
            }
        }

        tx.commit()
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?;
        // One post-commit route refresh covers every app node touched.
        if any_app {
            if let Err(e) = self.refresh_route_table().await {
                tracing::warn!(error = %e, "apply_tree: route table refresh failed");
                report
                    .warnings
                    .push("route table refresh failed; it will self-heal".into());
            }
        }
        Ok(report)
    }

    /// Resolve nodes to owners, validate each (apps' route/trigger targets may
    /// bind to in-tree ancestor group scripts), compute each node's plan, and
    /// fold a combined bound-plan token. Shared by `plan_tree` / `apply_tree`.
    #[allow(clippy::too_many_lines)]
    async fn prepare_tree<'a>(
        &self,
        bundle: &'a TreeBundle,
    ) -> Result<(Vec<PreparedNode<'a>>, String), ApplyError> {
        if bundle.nodes.is_empty() {
            return Err(ApplyError::Invalid("project tree has no nodes".into()));
        }
        // Register in-tree group nodes first: their (resolved) ids and endpoint
        // script names, which an app node's binding validation may reference.
        let mut group_id_by_slug: HashMap<String, GroupId> = HashMap::new();
        let mut group_script_names: HashMap<GroupId, HashSet<String>> = HashMap::new();
        let mut seen_slugs: HashSet<String> = HashSet::new();
        for n in &bundle.nodes {
            let key = format!("{:?}:{}", n.kind, n.slug.to_lowercase());
            if !seen_slugs.insert(key) {
                return Err(ApplyError::Invalid(format!(
                    "project tree lists node `{}` more than once",
                    n.slug
                )));
            }
            if n.kind == NodeKind::Group {
                let gid = self.resolve_group(&n.slug).await?;
                group_id_by_slug.insert(n.slug.to_lowercase(), gid);
                group_script_names.insert(
                    gid,
                    n.bundle
                        .scripts
                        .iter()
                        .filter(|s| s.kind == ScriptKind::Endpoint)
                        .map(|s| s.name.to_lowercase())
                        .collect(),
                );
            }
        }

        let mut prepared = Vec::with_capacity(bundle.nodes.len());
        let mut token_parts: Vec<String> = Vec::new();
        let mut versioned_groups: BTreeSet<GroupId> = BTreeSet::new();

        for n in &bundle.nodes {
            let (owner, app_chain) = match n.kind {
                NodeKind::Group => {
                    let gid = group_id_by_slug[&n.slug.to_lowercase()];
                    versioned_groups.insert(gid);
                    (ApplyOwner::Group(gid), Vec::new())
                }
                NodeKind::App => {
                    let app = self.resolve_app(&n.slug).await?;
                    let chain = self
                        .groups
                        .ancestors(app.group_id)
                        .await
                        .map_err(|e| ApplyError::Backend(e.to_string()))?;
                    for g in &chain {
                        versioned_groups.insert(g.id);
                    }
                    (
                        ApplyOwner::App(app.id),
                        chain.iter().map(|g| g.id).collect::<Vec<_>>(),
                    )
                }
            };

            // Inherited endpoint NAMES the bindings may reference: pre-existing
            // (pool resolver) plus in-tree ancestor group scripts.
            let inherited_names: HashSet<String> = match owner {
                ApplyOwner::App(app_id) => {
                    let mut names =
                        keys_set(&self.resolve_inherited_targets(app_id, &n.bundle).await?);
                    for gid in &app_chain {
                        if let Some(s) = group_script_names.get(gid) {
                            names.extend(s.iter().cloned());
                        }
                    }
                    names
                }
                ApplyOwner::Group(_) => HashSet::new(),
            };
            self.validate_bundle_for(owner, &n.bundle, &inherited_names)?;
            if let ApplyOwner::App(app_id) = owner {
                self.validate_route_hosts(app_id, &n.bundle).await?;
            }

            let current = self.load_current(owner).await?;
            let current_names = self.resolve_current_target_names(&current).await?;
            validate_email_secrets_present(&n.bundle, &current.secret_names)?;
            let plan = compute_diff_with_names(&current, &n.bundle, &current_names);
            token_parts.push(format!(
                "n|{}|{}",
                n.slug.to_lowercase(),
                state_token_with_names(&current, &current_names)
            ));
            prepared.push(PreparedNode {
                owner,
                node: n,
                current,
                current_names,
                plan,
                app_chain,
            });
        }

        // §5.5 / Phase 4b: the single-node path runs `check_imports_resolve` and
        // `check_extension_points_provided`, but the tree path cannot — a node
        // may legitimately import a module another in-tree (uncommitted) node
        // declares, so a pool-based provider check would false-positive. Both
        // lean on the runtime backstop here. Log it so the gap isn't silent.
        if prepared
            .iter()
            .any(|p| matches!(p.owner, ApplyOwner::App(_)))
        {
            tracing::debug!(
                nodes = prepared.len(),
                "tree apply: per-node import / extension-point provider checks \
                 deferred to the runtime backstop (single-node-only at plan time)"
            );
        }

        // Fold in every in-scope group's structure version, so a reparent or a
        // new app under the subtree between plan and apply trips StateMoved.
        for gid in &versioned_groups {
            if let Some(g) = self
                .groups
                .get_by_id(*gid)
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?
            {
                token_parts.push(format!("gs|{gid}|{}", g.structure_version));
            }
        }
        token_parts.sort_unstable();
        Ok((prepared, combine_tokens(&token_parts)))
    }

    /// Resolve an app node's inherited route/trigger targets to script ids,
    /// nearest-ancestor-wins, considering BOTH in-tree group nodes (their
    /// in-tx `name -> id`, incl. just-created scripts) and pre-existing
    /// out-of-tree ancestor groups (their committed rows).
    async fn tree_inherited_for_app(
        &self,
        app_chain: &[GroupId],
        bundle: &Bundle,
        index: &HashMap<GroupId, HashMap<String, ScriptId>>,
    ) -> Result<HashMap<String, ScriptId>, ApplyError> {
        let own: HashSet<String> = bundle
            .scripts
            .iter()
            .map(|s| s.name.to_lowercase())
            .collect();
        let mut referenced: HashSet<String> = HashSet::new();
        for r in &bundle.routes {
            referenced.insert(r.script.to_lowercase());
        }
        for t in &bundle.triggers {
            referenced.insert(t.script().to_lowercase());
        }
        referenced.retain(|n| !own.contains(n));
        if referenced.is_empty() {
            return Ok(HashMap::new());
        }
        // Per-ancestor name→id maps, nearest-first.
        let mut chain_maps: Vec<HashMap<String, ScriptId>> = Vec::with_capacity(app_chain.len());
        for gid in app_chain {
            if let Some(idx) = index.get(gid) {
                chain_maps.push(idx.clone());
            } else {
                let m = self
                    .scripts
                    .list_for_group(*gid)
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?
                    .into_iter()
                    .filter(|s| s.kind == ScriptKind::Endpoint)
                    .map(|s| (s.name.to_lowercase(), s.id))
                    .collect();
                chain_maps.push(m);
            }
        }
        let mut out = HashMap::new();
        for name in referenced {
            for m in &chain_maps {
                if let Some(id) = m.get(&name) {
                    out.insert(name.clone(), *id);
                    break;
                }
            }
        }
        Ok(out)
    }

    async fn resolve_app(&self, ident: &str) -> Result<picloud_shared::App, ApplyError> {
        crate::app_repo::resolve_app(self.apps.as_ref(), ident)
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?
            .map(|l| l.app)
            .ok_or_else(|| ApplyError::AppNotFound(ident.to_string()))
    }

    async fn resolve_group(&self, ident: &str) -> Result<GroupId, ApplyError> {
        let found = if let Ok(uuid) = ident.parse::<uuid::Uuid>() {
            self.groups.get_by_id(uuid.into()).await
        } else {
            self.groups.get_by_slug(ident).await
        };
        found
            .map_err(|e| ApplyError::Backend(e.to_string()))?
            .map(|g| g.id)
            .ok_or_else(|| ApplyError::AppNotFound(format!("group {ident}")))
    }

    fn script_imports(&self, bs: &BundleScript) -> Result<Vec<String>, ApplyError> {
        let res = if bs.kind == ScriptKind::Module {
            self.validator.validate_module(&bs.source)
        } else {
            self.validator.validate(&bs.source)
        };
        res.map(|v| v.imports)
            .map_err(|e| ApplyError::Invalid(format!("script `{}`: {e}", bs.name)))
    }

    fn trigger_settings(
        &self,
        bt: &BundleTrigger,
    ) -> (
        TriggerDispatchMode,
        u32,
        crate::trigger_config::BackoffShape,
        u32,
    ) {
        let (dispatch, retry_max) = match bt {
            BundleTrigger::Kv {
                dispatch_mode,
                retry_max_attempts,
                ..
            }
            | BundleTrigger::Docs {
                dispatch_mode,
                retry_max_attempts,
                ..
            }
            | BundleTrigger::Files {
                dispatch_mode,
                retry_max_attempts,
                ..
            }
            | BundleTrigger::Cron {
                dispatch_mode,
                retry_max_attempts,
                ..
            }
            | BundleTrigger::Pubsub {
                dispatch_mode,
                retry_max_attempts,
                ..
            }
            | BundleTrigger::Email {
                dispatch_mode,
                retry_max_attempts,
                ..
            }
            | BundleTrigger::Queue {
                dispatch_mode,
                retry_max_attempts,
                ..
            } => (
                dispatch_mode.unwrap_or(TriggerDispatchMode::Async),
                retry_max_attempts.unwrap_or(self.trigger_config.retry_max_attempts),
            ),
        };
        (
            dispatch,
            retry_max,
            self.trigger_config.retry_backoff,
            self.trigger_config.retry_base_ms,
        )
    }

    async fn refresh_route_table(&self) -> Result<(), ApplyError> {
        let rows = self
            .routes
            .list_all()
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?;
        let compiled = crate::route_admin::compile_routes(&rows);
        self.route_table.replace_all(compiled);
        Ok(())
    }

    /// Resolve a referenced secret to plaintext, then re-seal it for an
    /// email trigger's inbound-secret column. The value never appears in
    /// the manifest — only the secret's name does.
    async fn resolve_and_seal(
        &self,
        app_id: AppId,
        name: &str,
    ) -> Result<(Vec<u8>, Vec<u8>), ApplyError> {
        let stored = self
            .secrets
            .get(crate::secrets_service::SecretOwner::App(app_id), "*", name)
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?
            .ok_or_else(|| {
                ApplyError::Invalid(format!(
                    "email trigger references secret `{name}`, which is not set \
                     (push it with `pic secret set {name}`)"
                ))
            })?;
        let plaintext = crate::secrets_service::open(
            &self.master_key,
            crate::secrets_service::SecretOwner::App(app_id),
            name,
            &stored,
        )
        .map_err(|e| ApplyError::Invalid(format!("could not read secret `{name}`: {e}")))?;
        // The inbound HMAC must be a non-empty string — an empty/whitespace
        // key is forgeable (preserves the spirit of audit 2026-06-11 H-B2).
        match plaintext.as_str() {
            Some(s) if !s.trim().is_empty() => {}
            _ => {
                return Err(ApplyError::Invalid(format!(
                    "secret `{name}` must be a non-empty string to sign an email trigger"
                )))
            }
        }
        let (ct, nonce) = crate::secrets_service::seal_legacy(
            &self.master_key,
            &plaintext,
            crate::secrets_service::DEFAULT_SECRET_MAX_VALUE_BYTES,
        )
        .map_err(|e| ApplyError::Invalid(format!("could not seal secret `{name}`: {e}")))?;
        Ok((ct, nonce.to_vec()))
    }

    /// Validate each bundle route's host against the app's domain claims —
    /// the cross-state check the interactive route API enforces. Async, so
    /// it runs alongside `plan`/`apply`, not in the sync `validate_bundle`.
    async fn validate_route_hosts(&self, app_id: AppId, bundle: &Bundle) -> Result<(), ApplyError> {
        for r in &bundle.routes {
            crate::route_admin::validate_route_host_against_app(
                self.domains.as_ref(),
                app_id,
                r.host_kind,
                &r.host,
            )
            .await
            .map_err(|e| ApplyError::Invalid(format!("route host claim: {e}")))?;
        }
        Ok(())
    }

    /// Bundle-local validation: catches the errors visible without the
    /// live state (syntax, ceilings, dangling/wrong-kind targets, intra-
    /// bundle duplicates, reserved paths, host/path structural validity).
    /// Host-claim validation runs separately in `validate_route_hosts`.
    ///
    /// NOTE (known gap vs. the interactive route API): cross-pattern route
    /// conflict-vs-live (e.g. a param route overlapping a live exact route)
    /// is NOT checked — the DB unique index catches only identical tuples,
    /// so overlapping patterns are accepted.
    #[allow(clippy::too_many_lines)]
    /// Phase 4: resolve route/trigger target names that the manifest does NOT
    /// declare as app-own scripts to inherited group-owned endpoint scripts on
    /// the app's chain. Returns `name(lowercased) → script_id` for the ones
    /// that resolve to a group endpoint. App-own names (in `bundle.scripts`)
    /// are skipped — they bind through the normal `name_to_id` path and take
    /// precedence (CoW). Resolved before the apply transaction; the group
    /// scripts pre-exist (committed), so a pool read is correct and stable.
    async fn resolve_inherited_targets(
        &self,
        app_id: AppId,
        bundle: &Bundle,
    ) -> Result<HashMap<String, ScriptId>, ApplyError> {
        let own: HashSet<String> = bundle
            .scripts
            .iter()
            .map(|s| s.name.to_lowercase())
            .collect();
        let mut referenced: HashSet<String> = HashSet::new();
        for r in &bundle.routes {
            referenced.insert(r.script.to_lowercase());
        }
        for t in &bundle.triggers {
            referenced.insert(t.script().to_lowercase());
        }
        let mut out = HashMap::new();
        for name in referenced {
            if own.contains(&name) {
                continue;
            }
            if let Some(s) = self
                .scripts
                .get_by_name_inherited(app_id, &name)
                .await
                .map_err(map_repo)?
            {
                // Only genuinely-inherited group endpoints qualify as
                // route/trigger targets. A depth-0 app-own match is already
                // covered by `name_to_id`; skip it so the manifest keeps
                // precedence. Group modules (Phase 4b) exist but are never
                // bindable targets — the kind guard excludes them.
                if s.group_id.is_some() && s.kind == ScriptKind::Endpoint {
                    out.insert(name, s.id);
                }
            }
        }
        Ok(out)
    }

    /// Phase 4: resolve the names of group-owned scripts the app's CURRENT
    /// routes/triggers are bound to, `script_id → name`. These ids aren't in
    /// `current.scripts` (app-own only), so this is what lets the diff /
    /// state-token / prune resolve an existing inherited binding back to its
    /// name — independent of whether the bundle still references it (so a
    /// dropped inherited binding still diffs to a Delete, and an inherited-bound
    /// trigger still moves the state token).
    async fn resolve_current_target_names(
        &self,
        current: &CurrentState,
    ) -> Result<HashMap<ScriptId, String>, ApplyError> {
        let own: HashSet<ScriptId> = current.scripts.iter().map(|s| s.id).collect();
        let mut ids: HashSet<ScriptId> = HashSet::new();
        for r in &current.routes {
            if !own.contains(&r.script_id) {
                ids.insert(r.script_id);
            }
        }
        for t in &current.triggers {
            if !own.contains(&t.script_id) {
                ids.insert(t.script_id);
            }
        }
        let mut out = HashMap::new();
        for id in ids {
            if let Some(s) = self.scripts.get(id).await.map_err(map_repo)? {
                out.insert(id, s.name);
            }
        }
        Ok(out)
    }

    /// §5.5 dangling-import plan check (Phase 4b, single-node path). Every
    /// `import "<m>"` in a bundle script must resolve to a module declared in
    /// the same bundle OR reachable lexically up the node's chain — otherwise
    /// the apply would write a script that 404s its import at runtime. Lexical:
    /// the node's owner IS the defining node for its own scripts, so resolution
    /// roots at `owner` (an app node reaches ancestor group modules; a group
    /// node walks its own ancestry, sealed from apps below).
    ///
    /// Not run on the tree path: a tree node may import a module declared by
    /// another node created in the same transaction (not yet committed), so the
    /// in-tree set would have to be threaded through; the runtime check is the
    /// backstop there.
    async fn check_imports_resolve(
        &self,
        owner: ApplyOwner,
        bundle: &Bundle,
    ) -> Result<(), ApplyError> {
        // Modules this bundle declares (lowercased) satisfy their own imports.
        let local: HashSet<String> = bundle
            .scripts
            .iter()
            .filter(|s| s.kind == ScriptKind::Module)
            .map(|s| s.name.to_lowercase())
            .collect();
        let origin = match owner {
            ApplyOwner::App(a) => picloud_shared::ScriptOwner::App(a),
            ApplyOwner::Group(g) => picloud_shared::ScriptOwner::Group(g),
        };
        // Resolve each distinct imported name once.
        let mut checked: HashSet<String> = HashSet::new();
        for s in &bundle.scripts {
            for imp in self.script_imports(s)? {
                let key = imp.to_lowercase();
                if local.contains(&key) || !checked.insert(key) {
                    continue;
                }
                let found = self
                    .modules
                    .resolve(origin, &imp)
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?;
                if found.is_none() {
                    return Err(ApplyError::Invalid(format!(
                        "script `{}` imports unknown module `{imp}` \
                         (not declared here and not inherited from an ancestor)",
                        s.name
                    )));
                }
            }
        }
        Ok(())
    }

    /// Read-only extension-point report for a node (§5.5). For an **app**:
    /// every EP visible on its chain, each flagged `declared_here` (owned by
    /// the app, vs inherited) with its resolved `provider` (`app override` /
    /// `inherited default` / `null` = unset). For a **group**: its own declared
    /// names (no single app to resolve a provider against). Backs the read-only
    /// `extension-points ls` and the `pull` round-trip.
    pub async fn extension_point_report(
        &self,
        owner: ApplyOwner,
    ) -> Result<Vec<ExtensionPointInfo>, ApplyError> {
        let pool = &self.pool;
        match owner {
            ApplyOwner::Group(g) => {
                let names =
                    crate::extension_point_repo::list_for_owner(pool, ScriptOwner::Group(g))
                        .await
                        .map_err(|e| ApplyError::Backend(e.to_string()))?;
                Ok(names
                    .into_iter()
                    .map(|name| ExtensionPointInfo {
                        name,
                        declared_here: true,
                        provider: None,
                    })
                    .collect())
            }
            ApplyOwner::App(a) => {
                let own: HashSet<String> =
                    crate::extension_point_repo::list_for_owner(pool, ScriptOwner::App(a))
                        .await
                        .map_err(|e| ApplyError::Backend(e.to_string()))?
                        .into_iter()
                        .map(|n| n.to_lowercase())
                        .collect();
                let visible = crate::extension_point_repo::list_on_app_chain(pool, a)
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?;
                let mut out = Vec::with_capacity(visible.len());
                for name in visible {
                    let provider = match self
                        .modules
                        .resolve(ScriptOwner::App(a), &name)
                        .await
                        .map_err(|e| ApplyError::Backend(e.to_string()))?
                        .and_then(|m| m.owner())
                    {
                        Some(ScriptOwner::App(_)) => Some("app override".to_string()),
                        Some(ScriptOwner::Group(_)) => Some("inherited default".to_string()),
                        None => None,
                    };
                    out.push(ExtensionPointInfo {
                        declared_here: own.contains(&name.to_lowercase()),
                        name,
                        provider,
                    });
                }
                Ok(out)
            }
        }
    }

    /// Read-only §11.6 shared-collection report for a node: the shared
    /// collections (`name` + `kind`) declared directly at it. Group-only in
    /// practice (the CLI rejects app-declared collections). Backs
    /// `pic collections ls`.
    pub async fn collection_report(
        &self,
        owner: ApplyOwner,
    ) -> Result<Vec<CollectionInfo>, ApplyError> {
        let rows =
            crate::group_collection_repo::list_all_for_owner(&self.pool, owner.as_script_owner())
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
        Ok(rows
            .into_iter()
            .map(|(name, kind)| CollectionInfo { name, kind })
            .collect())
    }

    /// §5.5 no-provider plan check (app nodes only). Every extension point
    /// visible to the app — declared in this bundle or inherited from an
    /// ancestor — must have a provider: a module of that name the app provides
    /// (in this bundle, or resolvable up its chain as an override or a default
    /// body). Otherwise the import would fail at runtime, so refuse at plan.
    /// Group nodes have no single inheriting app, so the check is per-app.
    /// Single-node only (the tree path leans on the runtime backstop).
    async fn check_extension_points_provided(
        &self,
        owner: ApplyOwner,
        bundle: &Bundle,
    ) -> Result<(), ApplyError> {
        let ApplyOwner::App(app_id) = owner else {
            return Ok(());
        };
        // A same-apply app module provides its EP without being committed yet.
        let local: HashSet<String> = bundle
            .scripts
            .iter()
            .filter(|s| s.kind == ScriptKind::Module)
            .map(|s| s.name.to_lowercase())
            .collect();
        // EP names visible to the app: its own (this bundle) ∪ inherited.
        let mut names: HashSet<String> = bundle
            .extension_points
            .iter()
            .map(|n| n.to_lowercase())
            .collect();
        for n in crate::extension_point_repo::list_on_app_chain(&self.pool, app_id)
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?
        {
            names.insert(n.to_lowercase());
        }
        for name in names {
            if local.contains(&name) {
                continue;
            }
            let found = self
                .modules
                .resolve(picloud_shared::ScriptOwner::App(app_id), &name)
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
            if found.is_none() {
                return Err(ApplyError::Invalid(format!(
                    "extension point `{name}` has no provider for this app — \
                     declare a module named `{name}` here or ensure a default \
                     body exists up-chain"
                )));
            }
        }
        Ok(())
    }

    /// `inherited_endpoints` (lowercased names) are group-owned endpoint
    /// scripts reachable from the app's chain that a route/trigger may bind to
    /// even though the manifest doesn't declare them (Phase 4). They count as
    /// known endpoints for the binding checks below.
    #[allow(clippy::too_many_lines)]
    fn validate_bundle(
        &self,
        bundle: &Bundle,
        inherited_endpoints: &HashSet<String>,
    ) -> Result<(), ApplyError> {
        // Scripts: unique names; each source compiles for its kind.
        let mut names: HashSet<String> = HashSet::new();
        for s in &bundle.scripts {
            if !names.insert(s.name.to_lowercase()) {
                return Err(ApplyError::Invalid(format!(
                    "duplicate script name `{}`",
                    s.name
                )));
            }
            let res = if s.kind == ScriptKind::Module {
                // Parity with the interactive create path (`api.rs`): a module
                // may not shadow a built-in SDK namespace, or `import "kv"`
                // could resolve to a user module instead of the real bridge.
                if crate::api::RESERVED_MODULE_NAMES.contains(&s.name.as_str()) {
                    return Err(ApplyError::Invalid(format!(
                        "script `{}`: reserved module name (shadows a built-in SDK namespace)",
                        s.name
                    )));
                }
                self.validator.validate_module(&s.source)
            } else {
                self.validator.validate(&s.source)
            };
            res.map_err(|e| ApplyError::Invalid(format!("script `{}`: {e}", s.name)))?;
            if let Some(sb) = &s.sandbox {
                self.sandbox_ceiling
                    .check(sb)
                    .map_err(|e| ApplyError::Invalid(format!("script `{}`: {e}", s.name)))?;
            }
        }
        let endpoints: HashSet<&str> = bundle
            .scripts
            .iter()
            .filter(|s| s.kind == ScriptKind::Endpoint)
            .map(|s| s.name.as_str())
            .collect();
        let all_names: HashSet<&str> = bundle.scripts.iter().map(|s| s.name.as_str()).collect();

        // Routes: reserved paths; bind to an endpoint script; unique tuples.
        let mut route_keys: HashSet<String> = HashSet::new();
        for r in &bundle.routes {
            reject_reserved_path(&r.path)?;
            // Structural validity (parity with the interactive route API):
            // a route whose path/host doesn't parse would otherwise be
            // written and then silently dropped by `compile_routes`.
            pattern::parse_path(r.path_kind, &r.path)
                .map_err(|e| ApplyError::Invalid(format!("route path `{}`: {e}", r.path)))?;
            pattern::parse_host(r.host_kind, &r.host, r.host_param_name.as_deref())
                .map_err(|e| ApplyError::Invalid(format!("route host `{}`: {e}", r.host)))?;
            let inherited = inherited_endpoints.contains(&r.script.to_lowercase());
            if !all_names.contains(r.script.as_str()) && !inherited {
                return Err(ApplyError::Invalid(format!(
                    "route {} binds to unknown script `{}`",
                    route_key(
                        r.method.as_deref(),
                        r.host_kind,
                        &r.host,
                        r.path_kind,
                        &r.path
                    ),
                    r.script
                )));
            }
            // Inherited targets are resolved as endpoints by construction
            // (the inherited-name resolver excludes group modules); only
            // check kind for app-own names.
            if !inherited && !endpoints.contains(r.script.as_str()) {
                return Err(ApplyError::Invalid(format!(
                    "route binds to `{}` which is a module, not an endpoint",
                    r.script
                )));
            }
            let key = route_key(
                r.method.as_deref(),
                r.host_kind,
                &r.host,
                r.path_kind,
                &r.path,
            );
            if !route_keys.insert(key.clone()) {
                return Err(ApplyError::Invalid(format!("duplicate route `{key}`")));
            }
        }

        // Triggers: bind to an endpoint script; unique semantic identity.
        let mut trig_keys: HashSet<String> = HashSet::new();
        for t in &bundle.triggers {
            let inherited = inherited_endpoints.contains(&t.script().to_lowercase());
            if !all_names.contains(t.script()) && !inherited {
                return Err(ApplyError::Invalid(format!(
                    "{} trigger binds to unknown script `{}`",
                    t.kind_str(),
                    t.script()
                )));
            }
            if !inherited && !endpoints.contains(t.script()) {
                return Err(ApplyError::Invalid(format!(
                    "{} trigger binds to `{}` which is a module, not an endpoint",
                    t.kind_str(),
                    t.script()
                )));
            }
            validate_trigger_shape(t)?;
            let id = t.identity();
            if !trig_keys.insert(id.clone()) {
                return Err(ApplyError::Invalid(format!("duplicate trigger `{id}`")));
            }
        }

        // Vars: kebab keys (same rule the vars admin API enforces), so a
        // manifest can't write a key the interactive surface would reject.
        for key in bundle.vars.keys() {
            validate_var_key(key)?;
        }

        // Extension points (§5.5): unique names; none may shadow a built-in
        // SDK namespace (same guard as module names — an `import "kv"` must
        // never resolve to a user-supplied provider).
        let mut ep_names: HashSet<String> = HashSet::new();
        for name in &bundle.extension_points {
            if !ep_names.insert(name.to_lowercase()) {
                return Err(ApplyError::Invalid(format!(
                    "duplicate extension point `{name}`"
                )));
            }
            if crate::api::RESERVED_MODULE_NAMES.contains(&name.as_str()) {
                return Err(ApplyError::Invalid(format!(
                    "extension point `{name}`: reserved module name \
                     (shadows a built-in SDK namespace)"
                )));
            }
        }

        // Shared group collections (§11.6): a known kind + unique (name, kind).
        // No reserved-name guard — a collection name is a data namespace, not an
        // importable module, so it can't shadow an SDK namespace.
        let mut seen_collections: HashSet<(String, &str)> = HashSet::new();
        for c in &bundle.collections {
            if c.name.is_empty() {
                return Err(ApplyError::Invalid(
                    "shared collection name must not be empty".into(),
                ));
            }
            if !COLLECTION_KINDS.contains(&c.kind.as_str()) {
                return Err(ApplyError::Invalid(format!(
                    "shared collection `{}`: unknown kind `{}` (want one of {COLLECTION_KINDS:?})",
                    c.name, c.kind
                )));
            }
            if !seen_collections.insert((c.name.to_lowercase(), c.kind.as_str())) {
                return Err(ApplyError::Invalid(format!(
                    "duplicate shared collection `{}` (kind `{}`)",
                    c.name, c.kind
                )));
            }
        }
        Ok(())
    }

    async fn load_current(&self, owner: ApplyOwner) -> Result<CurrentState, ApplyError> {
        // A group node has only scripts + vars; routes / triggers / secrets are
        // app-only and stay empty (the bundle for a group declares none).
        let (scripts, routes, triggers, secret_names, var_owner) = match owner {
            ApplyOwner::App(app_id) => (
                self.scripts
                    .list_for_app(app_id)
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?,
                self.routes
                    .list_for_app(app_id)
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?,
                self.triggers
                    .list_for_app(app_id)
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?,
                self.all_secret_names(app_id).await?,
                VarOwner::App(app_id),
            ),
            ApplyOwner::Group(group_id) => (
                self.scripts
                    .list_for_group(group_id)
                    .await
                    .map_err(|e| ApplyError::Backend(e.to_string()))?,
                Vec::new(),
                Vec::new(),
                Vec::new(),
                VarOwner::Group(group_id),
            ),
        };
        // The owner's OWN vars, narrowed to scope `*` and real values: the
        // manifest reconciles env-agnostic config, not inherited vars (not in
        // this owner's rows) nor tombstones (an imperative suppress the manifest
        // can't express).
        let vars = self
            .vars
            .list_for_owner(var_owner)
            .await
            .map_err(|e| ApplyError::Backend(e.to_string()))?
            .into_iter()
            .filter(|r| r.environment_scope == "*" && !r.is_tombstone)
            .map(|r| (r.key, r.value))
            .collect();
        // Extension-point markers declared directly at this node (§5.5).
        let extension_point_names =
            crate::extension_point_repo::list_for_owner(&self.pool, owner.as_script_owner())
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
        // Shared group-collection markers declared directly at this node
        // (§11.6), all kinds.
        let collections =
            crate::group_collection_repo::list_all_for_owner(&self.pool, owner.as_script_owner())
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
        Ok(CurrentState {
            scripts,
            routes,
            triggers,
            secret_names,
            vars,
            extension_point_names,
            collections,
        })
    }

    async fn all_secret_names(&self, app_id: AppId) -> Result<Vec<String>, ApplyError> {
        let mut names = Vec::new();
        let mut cursor: Option<String> = None;
        loop {
            let page = self
                .secrets
                .list_names(
                    crate::secrets_service::SecretOwner::App(app_id),
                    cursor.as_deref(),
                    200,
                )
                .await
                .map_err(|e| ApplyError::Backend(e.to_string()))?;
            names.extend(page.names);
            match page.next_cursor {
                Some(c) => cursor = Some(c),
                None => break,
            }
        }
        Ok(names)
    }
}

// ----------------------------------------------------------------------------
// Diff (pure)
// ----------------------------------------------------------------------------

/// Diff desired `bundle` against `current`. Pure — unit-tested directly.
#[must_use]
pub fn compute_diff(current: &CurrentState, bundle: &Bundle) -> Plan {
    compute_diff_with_names(current, bundle, &HashMap::new())
}

/// `compute_diff` with `current_names` — the names of group-owned (inherited)
/// scripts the app's CURRENT routes/triggers are bound to, `script_id → name`
/// (Phase 4). Those ids aren't in `current.scripts` (app-own only), so without
/// them a current binding to a group script resolves to no name: it reads as a
/// perpetual rebind, and a binding dropped from the manifest never diffs to a
/// Delete (so `--prune` can't remove it). The public `compute_diff` passes an
/// empty map (no inheritance).
fn compute_diff_with_names(
    current: &CurrentState,
    bundle: &Bundle,
    current_names: &HashMap<ScriptId, String>,
) -> Plan {
    let script_name_by_id: HashMap<ScriptId, String> = current
        .scripts
        .iter()
        .map(|s| (s.id, s.name.clone()))
        // Inherited group bindings' ids → names. Disjoint from app-own ids by
        // construction, so order doesn't matter.
        .chain(current_names.iter().map(|(id, n)| (*id, n.clone())))
        .collect();

    Plan {
        scripts: diff_scripts(current, bundle),
        routes: diff_routes(current, bundle, &script_name_by_id),
        triggers: diff_triggers(current, bundle, &script_name_by_id),
        secrets: diff_secrets(current, bundle),
        vars: diff_vars(current, bundle),
        extension_points: diff_extension_points(current, bundle),
        collections: diff_collections(current, bundle),
    }
}

/// Diff app-owned vars by key. Unlike secrets (whose values live out-of-band),
/// a var's value IS in the manifest, so equality is value-sensitive: a changed
/// value is an `Update`. Live vars absent from the manifest are `Delete`
/// (applied only under `--prune`).
fn diff_vars(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
    let live: HashMap<&str, &serde_json::Value> =
        current.vars.iter().map(|(k, v)| (k.as_str(), v)).collect();

    let mut out = Vec::new();
    for (key, value) in &bundle.vars {
        match live.get(key.as_str()) {
            Some(cur) if *cur == value => out.push(ResourceChange {
                op: Op::NoOp,
                key: key.clone(),
                detail: None,
            }),
            Some(_) => out.push(ResourceChange {
                op: Op::Update,
                key: key.clone(),
                detail: Some("value changed".into()),
            }),
            None => out.push(ResourceChange {
                op: Op::Create,
                key: key.clone(),
                detail: None,
            }),
        }
    }
    for (key, _) in &current.vars {
        if !bundle.vars.contains_key(key) {
            out.push(ResourceChange {
                op: Op::Delete,
                key: key.clone(),
                detail: Some("on server, not declared".into()),
            });
        }
    }
    out
}

/// Kebab var key (`^[a-z0-9][a-z0-9-]{0,127}$`) — mirrors `vars_api`'s
/// `validate_key` so the manifest and the interactive surface agree.
fn validate_var_key(key: &str) -> Result<(), ApplyError> {
    let ok = !key.is_empty()
        && key.len() <= 128
        && key
            .chars()
            .next()
            .is_some_and(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
        && key
            .chars()
            .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-');
    if ok {
        Ok(())
    } else {
        Err(ApplyError::Invalid(format!(
            "var key `{key}` must be 1–128 chars of lowercase letters, digits, and hyphens, \
             starting with a letter or digit"
        )))
    }
}

/// Upsert one app-owned var at scope `*`, in the apply transaction. Mirrors
/// `PostgresVarsRepo::set` for `VarOwner::App` (the partial-index predicate is
/// restated in the conflict target) but runs against `&mut tx` so it commits
/// atomically with the rest of the apply. Clears any tombstone.
async fn set_app_var_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    app_id: AppId,
    key: &str,
    value: &serde_json::Value,
) -> Result<(), ApplyError> {
    sqlx::query(
        "INSERT INTO vars (app_id, environment_scope, key, value, is_tombstone) \
         VALUES ($1, '*', $2, $3, FALSE) \
         ON CONFLICT (app_id, environment_scope, key) WHERE app_id IS NOT NULL DO UPDATE \
           SET value = EXCLUDED.value, is_tombstone = FALSE, updated_at = NOW()",
    )
    .bind(app_id.into_inner())
    .bind(key)
    .bind(value)
    .execute(&mut **tx)
    .await
    .map_err(|e| ApplyError::Backend(e.to_string()))?;
    Ok(())
}

/// Delete one app-owned var at scope `*`, in the apply transaction.
async fn delete_app_var_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    app_id: AppId,
    key: &str,
) -> Result<(), ApplyError> {
    sqlx::query("DELETE FROM vars WHERE app_id = $1 AND environment_scope = '*' AND key = $2")
        .bind(app_id.into_inner())
        .bind(key)
        .execute(&mut **tx)
        .await
        .map_err(|e| ApplyError::Backend(e.to_string()))?;
    Ok(())
}

/// Upsert one group-owned var at scope `*`, in the apply transaction (Phase 5).
/// Mirrors `set_app_var_tx` for `VarOwner::Group`. Env-scoped group vars
/// (`@staging`) stay imperative for now — a group manifest declares only
/// env-agnostic (`*`) group config.
async fn set_group_var_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    group_id: GroupId,
    key: &str,
    value: &serde_json::Value,
) -> Result<(), ApplyError> {
    sqlx::query(
        "INSERT INTO vars (group_id, environment_scope, key, value, is_tombstone) \
         VALUES ($1, '*', $2, $3, FALSE) \
         ON CONFLICT (group_id, environment_scope, key) WHERE group_id IS NOT NULL DO UPDATE \
           SET value = EXCLUDED.value, is_tombstone = FALSE, updated_at = NOW()",
    )
    .bind(group_id.into_inner())
    .bind(key)
    .bind(value)
    .execute(&mut **tx)
    .await
    .map_err(|e| ApplyError::Backend(e.to_string()))?;
    Ok(())
}

/// Delete one group-owned var at scope `*`, in the apply transaction (Phase 5).
async fn delete_group_var_tx(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    group_id: GroupId,
    key: &str,
) -> Result<(), ApplyError> {
    sqlx::query("DELETE FROM vars WHERE group_id = $1 AND environment_scope = '*' AND key = $2")
        .bind(group_id.into_inner())
        .bind(key)
        .execute(&mut **tx)
        .await
        .map_err(|e| ApplyError::Backend(e.to_string()))?;
    Ok(())
}

/// Owner of an apply node (Phase 5): an app or a group. The apply engine is
/// owner-generic — only script-create ownership, var writes, the current-state
/// load, and the advisory lock differ; the diff and all other CRUD are shared.
/// A group node has only scripts + vars (no routes / triggers / secrets).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApplyOwner {
    App(AppId),
    Group(GroupId),
}

impl ApplyOwner {
    /// `(app_id, group_id)` for a `NewScript` — exactly one is `Some`.
    fn script_owner(self) -> (Option<AppId>, Option<GroupId>) {
        match self {
            Self::App(a) => (Some(a), None),
            Self::Group(g) => (None, Some(g)),
        }
    }

    /// As the shared [`ScriptOwner`] — for the module/extension-point repos.
    fn as_script_owner(self) -> picloud_shared::ScriptOwner {
        match self {
            Self::App(a) => picloud_shared::ScriptOwner::App(a),
            Self::Group(g) => picloud_shared::ScriptOwner::Group(g),
        }
    }

    /// The app id, when this is an app node. Routes/triggers/email-secrets only
    /// exist on app nodes, so their reconcile paths call this with an
    /// `expect` — guarded by validation that rejects them on a group bundle.
    fn app_id(self) -> Option<AppId> {
        match self {
            Self::App(a) => Some(a),
            Self::Group(_) => None,
        }
    }

    async fn set_var_tx(
        self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        key: &str,
        value: &serde_json::Value,
    ) -> Result<(), ApplyError> {
        match self {
            Self::App(a) => set_app_var_tx(tx, a, key, value).await,
            Self::Group(g) => set_group_var_tx(tx, g, key, value).await,
        }
    }

    async fn delete_var_tx(
        self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        key: &str,
    ) -> Result<(), ApplyError> {
        match self {
            Self::App(a) => delete_app_var_tx(tx, a, key).await,
            Self::Group(g) => delete_group_var_tx(tx, g, key).await,
        }
    }
}

fn diff_scripts(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
    let by_name: HashMap<String, &Script> = current
        .scripts
        .iter()
        .map(|s| (s.name.to_lowercase(), s))
        .collect();
    let desired: HashSet<String> = bundle
        .scripts
        .iter()
        .map(|s| s.name.to_lowercase())
        .collect();

    let mut out = Vec::new();
    for s in &bundle.scripts {
        match by_name.get(&s.name.to_lowercase()) {
            None => out.push(ResourceChange {
                op: Op::Create,
                key: s.name.clone(),
                detail: None,
            }),
            Some(cur) => {
                if let Some(reason) = script_update_reason(cur, s) {
                    out.push(ResourceChange {
                        op: Op::Update,
                        key: s.name.clone(),
                        detail: Some(reason),
                    });
                } else {
                    out.push(ResourceChange {
                        op: Op::NoOp,
                        key: s.name.clone(),
                        detail: None,
                    });
                }
            }
        }
    }
    for s in &current.scripts {
        if !desired.contains(&s.name.to_lowercase()) {
            out.push(ResourceChange {
                op: Op::Delete,
                key: s.name.clone(),
                detail: None,
            });
        }
    }
    out
}

/// Returns `Some(reason)` if the desired script differs from current. A
/// `None` optional field means "unspecified — leave as-is", so it never
/// forces an update.
fn script_update_reason(cur: &Script, desired: &BundleScript) -> Option<String> {
    if cur.source != desired.source {
        return Some("source changed".into());
    }
    if cur.kind != desired.kind {
        return Some("kind changed".into());
    }
    // Declarative (default true): always reconcile to the manifest's value.
    if cur.enabled != desired.enabled {
        return Some(if desired.enabled {
            "enabled".into()
        } else {
            "disabled".into()
        });
    }
    // Sparse like the other optional fields: an omitted (`None`) description
    // means "leave as-is", so only an explicitly-set value that differs
    // forces an update. (Clearing a description is done in the dashboard.)
    if let Some(d) = &desired.description {
        if cur.description.as_ref() != Some(d) {
            return Some("description changed".into());
        }
    }
    if let Some(t) = desired.timeout_seconds {
        if i64::from(cur.timeout_seconds) != i64::from(t) {
            return Some("timeout changed".into());
        }
    }
    if let Some(m) = desired.memory_limit_mb {
        if i64::from(cur.memory_limit_mb) != i64::from(m) {
            return Some("memory changed".into());
        }
    }
    if let Some(sb) = desired.sandbox {
        if cur.sandbox != sb {
            return Some("sandbox changed".into());
        }
    }
    None
}

fn diff_routes(
    current: &CurrentState,
    bundle: &Bundle,
    name_by_id: &HashMap<ScriptId, String>,
) -> Vec<ResourceChange> {
    let by_key: HashMap<String, &Route> = current
        .routes
        .iter()
        .map(|r| {
            (
                route_key(
                    r.method.as_deref(),
                    r.host_kind,
                    &r.host,
                    r.path_kind,
                    &r.path,
                ),
                r,
            )
        })
        .collect();
    let desired: HashSet<String> = bundle
        .routes
        .iter()
        .map(|r| {
            route_key(
                r.method.as_deref(),
                r.host_kind,
                &r.host,
                r.path_kind,
                &r.path,
            )
        })
        .collect();

    let mut out = Vec::new();
    for r in &bundle.routes {
        let key = route_key(
            r.method.as_deref(),
            r.host_kind,
            &r.host,
            r.path_kind,
            &r.path,
        );
        match by_key.get(&key) {
            None => out.push(ResourceChange {
                op: Op::Create,
                key,
                detail: Some(format!("→ {}", r.script)),
            }),
            Some(cur) => {
                let cur_script = name_by_id.get(&cur.script_id).map(String::as_str);
                if cur_script != Some(r.script.as_str())
                    || cur.dispatch_mode != r.dispatch_mode
                    || cur.host_param_name != r.host_param_name
                    || cur.enabled != r.enabled
                {
                    out.push(ResourceChange {
                        op: Op::Update,
                        key,
                        detail: Some(format!("→ {}", r.script)),
                    });
                } else {
                    out.push(ResourceChange {
                        op: Op::NoOp,
                        key,
                        detail: None,
                    });
                }
            }
        }
    }
    for r in &current.routes {
        let key = route_key(
            r.method.as_deref(),
            r.host_kind,
            &r.host,
            r.path_kind,
            &r.path,
        );
        if !desired.contains(&key) {
            out.push(ResourceChange {
                op: Op::Delete,
                key,
                detail: None,
            });
        }
    }
    out
}

fn diff_triggers(
    current: &CurrentState,
    bundle: &Bundle,
    name_by_id: &HashMap<ScriptId, String>,
) -> Vec<ResourceChange> {
    let by_id: HashMap<String, &Trigger> = current
        .triggers
        .iter()
        .filter_map(|t| current_trigger_identity(t, name_by_id).map(|id| (id, t)))
        .collect();
    let desired: HashSet<String> = bundle
        .triggers
        .iter()
        .map(BundleTrigger::identity)
        .collect();

    let mut out = Vec::new();
    for t in &bundle.triggers {
        let id = t.identity();
        // Semantic identity captures the whole definition for these kinds,
        // so a match is a no-op (changing a field changes the identity).
        if by_id.contains_key(&id) {
            out.push(ResourceChange {
                op: Op::NoOp,
                key: id,
                detail: None,
            });
        } else {
            out.push(ResourceChange {
                op: Op::Create,
                key: id,
                detail: None,
            });
        }
    }
    for t in &current.triggers {
        // Never prune email triggers: `pull` can't represent them (the
        // server stores the sealed secret, not the `inbound_secret_ref`), so
        // their absence from the manifest is ambiguous, not a delete intent.
        // Manage email triggers with `pic triggers rm`.
        if matches!(t.details, TriggerDetails::Email { .. }) {
            continue;
        }
        if let Some(id) = current_trigger_identity(t, name_by_id) {
            if !desired.contains(&id) {
                out.push(ResourceChange {
                    op: Op::Delete,
                    key: id,
                    detail: None,
                });
            }
        }
    }
    out
}

fn diff_secrets(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
    let live: HashSet<&str> = current.secret_names.iter().map(String::as_str).collect();
    let declared: HashSet<&str> = bundle.secrets.iter().map(String::as_str).collect();

    let mut out = Vec::new();
    for name in &bundle.secrets {
        if live.contains(name.as_str()) {
            out.push(ResourceChange {
                op: Op::NoOp,
                key: name.clone(),
                detail: None,
            });
        } else {
            out.push(ResourceChange {
                op: Op::Create,
                key: name.clone(),
                detail: Some("declared but unset — push with `pic secret set`".into()),
            });
        }
    }
    for name in &current.secret_names {
        if !declared.contains(name.as_str()) {
            out.push(ResourceChange {
                op: Op::Delete,
                key: name.clone(),
                detail: Some("on server, not declared".into()),
            });
        }
    }
    out
}

/// Diff extension-point markers by name (§5.5). Name-only identity like
/// secrets, but — unlike secrets — the markers are real declarations the
/// manifest owns: a declared-but-absent name is a Create (the apply inserts
/// it), and a live-but-undeclared name is a Delete (pruned under `--prune`),
/// mirroring `vars`.
fn diff_extension_points(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
    let live: HashSet<&str> = current
        .extension_point_names
        .iter()
        .map(String::as_str)
        .collect();
    let declared: HashSet<&str> = bundle.extension_points.iter().map(String::as_str).collect();

    let mut out = Vec::new();
    for name in &bundle.extension_points {
        out.push(ResourceChange {
            op: if live.contains(name.as_str()) {
                Op::NoOp
            } else {
                Op::Create
            },
            key: name.clone(),
            detail: None,
        });
    }
    for name in &current.extension_point_names {
        if !declared.contains(name.as_str()) {
            out.push(ResourceChange {
                op: Op::Delete,
                key: name.clone(),
                detail: Some("on server, not declared".into()),
            });
        }
    }
    out
}

/// Diff shared group-collection markers by `(name, kind)` (§11.6). A
/// declared-but-absent marker is a Create, a live-but-undeclared one is a
/// Delete (pruned under `--prune`). Each change carries `key = name`,
/// `detail = Some(kind)` so the reconcile knows which store to touch and
/// `pic plan` renders the kind. Identity is `(LOWER(name), kind)`, so the same
/// name as both a `kv` and a `docs` collection are distinct markers.
fn diff_collections(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange> {
    let live: HashSet<(String, &str)> = current
        .collections
        .iter()
        .map(|(n, k)| (n.to_lowercase(), k.as_str()))
        .collect();
    let declared: HashSet<(String, &str)> = bundle
        .collections
        .iter()
        .map(|c| (c.name.to_lowercase(), c.kind.as_str()))
        .collect();

    let mut out = Vec::new();
    for c in &bundle.collections {
        out.push(ResourceChange {
            op: if live.contains(&(c.name.to_lowercase(), c.kind.as_str())) {
                Op::NoOp
            } else {
                Op::Create
            },
            key: c.name.clone(),
            detail: Some(c.kind.clone()),
        });
    }
    for (name, kind) in &current.collections {
        if !declared.contains(&(name.to_lowercase(), kind.as_str())) {
            out.push(ResourceChange {
                op: Op::Delete,
                key: name.clone(),
                detail: Some(kind.clone()),
            });
        }
    }
    out
}

/// Reject any email trigger whose referenced secret isn't set, so `plan` and
/// `apply` give the same answer (apply also fails in `resolve_and_seal`, but
/// only after taking the lock — surfacing it here keeps plan honest).
///
/// This checks the secret *reference exists*. `resolve_and_seal` additionally
/// requires the resolved value to be a non-empty string; `plan` deliberately
/// does not decrypt secrets, so a reference to a set-but-empty secret passes
/// `plan` and is caught at `apply` — the one residual plan/apply divergence,
/// and a deliberate one (plan stays read-only and crypto-free).
fn validate_email_secrets_present(
    bundle: &Bundle,
    secret_names: &[String],
) -> Result<(), ApplyError> {
    for t in &bundle.triggers {
        if let BundleTrigger::Email {
            script,
            inbound_secret_ref,
            ..
        } = t
        {
            if !secret_names.iter().any(|n| n == inbound_secret_ref) {
                return Err(ApplyError::Invalid(format!(
                    "email trigger on `{script}` references secret \
                     `{inbound_secret_ref}`, which is not set; push it with \
                     `pic secret set` first"
                )));
            }
        }
    }
    Ok(())
}

/// Semantic identity for a live trigger. Returns `None` only for
/// `dead_letter` (not representable in the manifest), so it's ignored by the
/// diff entirely. `email` returns `Some` so a declared email trigger matches
/// as NoOp, but `diff_triggers` separately refuses to emit a Delete for it.
fn current_trigger_identity(t: &Trigger, name_by_id: &HashMap<ScriptId, String>) -> Option<String> {
    let script = name_by_id.get(&t.script_id)?;
    match &t.details {
        TriggerDetails::Kv {
            collection_glob,
            ops,
        } => Some(format!(
            "kv|{script}|{collection_glob}|{}",
            sorted_csv(ops.iter().copied().map(KvEventOp::as_str))
        )),
        TriggerDetails::Docs {
            collection_glob,
            ops,
        } => Some(format!(
            "docs|{script}|{collection_glob}|{}",
            sorted_csv(ops.iter().copied().map(DocsEventOp::as_str))
        )),
        TriggerDetails::Files {
            collection_glob,
            ops,
        } => Some(format!(
            "files|{script}|{collection_glob}|{}",
            sorted_csv(ops.iter().copied().map(FilesEventOp::as_str))
        )),
        TriggerDetails::Cron {
            schedule, timezone, ..
        } => Some(format!("cron|{script}|{schedule}|{timezone}")),
        TriggerDetails::Pubsub { topic_pattern } => {
            Some(format!("pubsub|{script}|{topic_pattern}"))
        }
        TriggerDetails::Email { .. } => Some(format!("email|{script}")),
        TriggerDetails::Queue { queue_name, .. } => Some(format!("queue|{queue_name}")),
        TriggerDetails::DeadLetter { .. } => None,
    }
}

fn route_key(
    method: Option<&str>,
    host_kind: HostKind,
    host: &str,
    path_kind: PathKind,
    path: &str,
) -> String {
    let m = method.map_or_else(|| "ANY".to_string(), str::to_uppercase);
    format!(
        "{m} {} {} {} {path}",
        host_kind_str(host_kind),
        host,
        path_kind_str(path_kind)
    )
}

fn host_kind_str(k: HostKind) -> &'static str {
    match k {
        HostKind::Any => "any",
        HostKind::Strict => "strict",
        HostKind::Wildcard => "wildcard",
    }
}

fn path_kind_str(k: PathKind) -> &'static str {
    match k {
        PathKind::Exact => "exact",
        PathKind::Prefix => "prefix",
        PathKind::Param => "param",
    }
}

/// Sort + comma-join op strings so order is insignificant for identity.
fn sorted_csv<'a>(ops: impl Iterator<Item = &'a str>) -> String {
    let set: BTreeSet<&str> = ops.collect();
    set.into_iter().collect::<Vec<_>>().join(",")
}

fn reject_reserved_path(path: &str) -> Result<(), ApplyError> {
    if RESERVED_PATH_EXACT.contains(&path)
        || RESERVED_PATH_PREFIXES.iter().any(|p| path.starts_with(p))
    {
        return Err(ApplyError::Invalid(format!("path `{path}` is reserved")));
    }
    Ok(())
}

/// §4.7 reachability warning: an *enabled* endpoint script with no route and
/// no trigger has no event surface — it's only reachable via the
/// execute-by-id bypass / `invoke()`. Modules are exempt (never invoked
/// directly); disabled endpoints are intentionally inert, so skip them.
fn unreachable_endpoint_warnings(bundle: &Bundle) -> Vec<String> {
    let bound: HashSet<&str> = bundle
        .routes
        .iter()
        .map(|r| r.script.as_str())
        .chain(bundle.triggers.iter().map(BundleTrigger::script))
        .collect();
    bundle
        .scripts
        .iter()
        .filter(|s| s.kind == ScriptKind::Endpoint && s.enabled && !bound.contains(s.name.as_str()))
        .map(|s| {
            format!(
                "endpoint `{}` has no route or trigger — only reachable via the \
                 execute-by-id bypass / invoke()",
                s.name
            )
        })
        .collect()
}

/// §4.7 reachability warnings: an *enabled* route or trigger bound to a
/// script the manifest marks *disabled* is deployed but unreachable (the
/// route 404s, the trigger won't fire). Valid desired state, so a warning —
/// not an error — keyed on the bundle alone.
fn disabled_target_warnings(bundle: &Bundle) -> Vec<String> {
    let disabled: HashSet<&str> = bundle
        .scripts
        .iter()
        .filter(|s| !s.enabled)
        .map(|s| s.name.as_str())
        .collect();
    if disabled.is_empty() {
        return Vec::new();
    }
    let mut out = Vec::new();
    for r in &bundle.routes {
        if r.enabled && disabled.contains(r.script.as_str()) {
            out.push(format!(
                "route `{}` is enabled but its script `{}` is disabled — it will 404",
                route_key(
                    r.method.as_deref(),
                    r.host_kind,
                    &r.host,
                    r.path_kind,
                    &r.path
                ),
                r.script
            ));
        }
    }
    for t in &bundle.triggers {
        if disabled.contains(t.script()) {
            out.push(format!(
                "{} trigger targets disabled script `{}` — it will not fire",
                t.kind_str(),
                t.script()
            ));
        }
    }
    out
}

/// Per-kind structural validation for one desired trigger — kept at parity
/// with the interactive trigger API so `apply` can't write a trigger the
/// dashboard would have rejected. Pure (no live state), so it's unit-tested
/// directly. The script-binding checks live in `validate_bundle`.
fn validate_trigger_shape(t: &BundleTrigger) -> Result<(), ApplyError> {
    match t {
        BundleTrigger::Email {
            inbound_secret_ref, ..
        } if inbound_secret_ref.trim().is_empty() => {
            return Err(ApplyError::Invalid(format!(
                "email trigger on `{}` needs inbound_secret_ref (a secret name)",
                t.script()
            )));
        }
        BundleTrigger::Queue {
            queue_name,
            visibility_timeout_secs,
            ..
        } => {
            if queue_name.trim().is_empty() {
                return Err(ApplyError::Invalid("queue trigger needs queue_name".into()));
            }
            if let Some(v) = visibility_timeout_secs {
                // Parity with the interactive API: the floor is the dispatcher
                // tick/reclaim-cadence minimum (`triggers_api`), and the ceiling
                // matches `trigger_repo`'s queue check. Apply previously allowed
                // a [5, 29] floor the dashboard rejects.
                let min = crate::triggers_api::MIN_QUEUE_VISIBILITY_TIMEOUT_SECS;
                if !(min..=3600).contains(v) {
                    return Err(ApplyError::Invalid(format!(
                        "queue `{queue_name}`: visibility_timeout_secs must be in [{min}, 3600]"
                    )));
                }
            }
        }
        BundleTrigger::Cron {
            schedule, timezone, ..
        } => {
            crate::cron_scheduler::validate_schedule(schedule).map_err(|e| {
                ApplyError::Invalid(format!(
                    "cron trigger on `{}`: invalid schedule: {e}",
                    t.script()
                ))
            })?;
            crate::cron_scheduler::validate_timezone(timezone).map_err(|e| {
                ApplyError::Invalid(format!(
                    "cron trigger on `{}`: invalid timezone: {e}",
                    t.script()
                ))
            })?;
        }
        // Parity with the interactive trigger API, which rejects an empty
        // collection_glob (`create_{kv,docs,files}_trigger`).
        BundleTrigger::Kv {
            collection_glob, ..
        }
        | BundleTrigger::Docs {
            collection_glob, ..
        }
        | BundleTrigger::Files {
            collection_glob, ..
        } if collection_glob.trim().is_empty() => {
            return Err(ApplyError::Invalid(format!(
                "{} trigger on `{}` needs a non-empty collection_glob",
                t.kind_str(),
                t.script()
            )));
        }
        // Parity with `create_pubsub_trigger`'s topic-pattern check — otherwise
        // a malformed pattern is written and silently never matches at dispatch.
        BundleTrigger::Pubsub { topic_pattern, .. } => {
            picloud_shared::validate_topic_pattern(topic_pattern).map_err(|e| {
                ApplyError::Invalid(format!(
                    "pubsub trigger on `{}`: invalid topic_pattern: {e}",
                    t.script()
                ))
            })?;
        }
        _ => {}
    }
    Ok(())
}

/// Summary of what an `apply` changed.
#[derive(Debug, Default, Serialize)]
pub struct ApplyReport {
    pub scripts_created: u32,
    pub scripts_updated: u32,
    pub scripts_deleted: u32,
    pub routes_created: u32,
    pub routes_updated: u32,
    pub routes_deleted: u32,
    pub triggers_created: u32,
    pub triggers_deleted: u32,
    #[serde(default)]
    pub vars_created: u32,
    #[serde(default)]
    pub vars_updated: u32,
    #[serde(default)]
    pub vars_deleted: u32,
    #[serde(default)]
    pub extension_points_created: u32,
    #[serde(default)]
    pub extension_points_deleted: u32,
    #[serde(default)]
    pub collections_created: u32,
    #[serde(default)]
    pub collections_deleted: u32,
    #[serde(skip_serializing_if = "Vec::is_empty")]
    pub warnings: Vec<String>,
}

/// The key set of a name→id map, for passing the inherited target *names*
/// into `validate_bundle` without leaking the ids.
fn keys_set(m: &HashMap<String, ScriptId>) -> HashSet<String> {
    m.keys().cloned().collect()
}

/// A tree node resolved to its owner + loaded current state + computed plan,
/// ready to reconcile. `app_chain` is the ordered (nearest-first) ancestor
/// group ids for an app node (empty for a group node).
struct PreparedNode<'a> {
    owner: ApplyOwner,
    node: &'a TreeNode,
    current: CurrentState,
    current_names: HashMap<ScriptId, String>,
    plan: Plan,
    app_chain: Vec<GroupId>,
}

/// FNV-1a over a set of already-sorted per-resource token parts — the same
/// hashing `state_token` uses, lifted to combine many nodes into one tree
/// token. Order-independent because the caller sorts `parts`.
fn combine_tokens(parts: &[String]) -> String {
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for p in parts {
        for b in p.as_bytes() {
            h ^= u64::from(*b);
            h = h.wrapping_mul(0x0000_0100_0000_01b3);
        }
        h ^= u64::from(b'\n');
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    format!("{h:016x}")
}

fn resolve_script(
    name_to_id: &HashMap<String, ScriptId>,
    name: &str,
) -> Result<ScriptId, ApplyError> {
    name_to_id
        .get(&name.to_lowercase())
        .copied()
        .ok_or_else(|| ApplyError::Invalid(format!("binding references unknown script `{name}`")))
}

async fn insert_bundle_route(
    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
    app_id: AppId,
    script_id: ScriptId,
    br: &BundleRoute,
) -> Result<(), ApplyError> {
    let new = NewRoute {
        app_id,
        script_id,
        host_kind: br.host_kind,
        host: br.host.clone(),
        host_param_name: br.host_param_name.clone(),
        path_kind: br.path_kind,
        path: br.path.clone(),
        method: br.method.clone(),
        dispatch_mode: br.dispatch_mode,
        enabled: br.enabled,
    };
    insert_route_tx(tx, &new).await.map_err(map_repo)?;
    Ok(())
}

fn bundle_trigger_details(bt: &BundleTrigger, default_visibility: u32) -> TriggerDetails {
    match bt {
        BundleTrigger::Kv {
            collection_glob,
            ops,
            ..
        } => TriggerDetails::Kv {
            collection_glob: collection_glob.clone(),
            ops: ops.clone(),
        },
        BundleTrigger::Docs {
            collection_glob,
            ops,
            ..
        } => TriggerDetails::Docs {
            collection_glob: collection_glob.clone(),
            ops: ops.clone(),
        },
        BundleTrigger::Files {
            collection_glob,
            ops,
            ..
        } => TriggerDetails::Files {
            collection_glob: collection_glob.clone(),
            ops: ops.clone(),
        },
        BundleTrigger::Cron {
            schedule, timezone, ..
        } => TriggerDetails::Cron {
            schedule: schedule.clone(),
            timezone: timezone.clone(),
            last_fired_at: None,
        },
        BundleTrigger::Pubsub { topic_pattern, .. } => TriggerDetails::Pubsub {
            topic_pattern: topic_pattern.clone(),
        },
        BundleTrigger::Queue {
            queue_name,
            visibility_timeout_secs,
            ..
        } => TriggerDetails::Queue {
            queue_name: queue_name.clone(),
            visibility_timeout_secs: visibility_timeout_secs.unwrap_or(default_visibility),
            last_fired_at: None,
        },
        BundleTrigger::Email { .. } => unreachable!("email handled separately"),
    }
}

fn map_repo(e: ScriptRepositoryError) -> ApplyError {
    match e {
        ScriptRepositoryError::Conflict(m) => ApplyError::Invalid(m),
        ScriptRepositoryError::NotFound(_) => {
            ApplyError::Invalid("a referenced resource was not found".into())
        }
        ScriptRepositoryError::Db(e) => ApplyError::Backend(e.to_string()),
    }
}

fn map_trig(e: TriggerRepoError) -> ApplyError {
    match e {
        TriggerRepoError::Invalid(m) => ApplyError::Invalid(m),
        TriggerRepoError::NotFound(_) => ApplyError::Invalid("trigger not found".into()),
        TriggerRepoError::Db(e) => ApplyError::Backend(e.to_string()),
    }
}

/// A stable fingerprint of an app's live state, covering exactly what the
/// reconcile diff keys on: script identity + write-counter (`version` bumps on
/// any script edit), route identity + binding/attrs, trigger semantic identity
/// (via `current_trigger_identity`, so dead-letter triggers — which the diff
/// ignores — are excluded), and secret names. Any out-of-band change to those
/// flips the token, so `plan`-then-`apply` can detect the app moving underneath.
///
/// Deliberately mirrors the diff's inputs so a mismatch means the *reviewed
/// plan* would now differ — not merely that some unrelated row changed.
///
/// Uses FNV-1a (deterministic across process restarts, unlike `DefaultHasher`'s
/// per-process seed) so a token stored by `pic plan` still matches on a later
/// `pic apply`. A hash collision can only ever yield a false "unchanged", which
/// is the same risk class as not checking at all — never a false refusal.
#[must_use]
pub fn state_token(current: &CurrentState) -> String {
    state_token_with_names(current, &HashMap::new())
}

/// `state_token` with `current_names` (group-bound binding ids → names, Phase
/// 4) so a trigger bound to an inherited group script contributes to the
/// fingerprint. Without it such a trigger is silently excluded and the
/// bound-plan (StateMoved) check has a false-negative for that binding class.
fn state_token_with_names(
    current: &CurrentState,
    current_names: &HashMap<ScriptId, String>,
) -> String {
    let mut parts: Vec<String> = Vec::with_capacity(
        current.scripts.len()
            + current.routes.len()
            + current.triggers.len()
            + current.secret_names.len()
            + current.vars.len()
            + current.extension_point_names.len()
            + current.collections.len(),
    );
    for s in &current.scripts {
        parts.push(format!(
            "s|{}|{}|{}",
            s.name.to_lowercase(),
            s.version,
            s.enabled
        ));
    }
    for r in &current.routes {
        parts.push(format!(
            "r|{}|{:?}|{:?}|{}|{}",
            route_key(
                r.method.as_deref(),
                r.host_kind,
                &r.host,
                r.path_kind,
                &r.path
            ),
            r.script_id,
            r.dispatch_mode,
            r.host_param_name.as_deref().unwrap_or(""),
            r.enabled,
        ));
    }
    // Mirror exactly what the trigger diff keys on: `current_trigger_identity`
    // excludes dead-letter triggers (the diff ignores them) and keys on the
    // semantic identity — NOT raw id/enabled. Hashing dead-letter rows or the
    // (currently inert) `enabled` flag would force a false refusal over a
    // change the manifest can't represent.
    let script_name_by_id: HashMap<ScriptId, String> = current
        .scripts
        .iter()
        .map(|s| (s.id, s.name.clone()))
        .chain(current_names.iter().map(|(id, n)| (*id, n.clone())))
        .collect();
    for t in &current.triggers {
        if let Some(id) = current_trigger_identity(t, &script_name_by_id) {
            parts.push(format!("t|{id}"));
        }
    }
    for n in &current.secret_names {
        parts.push(format!("k|{n}"));
    }
    // Vars are value-sensitive (the value lives in the manifest), so the token
    // moves when a value changes — the bound-plan check then catches an edit
    // between plan and apply. `to_string` on a serde_json::Value is stable
    // (serde_json sorts object keys by default).
    for (key, value) in &current.vars {
        parts.push(format!(
            "v|{key}|{}",
            serde_json::to_string(value).unwrap_or_default()
        ));
    }
    // Extension-point markers are name-only, like secret names.
    for n in &current.extension_point_names {
        parts.push(format!("ep|{n}"));
    }
    // Shared group-collection markers carry a kind (§11.6), so a kind change
    // moves the token.
    for (name, kind) in &current.collections {
        parts.push(format!("coll|{kind}|{name}"));
    }
    // Order-independent: sort the per-resource tokens before hashing.
    parts.sort_unstable();
    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
    for p in &parts {
        for b in p.as_bytes() {
            h ^= u64::from(*b);
            h = h.wrapping_mul(0x0000_0100_0000_01b3);
        }
        h ^= u64::from(b'\n');
        h = h.wrapping_mul(0x0000_0100_0000_01b3);
    }
    format!("{h:016x}")
}

/// Per-node advisory lock key, namespaced so it can't collide with the
/// queue-trigger lock space. App and group ids live in disjoint UUID spaces and
/// are tagged by owner kind, so an app and a group never share a key.
fn apply_lock_key(owner: ApplyOwner) -> i64 {
    use std::hash::{Hash, Hasher};
    let mut h = std::collections::hash_map::DefaultHasher::new();
    "picloud-apply".hash(&mut h);
    match owner {
        ApplyOwner::App(a) => {
            "app".hash(&mut h);
            a.into_inner().hash(&mut h);
        }
        ApplyOwner::Group(g) => {
            "group".hash(&mut h);
            g.into_inner().hash(&mut h);
        }
    }
    i64::from_ne_bytes(h.finish().to_ne_bytes())
}

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

    fn script(name: &str, source: &str) -> Script {
        Script {
            id: ScriptId::from(uuid::Uuid::new_v4()),
            app_id: Some(AppId::from(uuid::Uuid::nil())),
            group_id: None,
            name: name.to_string(),
            description: None,
            version: 1,
            source: source.to_string(),
            kind: ScriptKind::Endpoint,
            timeout_seconds: 30,
            sandbox: ScriptSandbox::empty(),
            memory_limit_mb: 256,
            enabled: true,
            created_at: Utc::now(),
            updated_at: Utc::now(),
        }
    }

    fn bundle_script(name: &str, source: &str) -> BundleScript {
        BundleScript {
            name: name.to_string(),
            source: source.to_string(),
            kind: ScriptKind::Endpoint,
            description: None,
            timeout_seconds: None,
            memory_limit_mb: None,
            sandbox: None,
            enabled: true,
        }
    }

    #[test]
    fn empty_app_is_all_create() {
        let current = CurrentState::default();
        let bundle = Bundle {
            scripts: vec![bundle_script("a", "1")],
            routes: vec![],
            triggers: vec![],
            secrets: vec!["S".into()],
            vars: std::collections::BTreeMap::new(),
            extension_points: Vec::new(),
            collections: Vec::new(),
        };
        let plan = compute_diff(&current, &bundle);
        assert_eq!(plan.scripts.len(), 1);
        assert_eq!(plan.scripts[0].op, Op::Create);
        assert_eq!(plan.secrets[0].op, Op::Create); // declared-but-unset
        assert!(!plan.is_noop());
    }

    #[test]
    fn identical_is_all_noop() {
        let s = script("a", "let x = 1;");
        let current = CurrentState {
            scripts: vec![s.clone()],
            secret_names: vec!["S".into()],
            ..CurrentState::default()
        };
        let bundle = Bundle {
            scripts: vec![bundle_script("a", "let x = 1;")],
            routes: vec![],
            triggers: vec![],
            secrets: vec!["S".into()],
            vars: std::collections::BTreeMap::new(),
            extension_points: Vec::new(),
            collections: Vec::new(),
        };
        let plan = compute_diff(&current, &bundle);
        assert!(plan.is_noop(), "expected all no-op, got {plan:?}");
    }

    #[test]
    fn changed_source_is_update() {
        let current = CurrentState {
            scripts: vec![script("a", "old")],
            ..CurrentState::default()
        };
        let bundle = Bundle {
            scripts: vec![bundle_script("a", "new")],
            routes: vec![],
            triggers: vec![],
            secrets: vec![],
            vars: std::collections::BTreeMap::new(),
            extension_points: Vec::new(),
            collections: Vec::new(),
        };
        let plan = compute_diff(&current, &bundle);
        assert_eq!(plan.scripts[0].op, Op::Update);
    }

    #[test]
    fn missing_from_bundle_is_delete() {
        let current = CurrentState {
            scripts: vec![script("gone", "x")],
            secret_names: vec!["OLD".into()],
            ..CurrentState::default()
        };
        let bundle = Bundle {
            scripts: vec![],
            routes: vec![],
            triggers: vec![],
            secrets: vec![],
            vars: std::collections::BTreeMap::new(),
            extension_points: Vec::new(),
            collections: Vec::new(),
        };
        let plan = compute_diff(&current, &bundle);
        assert_eq!(plan.scripts[0].op, Op::Delete);
        assert_eq!(plan.secrets[0].op, Op::Delete);
    }

    #[test]
    fn route_rebind_is_update() {
        let s = script("a", "x");
        let sid = s.id;
        let route = Route {
            id: uuid::Uuid::new_v4(),
            app_id: AppId::from(uuid::Uuid::nil()),
            script_id: sid,
            host_kind: HostKind::Any,
            host: String::new(),
            host_param_name: None,
            path_kind: PathKind::Exact,
            path: "/p".into(),
            method: Some("POST".into()),
            dispatch_mode: DispatchMode::Sync,
            enabled: true,
            created_at: Utc::now(),
        };
        let current = CurrentState {
            scripts: vec![s, script("b", "y")],
            routes: vec![route],
            ..CurrentState::default()
        };
        // Same tuple, bound to a different script → update.
        let bundle = Bundle {
            scripts: vec![],
            routes: vec![BundleRoute {
                script: "b".into(),
                method: Some("POST".into()),
                host_kind: HostKind::Any,
                host: String::new(),
                host_param_name: None,
                path_kind: PathKind::Exact,
                path: "/p".into(),
                dispatch_mode: DispatchMode::Sync,
                enabled: true,
            }],
            triggers: vec![],
            secrets: vec![],
            vars: std::collections::BTreeMap::new(),
            extension_points: Vec::new(),
            collections: Vec::new(),
        };
        let plan = compute_diff(&current, &bundle);
        assert_eq!(plan.routes[0].op, Op::Update);
    }

    fn empty_bundle() -> Bundle {
        Bundle {
            scripts: vec![],
            routes: vec![],
            triggers: vec![],
            secrets: vec![],
            vars: std::collections::BTreeMap::new(),
            extension_points: Vec::new(),
            collections: Vec::new(),
        }
    }

    fn trig(script_id: ScriptId, details: TriggerDetails) -> Trigger {
        use crate::trigger_config::BackoffShape;
        use crate::trigger_repo::TriggerKind;
        let kind = match &details {
            TriggerDetails::Cron { .. } => TriggerKind::Cron,
            TriggerDetails::DeadLetter { .. } => TriggerKind::DeadLetter,
            TriggerDetails::Queue { .. } => TriggerKind::Queue,
            // Other kinds aren't distinguished by these tests (the diff keys
            // on `details`, not `kind`).
            _ => TriggerKind::Kv,
        };
        Trigger {
            id: TriggerId::from(uuid::Uuid::new_v4()),
            app_id: AppId::from(uuid::Uuid::nil()),
            script_id,
            name: "t".into(),
            kind,
            enabled: true,
            dispatch_mode: TriggerDispatchMode::Async,
            retry_max_attempts: 3,
            retry_backoff: BackoffShape::Exponential,
            retry_base_ms: 1000,
            registered_by_principal: AdminUserId::from(uuid::Uuid::nil()),
            created_at: Utc::now(),
            updated_at: Utc::now(),
            details,
        }
    }

    #[test]
    fn script_update_only_on_real_change() {
        // `script("a", ..)` defaults to timeout 30, mem 256, empty sandbox.
        let current = CurrentState {
            scripts: vec![script("a", "let x = 1;")],
            ..CurrentState::default()
        };
        // Same source, optional fields unset → NoOp ("None = leave as-is").
        let mut b = empty_bundle();
        b.scripts = vec![bundle_script("a", "let x = 1;")];
        assert_eq!(compute_diff(&current, &b).scripts[0].op, Op::NoOp);
        // Explicit timeout that MATCHES current → still NoOp.
        let mut same = empty_bundle();
        let mut s = bundle_script("a", "let x = 1;");
        s.timeout_seconds = Some(30);
        same.scripts = vec![s];
        assert_eq!(compute_diff(&current, &same).scripts[0].op, Op::NoOp);
        // Explicit timeout that DIFFERS → Update.
        let mut chg = empty_bundle();
        let mut s2 = bundle_script("a", "let x = 1;");
        s2.timeout_seconds = Some(99);
        chg.scripts = vec![s2];
        assert_eq!(compute_diff(&current, &chg).scripts[0].op, Op::Update);
        // kind change → Update.
        let mut chg2 = empty_bundle();
        let mut s3 = bundle_script("a", "let x = 1;");
        s3.kind = ScriptKind::Module;
        chg2.scripts = vec![s3];
        assert_eq!(compute_diff(&current, &chg2).scripts[0].op, Op::Update);
    }

    #[test]
    fn description_is_sparse() {
        // An omitted (`None`) description must mean "leave as-is", matching
        // the other optional fields — not "clear it".
        let mut cur = script("a", "let x = 1;");
        cur.description = Some("kept".into());
        let current = CurrentState {
            scripts: vec![cur],
            ..CurrentState::default()
        };
        // Omitted → NoOp (leave the stored description alone).
        let mut omit = empty_bundle();
        omit.scripts = vec![bundle_script("a", "let x = 1;")];
        assert_eq!(compute_diff(&current, &omit).scripts[0].op, Op::NoOp);
        // Same explicit value → NoOp.
        let mut same = empty_bundle();
        let mut s = bundle_script("a", "let x = 1;");
        s.description = Some("kept".into());
        same.scripts = vec![s];
        assert_eq!(compute_diff(&current, &same).scripts[0].op, Op::NoOp);
        // Different explicit value → Update.
        let mut chg = empty_bundle();
        let mut s2 = bundle_script("a", "let x = 1;");
        s2.description = Some("changed".into());
        chg.scripts = vec![s2];
        assert_eq!(compute_diff(&current, &chg).scripts[0].op, Op::Update);
    }

    #[test]
    fn trigger_shape_validation_matches_interactive_api() {
        // Empty collection_glob is rejected (kv/docs/files).
        assert!(validate_trigger_shape(&BundleTrigger::Kv {
            script: "h".into(),
            collection_glob: "  ".into(),
            ops: vec![],
            dispatch_mode: None,
            retry_max_attempts: None,
        })
        .is_err());
        // A non-empty glob passes.
        assert!(validate_trigger_shape(&BundleTrigger::Docs {
            script: "h".into(),
            collection_glob: "users".into(),
            ops: vec![],
            dispatch_mode: None,
            retry_max_attempts: None,
        })
        .is_ok());
        // A malformed pubsub topic_pattern (mid-pattern wildcard) is rejected;
        // a valid one passes.
        assert!(validate_trigger_shape(&BundleTrigger::Pubsub {
            script: "h".into(),
            topic_pattern: "user.*.created".into(),
            dispatch_mode: None,
            retry_max_attempts: None,
        })
        .is_err());
        assert!(validate_trigger_shape(&BundleTrigger::Pubsub {
            script: "h".into(),
            topic_pattern: "orders.created".into(),
            dispatch_mode: None,
            retry_max_attempts: None,
        })
        .is_ok());
        // Queue visibility floor matches the interactive API (>= 30): a value
        // below the floor is rejected; the floor itself and `None` are ok.
        let queue = |vis| BundleTrigger::Queue {
            script: "h".into(),
            queue_name: "jobs".into(),
            visibility_timeout_secs: vis,
            dispatch_mode: None,
            retry_max_attempts: None,
        };
        assert!(validate_trigger_shape(&queue(Some(10))).is_err());
        assert!(validate_trigger_shape(&queue(Some(
            crate::triggers_api::MIN_QUEUE_VISIBILITY_TIMEOUT_SECS
        )))
        .is_ok());
        assert!(validate_trigger_shape(&queue(None)).is_ok());
        // Empty email inbound_secret_ref is rejected.
        assert!(validate_trigger_shape(&BundleTrigger::Email {
            script: "h".into(),
            inbound_secret_ref: "  ".into(),
            dispatch_mode: None,
            retry_max_attempts: None,
        })
        .is_err());
    }

    #[test]
    fn state_token_is_stable_order_independent_and_sensitive() {
        let a = script("a", "x"); // version 1
        let b = script("b", "y");
        let base = CurrentState {
            scripts: vec![a.clone(), b.clone()],
            secret_names: vec!["S".into()],
            ..CurrentState::default()
        };
        // Deterministic + order-independent.
        let reordered = CurrentState {
            scripts: vec![b.clone(), a.clone()],
            secret_names: vec!["S".into()],
            ..CurrentState::default()
        };
        assert_eq!(state_token(&base), state_token(&reordered));
        // A script edit bumps `version`, which must flip the token even if the
        // source-by-name set is otherwise unchanged (out-of-band redeploy).
        let mut a2 = a.clone();
        a2.version += 1;
        let bumped = CurrentState {
            scripts: vec![a2, b.clone()],
            secret_names: vec!["S".into()],
            ..CurrentState::default()
        };
        assert_ne!(state_token(&base), state_token(&bumped));
        // Adding a secret name flips it too.
        let with_secret = CurrentState {
            scripts: vec![a, b],
            secret_names: vec!["S".into(), "T".into()],
            ..CurrentState::default()
        };
        assert_ne!(state_token(&base), state_token(&with_secret));
    }

    #[test]
    fn diff_vars_create_update_noop_delete() {
        let current = CurrentState {
            vars: vec![
                ("region".into(), serde_json::json!("eu")),
                ("keep".into(), serde_json::json!(1)),
                ("stale".into(), serde_json::json!(true)),
            ],
            ..CurrentState::default()
        };
        let mut vars = std::collections::BTreeMap::new();
        vars.insert("region".to_string(), serde_json::json!("us")); // value changed
        vars.insert("keep".to_string(), serde_json::json!(1)); // identical
        vars.insert("new".to_string(), serde_json::json!("x")); // not live
                                                                // "stale" is live but undeclared → Delete.
        let bundle = Bundle {
            vars,
            ..empty_bundle()
        };
        let plan = compute_diff(&current, &bundle);
        let op = |k: &str| plan.vars.iter().find(|c| c.key == k).unwrap().op;
        assert_eq!(op("region"), Op::Update);
        assert_eq!(op("keep"), Op::NoOp);
        assert_eq!(op("new"), Op::Create);
        assert_eq!(op("stale"), Op::Delete);
        assert!(!plan.is_noop());
    }

    #[test]
    fn state_token_is_sensitive_to_var_values_and_order_independent() {
        let base = CurrentState {
            vars: vec![("region".into(), serde_json::json!("eu"))],
            ..CurrentState::default()
        };
        // A value change must flip the token (covers the bound-plan check for
        // an out-of-band `pic vars set` between plan and apply).
        let changed = CurrentState {
            vars: vec![("region".into(), serde_json::json!("us"))],
            ..CurrentState::default()
        };
        assert_ne!(state_token(&base), state_token(&changed));
        // Order-independent across multiple vars.
        let ab = CurrentState {
            vars: vec![
                ("a".into(), serde_json::json!(1)),
                ("b".into(), serde_json::json!(2)),
            ],
            ..CurrentState::default()
        };
        let ba = CurrentState {
            vars: vec![
                ("b".into(), serde_json::json!(2)),
                ("a".into(), serde_json::json!(1)),
            ],
            ..CurrentState::default()
        };
        assert_eq!(state_token(&ab), state_token(&ba));
    }

    #[test]
    fn state_token_ignores_dead_letter_triggers() {
        // The diff ignores dead-letter triggers (not manifest-representable),
        // so the token must too — otherwise an out-of-band dead-letter change
        // would force a spurious `--force`.
        let s = script("h", "x");
        let sid = s.id;
        let base = CurrentState {
            scripts: vec![s.clone()],
            ..CurrentState::default()
        };
        let with_dl = CurrentState {
            scripts: vec![s],
            triggers: vec![trig(
                sid,
                TriggerDetails::DeadLetter {
                    source_filter: None,
                    trigger_id_filter: None,
                    script_id_filter: None,
                },
            )],
            ..CurrentState::default()
        };
        assert_eq!(
            state_token(&base),
            state_token(&with_dl),
            "dead-letter triggers must not affect the token"
        );
    }

    #[test]
    fn enabled_is_declarative_and_diffed() {
        // A live-active script the manifest marks disabled → Update; the token
        // is sensitive to the flag so an out-of-band toggle is detectable.
        let cur = script("a", "let x = 1;"); // enabled: true
        let base = CurrentState {
            scripts: vec![cur.clone()],
            ..CurrentState::default()
        };
        let mut disable = empty_bundle();
        let mut bs = bundle_script("a", "let x = 1;");
        bs.enabled = false;
        disable.scripts = vec![bs];
        assert_eq!(compute_diff(&base, &disable).scripts[0].op, Op::Update);

        let mut flipped = cur;
        flipped.enabled = false;
        let toggled = CurrentState {
            scripts: vec![flipped],
            ..CurrentState::default()
        };
        assert_ne!(
            state_token(&base),
            state_token(&toggled),
            "token must flip on an enabled change"
        );
    }

    #[test]
    fn warns_on_enabled_binding_to_disabled_script() {
        let mut b = empty_bundle();
        let mut s = bundle_script("h", "x");
        s.enabled = false;
        b.scripts = vec![s];
        b.routes = vec![BundleRoute {
            script: "h".into(),
            method: None,
            host_kind: HostKind::Any,
            host: String::new(),
            host_param_name: None,
            path_kind: PathKind::Exact,
            path: "/h".into(),
            dispatch_mode: DispatchMode::Sync,
            enabled: true,
        }];
        let w = disabled_target_warnings(&b);
        assert!(
            w.iter().any(|m| m.contains("will 404")),
            "expected a 404 reachability warning: {w:?}"
        );
        // No warning when the script is active.
        b.scripts[0].enabled = true;
        assert!(disabled_target_warnings(&b).is_empty());
    }

    #[test]
    fn warns_on_unreachable_endpoint() {
        // §4.7: an enabled endpoint with no route and no trigger is flagged.
        let mut b = empty_bundle();
        b.scripts = vec![bundle_script("orphan", "x")];
        let w = unreachable_endpoint_warnings(&b);
        assert!(
            w.iter().any(|m| m.contains("no route or trigger")),
            "expected an unreachable-endpoint warning: {w:?}"
        );
        // Bound by a route → no warning.
        b.routes = vec![BundleRoute {
            script: "orphan".into(),
            method: None,
            host_kind: HostKind::Any,
            host: String::new(),
            host_param_name: None,
            path_kind: PathKind::Exact,
            path: "/o".into(),
            dispatch_mode: DispatchMode::Sync,
            enabled: true,
        }];
        assert!(unreachable_endpoint_warnings(&b).is_empty());
        // A module is exempt even with no binding.
        let mut m = empty_bundle();
        let mut lib = bundle_script("lib", "x");
        lib.kind = ScriptKind::Module;
        m.scripts = vec![lib];
        assert!(unreachable_endpoint_warnings(&m).is_empty());
    }

    #[test]
    fn trigger_diff_create_noop_delete() {
        let s = script("h", "x");
        let sid = s.id;
        let cron = |sched: &str| TriggerDetails::Cron {
            schedule: sched.into(),
            timezone: "UTC".into(),
            last_fired_at: None,
        };
        let current = CurrentState {
            scripts: vec![s],
            triggers: vec![trig(sid, cron("0 0 * * * *"))],
            ..CurrentState::default()
        };
        let bundle_cron = |sched: &str| BundleTrigger::Cron {
            script: "h".into(),
            schedule: sched.into(),
            timezone: "UTC".into(),
            dispatch_mode: None,
            retry_max_attempts: None,
        };
        // Same schedule → NoOp.
        let mut same = empty_bundle();
        same.triggers = vec![bundle_cron("0 0 * * * *")];
        let p = compute_diff(&current, &same);
        assert!(p.triggers.iter().all(|c| c.op == Op::NoOp), "{p:?}");
        // Removed → Delete.
        let p = compute_diff(&current, &empty_bundle());
        assert_eq!(p.triggers.len(), 1);
        assert_eq!(p.triggers[0].op, Op::Delete);
        // Different schedule → Create (new) + Delete (old).
        let mut chg = empty_bundle();
        chg.triggers = vec![bundle_cron("0 5 * * * *")];
        let p = compute_diff(&current, &chg);
        assert!(p.triggers.iter().any(|c| c.op == Op::Create), "{p:?}");
        assert!(p.triggers.iter().any(|c| c.op == Op::Delete), "{p:?}");
    }

    #[test]
    fn kv_ops_order_insensitive() {
        let s = script("h", "x");
        let sid = s.id;
        let current = CurrentState {
            scripts: vec![s],
            triggers: vec![trig(
                sid,
                TriggerDetails::Kv {
                    collection_glob: "users".into(),
                    ops: vec![KvEventOp::Update, KvEventOp::Insert],
                },
            )],
            ..CurrentState::default()
        };
        let mut b = empty_bundle();
        b.triggers = vec![BundleTrigger::Kv {
            script: "h".into(),
            collection_glob: "users".into(),
            ops: vec![KvEventOp::Insert, KvEventOp::Update], // reversed
            dispatch_mode: None,
            retry_max_attempts: None,
        }];
        let p = compute_diff(&current, &b);
        assert!(
            p.triggers.iter().all(|c| c.op == Op::NoOp),
            "ops order must not matter: {p:?}"
        );
    }

    #[test]
    fn dead_letter_trigger_is_ignored_by_diff() {
        // Dead-letter triggers can't be expressed in the manifest, so the
        // diff must neither match nor prune them.
        let s = script("h", "x");
        let sid = s.id;
        let current = CurrentState {
            scripts: vec![s],
            triggers: vec![trig(
                sid,
                TriggerDetails::DeadLetter {
                    source_filter: None,
                    trigger_id_filter: None,
                    script_id_filter: None,
                },
            )],
            ..CurrentState::default()
        };
        let p = compute_diff(&current, &empty_bundle());
        assert!(p.triggers.is_empty(), "dead-letter must be ignored: {p:?}");
    }

    #[test]
    fn email_trigger_is_never_pruned() {
        // `pull` can't represent email triggers (the server keeps the sealed
        // secret, not the `inbound_secret_ref`), so their absence from a
        // bundle is ambiguous — the diff must not emit a Delete for them.
        let s = script("h", "x");
        let sid = s.id;
        let current = CurrentState {
            scripts: vec![s],
            triggers: vec![trig(
                sid,
                TriggerDetails::Email {
                    has_inbound_secret: true,
                },
            )],
            ..CurrentState::default()
        };
        let p = compute_diff(&current, &empty_bundle());
        assert!(
            !p.triggers.iter().any(|c| c.op == Op::Delete),
            "email trigger must not be pruned: {p:?}"
        );
    }

    #[test]
    fn queue_rebind_is_noop_documented_limitation() {
        // Queue identity is `queue|{queue_name}` (script-independent), so a
        // rebind to a different script diffs as NoOp — a known limitation.
        let a = script("a", "x");
        let aid = a.id;
        let current = CurrentState {
            scripts: vec![a, script("b", "y")],
            triggers: vec![trig(
                aid,
                TriggerDetails::Queue {
                    queue_name: "jobs".into(),
                    visibility_timeout_secs: 30,
                    last_fired_at: None,
                },
            )],
            ..CurrentState::default()
        };
        let mut b = empty_bundle();
        b.triggers = vec![BundleTrigger::Queue {
            script: "b".into(), // rebind a→b
            queue_name: "jobs".into(),
            visibility_timeout_secs: None,
            dispatch_mode: None,
            retry_max_attempts: None,
        }];
        let p = compute_diff(&current, &b);
        assert!(
            p.triggers.iter().all(|c| c.op == Op::NoOp),
            "queue identity is queue_name-only: {p:?}"
        );
    }
}