docs(design): record async disabled-execution + cross-app backstops

Update §4.3: the fire-time `enabled` re-check is shipped and test-backed on
BOTH async arms (the unified outbox `!resolved.active` gate and the queue
arm's fresh-script nack), with the specific regression tests cited, plus the
same-app backstop now present on every async build path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/migrations/0045_scripts_routes_enabled.sql

@@ -0,0 +1,8 @@
+-- Three-state `enabled` lifecycle (blueprint §4.3), scripts + routes half.
+-- Triggers already carry `enabled` (0008) honored at match/schedule time;
+-- this adds the same toggle to scripts and routes. A disabled script is not
+-- invocable; a disabled route 404s (indistinguishable from absent). Default
+-- TRUE so every existing row stays active — no behavior change on migrate.
+
+ALTER TABLE scripts ADD COLUMN enabled BOOLEAN NOT NULL DEFAULT TRUE;
+ALTER TABLE routes  ADD COLUMN enabled BOOLEAN NOT NULL DEFAULT TRUE;

crates/manager-core/migrations/0046_trigger_name.sql

@@ -0,0 +1,29 @@
+-- Trigger `name` (§4.5): an explicit per-app identifier that becomes the
+-- manifest merge/upsert key (Step B switches the apply diff to key on it).
+-- Until now triggers were identified only by their per-kind semantic tuple,
+-- so the declarative tool could only Create/Delete them, never Update.
+--
+-- Backfill existing rows with `{kind}-{n}` (n = per-(app,kind) sequence by
+-- creation order) — the spec-sanctioned fallback when there's no cleaner
+-- entity token. Then enforce NOT NULL + UNIQUE(app_id, name).
+
+-- A `gen_random_uuid()` default keeps existing INSERTs (which don't yet
+-- supply a name) valid and unique — the manager-core write paths start
+-- providing meaningful names in the follow-up; new rows until then get a
+-- harmless unique placeholder.
+ALTER TABLE triggers
+    ADD COLUMN name TEXT NOT NULL DEFAULT gen_random_uuid()::text;
+
+-- Rewrite the just-defaulted existing rows to the readable `{kind}-{n}` form.
+UPDATE triggers t
+SET name = sub.nm
+FROM (
+    SELECT id,
+           kind || '-' || row_number() OVER (
+               PARTITION BY app_id, kind ORDER BY created_at, id
+           ) AS nm
+    FROM triggers
+) sub
+WHERE t.id = sub.id;
+
+CREATE UNIQUE INDEX triggers_app_name_uniq ON triggers (app_id, name);

crates/manager-core/src/api.rs

@@ -246,6 +246,8 @@ async fn create_script<R: ScriptRepository, L: ExecutionLogRepository>(
             } else {
                 Some(input.sandbox)
             },
+            // Scripts are created active; toggling is a dedicated path.
+            enabled: true,
             imports: validated.imports,
         })
         .await?;
@@ -258,7 +260,7 @@ async fn create_script<R: ScriptRepository, L: ExecutionLogRepository>(
 /// real KV bridge — defense against author confusion, not a security
 /// boundary (stdlib namespaces and module imports already live in
 /// disjoint Rhai scopes).
-const RESERVED_MODULE_NAMES: &[&str] = &[
+pub(crate) const RESERVED_MODULE_NAMES: &[&str] = &[
     "log",
     "regex",
     "random",
@@ -345,6 +347,7 @@ async fn update_script<R: ScriptRepository, L: ExecutionLogRepository>(
                 memory_limit_mb: input.memory_limit_mb,
                 sandbox: input.sandbox,
                 kind: input.kind,
+                enabled: None,
                 imports: imports_for_patch,
             },
         )
@@ -476,10 +479,9 @@ impl IntoResponse for ApiError {
     fn into_response(self) -> Response {
         let (status, message) = match &self {
             Self::NotFound(_) => (StatusCode::NOT_FOUND, self.to_string()),
-            Self::AppNotFound(_)
+            Self::AppNotFound(_) | Self::BadRequest(_) | Self::Invalid(_) | Self::Ceiling(_) => {
-            | Self::BadRequest(_)
+                (StatusCode::UNPROCESSABLE_ENTITY, self.to_string())
-            | Self::Invalid(_)
+            }
-            | Self::Ceiling(_) => (StatusCode::UNPROCESSABLE_ENTITY, self.to_string()),
             Self::Conflict(_) => (StatusCode::CONFLICT, self.to_string()),
             Self::Forbidden => (StatusCode::FORBIDDEN, self.to_string()),
             Self::AuthzRepo(e) => {

crates/manager-core/src/app_bootstrap.rs

@@ -68,6 +68,7 @@ async fn seed_into(
             timeout_seconds: Some(5),
             memory_limit_mb: None,
             sandbox: None,
+            enabled: true,
             imports: Vec::new(),
         })
         .await?;
@@ -85,6 +86,7 @@ async fn seed_into(
             // `curl -d '{"name":"X"}' /hello` work out of the box.
             method: None,
             dispatch_mode: picloud_shared::DispatchMode::Sync,
+            enabled: true,
         })
         .await?;
 

crates/manager-core/src/apply_api.rs

@@ -0,0 +1,173 @@
+//! Admin HTTP surface for the declarative reconcile engine.
+//!
+//! `POST /api/v1/admin/apps/{id}/plan` — diff a desired-state bundle
+//! against the app's live state and return the plan. Read-only; requires
+//! `AppRead`. The `apply` route (write path) lands in the next milestone.
+
+use axum::{
+    extract::{Path, State},
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    routing::post,
+    Extension, Json, Router,
+};
+use picloud_shared::{AppId, Principal};
+use serde::Deserialize;
+use serde_json::json;
+
+use crate::app_repo::AppRepository;
+use crate::apply_service::{
+    ApplyError, ApplyReport, ApplyService, Bundle, BundleTrigger, PlanResult,
+};
+use crate::authz::{require, AuthzDenied, Capability};
+
+/// Build the apply/plan router. Mounted under `/api/v1/admin`.
+pub fn apply_router(service: ApplyService) -> Router {
+    Router::new()
+        .route("/apps/{id}/plan", post(plan_handler))
+        .route("/apps/{id}/apply", post(apply_handler))
+        .with_state(service)
+}
+
+#[derive(Deserialize)]
+pub struct ApplyRequest {
+    pub bundle: Bundle,
+    #[serde(default)]
+    pub prune: bool,
+    /// Optional bound-plan token from a prior `plan`. When present, apply
+    /// refuses (409) if the app's live state has changed since.
+    #[serde(default)]
+    pub expected_token: Option<String>,
+}
+
+async fn apply_handler(
+    State(svc): State<ApplyService>,
+    Extension(principal): Extension<Principal>,
+    Path(id_or_slug): Path<String>,
+    Json(req): Json<ApplyRequest>,
+) -> Result<Json<ApplyReport>, ApplyError> {
+    let app_id = resolve_app_id(svc.apps.as_ref(), &id_or_slug).await?;
+    // Read is always needed; write caps are required for the resource kinds
+    // the bundle touches — and for ALL kinds when `prune` is set, since
+    // pruning deletes resources whose bundle section is empty (and a script
+    // delete cascades its routes/triggers).
+    require(svc.authz.as_ref(), &principal, Capability::AppRead(app_id))
+        .await
+        .map_err(map_authz)?;
+    if req.prune || !req.bundle.scripts.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            &principal,
+            Capability::AppWriteScript(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    if req.prune || !req.bundle.routes.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            &principal,
+            Capability::AppWriteRoute(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    if req.prune || !req.bundle.triggers.is_empty() {
+        require(
+            svc.authz.as_ref(),
+            &principal,
+            Capability::AppManageTriggers(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    // Email triggers resolve and decrypt a stored secret by name server-side,
+    // which the secrets API guards with `AppSecretsRead`. Require it here too
+    // so apply can't bind a secret a principal couldn't otherwise read — the
+    // caps aren't strictly nested on the API-key scope path.
+    if req.bundle.triggers.iter().any(BundleTrigger::is_email) {
+        require(
+            svc.authz.as_ref(),
+            &principal,
+            Capability::AppSecretsRead(app_id),
+        )
+        .await
+        .map_err(map_authz)?;
+    }
+    let report = svc
+        .apply(
+            app_id,
+            &req.bundle,
+            req.prune,
+            principal.user_id,
+            req.expected_token.as_deref(),
+        )
+        .await?;
+    Ok(Json(report))
+}
+
+async fn plan_handler(
+    State(svc): State<ApplyService>,
+    Extension(principal): Extension<Principal>,
+    Path(id_or_slug): Path<String>,
+    Json(bundle): Json<Bundle>,
+) -> Result<Json<PlanResult>, ApplyError> {
+    let app_id = resolve_app_id(svc.apps.as_ref(), &id_or_slug).await?;
+    // NOTE: the returned `Plan` discloses live secret NAMES (not values). That
+    // is safe today only because `AppRead` and `AppSecretsRead` are co-granted
+    // at every tier (same `script:read` scope, both in the viewer role). If a
+    // future authz split puts `AppSecretsRead` on its own tier, this handler
+    // must additionally require it — otherwise it leaks names a principal
+    // couldn't enumerate via the secrets API.
+    require(svc.authz.as_ref(), &principal, Capability::AppRead(app_id))
+        .await
+        .map_err(map_authz)?;
+    let plan = svc.plan(app_id, &bundle).await?;
+    Ok(Json(plan))
+}
+
+/// Resolve a slug-or-id path param to an `AppId`, mapping miss → 404.
+/// Mirrors the `triggers_api` helper of the same shape.
+async fn resolve_app_id(apps: &dyn AppRepository, ident: &str) -> Result<AppId, ApplyError> {
+    crate::app_repo::resolve_app(apps, ident)
+        .await
+        .map_err(|e| ApplyError::Backend(e.to_string()))?
+        .map(|l| l.app.id)
+        .ok_or_else(|| ApplyError::AppNotFound(ident.to_string()))
+}
+
+fn map_authz(denied: AuthzDenied) -> ApplyError {
+    match denied {
+        AuthzDenied::Denied => ApplyError::Forbidden,
+        AuthzDenied::Repo(e) => ApplyError::AuthzRepo(e.to_string()),
+    }
+}
+
+impl IntoResponse for ApplyError {
+    fn into_response(self) -> Response {
+        let (status, body) = match &self {
+            Self::AppNotFound(_) => (StatusCode::NOT_FOUND, json!({ "error": self.to_string() })),
+            Self::Invalid(_) => (
+                StatusCode::UNPROCESSABLE_ENTITY,
+                json!({ "error": self.to_string() }),
+            ),
+            Self::StateMoved => (StatusCode::CONFLICT, json!({ "error": self.to_string() })),
+            Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })),
+            Self::AuthzRepo(e) => {
+                tracing::error!(error = %e, "apply authz repo error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+            Self::Backend(e) => {
+                tracing::error!(error = %e, "apply backend error");
+                (
+                    StatusCode::INTERNAL_SERVER_ERROR,
+                    json!({ "error": "internal error" }),
+                )
+            }
+        };
+        (status, Json(body)).into_response()
+    }
+}

crates/manager-core/src/dispatcher.rs

@@ -384,6 +384,39 @@ impl Dispatcher {
             return Ok(());
         }
 
+        // Fire-time `enabled` re-check (§4.3). The queue arm does not flow
+        // through `dispatch_one`'s unified `!active` gate; its only other
+        // `enabled` guard is the per-tick `list_active_queue_consumers`
+        // snapshot, which is stale for messages already claimed in this
+        // tick. A script disabled after the list query but before this
+        // claimed message executes must NOT run (`script` here is a fresh
+        // read from line ~331, so `enabled` is current). Release the claim
+        // (nack) rather than ack/dead-letter: the message stays queued and
+        // is processed when the script is re-enabled, and the next tick
+        // won't re-claim it because the list filters `s.enabled`. Without
+        // this nack the message would sit claimed indefinitely —
+        // `reclaim_visibility_timeouts` only reclaims for enabled triggers.
+        if !script.enabled {
+            tracing::info!(
+                script_id = %consumer.script_id,
+                trigger_id = %consumer.trigger_id,
+                "queue consumer script disabled at fire time; releasing claim"
+            );
+            if let Err(e) = self
+                .queue
+                .nack(
+                    claimed.id,
+                    claimed.claim_token,
+                    chrono::Duration::seconds(1),
+                )
+                .await
+            {
+                tracing::warn!(?e, "queue nack on disabled consumer failed");
+            }
+            drop(permit);
+            return Ok(());
+        }
+
         let principal = self
             .principals
             .resolve(consumer.registered_by_principal)
@@ -546,6 +579,7 @@ impl Dispatcher {
         // TODO(metrics): bump picloud_queue_dead_letters_total{app_id, queue_name}.
     }
 
+    #[allow(clippy::too_many_lines)]
     async fn dispatch_one(&self, row: OutboxRow) -> Result<(), DispatcherError> {
         // Depth-limit check — design notes §4: loops aren't DL'd.
         if row.trigger_depth > self.config.max_trigger_depth {
@@ -626,6 +660,26 @@ impl Dispatcher {
             }
         };
 
+        // §4.3 fire-time re-check, for EVERY outbox source (trigger, async
+        // HTTP/202, and invoke): if the target script (or, for triggers, the
+        // trigger) was disabled after this row was enqueued, drop it rather
+        // than fire a stale event. The match-time `enabled` check can't see a
+        // later toggle, so this is the gate that makes a disabled script
+        // genuinely non-invocable on the async paths too.
+        if !resolved.active {
+            tracing::debug!(
+                outbox_id = %row.id,
+                app_id = %row.app_id,
+                "target script/trigger disabled since enqueue; dropping outbox row"
+            );
+            self.outbox
+                .delete(row.id)
+                .await
+                .map_err(|e| DispatcherError::Outbox(e.to_string()))?;
+            drop(permit);
+            return Ok(());
+        }
+
         // The gate permit auto-releases when this scope ends or when
         // the executor finishes. We hand control to the executor and
         // wait synchronously here — sync HTTP and dispatcher share the
@@ -722,9 +776,26 @@ impl Dispatcher {
                 DispatcherError::ResolveTrigger(format!("script {} not found", trigger.script_id))
             })?;
 
+        // Audit 2026-06-11 H-F1 sibling — same-app guard mirroring
+        // build_http_request / build_invoke_request / dispatch_one_queue.
+        // `build_exec_request` stamps `ExecRequest.app_id = row.app_id`
+        // while sourcing the body from `trigger.script_id`; without this
+        // check a hand-edited outbox/trigger row (or a partial restore, or
+        // a script re-pointed across apps) could run one app's script under
+        // another app's `SdkCallCx.app_id` — the cross-app isolation
+        // boundary. Not reachable via the trigger-create or `apply` paths
+        // (both resolve the script within the app's own scope), so this is
+        // the runtime backstop the other arms already carry.
+        if script.app_id != row.app_id {
+            return Err(DispatcherError::ResolveTrigger(
+                "trigger outbox target belongs to a different app".into(),
+            ));
+        }
+
         Ok(ResolvedTrigger {
             trigger_kind: trigger.kind,
             is_dead_letter_handler: matches!(trigger.kind, TriggerKind::DeadLetter),
+            active: trigger.enabled && script.enabled,
             script_id: script.id,
             script_source: script.source,
             script_name: script.name,
@@ -844,6 +915,9 @@ impl Dispatcher {
         let resolved = ResolvedTrigger {
             trigger_kind: TriggerKind::Kv, // placeholder; HTTP doesn't have a kind
             is_dead_letter_handler: false,
+            // §4.3: an async-HTTP (202) row whose script was disabled after
+            // enqueue is dropped at fire time by the post-match active check.
+            active: script.enabled,
             script_id,
             script_source: script.source,
             script_name: payload.script_name,
@@ -941,6 +1015,9 @@ impl Dispatcher {
         let resolved = ResolvedTrigger {
             trigger_kind: TriggerKind::Cron, // placeholder; not used downstream
             is_dead_letter_handler: false,
+            // §4.3: a queued invoke() whose target script was disabled after
+            // enqueue is dropped at fire time by the post-match active check.
+            active: script.enabled,
             script_id: script.id,
             script_source: script.source,
             script_name: script.name,
@@ -1238,6 +1315,9 @@ impl Dispatcher {
 pub struct ResolvedTrigger {
     pub trigger_kind: TriggerKind,
     pub is_dead_letter_handler: bool,
+    /// §4.3 fire-time gate: `trigger.enabled && script.enabled` at resolve
+    /// time. A row enqueued before either was disabled is dropped, not fired.
+    pub active: bool,
     pub script_id: ScriptId,
     pub script_source: String,
     pub script_name: String,
@@ -1504,4 +1584,837 @@ mod tests {
         );
         assert_eq!(failure_kind_to_status(InboxFailureKind::Platform), 500);
     }
+
+    // ----------------------------------------------------------------
+    // Queue-arm fire-time `enabled` gate (§4.3).
+    //
+    // `dispatch_one_queue` re-reads the script fresh after claiming a
+    // message and, if it has been disabled since the per-tick
+    // `list_active_queue_consumers` snapshot, releases the claim (nack)
+    // without resolving a principal or executing. This test proves that
+    // gate end-to-end with in-memory stubs.
+    //
+    // Regression property: the principal resolver returns a valid
+    // `Principal` and the executor records `executed = true` before
+    // erroring, so DELETING the `if !script.enabled` gate makes the flow
+    // fall through to resolve → execute, flipping `executed` and failing
+    // `assert!(!executed)`. (Verified by temporarily removing the gate.)
+    // ----------------------------------------------------------------
+    mod queue_enabled_gate {
+        use super::*;
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::Arc;
+
+        use async_trait::async_trait;
+        use chrono::Utc;
+        use picloud_executor_core::{ExecError, ExecRequest, ExecResponse};
+        use picloud_orchestrator_core::{ExecutionGate, ExecutorClient};
+        use picloud_shared::{
+            AdminUserId, AppId, InstanceRole, Principal, Script, ScriptId, ScriptSandbox,
+        };
+        use uuid::Uuid;
+
+        use crate::abandoned_repo::{AbandonedRepo, NewAbandonedExecution};
+        use crate::dead_letter_repo::{DeadLetterRepo, NewDeadLetter};
+        use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow};
+        use crate::principal_resolver::{PrincipalResolver, PrincipalResolverError};
+        use crate::queue_repo::{ClaimedMessage, NewQueueMessage, QueueRepo, QueueStats};
+        use crate::repo::{NewScript, ScriptPatch, ScriptRepository, ScriptRepositoryError};
+        use crate::trigger_config::BackoffShape;
+        use crate::trigger_repo::{ActiveQueueConsumer, TriggerRepo};
+
+        // ---- ScriptRepository: only `get` is exercised. ----
+        pub(super) struct DisabledScriptRepo {
+            pub(super) script: Script,
+        }
+
+        #[async_trait]
+        impl ScriptRepository for DisabledScriptRepo {
+            async fn get(&self, _id: ScriptId) -> Result<Option<Script>, ScriptRepositoryError> {
+                Ok(Some(self.script.clone()))
+            }
+            async fn get_by_name(
+                &self,
+                _app_id: AppId,
+                _name: &str,
+            ) -> Result<Option<Script>, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn list(&self) -> Result<Vec<Script>, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn list_for_app(
+                &self,
+                _app_id: AppId,
+            ) -> Result<Vec<Script>, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn list_for_user(
+                &self,
+                _user_id: AdminUserId,
+            ) -> Result<Vec<Script>, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn create(&self, _input: NewScript) -> Result<Script, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn update(
+                &self,
+                _id: ScriptId,
+                _patch: ScriptPatch,
+            ) -> Result<Script, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn delete(&self, _id: ScriptId) -> Result<(), ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn count_routes_for_script(
+                &self,
+                _script_id: ScriptId,
+            ) -> Result<i64, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn count_triggers_for_script(
+                &self,
+                _script_id: ScriptId,
+            ) -> Result<i64, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+            async fn list_imports(
+                &self,
+                _script_id: ScriptId,
+            ) -> Result<Vec<Script>, ScriptRepositoryError> {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        // ---- QueueRepo: `claim` returns the message, `nack` records. ----
+        struct ClaimNackQueue {
+            claimed: ClaimedMessage,
+            nacked: Arc<AtomicBool>,
+        }
+
+        #[async_trait]
+        impl QueueRepo for ClaimNackQueue {
+            async fn enqueue(
+                &self,
+                _msg: NewQueueMessage,
+            ) -> Result<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn claim(
+                &self,
+                _app_id: AppId,
+                _queue_name: &str,
+            ) -> Result<Option<ClaimedMessage>, crate::queue_repo::QueueRepoError> {
+                Ok(Some(self.claimed.clone()))
+            }
+            async fn ack(
+                &self,
+                _message_id: picloud_shared::QueueMessageId,
+                _claim_token: Uuid,
+            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn nack(
+                &self,
+                _message_id: picloud_shared::QueueMessageId,
+                _claim_token: Uuid,
+                _retry_delay: chrono::Duration,
+            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
+                self.nacked.store(true, Ordering::SeqCst);
+                Ok(true)
+            }
+            async fn reclaim_visibility_timeouts(
+                &self,
+            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn depth(
+                &self,
+                _app_id: AppId,
+                _queue_name: &str,
+            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn depth_pending(
+                &self,
+                _app_id: AppId,
+                _queue_name: &str,
+            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn list_for_app(
+                &self,
+                _app_id: AppId,
+            ) -> Result<Vec<(String, QueueStats)>, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            #[allow(clippy::too_many_arguments)]
+            async fn dead_letter(
+                &self,
+                _message_id: picloud_shared::QueueMessageId,
+                _claim_token: Uuid,
+                _app_id: AppId,
+                _queue_name: &str,
+                _trigger_id: Option<picloud_shared::TriggerId>,
+                _script_id: Option<ScriptId>,
+                _attempt: u32,
+                _first_attempt_at: chrono::DateTime<Utc>,
+                _last_error: &str,
+            ) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        // ---- PrincipalResolver: returns a valid Principal (regression). ----
+        pub(super) struct OkPrincipals;
+
+        #[async_trait]
+        impl PrincipalResolver for OkPrincipals {
+            async fn resolve(
+                &self,
+                user_id: AdminUserId,
+            ) -> Result<Principal, PrincipalResolverError> {
+                Ok(Principal {
+                    user_id,
+                    instance_role: InstanceRole::Owner,
+                    scopes: None,
+                    app_binding: None,
+                })
+            }
+        }
+
+        // ---- ExecutorClient: records execution then errors (regression). ----
+        pub(super) struct RecordingExecutor {
+            pub(super) executed: Arc<AtomicBool>,
+        }
+
+        #[async_trait]
+        impl ExecutorClient for RecordingExecutor {
+            async fn execute(
+                &self,
+                _source: &str,
+                _req: ExecRequest,
+                _timeout: std::time::Duration,
+            ) -> Result<ExecResponse, ExecError> {
+                self.executed.store(true, Ordering::SeqCst);
+                Err(ExecError::Runtime("stub".into()))
+            }
+            // Intentionally NOT overriding `execute_with_identity`: the
+            // default impl forwards to `execute`, which is what gate
+            // removal would reach.
+        }
+
+        // ---- Remaining deps: never exercised by the disabled path. ----
+        pub(super) struct UnusedTriggers;
+
+        #[async_trait]
+        impl TriggerRepo for UnusedTriggers {
+            async fn create_kv_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreateKvTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn create_docs_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreateDocsTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn create_dead_letter_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreateDeadLetterTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn create_cron_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreateCronTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn create_files_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreateFilesTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn create_pubsub_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreatePubsubTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn create_email_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreateEmailTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn email_inbound_target(
+                &self,
+                _trigger_id: picloud_shared::TriggerId,
+            ) -> Result<
+                Option<crate::trigger_repo::EmailInboundTarget>,
+                crate::trigger_repo::TriggerRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            async fn list_for_app(
+                &self,
+                _app_id: AppId,
+            ) -> Result<Vec<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn get(
+                &self,
+                _id: picloud_shared::TriggerId,
+            ) -> Result<Option<crate::trigger_repo::Trigger>, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn delete(
+                &self,
+                _id: picloud_shared::TriggerId,
+            ) -> Result<bool, crate::trigger_repo::TriggerRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn list_matching_kv(
+                &self,
+                _app_id: AppId,
+                _collection: &str,
+                _op: picloud_shared::KvEventOp,
+            ) -> Result<
+                Vec<crate::trigger_repo::KvTriggerMatch>,
+                crate::trigger_repo::TriggerRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            async fn list_matching_docs(
+                &self,
+                _app_id: AppId,
+                _collection: &str,
+                _op: picloud_shared::DocsEventOp,
+            ) -> Result<
+                Vec<crate::trigger_repo::DocsTriggerMatch>,
+                crate::trigger_repo::TriggerRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            async fn list_matching_files(
+                &self,
+                _app_id: AppId,
+                _collection: &str,
+                _op: picloud_shared::FilesEventOp,
+            ) -> Result<
+                Vec<crate::trigger_repo::FilesTriggerMatch>,
+                crate::trigger_repo::TriggerRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            async fn list_matching_dead_letter(
+                &self,
+                _app_id: AppId,
+                _source: &str,
+                _trigger_id: Option<picloud_shared::TriggerId>,
+                _script_id: Option<ScriptId>,
+            ) -> Result<
+                Vec<crate::trigger_repo::DeadLetterTriggerMatch>,
+                crate::trigger_repo::TriggerRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            async fn create_queue_trigger(
+                &self,
+                _app_id: AppId,
+                _req: crate::trigger_repo::CreateQueueTrigger,
+            ) -> Result<crate::trigger_repo::Trigger, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn list_active_queue_consumers(
+                &self,
+            ) -> Result<Vec<ActiveQueueConsumer>, crate::trigger_repo::TriggerRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn touch_queue_trigger_last_fired_at(
+                &self,
+                _trigger_id: picloud_shared::TriggerId,
+                _at: chrono::DateTime<Utc>,
+            ) -> Result<(), crate::trigger_repo::TriggerRepoError> {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        pub(super) struct UnusedDeadLetters;
+
+        #[async_trait]
+        impl DeadLetterRepo for UnusedDeadLetters {
+            async fn insert(
+                &self,
+                _row: NewDeadLetter,
+            ) -> Result<picloud_shared::DeadLetterId, crate::dead_letter_repo::DeadLetterRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn get(
+                &self,
+                _id: picloud_shared::DeadLetterId,
+            ) -> Result<
+                Option<crate::dead_letter_repo::DeadLetterRow>,
+                crate::dead_letter_repo::DeadLetterRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            async fn list_for_app(
+                &self,
+                _app_id: AppId,
+                _unresolved_only: bool,
+                _limit: i64,
+                _offset: i64,
+            ) -> Result<
+                Vec<crate::dead_letter_repo::DeadLetterRow>,
+                crate::dead_letter_repo::DeadLetterRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            async fn unresolved_count(
+                &self,
+                _app_id: AppId,
+            ) -> Result<i64, crate::dead_letter_repo::DeadLetterRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn resolve(
+                &self,
+                _id: picloud_shared::DeadLetterId,
+                _reason: &str,
+            ) -> Result<(), crate::dead_letter_repo::DeadLetterRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn gc(
+                &self,
+                _older_than: chrono::DateTime<Utc>,
+                _limit: i64,
+            ) -> Result<u64, crate::dead_letter_repo::DeadLetterRepoError> {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        struct UnusedOutbox;
+
+        #[async_trait]
+        impl OutboxRepo for UnusedOutbox {
+            async fn insert(
+                &self,
+                _row: NewOutboxRow,
+            ) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn claim_due(
+                &self,
+                _claimed_by: &str,
+                _limit: i64,
+            ) -> Result<Vec<OutboxRow>, crate::outbox_repo::OutboxRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn delete(&self, _id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn reschedule(
+                &self,
+                _id: Uuid,
+                _attempt_count: u32,
+                _next_attempt_at: chrono::DateTime<Utc>,
+            ) -> Result<(), crate::outbox_repo::OutboxRepoError> {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        pub(super) struct UnusedAbandoned;
+
+        #[async_trait]
+        impl AbandonedRepo for UnusedAbandoned {
+            async fn insert(
+                &self,
+                _row: NewAbandonedExecution,
+            ) -> Result<Uuid, crate::abandoned_repo::AbandonedRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn gc(
+                &self,
+                _older_than: chrono::DateTime<Utc>,
+                _limit: i64,
+            ) -> Result<u64, crate::abandoned_repo::AbandonedRepoError> {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        pub(super) struct UnusedLogSink;
+
+        #[async_trait]
+        impl ExecutionLogSink for UnusedLogSink {
+            async fn record(
+                &self,
+                _log: picloud_shared::ExecutionLog,
+            ) -> Result<(), picloud_shared::LogSinkError> {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        pub(super) struct UnusedInbox;
+
+        #[async_trait]
+        impl InboxResolver for UnusedInbox {
+            async fn deliver(
+                &self,
+                _inbox_id: Uuid,
+                _result: picloud_shared::InboxResult,
+            ) -> picloud_shared::InboxDeliveryOutcome {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        pub(super) fn disabled_script(app_id: AppId) -> Script {
+            Script {
+                id: ScriptId::new(),
+                app_id,
+                name: "worker".into(),
+                description: None,
+                version: 1,
+                source: "0".into(),
+                kind: picloud_shared::ScriptKind::Endpoint,
+                timeout_seconds: 30,
+                memory_limit_mb: 64,
+                sandbox: ScriptSandbox::default(),
+                enabled: false,
+                created_at: Utc::now(),
+                updated_at: Utc::now(),
+            }
+        }
+
+        #[tokio::test]
+        async fn disabled_queue_consumer_releases_claim_without_executing() {
+            let app_id = AppId::new();
+            let script = disabled_script(app_id);
+
+            let claimed = ClaimedMessage {
+                id: picloud_shared::QueueMessageId::new(),
+                app_id,
+                queue_name: "jobs".into(),
+                payload: serde_json::Value::Null,
+                enqueued_at: Utc::now(),
+                attempt: 1,
+                max_attempts: 5,
+                claim_token: Uuid::new_v4(),
+            };
+
+            let consumer = ActiveQueueConsumer {
+                trigger_id: picloud_shared::TriggerId::new(),
+                app_id,
+                script_id: script.id,
+                queue_name: "jobs".into(),
+                visibility_timeout_secs: 30,
+                retry_max_attempts: 5,
+                retry_backoff: BackoffShape::Exponential,
+                retry_base_ms: 1000,
+                registered_by_principal: AdminUserId::new(),
+            };
+
+            let nacked = Arc::new(AtomicBool::new(false));
+            let executed = Arc::new(AtomicBool::new(false));
+
+            let dispatcher = Dispatcher {
+                outbox: Arc::new(UnusedOutbox),
+                triggers: Arc::new(UnusedTriggers),
+                scripts: Arc::new(DisabledScriptRepo {
+                    script: script.clone(),
+                }),
+                dead_letters: Arc::new(UnusedDeadLetters),
+                abandoned: Arc::new(UnusedAbandoned),
+                principals: Arc::new(OkPrincipals),
+                executor: Arc::new(RecordingExecutor {
+                    executed: executed.clone(),
+                }),
+                gate: Arc::new(ExecutionGate::new(1)),
+                log_sink: Arc::new(UnusedLogSink),
+                inbox: Arc::new(UnusedInbox),
+                queue: Arc::new(ClaimNackQueue {
+                    claimed,
+                    nacked: nacked.clone(),
+                }),
+                config: TriggerConfig::from_env(),
+                instance_id: "test-instance".into(),
+            };
+
+            let result = dispatcher.dispatch_one_queue(&consumer).await;
+
+            // 1. The disabled path returns Ok(()).
+            assert!(result.is_ok(), "dispatch_one_queue returned {result:?}");
+            // 2. The claim was released via nack.
+            assert!(
+                nacked.load(Ordering::SeqCst),
+                "expected nack to release the claim for a disabled consumer"
+            );
+            // 3. The executor was never reached. If the `if !script.enabled`
+            //    gate is deleted, the flow falls through to resolve →
+            //    execute and this flips to true.
+            assert!(
+                !executed.load(Ordering::SeqCst),
+                "executor ran for a disabled queue consumer; fire-time gate missing"
+            );
+        }
+    }
+
+    // ----------------------------------------------------------------
+    // OUTBOX (async-HTTP) arm fire-time `enabled` gate (§4.3).
+    //
+    // `dispatch_one` builds an `ExecRequest` from an HTTP outbox row via
+    // `build_http_request`, which sets `resolved.active = script.enabled`
+    // but does NOT itself reject a disabled script. The unified gate at
+    // `if !resolved.active` then drops the row (delete) without executing.
+    // This test proves that gate end-to-end for an HTTP-source row whose
+    // target script was disabled after the row was enqueued.
+    //
+    // Regression property (mirrors the queue test): `RecordingExecutor`
+    // flips `executed = true` before erroring, so DELETING the
+    // `if !resolved.active` gate makes the flow fall through to execute,
+    // flipping `executed` and failing `assert!(!executed)`.
+    // ----------------------------------------------------------------
+    mod outbox_enabled_gate {
+        use super::*;
+        use std::sync::atomic::{AtomicBool, Ordering};
+        use std::sync::{Arc, Mutex};
+
+        use async_trait::async_trait;
+        use chrono::Utc;
+        use picloud_orchestrator_core::ExecutionGate;
+        use picloud_shared::AppId;
+        use uuid::Uuid;
+
+        use crate::outbox_repo::{NewOutboxRow, OutboxRepo, OutboxRow, OutboxSourceKind};
+
+        // Reuse the stub trait impls + helpers from the queue test so the
+        // two gate tests share one set of in-memory deps.
+        use super::queue_enabled_gate::{
+            disabled_script, DisabledScriptRepo, OkPrincipals, RecordingExecutor, UnusedAbandoned,
+            UnusedDeadLetters, UnusedInbox, UnusedLogSink, UnusedTriggers,
+        };
+
+        // QueueRepo is never reached on the outbox arm — a panic stub keeps
+        // the surface honest without pulling the queue test's claim stub.
+        struct UnusedQueue;
+
+        #[async_trait]
+        impl crate::queue_repo::QueueRepo for UnusedQueue {
+            async fn enqueue(
+                &self,
+                _msg: crate::queue_repo::NewQueueMessage,
+            ) -> Result<picloud_shared::QueueMessageId, crate::queue_repo::QueueRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn claim(
+                &self,
+                _app_id: AppId,
+                _queue_name: &str,
+            ) -> Result<Option<crate::queue_repo::ClaimedMessage>, crate::queue_repo::QueueRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+            async fn ack(
+                &self,
+                _message_id: picloud_shared::QueueMessageId,
+                _claim_token: Uuid,
+            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn nack(
+                &self,
+                _message_id: picloud_shared::QueueMessageId,
+                _claim_token: Uuid,
+                _retry_delay: chrono::Duration,
+            ) -> Result<bool, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn reclaim_visibility_timeouts(
+                &self,
+            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn depth(
+                &self,
+                _app_id: AppId,
+                _queue_name: &str,
+            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn depth_pending(
+                &self,
+                _app_id: AppId,
+                _queue_name: &str,
+            ) -> Result<u64, crate::queue_repo::QueueRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn list_for_app(
+                &self,
+                _app_id: AppId,
+            ) -> Result<
+                Vec<(String, crate::queue_repo::QueueStats)>,
+                crate::queue_repo::QueueRepoError,
+            > {
+                unimplemented!("not used by this test")
+            }
+            #[allow(clippy::too_many_arguments)]
+            async fn dead_letter(
+                &self,
+                _message_id: picloud_shared::QueueMessageId,
+                _claim_token: Uuid,
+                _app_id: AppId,
+                _queue_name: &str,
+                _trigger_id: Option<picloud_shared::TriggerId>,
+                _script_id: Option<picloud_shared::ScriptId>,
+                _attempt: u32,
+                _first_attempt_at: chrono::DateTime<Utc>,
+                _last_error: &str,
+            ) -> Result<picloud_shared::DeadLetterId, crate::queue_repo::QueueRepoError>
+            {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        // OutboxRepo whose `delete` records the id it was called with; the
+        // other methods are never reached on the disabled-drop path.
+        struct RecordingOutbox {
+            deleted: Arc<Mutex<Option<Uuid>>>,
+        }
+
+        #[async_trait]
+        impl OutboxRepo for RecordingOutbox {
+            async fn insert(
+                &self,
+                _row: NewOutboxRow,
+            ) -> Result<Uuid, crate::outbox_repo::OutboxRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn claim_due(
+                &self,
+                _claimed_by: &str,
+                _limit: i64,
+            ) -> Result<Vec<OutboxRow>, crate::outbox_repo::OutboxRepoError> {
+                unimplemented!("not used by this test")
+            }
+            async fn delete(&self, id: Uuid) -> Result<(), crate::outbox_repo::OutboxRepoError> {
+                *self.deleted.lock().unwrap() = Some(id);
+                Ok(())
+            }
+            async fn reschedule(
+                &self,
+                _id: Uuid,
+                _attempt_count: u32,
+                _next_attempt_at: chrono::DateTime<Utc>,
+            ) -> Result<(), crate::outbox_repo::OutboxRepoError> {
+                unimplemented!("not used by this test")
+            }
+        }
+
+        #[tokio::test]
+        async fn disabled_http_outbox_row_is_dropped_without_executing() {
+            let app_id = AppId::new();
+            let script = disabled_script(app_id);
+
+            // Minimal valid HttpDispatchPayload (see shared::outbox_writer).
+            let payload = serde_json::json!({
+                "script_name": "worker",
+                "path": "/hook",
+                "method": "POST",
+                "headers": {},
+                "body": null,
+                "params": {},
+                "query": {},
+                "rest": "",
+                "timeout_seconds": 30,
+            });
+
+            let row_id = Uuid::new_v4();
+            let row = OutboxRow {
+                id: row_id,
+                // Same app_id as the script so the same-app guard passes.
+                app_id,
+                source_kind: OutboxSourceKind::Http,
+                trigger_id: None,
+                script_id: Some(script.id),
+                reply_to: None,
+                payload,
+                origin_principal: None,
+                trigger_depth: 0,
+                root_execution_id: None,
+                attempt_count: 0,
+                next_attempt_at: Utc::now(),
+                created_at: Utc::now(),
+            };
+
+            let deleted = Arc::new(Mutex::new(None));
+            let executed = Arc::new(AtomicBool::new(false));
+
+            let dispatcher = Dispatcher {
+                outbox: Arc::new(RecordingOutbox {
+                    deleted: deleted.clone(),
+                }),
+                triggers: Arc::new(UnusedTriggers),
+                scripts: Arc::new(DisabledScriptRepo {
+                    script: script.clone(),
+                }),
+                dead_letters: Arc::new(UnusedDeadLetters),
+                abandoned: Arc::new(UnusedAbandoned),
+                principals: Arc::new(OkPrincipals),
+                executor: Arc::new(RecordingExecutor {
+                    executed: executed.clone(),
+                }),
+                gate: Arc::new(ExecutionGate::new(1)),
+                log_sink: Arc::new(UnusedLogSink),
+                inbox: Arc::new(UnusedInbox),
+                queue: Arc::new(UnusedQueue),
+                config: TriggerConfig::from_env(),
+                instance_id: "test-instance".into(),
+            };
+
+            let result = dispatcher.dispatch_one(row).await;
+
+            // 1. The disabled-drop path returns Ok(()).
+            assert!(result.is_ok(), "dispatch_one returned {result:?}");
+            // 2. The outbox row was deleted with its own id.
+            assert_eq!(
+                *deleted.lock().unwrap(),
+                Some(row_id),
+                "expected the disabled HTTP outbox row to be deleted"
+            );
+            // 3. The executor was never reached. If the `if !resolved.active`
+            //    gate at dispatch_one is deleted, the flow falls through to
+            //    execute and this flips to true.
+            assert!(
+                !executed.load(Ordering::SeqCst),
+                "executor ran for a disabled HTTP outbox row; fire-time gate missing"
+            );
+        }
+    }
 }

crates/manager-core/src/email_service.rs

@@ -341,7 +341,10 @@ impl DevEmailSink {
     }
 
     fn push(&self, email: CapturedEmail) {
-        let mut q = self.captured.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        let mut q = self
+            .captured
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
         while q.len() >= self.capacity {
             q.pop_front();
         }
@@ -351,7 +354,10 @@ impl DevEmailSink {
     /// Newest-first snapshot of the captured mail.
     #[must_use]
     pub fn snapshot(&self) -> Vec<CapturedEmail> {
-        let q = self.captured.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+        let q = self
+            .captured
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
         q.iter().rev().cloned().collect()
     }
 }

crates/manager-core/src/invoke_service.rs

@@ -83,6 +83,11 @@ impl InvokeServiceImpl {
         if script.app_id != cx.app_id {
             return Err(InvokeError::CrossApp);
         }
+        if !script.enabled {
+            // §4.3: a disabled script is not invocable via any path. Surface as
+            // NotFound (indistinguishable from absent), like the data plane.
+            return Err(InvokeError::NotFound(format!("id {script_id}")));
+        }
         Ok(ResolvedScript {
             script_id: script.id,
             app_id: script.app_id,
@@ -103,6 +108,9 @@ impl InvokeServiceImpl {
             .await
             .map_err(|e| InvokeError::Backend(e.to_string()))?
             .ok_or_else(|| InvokeError::NotFound(format!("name {name:?}")))?;
+        if !script.enabled {
+            return Err(InvokeError::NotFound(format!("name {name:?}")));
+        }
         Ok(ResolvedScript {
             script_id: script.id,
             app_id: script.app_id,
@@ -313,6 +321,7 @@ mod tests {
             timeout_seconds: 30,
             memory_limit_mb: 64,
             sandbox: ScriptSandbox::default(),
+            enabled: true,
             created_at: Utc::now(),
             updated_at: Utc::now(),
         }

crates/manager-core/src/lib.rs

@@ -23,6 +23,8 @@ pub mod app_user_repo;
 pub mod app_user_role_repo;
 pub mod app_user_session_repo;
 pub mod app_user_verification_repo;
+pub mod apply_api;
+pub mod apply_service;
 pub mod apps_api;
 pub mod auth;
 pub mod auth_api;
@@ -128,6 +130,8 @@ pub use app_user_session_repo::{
 pub use app_user_verification_repo::{
     AppUserVerificationRepo, AppUserVerificationRepoError, PostgresAppUserVerificationRepo,
 };
+pub use apply_api::apply_router;
+pub use apply_service::{ApplyError, ApplyService, Bundle, Plan};
 pub use apps_api::{apps_router, AppsState};
 pub use auth_api::auth_router;
 pub use auth_bootstrap::{

crates/manager-core/src/repo.rs

@@ -154,6 +154,8 @@ pub struct NewScript {
     /// Sandbox overrides; `None` means store an empty object (use
     /// platform defaults at exec time).
     pub sandbox: Option<ScriptSandbox>,
+    /// Three-state lifecycle (§4.3). Create active by default.
+    pub enabled: bool,
     /// v1.1.3: literal-path `import "<name>"` declarations extracted
     /// from the source. The repo writes these into `script_imports`
     /// transactionally with the script row. Empty when validation
@@ -176,6 +178,8 @@ pub struct ScriptPatch {
     /// rejects unsafe transitions (e.g. endpoint→module when routes
     /// or triggers reference the script).
     pub kind: Option<ScriptKind>,
+    /// `Some(v)` toggles the three-state lifecycle flag; `None` leaves it.
+    pub enabled: Option<bool>,
     /// v1.1.3: when `source` is also `Some`, the repo replaces the
     /// `script_imports` edges for this script with these names.
     /// `None` keeps the existing edges untouched (a name/description
@@ -203,7 +207,8 @@ impl PostgresScriptRepository {
 /// adding `kind` (v1.1.3) and future columns can't accidentally skip
 /// one query.
 const SCRIPT_SELECT_COLS: &str = "id, app_id, name, description, version, source, kind, \
-                                  timeout_seconds, memory_limit_mb, sandbox, created_at, updated_at";
+                                  timeout_seconds, memory_limit_mb, sandbox, enabled, \
+                                  created_at, updated_at";
 
 #[async_trait]
 impl ScriptRepository for PostgresScriptRepository {
@@ -273,42 +278,8 @@ impl ScriptRepository for PostgresScriptRepository {
     }
 
     async fn create(&self, input: NewScript) -> Result<Script, ScriptRepositoryError> {
-        let sandbox_json = serde_json::to_value(input.sandbox.unwrap_or_default())
-            .unwrap_or_else(|_| serde_json::json!({}));
         let mut tx = self.pool.begin().await?;
-        let res = sqlx::query_as::<_, ScriptRow>(&format!(
+        let script = insert_script_tx(&mut tx, &input).await?;
-            "INSERT INTO scripts ( \
-                app_id, name, description, source, kind, \
-                timeout_seconds, memory_limit_mb, sandbox \
-             ) VALUES ($1, $2, $3, $4, $5, COALESCE($6, 30), COALESCE($7, 256), $8) \
-             RETURNING {SCRIPT_SELECT_COLS}"
-        ))
-        .bind(input.app_id.into_inner())
-        .bind(&input.name)
-        .bind(input.description.as_deref())
-        .bind(&input.source)
-        .bind(input.kind.as_str())
-        .bind(input.timeout_seconds)
-        .bind(input.memory_limit_mb)
-        .bind(sandbox_json)
-        .fetch_one(&mut *tx)
-        .await;
-
-        let script: Script = match res {
-            Ok(row) => row.into(),
-            Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
-                return Err(ScriptRepositoryError::Conflict(format!(
-                    "a script named {:?} already exists in this app",
-                    input.name
-                )));
-            }
-            Err(e) => return Err(e.into()),
-        };
-
-        // Dep-graph: write any literal-path imports declared in the
-        // source. Unresolved names (the referenced module doesn't
-        // exist yet) are silently skipped — best-effort.
-        replace_imports_tx(&mut tx, script.id, script.app_id, &input.imports).await?;
         tx.commit().await?;
         Ok(script)
     }
@@ -318,62 +289,8 @@ impl ScriptRepository for PostgresScriptRepository {
         id: ScriptId,
         patch: ScriptPatch,
     ) -> Result<Script, ScriptRepositoryError> {
-        // COALESCE-based partial update: `NULL` parameters leave columns
-        // untouched. Description is double-Optioned so callers can
-        // explicitly set it to NULL (Some(None)) vs leave it alone (None).
-        // Sandbox is replaced wholesale when present; per-field merging
-        // happens in the API layer (clearer semantics for a "PUT a new
-        // sandbox config" call). app_id is immutable — moving a script
-        // to another app is a copy-and-delete, not an in-place edit.
-        let sandbox_json = patch
-            .sandbox
-            .as_ref()
-            .map(|s| serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({})));
         let mut tx = self.pool.begin().await?;
-        let res = sqlx::query_as::<_, ScriptRow>(&format!(
+        let script = update_script_tx(&mut tx, id, &patch).await?;
-            "UPDATE scripts SET \
-                name = COALESCE($2, name), \
-                description = CASE WHEN $3::bool THEN $4 ELSE description END, \
-                source = COALESCE($5, source), \
-                timeout_seconds = COALESCE($6, timeout_seconds), \
-                memory_limit_mb = COALESCE($7, memory_limit_mb), \
-                sandbox = COALESCE($8, sandbox), \
-                kind = COALESCE($9, kind), \
-                version = version + 1, \
-                updated_at = NOW() \
-             WHERE id = $1 \
-             RETURNING {SCRIPT_SELECT_COLS}"
-        ))
-        .bind(id.into_inner())
-        .bind(patch.name.as_deref())
-        .bind(patch.description.is_some())
-        .bind(patch.description.as_ref().and_then(|d| d.as_deref()))
-        .bind(patch.source.as_deref())
-        .bind(patch.timeout_seconds)
-        .bind(patch.memory_limit_mb)
-        .bind(sandbox_json)
-        .bind(patch.kind.map(ScriptKind::as_str))
-        .fetch_optional(&mut *tx)
-        .await;
-
-        let script: Script = match res {
-            Ok(Some(row)) => row.into(),
-            Ok(None) => return Err(ScriptRepositoryError::NotFound(id)),
-            Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
-                return Err(ScriptRepositoryError::Conflict(
-                    "a script with that name already exists in this app".into(),
-                ));
-            }
-            Err(e) => return Err(e.into()),
-        };
-
-        // Replace imports only when the caller has a fresh list (i.e.
-        // the source actually changed and the validator re-extracted
-        // imports). A name-only or description-only edit leaves the
-        // dep graph alone.
-        if let Some(imports) = patch.imports.as_deref() {
-            replace_imports_tx(&mut tx, script.id, script.app_id, imports).await?;
-        }
         tx.commit().await?;
         Ok(script)
     }
@@ -469,6 +386,117 @@ async fn replace_imports_tx(
     Ok(())
 }
 
+/// Insert a script within an existing transaction — the declarative
+/// `apply` engine composes scripts + routes + triggers into one tx.
+/// Mirrors `create` minus the `begin`/`commit`.
+pub(crate) async fn insert_script_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    input: &NewScript,
+) -> Result<Script, ScriptRepositoryError> {
+    let sandbox_json = serde_json::to_value(input.sandbox.unwrap_or_default())
+        .unwrap_or_else(|_| serde_json::json!({}));
+    let res = sqlx::query_as::<_, ScriptRow>(&format!(
+        "INSERT INTO scripts ( \
+            app_id, name, description, source, kind, \
+            timeout_seconds, memory_limit_mb, sandbox, enabled \
+         ) VALUES ($1, $2, $3, $4, $5, COALESCE($6, 30), COALESCE($7, 256), $8, $9) \
+         RETURNING {SCRIPT_SELECT_COLS}"
+    ))
+    .bind(input.app_id.into_inner())
+    .bind(&input.name)
+    .bind(input.description.as_deref())
+    .bind(&input.source)
+    .bind(input.kind.as_str())
+    .bind(input.timeout_seconds)
+    .bind(input.memory_limit_mb)
+    .bind(sandbox_json)
+    .bind(input.enabled)
+    .fetch_one(&mut **tx)
+    .await;
+    let script: Script = match res {
+        Ok(row) => row.into(),
+        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
+            return Err(ScriptRepositoryError::Conflict(format!(
+                "a script named {:?} already exists in this app",
+                input.name
+            )));
+        }
+        Err(e) => return Err(e.into()),
+    };
+    replace_imports_tx(tx, script.id, script.app_id, &input.imports).await?;
+    Ok(script)
+}
+
+/// Update a script within an existing transaction. Mirrors `update`
+/// minus the `begin`/`commit`.
+pub(crate) async fn update_script_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: ScriptId,
+    patch: &ScriptPatch,
+) -> Result<Script, ScriptRepositoryError> {
+    let sandbox_json = patch
+        .sandbox
+        .as_ref()
+        .map(|s| serde_json::to_value(s).unwrap_or_else(|_| serde_json::json!({})));
+    let res = sqlx::query_as::<_, ScriptRow>(&format!(
+        "UPDATE scripts SET \
+            name = COALESCE($2, name), \
+            description = CASE WHEN $3::bool THEN $4 ELSE description END, \
+            source = COALESCE($5, source), \
+            timeout_seconds = COALESCE($6, timeout_seconds), \
+            memory_limit_mb = COALESCE($7, memory_limit_mb), \
+            sandbox = COALESCE($8, sandbox), \
+            kind = COALESCE($9, kind), \
+            enabled = COALESCE($10, enabled), \
+            version = version + 1, \
+            updated_at = NOW() \
+         WHERE id = $1 \
+         RETURNING {SCRIPT_SELECT_COLS}"
+    ))
+    .bind(id.into_inner())
+    .bind(patch.name.as_deref())
+    .bind(patch.description.is_some())
+    .bind(patch.description.as_ref().and_then(|d| d.as_deref()))
+    .bind(patch.source.as_deref())
+    .bind(patch.timeout_seconds)
+    .bind(patch.memory_limit_mb)
+    .bind(sandbox_json)
+    .bind(patch.kind.map(ScriptKind::as_str))
+    .bind(patch.enabled)
+    .fetch_optional(&mut **tx)
+    .await;
+    let script: Script = match res {
+        Ok(Some(row)) => row.into(),
+        Ok(None) => return Err(ScriptRepositoryError::NotFound(id)),
+        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => {
+            return Err(ScriptRepositoryError::Conflict(
+                "a script with that name already exists in this app".into(),
+            ));
+        }
+        Err(e) => return Err(e.into()),
+    };
+    if let Some(imports) = patch.imports.as_deref() {
+        replace_imports_tx(tx, script.id, script.app_id, imports).await?;
+    }
+    Ok(script)
+}
+
+/// Delete a script within an existing transaction (its routes/triggers
+/// cascade via their FKs). Mirrors `delete` minus the pool.
+pub(crate) async fn delete_script_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: ScriptId,
+) -> Result<(), ScriptRepositoryError> {
+    let res = sqlx::query("DELETE FROM scripts WHERE id = $1")
+        .bind(id.into_inner())
+        .execute(&mut **tx)
+        .await?;
+    if res.rows_affected() == 0 {
+        return Err(ScriptRepositoryError::NotFound(id));
+    }
+    Ok(())
+}
+
 /// Row shape mirroring the `scripts` table for sqlx FromRow.
 #[derive(sqlx::FromRow)]
 struct ScriptRow {
@@ -485,6 +513,7 @@ struct ScriptRow {
     timeout_seconds: i32,
     memory_limit_mb: i32,
     sandbox: serde_json::Value,
+    enabled: bool,
     created_at: chrono::DateTime<chrono::Utc>,
     updated_at: chrono::DateTime<chrono::Utc>,
 }
@@ -511,6 +540,7 @@ impl From<ScriptRow> for Script {
             timeout_seconds: u32::try_from(r.timeout_seconds).unwrap_or(30),
             memory_limit_mb: u32::try_from(r.memory_limit_mb).unwrap_or(256),
             sandbox,
+            enabled: r.enabled,
             created_at: r.created_at,
             updated_at: r.updated_at,
         }

crates/manager-core/src/route_admin.rs

@@ -229,6 +229,8 @@ async fn create_route<RR: RouteRepository, SR: ScriptRepository>(
             path: normalized_path,
             method: input.method,
             dispatch_mode: input.dispatch_mode,
+            // Routes are created active; toggling is a dedicated path.
+            enabled: true,
         })
         .await?;
     refresh_table(&state).await?;
@@ -394,6 +396,9 @@ async fn refresh_table<RR: RouteRepository, SR: ScriptRepository>(
 #[must_use]
 pub fn compile_routes(rows: &[Route]) -> Vec<CompiledRoute> {
     rows.iter()
+        // A disabled route (§4.3) is dropped from the match table entirely, so
+        // a request to it 404s indistinguishably from an absent route.
+        .filter(|r| r.enabled)
         .filter_map(|r| match compile_route(r) {
             Ok(compiled) => Some(compiled),
             Err(e) => {
@@ -426,7 +431,7 @@ fn compile_route(r: &Route) -> Result<CompiledRoute, pattern::ParseError> {
 /// Validate that a new route's (host_kind, host) is consistent with at
 /// least one of the parent app's domain claims. `HostKind::Any` is
 /// always permitted — it catches every host the app already owns.
-async fn validate_route_host_against_app(
+pub(crate) async fn validate_route_host_against_app(
     domains: &dyn AppDomainRepository,
     app_id: AppId,
     host_kind: HostKind,
@@ -625,6 +630,7 @@ mod tests {
             path: path.to_string(),
             method: None,
             dispatch_mode: DispatchMode::default(),
+            enabled: true,
             created_at: chrono::Utc::now(),
         }
     }
@@ -651,4 +657,16 @@ mod tests {
             "a reserved-path route must be skipped, never abort the compile"
         );
     }
+
+    #[test]
+    fn disabled_route_is_dropped_from_compiled_table() {
+        // §4.3: a disabled route is excluded from the match table, so a
+        // request to it 404s indistinguishably from an absent route.
+        let active = route_with_path("/on");
+        let mut disabled = route_with_path("/off");
+        disabled.enabled = false;
+        let compiled = compile_routes(&[active.clone(), disabled.clone()]);
+        let ids: Vec<Uuid> = compiled.iter().map(|c| c.route_id).collect();
+        assert_eq!(ids, vec![active.id], "only the enabled route compiles");
+    }
 }

crates/manager-core/src/route_repo.rs

@@ -21,6 +21,8 @@ pub struct NewRoute {
     pub path: String,
     pub method: Option<String>,
     pub dispatch_mode: DispatchMode,
+    /// Three-state lifecycle (§4.3). Create active by default.
+    pub enabled: bool,
 }
 
 #[async_trait]
@@ -63,7 +65,7 @@ impl RouteRepository for PostgresRouteRepository {
     async fn list_all(&self) -> Result<Vec<Route>, ScriptRepositoryError> {
         let rows = sqlx::query_as::<_, RouteRow>(
             "SELECT id, app_id, script_id, host_kind, host, host_param_name, \
-                    path_kind, path, method, dispatch_mode, created_at \
+                    path_kind, path, method, dispatch_mode, enabled, created_at \
              FROM routes ORDER BY created_at",
         )
         .fetch_all(&self.pool)
@@ -74,7 +76,7 @@ impl RouteRepository for PostgresRouteRepository {
     async fn get(&self, route_id: Uuid) -> Result<Option<Route>, ScriptRepositoryError> {
         let row = sqlx::query_as::<_, RouteRow>(
             "SELECT id, app_id, script_id, host_kind, host, host_param_name, \
-                    path_kind, path, method, dispatch_mode, created_at \
+                    path_kind, path, method, dispatch_mode, enabled, created_at \
              FROM routes WHERE id = $1",
         )
         .bind(route_id)
@@ -86,7 +88,7 @@ impl RouteRepository for PostgresRouteRepository {
     async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Route>, ScriptRepositoryError> {
         let rows = sqlx::query_as::<_, RouteRow>(
             "SELECT id, app_id, script_id, host_kind, host, host_param_name, \
-                    path_kind, path, method, dispatch_mode, created_at \
+                    path_kind, path, method, dispatch_mode, enabled, created_at \
              FROM routes WHERE app_id = $1 ORDER BY created_at",
         )
         .bind(app_id.into_inner())
@@ -101,7 +103,7 @@ impl RouteRepository for PostgresRouteRepository {
     ) -> Result<Vec<Route>, ScriptRepositoryError> {
         let rows = sqlx::query_as::<_, RouteRow>(
             "SELECT id, app_id, script_id, host_kind, host, host_param_name, \
-                    path_kind, path, method, dispatch_mode, created_at \
+                    path_kind, path, method, dispatch_mode, enabled, created_at \
              FROM routes WHERE script_id = $1 ORDER BY created_at",
         )
         .bind(script_id.into_inner())
@@ -111,36 +113,10 @@ impl RouteRepository for PostgresRouteRepository {
     }
 
     async fn create(&self, input: NewRoute) -> Result<Route, ScriptRepositoryError> {
-        let res = sqlx::query_as::<_, RouteRow>(
+        let mut tx = self.pool.begin().await?;
-            "INSERT INTO routes ( \
+        let route = insert_route_tx(&mut tx, &input).await?;
-                app_id, script_id, host_kind, host, host_param_name, \
+        tx.commit().await?;
-                path_kind, path, method, dispatch_mode \
+        Ok(route)
-             ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) \
-             RETURNING id, app_id, script_id, host_kind, host, host_param_name, \
-                       path_kind, path, method, dispatch_mode, created_at",
-        )
-        .bind(input.app_id.into_inner())
-        .bind(input.script_id.into_inner())
-        .bind(host_kind_str(input.host_kind))
-        .bind(&input.host)
-        .bind(input.host_param_name.as_deref())
-        .bind(path_kind_str(input.path_kind))
-        .bind(&input.path)
-        .bind(input.method.as_deref())
-        .bind(input.dispatch_mode.as_str())
-        .fetch_one(&self.pool)
-        .await;
-
-        match res {
-            Ok(row) => Ok(row.into()),
-            Err(sqlx::Error::Database(e)) if e.is_unique_violation() => Err(
-                ScriptRepositoryError::Conflict("a route with this binding already exists".into()),
-            ),
-            Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => {
-                Err(ScriptRepositoryError::NotFound(input.script_id))
-            }
-            Err(e) => Err(e.into()),
-        }
     }
 
     async fn delete(&self, route_id: Uuid) -> Result<(), ScriptRepositoryError> {
@@ -189,6 +165,63 @@ const fn path_kind_str(k: PathKind) -> &'static str {
     }
 }
 
+/// Insert a route within an existing transaction (declarative apply
+/// composes scripts + routes + triggers into one tx). Mirrors `create`
+/// minus the `begin`/`commit`.
+pub(crate) async fn insert_route_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    input: &NewRoute,
+) -> Result<Route, ScriptRepositoryError> {
+    let res = sqlx::query_as::<_, RouteRow>(
+        "INSERT INTO routes ( \
+            app_id, script_id, host_kind, host, host_param_name, \
+            path_kind, path, method, dispatch_mode, enabled \
+         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10) \
+         RETURNING id, app_id, script_id, host_kind, host, host_param_name, \
+                   path_kind, path, method, dispatch_mode, enabled, created_at",
+    )
+    .bind(input.app_id.into_inner())
+    .bind(input.script_id.into_inner())
+    .bind(host_kind_str(input.host_kind))
+    .bind(&input.host)
+    .bind(input.host_param_name.as_deref())
+    .bind(path_kind_str(input.path_kind))
+    .bind(&input.path)
+    .bind(input.method.as_deref())
+    .bind(input.dispatch_mode.as_str())
+    .bind(input.enabled)
+    .fetch_one(&mut **tx)
+    .await;
+    match res {
+        Ok(row) => Ok(row.into()),
+        Err(sqlx::Error::Database(e)) if e.is_unique_violation() => Err(
+            ScriptRepositoryError::Conflict("a route with this binding already exists".into()),
+        ),
+        Err(sqlx::Error::Database(e)) if e.is_foreign_key_violation() => {
+            Err(ScriptRepositoryError::NotFound(input.script_id))
+        }
+        Err(e) => Err(e.into()),
+    }
+}
+
+/// Delete a route by id within an existing transaction.
+///
+/// Unlike the non-tx [`RouteRepository::delete`], this is intentionally
+/// idempotent: a missing row is not an error. The only caller is the
+/// reconcile engine (`ApplyService`), where "delete a route already gone"
+/// (e.g. removed out-of-band between the diff read and the write) is a
+/// no-op to converge on, not a failure to roll back the whole apply.
+pub(crate) async fn delete_route_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    route_id: Uuid,
+) -> Result<(), ScriptRepositoryError> {
+    sqlx::query("DELETE FROM routes WHERE id = $1")
+        .bind(route_id)
+        .execute(&mut **tx)
+        .await?;
+    Ok(())
+}
+
 #[derive(sqlx::FromRow)]
 struct RouteRow {
     id: Uuid,
@@ -201,6 +234,7 @@ struct RouteRow {
     path: String,
     method: Option<String>,
     dispatch_mode: String,
+    enabled: bool,
     created_at: chrono::DateTime<chrono::Utc>,
 }
 
@@ -225,6 +259,7 @@ impl From<RouteRow> for Route {
             path: r.path,
             method: r.method,
             dispatch_mode: DispatchMode::from_wire(&r.dispatch_mode).unwrap_or(DispatchMode::Sync),
+            enabled: r.enabled,
             created_at: r.created_at,
         }
     }

crates/manager-core/src/trigger_repo.rs

@@ -34,6 +34,8 @@ pub struct Trigger {
     pub id: TriggerId,
     pub app_id: AppId,
     pub script_id: ScriptId,
+    /// §4.5 per-app trigger identifier; the manifest merge/upsert key.
+    pub name: String,
     pub kind: TriggerKind,
     pub enabled: bool,
     pub dispatch_mode: TriggerDispatchMode,
@@ -502,6 +504,220 @@ impl PostgresTriggerRepo {
     }
 }
 
+/// Insert a trigger (parent row + per-kind detail) within an existing
+/// transaction — used by the declarative `apply` engine. Supports the
+/// five settled kinds; `email`/`queue`/`dead_letter` have their own
+/// create paths and are rejected here.
+#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
+pub(crate) async fn insert_trigger_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    app_id: AppId,
+    script_id: ScriptId,
+    registered_by: AdminUserId,
+    dispatch_mode: TriggerDispatchMode,
+    retry_max_attempts: u32,
+    retry_backoff: BackoffShape,
+    retry_base_ms: u32,
+    details: &TriggerDetails,
+) -> Result<TriggerId, TriggerRepoError> {
+    let kind = match details {
+        TriggerDetails::Kv { .. } => "kv",
+        TriggerDetails::Docs { .. } => "docs",
+        TriggerDetails::Files { .. } => "files",
+        TriggerDetails::Cron { .. } => "cron",
+        TriggerDetails::Pubsub { .. } => "pubsub",
+        TriggerDetails::Queue { .. } => "queue",
+        TriggerDetails::DeadLetter { .. } | TriggerDetails::Email { .. } => {
+            return Err(TriggerRepoError::Invalid(
+                "trigger kind not supported by declarative apply".into(),
+            ));
+        }
+    };
+    // Queue: enforce the one-consumer-per-(app_id, queue_name) invariant —
+    // the same advisory-lock + existence guard the interactive
+    // `create_queue_trigger` uses. Without this, a concurrent apply +
+    // interactive create on disjoint locks could double-register a queue
+    // consumer (there is no DB unique constraint backing the invariant).
+    if let TriggerDetails::Queue { queue_name, .. } = details {
+        sqlx::query("SELECT pg_advisory_xact_lock($1)")
+            .bind(advisory_lock_key(app_id, queue_name))
+            .execute(&mut **tx)
+            .await?;
+        let existing: Option<(Uuid,)> = sqlx::query_as(
+            "SELECT t.id FROM triggers t \
+             JOIN queue_trigger_details d ON d.trigger_id = t.id \
+             WHERE t.app_id = $1 AND t.kind = 'queue' AND d.queue_name = $2",
+        )
+        .bind(app_id.into_inner())
+        .bind(queue_name)
+        .fetch_optional(&mut **tx)
+        .await?;
+        if existing.is_some() {
+            return Err(TriggerRepoError::Invalid(format!(
+                "queue '{queue_name}' already has a consumer trigger; remove the existing one first"
+            )));
+        }
+    }
+    let row: (Uuid,) = sqlx::query_as(
+        "INSERT INTO triggers ( \
+            app_id, script_id, kind, enabled, dispatch_mode, \
+            retry_max_attempts, retry_backoff, retry_base_ms, \
+            registered_by_principal \
+         ) VALUES ($1, $2, $3, TRUE, $4, $5, $6, $7, $8) RETURNING id",
+    )
+    .bind(app_id.into_inner())
+    .bind(script_id.into_inner())
+    .bind(kind)
+    .bind(dispatch_mode.as_str())
+    .bind(i32::try_from(retry_max_attempts).unwrap_or(3))
+    .bind(retry_backoff.as_str())
+    .bind(i32::try_from(retry_base_ms).unwrap_or(1000))
+    .bind(registered_by.into_inner())
+    .fetch_one(&mut **tx)
+    .await?;
+    let tid = row.0;
+
+    match details {
+        TriggerDetails::Kv {
+            collection_glob,
+            ops,
+        } => {
+            let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
+            sqlx::query(
+                "INSERT INTO kv_trigger_details (trigger_id, collection_glob, ops) \
+                 VALUES ($1, $2, $3)",
+            )
+            .bind(tid)
+            .bind(collection_glob)
+            .bind(&ops_str)
+            .execute(&mut **tx)
+            .await?;
+        }
+        TriggerDetails::Docs {
+            collection_glob,
+            ops,
+        } => {
+            let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
+            sqlx::query(
+                "INSERT INTO docs_trigger_details (trigger_id, collection_glob, ops) \
+                 VALUES ($1, $2, $3)",
+            )
+            .bind(tid)
+            .bind(collection_glob)
+            .bind(&ops_str)
+            .execute(&mut **tx)
+            .await?;
+        }
+        TriggerDetails::Files {
+            collection_glob,
+            ops,
+        } => {
+            let ops_str: Vec<String> = ops.iter().map(|o| o.as_str().to_string()).collect();
+            sqlx::query(
+                "INSERT INTO files_trigger_details (trigger_id, collection_glob, ops) \
+                 VALUES ($1, $2, $3)",
+            )
+            .bind(tid)
+            .bind(collection_glob)
+            .bind(&ops_str)
+            .execute(&mut **tx)
+            .await?;
+        }
+        TriggerDetails::Cron {
+            schedule, timezone, ..
+        } => {
+            sqlx::query(
+                "INSERT INTO cron_trigger_details (trigger_id, schedule, timezone) \
+                 VALUES ($1, $2, $3)",
+            )
+            .bind(tid)
+            .bind(schedule)
+            .bind(timezone)
+            .execute(&mut **tx)
+            .await?;
+        }
+        TriggerDetails::Pubsub { topic_pattern } => {
+            sqlx::query(
+                "INSERT INTO pubsub_trigger_details (trigger_id, topic_pattern) VALUES ($1, $2)",
+            )
+            .bind(tid)
+            .bind(topic_pattern)
+            .execute(&mut **tx)
+            .await?;
+        }
+        TriggerDetails::Queue {
+            queue_name,
+            visibility_timeout_secs,
+            ..
+        } => {
+            sqlx::query(
+                "INSERT INTO queue_trigger_details \
+                    (trigger_id, queue_name, visibility_timeout_secs) \
+                 VALUES ($1, $2, $3)",
+            )
+            .bind(tid)
+            .bind(queue_name)
+            .bind(i32::try_from(*visibility_timeout_secs).unwrap_or(30))
+            .execute(&mut **tx)
+            .await?;
+        }
+        TriggerDetails::DeadLetter { .. } | TriggerDetails::Email { .. } => {
+            unreachable!("guarded above")
+        }
+    }
+    Ok(tid.into())
+}
+
+/// Insert an email trigger within a transaction. The inbound HMAC secret
+/// is sealed by the apply engine (resolved from the app's secret store);
+/// this writes the ciphertext. Parent retry settings match the
+/// interactive `create_email_trigger` path (async, 3, exponential, 1000).
+pub(crate) async fn insert_email_trigger_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    app_id: AppId,
+    script_id: ScriptId,
+    registered_by: AdminUserId,
+    inbound_secret_encrypted: &[u8],
+    inbound_secret_nonce: &[u8],
+) -> Result<TriggerId, TriggerRepoError> {
+    let row: (Uuid,) = sqlx::query_as(
+        "INSERT INTO triggers ( \
+            app_id, script_id, kind, enabled, dispatch_mode, \
+            retry_max_attempts, retry_backoff, retry_base_ms, \
+            registered_by_principal \
+         ) VALUES ($1, $2, 'email', TRUE, 'async', 3, 'exponential', 1000, $3) RETURNING id",
+    )
+    .bind(app_id.into_inner())
+    .bind(script_id.into_inner())
+    .bind(registered_by.into_inner())
+    .fetch_one(&mut **tx)
+    .await?;
+    sqlx::query(
+        "INSERT INTO email_trigger_details \
+            (trigger_id, inbound_secret_encrypted, inbound_secret_nonce) \
+         VALUES ($1, $2, $3)",
+    )
+    .bind(row.0)
+    .bind(inbound_secret_encrypted)
+    .bind(inbound_secret_nonce)
+    .execute(&mut **tx)
+    .await?;
+    Ok(row.0.into())
+}
+
+/// Delete a trigger by id within an existing transaction (its detail row
+/// cascades via the FK). Used by `apply --prune`.
+pub(crate) async fn delete_trigger_tx(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    id: TriggerId,
+) -> Result<(), TriggerRepoError> {
+    sqlx::query("DELETE FROM triggers WHERE id = $1")
+        .bind(id.into_inner())
+        .execute(&mut **tx)
+        .await?;
+    Ok(())
+}
+
 #[async_trait]
 impl TriggerRepo for PostgresTriggerRepo {
     async fn create_kv_trigger(
@@ -521,7 +737,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'kv', TRUE, $3, $4, $5, $6, $7) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -552,6 +768,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::Kv,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -586,7 +803,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'docs', TRUE, $3, $4, $5, $6, $7) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -617,6 +834,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::Docs,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -649,7 +867,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'dead_letter', TRUE, 'async', 1, 'constant', 0, $3) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -677,6 +895,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::DeadLetter,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -713,7 +932,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'cron', TRUE, $3, $4, $5, $6, $7) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -743,6 +962,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::Cron,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -778,7 +998,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'files', TRUE, $3, $4, $5, $6, $7) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -809,6 +1029,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::Files,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -842,7 +1063,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'pubsub', TRUE, $3, $4, $5, $6, $7) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -870,6 +1091,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::Pubsub,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -902,7 +1124,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'email', TRUE, 'async', 3, 'exponential', 1000, $3) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -929,6 +1151,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::Email,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -971,7 +1194,7 @@ impl TriggerRepo for PostgresTriggerRepo {
 
     async fn list_for_app(&self, app_id: AppId) -> Result<Vec<Trigger>, TriggerRepoError> {
         let parents: Vec<TriggerRow> = sqlx::query_as(
-            "SELECT id, app_id, script_id, kind, enabled, dispatch_mode, \
+            "SELECT id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                     retry_max_attempts, retry_backoff, retry_base_ms, \
                     registered_by_principal, created_at, updated_at \
              FROM triggers WHERE app_id = $1 ORDER BY created_at DESC",
@@ -999,7 +1222,7 @@ impl TriggerRepo for PostgresTriggerRepo {
 
     async fn get(&self, id: TriggerId) -> Result<Option<Trigger>, TriggerRepoError> {
         let parent: Option<TriggerRow> = sqlx::query_as(
-            "SELECT id, app_id, script_id, kind, enabled, dispatch_mode, \
+            "SELECT id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                     retry_max_attempts, retry_backoff, retry_base_ms, \
                     registered_by_principal, created_at, updated_at \
              FROM triggers WHERE id = $1",
@@ -1249,7 +1472,7 @@ impl TriggerRepo for PostgresTriggerRepo {
                 retry_max_attempts, retry_backoff, retry_base_ms, \
                 registered_by_principal \
              ) VALUES ($1, $2, 'queue', TRUE, $3, $4, $5, $6, $7) \
-             RETURNING id, app_id, script_id, kind, enabled, dispatch_mode, \
+             RETURNING id, app_id, script_id, name, kind, enabled, dispatch_mode, \
                        retry_max_attempts, retry_backoff, retry_base_ms, \
                        registered_by_principal, created_at, updated_at",
         )
@@ -1280,6 +1503,7 @@ impl TriggerRepo for PostgresTriggerRepo {
             id: parent.id.into(),
             app_id: parent.app_id.into(),
             script_id: parent.script_id.into(),
+            name: parent.name.clone(),
             kind: TriggerKind::Queue,
             enabled: parent.enabled,
             dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -1308,7 +1532,8 @@ impl TriggerRepo for PostgresTriggerRepo {
                     t.registered_by_principal \
              FROM triggers t \
              JOIN queue_trigger_details d ON d.trigger_id = t.id \
-             WHERE t.kind = 'queue' AND t.enabled = TRUE",
+             JOIN scripts s ON s.id = t.script_id \
+             WHERE t.kind = 'queue' AND t.enabled = TRUE AND s.enabled = TRUE",
         )
         .fetch_all(&self.pool)
         .await?;
@@ -1486,6 +1711,7 @@ async fn hydrate_one(pool: &PgPool, parent: TriggerRow) -> Result<Trigger, Trigg
         id: parent.id.into(),
         app_id: parent.app_id.into(),
         script_id: parent.script_id.into(),
+        name: parent.name,
         kind,
         enabled: parent.enabled,
         dispatch_mode: dispatch_from_str(&parent.dispatch_mode),
@@ -1528,6 +1754,7 @@ struct TriggerRow {
     id: Uuid,
     app_id: Uuid,
     script_id: Uuid,
+    name: String,
     kind: String,
     enabled: bool,
     dispatch_mode: String,

crates/manager-core/src/triggers_api.rs

@@ -38,7 +38,7 @@ use crate::trigger_repo::{
 /// in practice (the dispatcher itself ticks every 100ms; reclaim ticks
 /// every 30s; an executor needs more wall-clock than this to do useful
 /// work). Reject below this with a 422 + actionable error.
-const MIN_QUEUE_VISIBILITY_TIMEOUT_SECS: u32 = 30;
+pub(crate) const MIN_QUEUE_VISIBILITY_TIMEOUT_SECS: u32 = 30;
 
 /// Default soft-warning ceiling when `PICLOUD_DISPATCHER_ASYNC_EXEC_TIMEOUT_SEC`
 /// is unset. Re-exported from the dispatcher so the single source of
@@ -892,6 +892,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: crate::trigger_repo::TriggerKind::Kv,
                 enabled: true,
                 dispatch_mode: req.dispatch_mode,
@@ -920,6 +921,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: crate::trigger_repo::TriggerKind::Docs,
                 enabled: true,
                 dispatch_mode: req.dispatch_mode,
@@ -948,6 +950,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: crate::trigger_repo::TriggerKind::DeadLetter,
                 enabled: true,
                 dispatch_mode: TriggerDispatchMode::Async,
@@ -977,6 +980,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: TriggerKind::Email,
                 enabled: true,
                 dispatch_mode: TriggerDispatchMode::Async,
@@ -1021,6 +1025,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: crate::trigger_repo::TriggerKind::Cron,
                 enabled: true,
                 dispatch_mode: req.dispatch_mode,
@@ -1050,6 +1055,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: crate::trigger_repo::TriggerKind::Files,
                 enabled: true,
                 dispatch_mode: req.dispatch_mode,
@@ -1078,6 +1084,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: crate::trigger_repo::TriggerKind::Pubsub,
                 enabled: true,
                 dispatch_mode: req.dispatch_mode,
@@ -1154,6 +1161,7 @@ mod tests {
                 id,
                 app_id,
                 script_id: req.script_id,
+                name: "mock".into(),
                 kind: TriggerKind::Queue,
                 enabled: true,
                 dispatch_mode: req.dispatch_mode,
@@ -1347,6 +1355,7 @@ mod tests {
                     timeout_seconds: 30,
                     sandbox: picloud_shared::ScriptSandbox::default(),
                     memory_limit_mb: 256,
+                    enabled: true,
                     created_at: now,
                     updated_at: now,
                 },

crates/orchestrator-core/src/api.rs

@@ -121,6 +121,11 @@ where
         .resolve(id)
         .await?
         .ok_or(ApiError::NotFound(id))?;
+    // A disabled script (§4.3) is not invocable — 404, indistinguishable
+    // from an absent one (no info leak that the id exists but is off).
+    if !script.enabled {
+        return Err(ApiError::NotFound(id));
+    }
 
     let mut req = build_exec_request(id, &script.name, &headers, &body, script.app_id, principal)?;
     req.sandbox_overrides = script.sandbox;
@@ -164,6 +169,7 @@ where
     Ok(exec_response_to_http(outcome?))
 }
 
+#[allow(clippy::too_many_lines)]
 async fn user_route_handler<E, R>(
     State(state): State<DataPlaneState<E, R>>,
     Extension(principal): Extension<Option<Principal>>,
@@ -221,6 +227,19 @@ where
         .resolve(matched.matched.script_id)
         .await?
         .ok_or(ApiError::NotFound(matched.matched.script_id))?;
+    // An enabled route bound to a disabled script (§4.3, §4.7) is unreachable.
+    // Return the SAME flat "no route matches" 404 as the unmatched case — not
+    // `NotFound(script_id)`, which would both leak the internal script id to an
+    // anonymous caller and be distinguishable from absent.
+    if !script.enabled {
+        return Ok((
+            StatusCode::NOT_FOUND,
+            Json(serde_json::json!({
+                "error": format!("no route matches {method} {path}")
+            })),
+        )
+            .into_response());
+    }
 
     // Drain the body now that we know we'll execute. 10 MiB cap matches
     // the conservative default response/request size in the blueprint.

crates/picloud-cli/src/client.rs

@@ -900,6 +900,54 @@ impl Client {
             .await?;
         decode(resp).await
     }
+
+    /// `POST /api/v1/admin/apps/{id_or_slug}/plan` — diff a desired-state
+    /// bundle against the app's live state. Read-only.
+    pub async fn plan(&self, app: &str, bundle: &serde_json::Value) -> Result<PlanDto> {
+        let app = seg(app);
+        let resp = self
+            .request(Method::POST, &format!("/api/v1/admin/apps/{app}/plan"))
+            .json(bundle)
+            .send()
+            .await?;
+        decode(resp).await
+    }
+
+    /// `POST /api/v1/admin/apps/{id_or_slug}/apply` — reconcile the live
+    /// app to the bundle in one transaction.
+    pub async fn apply(
+        &self,
+        app: &str,
+        bundle: &serde_json::Value,
+        prune: bool,
+        expected_token: Option<&str>,
+    ) -> Result<ApplyReportDto> {
+        let app = seg(app);
+        let body = serde_json::json!({
+            "bundle": bundle,
+            "prune": prune,
+            "expected_token": expected_token,
+        });
+        let resp = self
+            .request(Method::POST, &format!("/api/v1/admin/apps/{app}/apply"))
+            .json(&body)
+            .send()
+            .await?;
+        // The apply endpoint returns 409 only for a stale bound plan
+        // (`StateMoved`): the app changed since `pic plan` recorded its
+        // token. Surface an actionable next step instead of a bare
+        // `HTTP 409`.
+        if resp.status() == reqwest::StatusCode::CONFLICT {
+            let body = resp.text().await.unwrap_or_default();
+            let msg = parse_error_body(&body).unwrap_or(body);
+            return Err(anyhow!(
+                "{msg}\nThe app changed since your last `pic plan`. Re-run \
+                 `pic plan` to review the new diff, then `pic apply` — or \
+                 `pic apply --force` to apply without re-reviewing."
+            ));
+        }
+        decode(resp).await
+    }
 }
 
 /// `POST /api/v1/admin/auth/login` — sits outside the `Client` because
@@ -924,6 +972,54 @@ pub async fn auth_login(url: &str, username: &str, password: &str) -> Result<Log
 
 // ---------- DTOs (CLI-local, wire-shape-matched) ----------
 
+/// Response of `POST .../plan`: per-resource diffs grouped by kind.
+#[derive(Debug, Deserialize)]
+pub struct PlanDto {
+    #[serde(default)]
+    pub scripts: Vec<ChangeDto>,
+    #[serde(default)]
+    pub routes: Vec<ChangeDto>,
+    #[serde(default)]
+    pub triggers: Vec<ChangeDto>,
+    #[serde(default)]
+    pub secrets: Vec<ChangeDto>,
+    /// Fingerprint of the live state this plan was computed against; carried
+    /// in `.picloud/` and replayed to `apply` for the bound-plan check.
+    #[serde(default)]
+    pub state_token: String,
+}
+
+#[derive(Debug, Deserialize)]
+pub struct ChangeDto {
+    pub op: String,
+    pub key: String,
+    #[serde(default)]
+    pub detail: Option<String>,
+}
+
+/// Response of `POST .../apply`: counts of what changed.
+#[derive(Debug, Default, Deserialize)]
+pub struct ApplyReportDto {
+    #[serde(default)]
+    pub scripts_created: u32,
+    #[serde(default)]
+    pub scripts_updated: u32,
+    #[serde(default)]
+    pub scripts_deleted: u32,
+    #[serde(default)]
+    pub routes_created: u32,
+    #[serde(default)]
+    pub routes_updated: u32,
+    #[serde(default)]
+    pub routes_deleted: u32,
+    #[serde(default)]
+    pub triggers_created: u32,
+    #[serde(default)]
+    pub triggers_deleted: u32,
+    #[serde(default)]
+    pub warnings: Vec<String>,
+}
+
 #[allow(dead_code)]
 #[derive(Debug, Deserialize)]
 pub struct AuthMeDto {

crates/picloud-cli/src/cmds/apply.rs

@@ -0,0 +1,111 @@
+//! `pic apply [--file picloud.toml]` — reconcile the live app to the
+//! manifest's desired state in one server-side transaction. Creates and
+//! updates are applied; `--prune` additionally deletes live scripts/routes/
+//! triggers absent from the manifest (secrets are never pruned).
+
+use std::io::{IsTerminal, Write};
+use std::path::Path;
+
+use anyhow::{Context, Result};
+
+use crate::client::Client;
+use crate::cmds::plan::build_bundle;
+use crate::config;
+use crate::manifest::Manifest;
+use crate::output::{KvBlock, OutputMode};
+
+pub async fn run(
+    manifest_path: &Path,
+    env: Option<&str>,
+    prune: bool,
+    yes: bool,
+    force: bool,
+    mode: OutputMode,
+) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+
+    let manifest = Manifest::load_with_env(manifest_path, env)?;
+    let base_dir = manifest_path.parent().unwrap_or_else(|| Path::new("."));
+    let bundle = build_bundle(&manifest, base_dir)?;
+
+    if prune && !yes {
+        confirm_prune(&manifest.app.slug)?;
+    }
+
+    // Bound-plan check: replay the token from the last `pic plan` (for this
+    // app) so the server refuses if the app changed since it was reviewed.
+    // `--force` skips it; no recorded plan means no check (apply still works
+    // standalone). The token is single-use — cleared after a successful apply.
+    let expected_token = if force {
+        None
+    } else {
+        crate::linkstate::read_plan(base_dir)
+            .filter(|l| l.app == manifest.app.slug)
+            .map(|l| l.state_token)
+    };
+
+    let report = client
+        .apply(
+            &manifest.app.slug,
+            &bundle,
+            prune,
+            expected_token.as_deref(),
+        )
+        .await?;
+    crate::linkstate::clear_plan(base_dir, &manifest.app.slug);
+
+    let mut block = KvBlock::new();
+    block
+        .field("app", manifest.app.slug.clone())
+        .field(
+            "scripts",
+            format!(
+                "+{} ~{} -{}",
+                report.scripts_created, report.scripts_updated, report.scripts_deleted
+            ),
+        )
+        .field(
+            "routes",
+            format!(
+                "+{} ~{} -{}",
+                report.routes_created, report.routes_updated, report.routes_deleted
+            ),
+        )
+        .field(
+            "triggers",
+            format!("+{} -{}", report.triggers_created, report.triggers_deleted),
+        );
+    for w in &report.warnings {
+        block.field("warning", w.clone());
+    }
+    block.print(mode);
+    Ok(())
+}
+
+/// `--prune` deletes live scripts/routes/triggers absent from the manifest —
+/// irreversible. Require an explicit go-ahead: an interactive `y`, or `--yes`
+/// for non-interactive/CI use. Refuse a non-interactive prune without `--yes`
+/// rather than silently deleting (review the deletions first with `pic plan`).
+fn confirm_prune(slug: &str) -> Result<()> {
+    if !std::io::stdin().is_terminal() {
+        anyhow::bail!(
+            "refusing to `apply --prune` non-interactively without `--yes`: prune \
+             deletes resources absent from the manifest and cannot be undone. \
+             Review with `pic plan`, then re-run with `--yes`."
+        );
+    }
+    eprint!(
+        "apply --prune will DELETE live scripts/routes/triggers on `{slug}` that are \
+         absent from the manifest. This cannot be undone. Continue? [y/N] "
+    );
+    std::io::stderr().flush().ok();
+    let mut answer = String::new();
+    std::io::stdin()
+        .read_line(&mut answer)
+        .context("read confirmation")?;
+    if !matches!(answer.trim(), "y" | "Y" | "yes" | "Yes") {
+        anyhow::bail!("aborted");
+    }
+    Ok(())
+}

crates/picloud-cli/src/cmds/config.rs

@@ -0,0 +1,58 @@
+//! `pic config --effective` — read-only view of an app's resolved
+//! configuration, with secret values masked (§4.6).
+//!
+//! Single-app today: "config" means the app's secrets, cross-referenced
+//! against the manifest so an operator can see at a glance which declared
+//! secrets are still unset and which live secrets aren't declared. Values are
+//! never fetched or shown — only `<set>` / `<unset>`. The command is shaped to
+//! grow an `--explain` mode and inherited `vars` once groups/vars land (the
+//! Phase-3 multi-level resolution).
+
+use std::collections::BTreeSet;
+use std::path::Path;
+
+use anyhow::{bail, Result};
+
+use crate::client::Client;
+use crate::config;
+use crate::manifest::Manifest;
+use crate::output::{OutputMode, Table};
+
+pub async fn run(
+    manifest_path: &Path,
+    effective: bool,
+    env: Option<&str>,
+    mode: OutputMode,
+) -> Result<()> {
+    if !effective {
+        bail!("`pic config` currently supports only `--effective`");
+    }
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+    // Resolve against the same env overlay as `pic plan`/`apply` so the
+    // masked report targets the app those commands would act on — not the
+    // base slug — when an overlay re-points slug/secrets.
+    let manifest = Manifest::load_with_env(manifest_path, env)?;
+
+    let declared: BTreeSet<String> = manifest.secrets.names.iter().cloned().collect();
+    let on_server: BTreeSet<String> = client
+        .secrets_list(&manifest.app.slug)
+        .await?
+        .secrets
+        .into_iter()
+        .map(|s| s.name)
+        .collect();
+
+    let mut table = Table::new(["secret", "value", "status"]);
+    for name in declared.union(&on_server) {
+        let (value, status) = match (declared.contains(name), on_server.contains(name)) {
+            (true, true) => ("<set>", "managed"),
+            (true, false) => ("<unset>", "declared, not pushed — `pic secret set`"),
+            (false, true) => ("<set>", "on server, not in manifest"),
+            (false, false) => unreachable!("name came from one of the two sets"),
+        };
+        table.row([name.clone(), value.to_string(), status.to_string()]);
+    }
+    table.print(mode);
+    Ok(())
+}

crates/picloud-cli/src/cmds/init.rs

@@ -0,0 +1,278 @@
+//! `pic init [slug] [--dir .]` — scaffold a new declarative project: a
+//! `picloud.toml` describing a minimal working app (one `hello` endpoint +
+//! route), its `scripts/hello.rhai` source, and a `.gitignore` that ignores
+//! the project tool's `.picloud/` link-state directory.
+//!
+//! Offline by design — it never contacts the server. Run `pic plan` to
+//! preview the create, then `pic apply` to deploy. Refuses to overwrite an
+//! existing `picloud.toml` unless `--force`.
+
+use std::fs;
+use std::path::Path;
+
+use anyhow::{bail, Context, Result};
+use picloud_shared::{DispatchMode, HostKind, PathKind, ScriptKind};
+
+use crate::manifest::{Manifest, ManifestApp, ManifestRoute, ManifestScript, MANIFEST_FILE};
+use crate::output::{KvBlock, OutputMode};
+
+const HEADER: &str = "\
+# picloud project manifest — the declarative desired state for one app.
+# Edit, then `pic plan` to preview changes and `pic apply` to reconcile.
+# Scripts live under scripts/ and are referenced by `file`.
+\n";
+
+const EXAMPLES: &str = "\
+\n# ---------------------------------------------------------------------------
+# More to add (uncomment and adapt):
+#
+# [[scripts]]
+# name = \"lib\"
+# file = \"scripts/lib.rhai\"
+# kind = \"module\"            # default: endpoint
+#
+# [[routes]]
+# script = \"hello\"
+# method = \"POST\"            # omit for ANY
+# host_kind = \"any\"
+# path_kind = \"param\"        # exact | prefix | param
+# path = \"/hello/:name\"
+#
+# [[triggers.cron]]
+# script = \"hello\"
+# schedule = \"0 0 * * * *\"   # 6-field cron (seconds first)
+# timezone = \"UTC\"
+#
+# [secrets]                   # names only — push values with `pic secret set`
+# names = [\"STRIPE_KEY\"]
+";
+
+const HELLO_RHAI: &str = "\
+// A minimal endpoint script. Its return value is the HTTP response body.
+// `ctx` exposes the request; see the stdlib reference for the full SDK.
+\"Hello from PiCloud!\"
+";
+
+const GITIGNORE_LINE: &str = ".picloud/";
+
+pub fn run(
+    dir: &Path,
+    slug_arg: Option<&str>,
+    name_arg: Option<&str>,
+    force: bool,
+    mode: OutputMode,
+) -> Result<()> {
+    let slug = resolve_slug(dir, slug_arg)?;
+    let name = name_arg.map_or_else(|| title_from_slug(&slug), str::to_string);
+
+    let manifest_path = dir.join(MANIFEST_FILE);
+    if manifest_path.exists() && !force {
+        bail!(
+            "{} already exists; refusing to overwrite (use --force)",
+            manifest_path.display()
+        );
+    }
+
+    let manifest = scaffold_manifest(&slug, &name);
+    // Build the file as a commented header + the (valid, round-trippable)
+    // active manifest + commented examples. Rendering the active part through
+    // `to_toml` guarantees it parses and matches the wire model.
+    let body = format!("{HEADER}{}{EXAMPLES}", manifest.to_toml()?);
+
+    fs::create_dir_all(dir.join("scripts")).context("creating scripts/ directory")?;
+    let hello_path = dir.join("scripts/hello.rhai");
+    let wrote_hello = !hello_path.exists();
+    if wrote_hello {
+        fs::write(&hello_path, HELLO_RHAI).context("writing scripts/hello.rhai")?;
+    }
+    fs::write(&manifest_path, body).with_context(|| format!("writing {MANIFEST_FILE}"))?;
+    ensure_gitignored(dir)?;
+
+    let mut block = KvBlock::new();
+    block
+        .field("manifest", manifest_path.display().to_string())
+        .field("app", slug)
+        .field(
+            "scripts",
+            if wrote_hello {
+                "scripts/hello.rhai"
+            } else {
+                "scripts/hello.rhai (kept existing)"
+            }
+            .to_string(),
+        )
+        .field("next", "pic plan   then   pic apply".to_string());
+    block.print(mode);
+    Ok(())
+}
+
+/// The minimal working project: one `hello` endpoint bound to `GET /hello`.
+fn scaffold_manifest(slug: &str, name: &str) -> Manifest {
+    Manifest {
+        app: ManifestApp {
+            slug: slug.to_string(),
+            name: name.to_string(),
+            description: None,
+        },
+        scripts: vec![ManifestScript {
+            name: "hello".into(),
+            file: "scripts/hello.rhai".into(),
+            kind: ScriptKind::Endpoint,
+            description: None,
+            timeout_seconds: None,
+            memory_limit_mb: None,
+            sandbox: None,
+            enabled: true,
+        }],
+        routes: vec![ManifestRoute {
+            script: "hello".into(),
+            method: Some("GET".into()),
+            host_kind: HostKind::Any,
+            host: String::new(),
+            host_param_name: None,
+            path_kind: PathKind::Exact,
+            path: "/hello".into(),
+            dispatch_mode: DispatchMode::Sync,
+            enabled: true,
+        }],
+        triggers: crate::manifest::ManifestTriggers::default(),
+        secrets: crate::manifest::ManifestSecrets::default(),
+    }
+}
+
+/// Use the explicit slug if given, else derive one from the target
+/// directory's name. Either way it must satisfy the app-slug rule.
+fn resolve_slug(dir: &Path, slug_arg: Option<&str>) -> Result<String> {
+    if let Some(s) = slug_arg {
+        if !is_valid_slug(s) {
+            bail!("invalid app slug `{s}`: use lowercase letters, digits, and dashes (start alphanumeric, max 63)");
+        }
+        return Ok(s.to_string());
+    }
+    // Derive from the directory's own name. Use the path as given first —
+    // `canonicalize` requires the dir to already exist, which breaks the
+    // natural `pic init --dir new-project` flow. Fall back to canonicalizing
+    // only when the path has no final component of its own (e.g. `.`).
+    let base = dir
+        .file_name()
+        .map(|n| n.to_string_lossy().into_owned())
+        .or_else(|| {
+            dir.canonicalize()
+                .ok()
+                .and_then(|p| p.file_name().map(|n| n.to_string_lossy().into_owned()))
+        })
+        .unwrap_or_default();
+    let derived = slugify(&base);
+    if !is_valid_slug(&derived) {
+        bail!(
+            "could not derive a valid app slug from directory `{base}`; \
+             pass one explicitly, e.g. `pic init my-app`"
+        );
+    }
+    Ok(derived)
+}
+
+/// Lowercase, map runs of non-`[a-z0-9]` to a single `-`, trim dashes.
+fn slugify(s: &str) -> String {
+    let mut out = String::with_capacity(s.len());
+    let mut prev_dash = false;
+    for c in s.chars() {
+        if c.is_ascii_alphanumeric() {
+            out.push(c.to_ascii_lowercase());
+            prev_dash = false;
+        } else if !prev_dash {
+            out.push('-');
+            prev_dash = true;
+        }
+    }
+    out.trim_matches('-').to_string()
+}
+
+/// The canonical app-slug rule (mirrors the server): `^[a-z0-9][a-z0-9-]{0,62}$`.
+fn is_valid_slug(s: &str) -> bool {
+    if s.is_empty() || s.len() > 63 {
+        return false;
+    }
+    let mut chars = s.chars();
+    let first = chars.next().expect("non-empty checked above");
+    if !(first.is_ascii_lowercase() || first.is_ascii_digit()) {
+        return false;
+    }
+    chars.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
+}
+
+/// Title-case a slug for a default display name: `my-blog` → `My Blog`.
+fn title_from_slug(slug: &str) -> String {
+    slug.split('-')
+        .filter(|w| !w.is_empty())
+        .map(|w| {
+            let mut c = w.chars();
+            c.next().map_or_else(String::new, |f| {
+                f.to_ascii_uppercase().to_string() + c.as_str()
+            })
+        })
+        .collect::<Vec<_>>()
+        .join(" ")
+}
+
+/// Ensure `.gitignore` ignores `.picloud/` (the project tool's link state).
+/// Appends the line if missing; creates the file if absent. A repo without
+/// git still gets a correct `.gitignore` for when it's initialized.
+fn ensure_gitignored(dir: &Path) -> Result<()> {
+    let path = dir.join(".gitignore");
+    // Only a genuinely-absent file is treated as empty; an existing-but-
+    // unreadable `.gitignore` must error rather than be silently clobbered.
+    let existing = match fs::read_to_string(&path) {
+        Ok(s) => s,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => String::new(),
+        Err(e) => return Err(e).context("reading .gitignore"),
+    };
+    if existing.lines().any(|l| l.trim() == GITIGNORE_LINE) {
+        return Ok(());
+    }
+    let mut next = existing;
+    if !next.is_empty() && !next.ends_with('\n') {
+        next.push('\n');
+    }
+    next.push_str(GITIGNORE_LINE);
+    next.push('\n');
+    fs::write(&path, next).context("updating .gitignore")?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn scaffold_is_valid_and_round_trips() {
+        let m = scaffold_manifest("blog", "Blog");
+        let body = format!("{HEADER}{}{EXAMPLES}", m.to_toml().unwrap());
+        // The active manifest (header/examples are comments) must parse back
+        // to exactly the scaffold — the commented examples are inert.
+        let parsed = Manifest::parse(&body).expect("scaffold must be valid TOML");
+        assert_eq!(parsed, m);
+        // And it's a deployable project: one endpoint + its route.
+        assert_eq!(parsed.scripts.len(), 1);
+        assert_eq!(parsed.routes.len(), 1);
+        assert_eq!(parsed.routes[0].script, "hello");
+    }
+
+    #[test]
+    fn slugify_and_validation() {
+        assert_eq!(slugify("My Blog!"), "my-blog");
+        assert_eq!(slugify("  weird__name  "), "weird-name");
+        assert_eq!(slugify("Project (2026)"), "project-2026");
+        assert!(is_valid_slug("blog"));
+        assert!(is_valid_slug("a1-b2"));
+        assert!(!is_valid_slug(""));
+        assert!(!is_valid_slug("-leading"));
+        assert!(!is_valid_slug(&"a".repeat(64)));
+    }
+
+    #[test]
+    fn title_from_slug_humanizes() {
+        assert_eq!(title_from_slug("my-blog"), "My Blog");
+        assert_eq!(title_from_slug("api"), "Api");
+    }
+}

crates/picloud-cli/src/cmds/mod.rs

@@ -1,14 +1,19 @@
 pub mod admins;
 pub mod api_keys;
+pub mod apply;
 pub mod apps;
 pub mod apps_domains;
+pub mod config;
 pub mod dead_letters;
 pub mod files;
+pub mod init;
 pub mod kv;
 pub mod login;
 pub mod logout;
 pub mod logs;
 pub mod members;
+pub mod plan;
+pub mod pull;
 pub mod queues;
 pub mod routes;
 pub mod scripts;

crates/picloud-cli/src/cmds/plan.rs

@@ -0,0 +1,134 @@
+//! `pic plan [--file picloud.toml]` — diff the manifest's desired state
+//! against the live app and print the per-resource changes. Read-only:
+//! builds a bundle (manifest + script sources) and POSTs it to the
+//! server's plan endpoint, which computes the diff.
+
+use std::path::Path;
+
+use anyhow::{Context, Result};
+use serde::Serialize;
+use serde_json::{json, Map, Value};
+
+use crate::client::{ChangeDto, Client, PlanDto};
+use crate::config;
+use crate::manifest::Manifest;
+use crate::output::{OutputMode, Table};
+
+pub async fn run(manifest_path: &Path, env: Option<&str>, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+
+    let manifest = Manifest::load_with_env(manifest_path, env)?;
+    let base_dir = manifest_path.parent().unwrap_or_else(|| Path::new("."));
+    let bundle = build_bundle(&manifest, base_dir)?;
+
+    let plan = client.plan(&manifest.app.slug, &bundle).await?;
+    // Record the bound-plan token so a subsequent `pic apply` can detect the
+    // app changing underneath the reviewed plan (best-effort — a read-only
+    // plan still succeeds if the project dir isn't writable).
+    if !plan.state_token.is_empty() {
+        if let Err(e) =
+            crate::linkstate::write_plan(base_dir, &manifest.app.slug, &plan.state_token)
+        {
+            eprintln!("warning: could not record plan state for `pic apply`: {e}");
+        }
+    }
+    render(&plan, mode);
+    Ok(())
+}
+
+/// Assemble the wire bundle: scripts carry inlined source (read from
+/// their `file`), routes pass through, triggers flatten into a tagged
+/// array, secrets are names only.
+pub fn build_bundle(manifest: &Manifest, base_dir: &Path) -> Result<Value> {
+    let mut scripts = Vec::with_capacity(manifest.scripts.len());
+    for s in &manifest.scripts {
+        let source = std::fs::read_to_string(base_dir.join(&s.file))
+            .with_context(|| format!("reading script source {}", s.file))?;
+        let mut obj = Map::new();
+        obj.insert("name".into(), json!(s.name));
+        obj.insert("source".into(), json!(source));
+        obj.insert("kind".into(), serde_json::to_value(s.kind)?);
+        if let Some(d) = &s.description {
+            obj.insert("description".into(), json!(d));
+        }
+        if let Some(t) = s.timeout_seconds {
+            obj.insert("timeout_seconds".into(), json!(t));
+        }
+        if let Some(m) = s.memory_limit_mb {
+            obj.insert("memory_limit_mb".into(), json!(m));
+        }
+        if let Some(sb) = &s.sandbox {
+            obj.insert("sandbox".into(), serde_json::to_value(sb)?);
+        }
+        obj.insert("enabled".into(), json!(s.enabled));
+        scripts.push(Value::Object(obj));
+    }
+
+    let routes = manifest
+        .routes
+        .iter()
+        .map(serde_json::to_value)
+        .collect::<Result<Vec<_>, _>>()?;
+
+    let t = &manifest.triggers;
+    let mut triggers = Vec::new();
+    for s in &t.kv {
+        triggers.push(tagged("kv", s)?);
+    }
+    for s in &t.docs {
+        triggers.push(tagged("docs", s)?);
+    }
+    for s in &t.files {
+        triggers.push(tagged("files", s)?);
+    }
+    for s in &t.cron {
+        triggers.push(tagged("cron", s)?);
+    }
+    for s in &t.pubsub {
+        triggers.push(tagged("pubsub", s)?);
+    }
+    for s in &t.email {
+        triggers.push(tagged("email", s)?);
+    }
+    for s in &t.queue {
+        triggers.push(tagged("queue", s)?);
+    }
+
+    Ok(json!({
+        "scripts": scripts,
+        "routes": routes,
+        "triggers": triggers,
+        "secrets": manifest.secrets.names,
+    }))
+}
+
+/// Serialize a trigger spec and stamp its `kind` discriminator.
+fn tagged(kind: &str, spec: impl Serialize) -> Result<Value> {
+    let mut v = serde_json::to_value(spec)?;
+    if let Value::Object(map) = &mut v {
+        map.insert("kind".into(), Value::String(kind.to_string()));
+    }
+    Ok(v)
+}
+
+fn render(plan: &PlanDto, mode: OutputMode) {
+    let mut table = Table::new(["kind", "op", "resource", "detail"]);
+    let groups: [(&str, &Vec<ChangeDto>); 4] = [
+        ("script", &plan.scripts),
+        ("route", &plan.routes),
+        ("trigger", &plan.triggers),
+        ("secret", &plan.secrets),
+    ];
+    for (kind, changes) in groups {
+        for c in changes {
+            table.row([
+                kind.to_string(),
+                c.op.clone(),
+                c.key.clone(),
+                c.detail.clone().unwrap_or_default(),
+            ]);
+        }
+    }
+    table.print(mode);
+}

crates/picloud-cli/src/cmds/pull.rs

@@ -0,0 +1,345 @@
+//! `pic pull <app> [--dir .]` — export an app's current server state into
+//! a `picloud.toml` manifest plus `scripts/<name>.rhai` source files, for
+//! declarative management with `pic plan` / `pic apply`.
+//!
+//! Read-only: issues `GET`s only and writes local files. Every trigger
+//! kind is exported except `email` — the server stores the *sealed secret
+//! value*, not the secret name, so the manifest's `inbound_secret_ref`
+//! can't be reconstructed (email triggers must be set up by hand).
+
+use std::collections::HashMap;
+use std::path::Path;
+
+use anyhow::{Context, Result};
+use picloud_shared::{DispatchMode, DocsEventOp, FilesEventOp, KvEventOp, ScriptId};
+use serde::Deserialize;
+
+use crate::client::Client;
+use crate::config;
+use crate::manifest::{
+    CronTriggerSpec, DocsTriggerSpec, FilesTriggerSpec, KvTriggerSpec, Manifest, ManifestApp,
+    ManifestRoute, ManifestScript, ManifestSecrets, ManifestTriggers, PubsubTriggerSpec,
+    QueueTriggerSpec, MANIFEST_FILE,
+};
+use crate::output::{KvBlock, OutputMode};
+
+pub async fn run(app_ident: &str, dir: &Path, force: bool, mode: OutputMode) -> Result<()> {
+    let creds = config::resolve()?;
+    let client = Client::from_creds(&creds)?;
+
+    // Refuse to clobber an existing project (mirrors `pic init`). `pull`
+    // overwrites `picloud.toml` and every colliding `scripts/*.rhai`, so a
+    // stray `pic pull <wrong-app>` in a populated dir would destroy local
+    // edits. Fail fast — before any network call or file write — unless the
+    // operator opted in with `--force`.
+    let manifest_path = dir.join(MANIFEST_FILE);
+    if !force && manifest_path.exists() {
+        anyhow::bail!(
+            "{} already exists; refusing to overwrite. Re-run with --force to \
+             replace it (and any colliding scripts/*.rhai).",
+            manifest_path.display()
+        );
+    }
+
+    // One GET per resource kind (routes are per-script, below).
+    let app = client.apps_get(app_ident).await?;
+    let scripts = client.scripts_list_by_app(app_ident).await?;
+    let triggers = client.triggers_list(app_ident).await?.triggers;
+    let secrets = client.secrets_list(app_ident).await?.secrets;
+
+    let name_by_id: HashMap<ScriptId, String> =
+        scripts.iter().map(|s| (s.id, s.name.clone())).collect();
+
+    // Routes: the admin surface lists them per script.
+    let mut routes = Vec::new();
+    for s in &scripts {
+        for r in client.routes_list_for_script(&s.id.to_string()).await? {
+            routes.push(ManifestRoute {
+                script: s.name.clone(),
+                method: r.method,
+                host_kind: r.host_kind,
+                host: r.host,
+                host_param_name: r.host_param_name,
+                path_kind: r.path_kind,
+                path: r.path,
+                dispatch_mode: r.dispatch_mode,
+                enabled: r.enabled,
+            });
+        }
+    }
+
+    // The server does not constrain script names to a filesystem-safe
+    // charset, so a name containing a path separator or `..` would let `pull`
+    // write outside the project dir. Validate ALL names up front, before any
+    // file is written, so a single bad name can't leave a half-written dir.
+    // Reject rather than sanitize: a silent rename would desync the manifest
+    // `name` from its `file`.
+    for s in &scripts {
+        if !is_safe_filename(&s.name) {
+            anyhow::bail!(
+                "script name {:?} is not filesystem-safe (a path separator, \
+                 `..`, a leading dot, a control character, or longer than 200 \
+                 bytes); cannot pull",
+                s.name
+            );
+        }
+    }
+
+    // Scripts: write each source next to the manifest and record a path ref.
+    let scripts_dir = dir.join("scripts");
+    std::fs::create_dir_all(&scripts_dir)
+        .with_context(|| format!("creating {}", scripts_dir.display()))?;
+    let mut manifest_scripts = Vec::with_capacity(scripts.len());
+    for s in &scripts {
+        let rel = format!("scripts/{}.rhai", s.name);
+        std::fs::write(dir.join(&rel), &s.source).with_context(|| format!("writing {rel}"))?;
+        manifest_scripts.push(ManifestScript {
+            name: s.name.clone(),
+            file: rel,
+            kind: s.kind,
+            description: s.description.clone(),
+            timeout_seconds: i32::try_from(s.timeout_seconds).ok(),
+            memory_limit_mb: i32::try_from(s.memory_limit_mb).ok(),
+            sandbox: if s.sandbox.is_empty() {
+                None
+            } else {
+                Some(s.sandbox)
+            },
+            enabled: s.enabled,
+        });
+    }
+
+    // Triggers: map the five settled kinds; warn + skip the rest.
+    let mut manifest_triggers = ManifestTriggers::default();
+    let mut skipped: Vec<String> = Vec::new();
+    for t in &triggers {
+        let script = name_by_id
+            .get(&t.script_id)
+            .cloned()
+            .unwrap_or_else(|| t.script_id.to_string());
+        let dispatch_mode = DispatchMode::from_wire(&t.dispatch_mode);
+        let retry_max_attempts = Some(t.retry_max_attempts);
+        match t.kind.as_str() {
+            "kv" => {
+                let d: CollectionDetails<KvEventOp> = decode_details(&t.details, &t.kind)?;
+                manifest_triggers.kv.push(KvTriggerSpec {
+                    script,
+                    collection_glob: d.collection_glob,
+                    ops: d.ops,
+                    dispatch_mode,
+                    retry_max_attempts,
+                });
+            }
+            "docs" => {
+                let d: CollectionDetails<DocsEventOp> = decode_details(&t.details, &t.kind)?;
+                manifest_triggers.docs.push(DocsTriggerSpec {
+                    script,
+                    collection_glob: d.collection_glob,
+                    ops: d.ops,
+                    dispatch_mode,
+                    retry_max_attempts,
+                });
+            }
+            "files" => {
+                let d: CollectionDetails<FilesEventOp> = decode_details(&t.details, &t.kind)?;
+                manifest_triggers.files.push(FilesTriggerSpec {
+                    script,
+                    collection_glob: d.collection_glob,
+                    ops: d.ops,
+                    dispatch_mode,
+                    retry_max_attempts,
+                });
+            }
+            "cron" => {
+                let d: CronDetails = decode_details(&t.details, &t.kind)?;
+                manifest_triggers.cron.push(CronTriggerSpec {
+                    script,
+                    schedule: d.schedule,
+                    timezone: d.timezone,
+                    dispatch_mode,
+                    retry_max_attempts,
+                });
+            }
+            "pubsub" => {
+                let d: PubsubDetails = decode_details(&t.details, &t.kind)?;
+                manifest_triggers.pubsub.push(PubsubTriggerSpec {
+                    script,
+                    topic_pattern: d.topic_pattern,
+                    dispatch_mode,
+                    retry_max_attempts,
+                });
+            }
+            "queue" => {
+                let d: QueueDetails = decode_details(&t.details, &t.kind)?;
+                manifest_triggers.queue.push(QueueTriggerSpec {
+                    script,
+                    queue_name: d.queue_name,
+                    visibility_timeout_secs: Some(d.visibility_timeout_secs),
+                    dispatch_mode,
+                    retry_max_attempts,
+                });
+            }
+            // `email` is skipped: the server stores the sealed secret value,
+            // not the secret *name*, so the manifest's `inbound_secret_ref`
+            // can't be reconstructed — set it up by hand.
+            other => skipped.push(format!("{other} ({})", t.id)),
+        }
+    }
+    for s in &skipped {
+        eprintln!("warning: skipping {s} trigger — not yet representable in the manifest");
+    }
+
+    let manifest = Manifest {
+        app: ManifestApp {
+            slug: app.app.slug.clone(),
+            name: app.app.name.clone(),
+            description: app.app.description.clone(),
+        },
+        scripts: manifest_scripts,
+        routes,
+        triggers: manifest_triggers,
+        secrets: ManifestSecrets {
+            names: secrets.iter().map(|s| s.name.clone()).collect(),
+        },
+    };
+
+    std::fs::write(&manifest_path, manifest.to_toml()?)
+        .with_context(|| format!("writing {}", manifest_path.display()))?;
+
+    let mut block = KvBlock::new();
+    block
+        .field("manifest", manifest_path.display().to_string())
+        .field("app", manifest.app.slug.clone())
+        .field("scripts", manifest.scripts.len().to_string())
+        .field("routes", manifest.routes.len().to_string())
+        .field("triggers", trigger_count(&manifest.triggers).to_string())
+        .field("secrets", manifest.secrets.names.len().to_string());
+    block.print(mode);
+    Ok(())
+}
+
+fn trigger_count(t: &ManifestTriggers) -> usize {
+    t.kv.len() + t.docs.len() + t.files.len() + t.cron.len() + t.pubsub.len() + t.queue.len()
+}
+
+/// True if `name` is safe to use as a single path component in `scripts/`.
+/// Rejects empty/over-long names, path separators, `.`/`..`, leading dots,
+/// and any deceptive display character — a server-returned name is otherwise
+/// written verbatim as a filename and printed to the operator's terminal.
+fn is_safe_filename(name: &str) -> bool {
+    // Leave headroom under NAME_MAX (255 bytes on common filesystems) for the
+    // `.rhai` suffix.
+    const MAX_LEN: usize = 200;
+    !name.is_empty()
+        && name.len() <= MAX_LEN
+        && !name.starts_with('.')
+        && !name.contains('/')
+        && !name.contains('\\')
+        && name != ".."
+        && !name.chars().any(is_deceptive_char)
+}
+
+/// Control characters (NUL, newlines, ANSI escapes) plus the Unicode
+/// bidirectional-override and zero-width/format characters used to spoof how a
+/// name renders in a terminal — both classes are unsafe to print verbatim.
+fn is_deceptive_char(c: char) -> bool {
+    c.is_control()
+        || matches!(c,
+            '\u{200B}'..='\u{200F}'   // zero-width space … LTR/RTL marks
+            | '\u{2028}'..='\u{2029}' // line / paragraph separators
+            | '\u{202A}'..='\u{202E}' // bidi embeddings / overrides
+            | '\u{2060}'              // word joiner
+            | '\u{2066}'..='\u{2069}' // bidi isolates
+            | '\u{FEFF}'              // BOM / zero-width no-break space
+        )
+}
+
+/// Deserialize a trigger's `details` JSON, attributing failures to the kind.
+/// The server tags details with a `kind` field which these structs ignore.
+fn decode_details<T: for<'de> Deserialize<'de>>(
+    details: &serde_json::Value,
+    kind: &str,
+) -> Result<T> {
+    serde_json::from_value(details.clone())
+        .with_context(|| format!("decoding {kind} trigger details"))
+}
+
+#[derive(Deserialize)]
+struct CollectionDetails<Op> {
+    collection_glob: String,
+    #[serde(default = "Vec::new")]
+    ops: Vec<Op>,
+}
+
+#[derive(Deserialize)]
+struct CronDetails {
+    schedule: String,
+    #[serde(default = "default_timezone")]
+    timezone: String,
+}
+
+#[derive(Deserialize)]
+struct PubsubDetails {
+    topic_pattern: String,
+}
+
+#[derive(Deserialize)]
+struct QueueDetails {
+    queue_name: String,
+    visibility_timeout_secs: u32,
+}
+
+fn default_timezone() -> String {
+    "UTC".to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::is_safe_filename;
+
+    #[test]
+    fn rejects_traversal_and_separators() {
+        for bad in [
+            "",
+            ".",
+            "..",
+            "../etc/passwd",
+            "a/b",
+            "a\\b",
+            ".hidden",
+            "with\0nul",
+        ] {
+            assert!(!is_safe_filename(bad), "expected {bad:?} to be rejected");
+        }
+    }
+
+    #[test]
+    fn rejects_control_chars_and_overlong() {
+        for bad in [
+            "a\nb",
+            "a\tb",
+            "line\rdrop",
+            "esc\x1b[2Jseq",
+            "rtl\u{202E}gpj.exe", // bidi override (filename spoof)
+            "zero\u{200B}width",  // zero-width space
+            "bom\u{FEFF}name",    // BOM
+        ] {
+            assert!(!is_safe_filename(bad), "expected {bad:?} to be rejected");
+        }
+        assert!(
+            !is_safe_filename(&"a".repeat(201)),
+            "expected an over-long name to be rejected"
+        );
+        assert!(
+            is_safe_filename(&"a".repeat(200)),
+            "a 200-char name is at the limit and allowed"
+        );
+    }
+
+    #[test]
+    fn accepts_normal_names() {
+        for ok in ["create-post", "nightly_digest", "Greet", "x", "a.b"] {
+            assert!(is_safe_filename(ok), "expected {ok:?} to be accepted");
+        }
+    }
+}

crates/picloud-cli/src/linkstate.rs

@@ -0,0 +1,67 @@
+//! `.picloud/` link state — gitignored, per-project metadata the project tool
+//! carries between CLI invocations. Today it holds just the bound-plan token:
+//! `pic plan` records the fingerprint of the live state it diffed against, and
+//! `pic apply` replays it so the server can refuse if the app moved underneath.
+//!
+//! All paths are relative to the manifest's directory (the project root).
+
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use anyhow::{Context, Result};
+use serde::{Deserialize, Serialize};
+
+const DIR: &str = ".picloud";
+const PLAN_FILE: &str = "plan.json";
+
+/// The recorded result of the last `pic plan`, scoped to the app it was for.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PlanLink {
+    /// App slug the token belongs to — guards against replaying a token from a
+    /// different app if the manifest's `slug` changed.
+    pub app: String,
+    pub state_token: String,
+}
+
+fn plan_path(base: &Path) -> PathBuf {
+    base.join(DIR).join(PLAN_FILE)
+}
+
+/// Record the bound-plan token for `app` under `base/.picloud/`.
+pub fn write_plan(base: &Path, app: &str, state_token: &str) -> Result<()> {
+    let dir = base.join(DIR);
+    fs::create_dir_all(&dir).with_context(|| format!("creating {}", dir.display()))?;
+    // Self-ignore: a `.gitignore` of `*` inside `.picloud/` keeps the whole
+    // dir out of git regardless of the project root's `.gitignore` — so this
+    // is safe even when reached via `pic plan`/`pull` (which, unlike `init`,
+    // don't touch the root `.gitignore`).
+    let ignore = dir.join(".gitignore");
+    if !ignore.exists() {
+        fs::write(&ignore, "*\n").context("writing .picloud/.gitignore")?;
+    }
+    let link = PlanLink {
+        app: app.to_string(),
+        state_token: state_token.to_string(),
+    };
+    let body = serde_json::to_vec_pretty(&link).context("encoding .picloud/plan.json")?;
+    fs::write(plan_path(base), body).context("writing .picloud/plan.json")?;
+    Ok(())
+}
+
+/// Read the recorded plan token, if any. Returns `None` when absent or
+/// unreadable (treated as "no prior plan" — never an error).
+#[must_use]
+pub fn read_plan(base: &Path) -> Option<PlanLink> {
+    let body = fs::read(plan_path(base)).ok()?;
+    serde_json::from_slice(&body).ok()
+}
+
+/// Remove the recorded plan token (best-effort) **iff it belongs to `app`**.
+/// Called after a successful apply consumes it, so the next apply requires a
+/// fresh plan — without clobbering a token recorded for a different app that
+/// happens to share the directory.
+pub fn clear_plan(base: &Path, app: &str) {
+    if read_plan(base).is_some_and(|l| l.app == app) {
+        let _ = fs::remove_file(plan_path(base));
+    }
+}

crates/picloud-cli/src/main.rs

@@ -12,6 +12,8 @@ use clap::{Args, Parser, Subcommand, ValueEnum};
 mod client;
 mod cmds;
 mod config;
+mod linkstate;
+mod manifest;
 mod output;
 
 use crate::output::OutputMode;
@@ -156,6 +158,105 @@ enum Cmd {
         #[command(subcommand)]
         cmd: KvCmd,
     },
+
+    /// Reconcile the live app to a `picloud.toml` manifest in one
+    /// transaction (creates + updates; `--prune` also deletes resources
+    /// absent from the manifest).
+    Apply(ApplyArgs),
+
+    /// Diff a `picloud.toml` manifest against the live app and print the
+    /// changes (create / update / no-op / delete). Read-only.
+    Plan(PlanArgs),
+
+    /// Export an app's current server state into a `picloud.toml` manifest
+    /// (+ `scripts/<name>.rhai` sources) for declarative management with
+    /// `pic plan` / `pic apply`.
+    Pull(PullArgs),
+
+    /// Scaffold a new declarative project: a `picloud.toml` (minimal working
+    /// app), `scripts/hello.rhai`, and a `.gitignore`. Offline; deploy with
+    /// `pic plan` then `pic apply`.
+    Init(InitArgs),
+
+    /// Show an app's resolved configuration. `--effective` lists secrets with
+    /// values masked (`<set>`/`<unset>`), cross-referenced against the manifest.
+    Config(ConfigArgs),
+}
+
+#[derive(Args)]
+struct ApplyArgs {
+    /// Path to the manifest.
+    #[arg(long, default_value = "picloud.toml")]
+    file: PathBuf,
+    /// Delete live scripts/routes/triggers absent from the manifest.
+    /// Secrets are never pruned.
+    #[arg(long)]
+    prune: bool,
+    /// Skip the `--prune` confirmation prompt. Required to prune
+    /// non-interactively (CI).
+    #[arg(long)]
+    yes: bool,
+    /// Skip the bound-plan staleness check (apply even if the app changed
+    /// since the last `pic plan`).
+    #[arg(long)]
+    force: bool,
+    /// Merge the `picloud.<env>.toml` overlay (per-env slug + secrets) on
+    /// top of the base manifest before applying.
+    #[arg(long)]
+    env: Option<String>,
+}
+
+#[derive(Args)]
+struct PlanArgs {
+    /// Path to the manifest.
+    #[arg(long, default_value = "picloud.toml")]
+    file: PathBuf,
+    /// Merge the `picloud.<env>.toml` overlay (per-env slug + secrets) on
+    /// top of the base manifest before diffing.
+    #[arg(long)]
+    env: Option<String>,
+}
+
+#[derive(Args)]
+struct PullArgs {
+    /// App slug or id to export.
+    app: String,
+    /// Directory to write `picloud.toml` + `scripts/` into.
+    #[arg(long, default_value = ".")]
+    dir: PathBuf,
+    /// Overwrite an existing `picloud.toml` (and colliding `scripts/*.rhai`).
+    #[arg(long)]
+    force: bool,
+}
+
+#[derive(Args)]
+struct InitArgs {
+    /// App slug for the new project. Defaults to a slug derived from the
+    /// target directory's name.
+    slug: Option<String>,
+    /// Directory to scaffold into.
+    #[arg(long, default_value = ".")]
+    dir: PathBuf,
+    /// Display name for the app. Defaults to a title-cased slug.
+    #[arg(long)]
+    name: Option<String>,
+    /// Overwrite an existing `picloud.toml`.
+    #[arg(long)]
+    force: bool,
+}
+
+#[derive(Args)]
+struct ConfigArgs {
+    /// Path to the manifest.
+    #[arg(long, default_value = "picloud.toml")]
+    file: PathBuf,
+    /// Show the effective (resolved) config with secrets masked.
+    #[arg(long)]
+    effective: bool,
+    /// Resolve against the `picloud.<env>.toml` overlay (per-env slug +
+    /// secrets), matching `pic plan --env` / `pic apply --env`.
+    #[arg(long)]
+    env: Option<String>,
 }
 
 #[derive(Subcommand)]
@@ -1045,6 +1146,29 @@ async fn main() -> ExitCode {
         }
         Cmd::Logout => cmds::logout::run().await,
         Cmd::Whoami => cmds::whoami::run(mode).await,
+        Cmd::Apply(args) => {
+            cmds::apply::run(
+                &args.file,
+                args.env.as_deref(),
+                args.prune,
+                args.yes,
+                args.force,
+                mode,
+            )
+            .await
+        }
+        Cmd::Plan(args) => cmds::plan::run(&args.file, args.env.as_deref(), mode).await,
+        Cmd::Pull(args) => cmds::pull::run(&args.app, &args.dir, args.force, mode).await,
+        Cmd::Config(args) => {
+            cmds::config::run(&args.file, args.effective, args.env.as_deref(), mode).await
+        }
+        Cmd::Init(args) => cmds::init::run(
+            &args.dir,
+            args.slug.as_deref(),
+            args.name.as_deref(),
+            args.force,
+            mode,
+        ),
         Cmd::Apps { cmd: AppsCmd::Ls } => cmds::apps::ls(mode).await,
         Cmd::Apps {
             cmd:

crates/picloud-cli/src/manifest.rs

@@ -0,0 +1,523 @@
+//! Declarative project manifest (`picloud.toml`).
+//!
+//! One manifest describes the desired state of a **single app** — its
+//! scripts, routes, triggers, and the *names* of the secrets it expects
+//! (values are pushed out-of-band via `pic secret set`, never committed).
+//!
+//! This is the foundation of the declarative project tool (`pic pull` /
+//! `pic plan` / `pic apply`). The types deliberately reuse `picloud_shared`
+//! enums (`HostKind`, `PathKind`, `DispatchMode`, `ScriptKind`,
+//! `ScriptSandbox`, the event-op enums) so the manifest's wire shape stays
+//! identical to the admin API — the CLI never depends on `manager-core`.
+//!
+//! All eight trigger kinds are representable except `dead_letter` (not
+//! exposed declaratively). `email` triggers carry an `inbound_secret_ref`
+//! (a secret name) resolved server-side at apply.
+
+use std::fs;
+use std::path::Path;
+
+use anyhow::{Context, Result};
+use picloud_shared::{
+    DispatchMode, DocsEventOp, FilesEventOp, HostKind, KvEventOp, PathKind, ScriptKind,
+    ScriptSandbox,
+};
+use serde::{Deserialize, Serialize};
+
+/// Conventional manifest filename at a project root.
+pub const MANIFEST_FILE: &str = "picloud.toml";
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct Manifest {
+    pub app: ManifestApp,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub scripts: Vec<ManifestScript>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub routes: Vec<ManifestRoute>,
+    #[serde(default, skip_serializing_if = "ManifestTriggers::is_empty")]
+    pub triggers: ManifestTriggers,
+    #[serde(default, skip_serializing_if = "ManifestSecrets::is_empty")]
+    pub secrets: ManifestSecrets,
+}
+
+impl Manifest {
+    /// Parse a manifest from TOML text.
+    pub fn parse(text: &str) -> Result<Self> {
+        toml::from_str(text).context("parsing manifest TOML")
+    }
+
+    /// Load and parse the manifest at `path`.
+    pub fn load(path: &Path) -> Result<Self> {
+        let body =
+            fs::read_to_string(path).with_context(|| format!("reading {}", path.display()))?;
+        Self::parse(&body)
+    }
+
+    /// Render to TOML text. Tables are emitted after scalars (the struct
+    /// field order already satisfies TOML's "values before tables" rule).
+    pub fn to_toml(&self) -> Result<String> {
+        toml::to_string_pretty(self).context("serializing manifest TOML")
+    }
+
+    /// Load the base manifest, then (if `env` is set) merge the sparse
+    /// `picloud.<env>.toml` overlay on top — the §4.1 base+overlay model
+    /// where "an environment is an app". The overlay carries per-env slug +
+    /// secrets; scripts/routes/triggers stay in the shared base. (Rich
+    /// per-key `vars` resolution is Phase 3.)
+    pub fn load_with_env(base_path: &Path, env: Option<&str>) -> Result<Self> {
+        let mut base = Self::load(base_path)?;
+        if let Some(env) = env {
+            let path = overlay_path(base_path, env);
+            let body = fs::read_to_string(&path).with_context(|| {
+                format!(
+                    "reading overlay {} for env `{env}` (expected next to the base manifest)",
+                    path.display()
+                )
+            })?;
+            let overlay: ManifestOverlay = toml::from_str(&body)
+                .with_context(|| format!("parsing overlay {}", path.display()))?;
+            base.apply_overlay(overlay);
+        }
+        Ok(base)
+    }
+
+    /// Merge a sparse overlay onto this manifest: overlay `app.slug`/`name`
+    /// replace the base's; overlay secret names union into the base set.
+    fn apply_overlay(&mut self, overlay: ManifestOverlay) {
+        if let Some(slug) = overlay.app.slug {
+            self.app.slug = slug;
+        }
+        if let Some(name) = overlay.app.name {
+            self.app.name = name;
+        }
+        for n in overlay.secrets.names {
+            if !self.secrets.names.contains(&n) {
+                self.secrets.names.push(n);
+            }
+        }
+    }
+}
+
+/// `picloud.toml` → `picloud.<env>.toml`, alongside the base (works for a
+/// custom `--file` too: `custom.toml` → `custom.<env>.toml`).
+fn overlay_path(base_path: &Path, env: &str) -> std::path::PathBuf {
+    let parent = base_path.parent().unwrap_or_else(|| Path::new("."));
+    let name = base_path
+        .file_name()
+        .map(|n| n.to_string_lossy().into_owned())
+        .unwrap_or_else(|| MANIFEST_FILE.to_string());
+    let stem = name.strip_suffix(".toml").unwrap_or(&name);
+    parent.join(format!("{stem}.{env}.toml"))
+}
+
+/// A sparse per-environment overlay (`picloud.<env>.toml`). Only the fields
+/// that vary per environment today — slug/name and secret names.
+///
+/// `deny_unknown_fields`: an overlay can carry *only* `[app]` and `[secrets]`.
+/// Scripts/routes/triggers belong in the shared base manifest, so a
+/// `[[scripts]]`/`[[routes]]`/`[[triggers]]` table (or a typo'd key) in an
+/// overlay is a mistake — error loudly rather than silently dropping it.
+#[derive(Debug, Clone, Default, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct ManifestOverlay {
+    #[serde(default)]
+    pub app: OverlayApp,
+    #[serde(default)]
+    pub secrets: ManifestSecrets,
+}
+
+#[derive(Debug, Clone, Default, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct OverlayApp {
+    #[serde(default)]
+    pub slug: Option<String>,
+    #[serde(default)]
+    pub name: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ManifestApp {
+    pub slug: String,
+    pub name: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ManifestScript {
+    pub name: String,
+    /// Path to the `.rhai` source, relative to the manifest's directory.
+    pub file: String,
+    #[serde(default, skip_serializing_if = "is_endpoint")]
+    pub kind: ScriptKind,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub timeout_seconds: Option<i32>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub memory_limit_mb: Option<i32>,
+    /// Per-script sandbox overrides; omitted entirely when no knob is set.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub sandbox: Option<ScriptSandbox>,
+    /// Three-state lifecycle (§4.3): `false` deploys the script inert (not
+    /// invocable). Omitted ⇒ active; only serialized when disabled.
+    #[serde(
+        default = "picloud_shared::default_true",
+        skip_serializing_if = "is_true"
+    )]
+    pub enabled: bool,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct ManifestRoute {
+    /// Name of the script this route binds to.
+    pub script: String,
+    /// HTTP method; omit for ANY.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub method: Option<String>,
+    pub host_kind: HostKind,
+    #[serde(default, skip_serializing_if = "String::is_empty")]
+    pub host: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub host_param_name: Option<String>,
+    pub path_kind: PathKind,
+    pub path: String,
+    #[serde(default, skip_serializing_if = "is_sync")]
+    pub dispatch_mode: DispatchMode,
+    /// Three-state lifecycle (§4.3): `false` deploys the route inert (404).
+    /// Omitted ⇒ active; only serialized when disabled.
+    #[serde(
+        default = "picloud_shared::default_true",
+        skip_serializing_if = "is_true"
+    )]
+    pub enabled: bool,
+}
+
+/// Triggers grouped by kind (arrays-of-tables: `[[triggers.cron]]`, …).
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct ManifestTriggers {
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub kv: Vec<KvTriggerSpec>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub docs: Vec<DocsTriggerSpec>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub files: Vec<FilesTriggerSpec>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub cron: Vec<CronTriggerSpec>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub pubsub: Vec<PubsubTriggerSpec>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub email: Vec<EmailTriggerSpec>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub queue: Vec<QueueTriggerSpec>,
+}
+
+impl ManifestTriggers {
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.kv.is_empty()
+            && self.docs.is_empty()
+            && self.files.is_empty()
+            && self.cron.is_empty()
+            && self.pubsub.is_empty()
+            && self.email.is_empty()
+            && self.queue.is_empty()
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct KvTriggerSpec {
+    pub script: String,
+    pub collection_glob: String,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub ops: Vec<KvEventOp>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dispatch_mode: Option<DispatchMode>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry_max_attempts: Option<u32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct DocsTriggerSpec {
+    pub script: String,
+    pub collection_glob: String,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub ops: Vec<DocsEventOp>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dispatch_mode: Option<DispatchMode>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry_max_attempts: Option<u32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct FilesTriggerSpec {
+    pub script: String,
+    pub collection_glob: String,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub ops: Vec<FilesEventOp>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dispatch_mode: Option<DispatchMode>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry_max_attempts: Option<u32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct CronTriggerSpec {
+    pub script: String,
+    /// 6-field cron expression (with seconds).
+    pub schedule: String,
+    #[serde(default = "default_timezone")]
+    pub timezone: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dispatch_mode: Option<DispatchMode>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry_max_attempts: Option<u32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct PubsubTriggerSpec {
+    pub script: String,
+    pub topic_pattern: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dispatch_mode: Option<DispatchMode>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry_max_attempts: Option<u32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct EmailTriggerSpec {
+    pub script: String,
+    /// Name of the secret (set via `pic secret set`) holding the inbound
+    /// HMAC value — resolved + sealed server-side at apply. Never the value.
+    pub inbound_secret_ref: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dispatch_mode: Option<DispatchMode>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry_max_attempts: Option<u32>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct QueueTriggerSpec {
+    pub script: String,
+    pub queue_name: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub visibility_timeout_secs: Option<u32>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub dispatch_mode: Option<DispatchMode>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub retry_max_attempts: Option<u32>,
+}
+
+/// `[secrets] names = [...]` — declares which secrets the app expects.
+/// Values are never in the manifest; `pic secret set` pushes them.
+#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
+pub struct ManifestSecrets {
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub names: Vec<String>,
+}
+
+impl ManifestSecrets {
+    #[must_use]
+    pub fn is_empty(&self) -> bool {
+        self.names.is_empty()
+    }
+}
+
+// ---- serde skip/default helpers ----
+
+fn is_endpoint(kind: &ScriptKind) -> bool {
+    *kind == ScriptKind::Endpoint
+}
+
+fn is_sync(mode: &DispatchMode) -> bool {
+    *mode == DispatchMode::Sync
+}
+
+/// Skip-serialize helper: `enabled` defaults true, so only emit it when false.
+fn is_true(b: &bool) -> bool {
+    *b
+}
+
+fn default_timezone() -> String {
+    "UTC".to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn sample() -> Manifest {
+        Manifest {
+            app: ManifestApp {
+                slug: "blog".into(),
+                name: "My Blog".into(),
+                description: Some("demo".into()),
+            },
+            scripts: vec![
+                ManifestScript {
+                    name: "create-post".into(),
+                    file: "scripts/create-post.rhai".into(),
+                    kind: ScriptKind::Endpoint,
+                    description: None,
+                    timeout_seconds: Some(10),
+                    memory_limit_mb: Some(256),
+                    sandbox: None,
+                    enabled: true,
+                },
+                ManifestScript {
+                    name: "lib".into(),
+                    file: "scripts/lib.rhai".into(),
+                    kind: ScriptKind::Module,
+                    description: None,
+                    timeout_seconds: None,
+                    memory_limit_mb: None,
+                    sandbox: Some(ScriptSandbox {
+                        max_operations: Some(5_000_000),
+                        ..ScriptSandbox::empty()
+                    }),
+                    enabled: false,
+                },
+            ],
+            routes: vec![ManifestRoute {
+                script: "create-post".into(),
+                method: Some("POST".into()),
+                host_kind: HostKind::Any,
+                host: String::new(),
+                host_param_name: None,
+                path_kind: PathKind::Exact,
+                path: "/posts".into(),
+                dispatch_mode: DispatchMode::Sync,
+                enabled: true,
+            }],
+            triggers: ManifestTriggers {
+                cron: vec![CronTriggerSpec {
+                    script: "create-post".into(),
+                    schedule: "0 6 * * * *".into(),
+                    timezone: "UTC".into(),
+                    dispatch_mode: None,
+                    retry_max_attempts: None,
+                }],
+                kv: vec![KvTriggerSpec {
+                    script: "create-post".into(),
+                    collection_glob: "users".into(),
+                    ops: vec![KvEventOp::Insert, KvEventOp::Update],
+                    dispatch_mode: Some(DispatchMode::Async),
+                    retry_max_attempts: Some(5),
+                }],
+                ..ManifestTriggers::default()
+            },
+            secrets: ManifestSecrets {
+                names: vec!["STRIPE_KEY".into()],
+            },
+        }
+    }
+
+    #[test]
+    fn round_trips_through_toml() {
+        let m = sample();
+        let text = m.to_toml().expect("serialize");
+        let back = Manifest::parse(&text).expect("parse");
+        assert_eq!(m, back, "manifest must survive a TOML round-trip");
+    }
+
+    #[test]
+    fn omits_defaulted_fields() {
+        let text = sample().to_toml().unwrap();
+        // Endpoint kind + sync dispatch are defaults → not emitted.
+        assert!(
+            !text.contains("kind = \"endpoint\""),
+            "default kind should be omitted:\n{text}"
+        );
+        assert!(
+            !text.contains("dispatch_mode = \"sync\""),
+            "default route dispatch should be omitted:\n{text}"
+        );
+        // Module kind IS non-default → emitted.
+        assert!(text.contains("kind = \"module\""), "got:\n{text}");
+        // `enabled` defaults true → omitted for the active script/route, but
+        // the disabled `lib` script emits `enabled = false`.
+        assert!(
+            text.contains("enabled = false"),
+            "a disabled entity must emit enabled:\n{text}"
+        );
+        assert!(
+            !text.contains("enabled = true"),
+            "active entities must omit the default:\n{text}"
+        );
+    }
+
+    #[test]
+    fn overlay_merges_slug_and_unions_secrets() {
+        let mut m = sample(); // slug "blog", secrets ["STRIPE_KEY"]
+        let overlay: ManifestOverlay = toml::from_str(
+            "[app]\nslug = \"blog-staging\"\n\n[secrets]\nnames = [\"STRIPE_KEY\", \"STAGING_ONLY\"]\n",
+        )
+        .unwrap();
+        m.apply_overlay(overlay);
+        assert_eq!(m.app.slug, "blog-staging", "overlay slug wins");
+        assert_eq!(
+            m.app.name, "My Blog",
+            "base name kept when overlay omits it"
+        );
+        assert_eq!(
+            m.secrets.names,
+            vec!["STRIPE_KEY".to_string(), "STAGING_ONLY".to_string()],
+            "secrets union, no dupes"
+        );
+        // Scripts/routes come from the base, untouched by the overlay.
+        assert_eq!(m.scripts.len(), 2);
+    }
+
+    #[test]
+    fn overlay_rejects_non_overlay_tables() {
+        // An overlay carries only [app]/[secrets]. Scripts/routes/triggers
+        // belong in the shared base, so a `[[scripts]]` table (or a typo'd
+        // key) in an overlay must error loudly, not be silently dropped.
+        let err = toml::from_str::<ManifestOverlay>(
+            "[app]\nslug = \"blog-staging\"\n\n\
+             [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\n",
+        )
+        .expect_err("overlay with [[scripts]] must be rejected");
+        assert!(
+            err.to_string().contains("scripts") || err.to_string().contains("unknown"),
+            "error should point at the offending table: {err}"
+        );
+
+        // Typo'd key inside [app] is likewise rejected.
+        toml::from_str::<ManifestOverlay>("[app]\nslag = \"oops\"\n")
+            .expect_err("overlay with a typo'd [app] key must be rejected");
+    }
+
+    #[test]
+    fn overlay_path_derivation() {
+        assert_eq!(
+            overlay_path(Path::new("proj/picloud.toml"), "staging"),
+            Path::new("proj/picloud.staging.toml")
+        );
+        assert_eq!(
+            overlay_path(Path::new("custom.toml"), "prod"),
+            Path::new("custom.prod.toml")
+        );
+    }
+
+    #[test]
+    fn empty_optional_sections_omitted() {
+        let m = Manifest {
+            app: ManifestApp {
+                slug: "x".into(),
+                name: "X".into(),
+                description: None,
+            },
+            scripts: vec![],
+            routes: vec![],
+            triggers: ManifestTriggers::default(),
+            secrets: ManifestSecrets::default(),
+        };
+        let text = m.to_toml().unwrap();
+        assert!(!text.contains("[[scripts]]"), "got:\n{text}");
+        assert!(!text.contains("triggers"), "got:\n{text}");
+        assert!(!text.contains("secrets"), "got:\n{text}");
+        // Still round-trips.
+        assert_eq!(m, Manifest::parse(&text).unwrap());
+    }
+}

crates/picloud-cli/tests/apply.rs

@@ -0,0 +1,153 @@
+//! `pic apply` journey: apply a manifest to an empty app (atomic create),
+//! re-apply is an idempotent no-op, and a bundle containing any invalid
+//! resource applies nothing (all-or-nothing).
+
+use std::fs;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::AppGuard;
+
+fn manifest_dir() -> TempDir {
+    let dir = TempDir::new().expect("tempdir");
+    fs::create_dir_all(dir.path().join("scripts")).expect("scripts dir");
+    dir
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn apply_creates_then_noop() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("apply");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = manifest_dir();
+    fs::write(
+        dir.path().join("scripts/greet.rhai"),
+        "let body = #{ ok: true }; body",
+    )
+    .unwrap();
+    let manifest = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"Apply Test\"\n\n\
+         [[scripts]]\nname = \"greet\"\nfile = \"scripts/greet.rhai\"\n\n\
+         [[routes]]\nscript = \"greet\"\nmethod = \"POST\"\n\
+         host_kind = \"any\"\npath_kind = \"exact\"\npath = \"/greet\"\n\n\
+         [[triggers.cron]]\nscript = \"greet\"\nschedule = \"0 0 * * * *\"\ntimezone = \"UTC\"\n"
+    );
+    let manifest_path = dir.path().join("picloud.toml");
+    fs::write(&manifest_path, &manifest).unwrap();
+
+    // First apply: creates script + route + trigger.
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .output()
+        .expect("apply");
+    assert!(
+        out.status.success(),
+        "apply failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let stdout = String::from_utf8(out.stdout).unwrap();
+    assert!(
+        stdout.contains("+1"),
+        "expected creations in report:\n{stdout}"
+    );
+
+    // The resources now exist.
+    let s = String::from_utf8(
+        common::pic_as(&env)
+            .args(["scripts", "ls", "--app", &slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(s.contains("greet"), "script not created:\n{s}");
+
+    // Plan is now clean (apply reached desired state).
+    let p = String::from_utf8(
+        common::pic_as(&env)
+            .args(["plan", "--file"])
+            .arg(&manifest_path)
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        !p.contains("create") && !p.contains("update"),
+        "expected clean plan after apply:\n{p}"
+    );
+
+    // Re-apply: idempotent — nothing created/updated.
+    let r = String::from_utf8(
+        common::pic_as(&env)
+            .args(["apply", "--file"])
+            .arg(&manifest_path)
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(!r.contains("+1"), "re-apply should be a no-op:\n{r}");
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn apply_rejects_bad_bundle_atomically() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("apply-atomic");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = manifest_dir();
+    fs::write(dir.path().join("scripts/good.rhai"), "let x = 1; x").unwrap();
+    // Invalid Rhai — fails validation, so the whole apply must abort.
+    fs::write(dir.path().join("scripts/bad.rhai"), "let x = ;").unwrap();
+    let manifest = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"Atomic Test\"\n\n\
+         [[scripts]]\nname = \"good\"\nfile = \"scripts/good.rhai\"\n\n\
+         [[scripts]]\nname = \"bad\"\nfile = \"scripts/bad.rhai\"\n"
+    );
+    let manifest_path = dir.path().join("picloud.toml");
+    fs::write(&manifest_path, &manifest).unwrap();
+
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .output()
+        .expect("apply");
+    assert!(
+        !out.status.success(),
+        "apply with an invalid script should fail"
+    );
+
+    // Atomic: the valid script must NOT have been created.
+    let s = String::from_utf8(
+        common::pic_as(&env)
+            .args(["scripts", "ls", "--app", &slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        !s.contains("good"),
+        "a failed apply must leave nothing behind:\n{s}"
+    );
+}

crates/picloud-cli/tests/cli.rs

@@ -15,14 +15,24 @@ mod common;
 
 mod admins;
 mod api_keys;
+mod apply;
 mod apps;
 mod auth;
+mod config;
 mod dead_letters;
+mod email_queue;
+mod enabled;
+mod env_overlay;
+mod init;
 mod invoke;
 mod logs;
 mod output;
+mod plan;
+mod prune;
+mod pull;
 mod roles;
 mod routes;
 mod scripts;
 mod secrets;
+mod staleness;
 mod triggers;

crates/picloud-cli/tests/config.rs

@@ -0,0 +1,57 @@
+//! `pic config --effective` — masked secret resolution against the manifest.
+
+use std::fs;
+
+use predicates::prelude::*;
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::AppGuard;
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn config_effective_masks_and_classifies_secrets() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("config");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = TempDir::new().unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+    fs::write(
+        &manifest_path,
+        format!("[app]\nslug = \"{slug}\"\nname = \"Cfg\"\n\n[secrets]\nnames = [\"API_KEY\"]\n"),
+    )
+    .unwrap();
+
+    // Declared but not pushed → masked + flagged unset; value never shown.
+    common::pic_as(&env)
+        .args(["config", "--effective", "--file"])
+        .arg(&manifest_path)
+        .assert()
+        .success()
+        .stdout(predicate::str::contains("API_KEY"))
+        .stdout(predicate::str::contains("<unset>"))
+        .stdout(predicate::str::contains("not pushed"));
+
+    // Push it → now masked as <set> / managed, still never the value.
+    common::pic_as(&env)
+        .args(["secrets", "set", "--app", &slug, "API_KEY"])
+        .write_stdin("super-secret-value")
+        .assert()
+        .success();
+    common::pic_as(&env)
+        .args(["config", "--effective", "--file"])
+        .arg(&manifest_path)
+        .assert()
+        .success()
+        .stdout(predicate::str::contains("<set>"))
+        .stdout(predicate::str::contains("managed"))
+        .stdout(predicate::str::contains("super-secret-value").not());
+}

crates/picloud-cli/tests/email_queue.rs

@@ -0,0 +1,222 @@
+//! M5: `pic apply` creates email + queue triggers. The email trigger's
+//! inbound secret is referenced by name (pushed via `pic secret set`) and
+//! resolved + re-sealed server-side — never written into the manifest.
+
+use std::fs;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::AppGuard;
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn apply_email_and_queue_triggers() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("m5");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    // The email trigger references this secret by name; push its value
+    // out-of-band first.
+    common::pic_as(&env)
+        .args(["secrets", "set", "--app", &slug, "email-hmac"])
+        .write_stdin("super-secret-hmac")
+        .assert()
+        .success();
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/handler.rhai"), "let x = 1; x").unwrap();
+    let manifest = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"M5\"\n\n\
+         [secrets]\nnames = [\"email-hmac\"]\n\n\
+         [[scripts]]\nname = \"handler\"\nfile = \"scripts/handler.rhai\"\n\n\
+         [[triggers.queue]]\nscript = \"handler\"\nqueue_name = \"jobs\"\n\n\
+         [[triggers.email]]\nscript = \"handler\"\ninbound_secret_ref = \"email-hmac\"\n"
+    );
+    let manifest_path = dir.path().join("picloud.toml");
+    fs::write(&manifest_path, &manifest).unwrap();
+
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .output()
+        .expect("apply");
+    assert!(
+        out.status.success(),
+        "apply failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+
+    // Both triggers exist.
+    let s = String::from_utf8(
+        common::pic_as(&env)
+            .args(["triggers", "ls", "--app", &slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        s.lines().any(|l| l.contains("queue")),
+        "queue trigger missing:\n{s}"
+    );
+    assert!(
+        s.lines().any(|l| l.contains("email")),
+        "email trigger missing:\n{s}"
+    );
+
+    // Re-apply is a no-op (both triggers match by identity).
+    let r = String::from_utf8(
+        common::pic_as(&env)
+            .args(["apply", "--file"])
+            .arg(&manifest_path)
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(!r.contains("+1"), "re-apply should be a no-op:\n{r}");
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn prune_refuses_to_orphan_email_trigger() {
+    // `pull` can't represent email triggers, so a manifest that omits the
+    // script owning one would, under `--prune`, cascade-delete the trigger
+    // (and its sealed secret) when the script is dropped. Apply must refuse.
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("m5-orphan");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    common::pic_as(&env)
+        .args(["secrets", "set", "--app", &slug, "email-hmac"])
+        .write_stdin("super-secret-hmac")
+        .assert()
+        .success();
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/handler.rhai"), "let x = 1; x").unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+
+    // v1: a script with an email trigger.
+    let v1 = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"M5\"\n\n\
+         [secrets]\nnames = [\"email-hmac\"]\n\n\
+         [[scripts]]\nname = \"handler\"\nfile = \"scripts/handler.rhai\"\n\n\
+         [[triggers.email]]\nscript = \"handler\"\ninbound_secret_ref = \"email-hmac\"\n"
+    );
+    fs::write(&manifest_path, &v1).unwrap();
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .assert()
+        .success();
+
+    // v2: drop the script (and, implicitly, its un-representable email
+    // trigger). A prune apply must REFUSE rather than cascade-destroy it.
+    let v2 = format!("[app]\nslug = \"{slug}\"\nname = \"M5\"\n");
+    fs::write(&manifest_path, &v2).unwrap();
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .args(["--prune", "--yes"])
+        .output()
+        .expect("apply --prune");
+    assert!(
+        !out.status.success(),
+        "prune must refuse to orphan an email trigger"
+    );
+
+    // The script and its email trigger both survive the refused apply.
+    let scripts = String::from_utf8(
+        common::pic_as(&env)
+            .args(["scripts", "ls", "--app", &slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        scripts.contains("handler"),
+        "script must survive:\n{scripts}"
+    );
+    let triggers = String::from_utf8(
+        common::pic_as(&env)
+            .args(["triggers", "ls", "--app", &slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        triggers.lines().any(|l| l.contains("email")),
+        "email trigger must survive:\n{triggers}"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn apply_email_unset_secret_fails() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("m5-nosecret");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/handler.rhai"), "let x = 1; x").unwrap();
+    let manifest = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"M5\"\n\n\
+         [[scripts]]\nname = \"handler\"\nfile = \"scripts/handler.rhai\"\n\n\
+         [[triggers.email]]\nscript = \"handler\"\ninbound_secret_ref = \"never-set\"\n"
+    );
+    let manifest_path = dir.path().join("picloud.toml");
+    fs::write(&manifest_path, &manifest).unwrap();
+
+    // The referenced secret was never set → apply must fail atomically.
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .output()
+        .expect("apply");
+    assert!(
+        !out.status.success(),
+        "apply must fail when an email secret is unset"
+    );
+
+    // Atomic: neither the script nor the email trigger was created.
+    let s = String::from_utf8(
+        common::pic_as(&env)
+            .args(["scripts", "ls", "--app", &slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        !s.contains("handler"),
+        "failed apply must leave nothing behind:\n{s}"
+    );
+}

crates/picloud-cli/tests/enabled.rs

@@ -0,0 +1,226 @@
+//! `enabled` three-state lifecycle, end to end: disabling a script via the
+//! manifest makes it non-invocable (404 on the execute-by-id bypass), and
+//! re-enabling restores it — proving the data path + runtime honoring.
+
+use std::fs;
+use std::path::Path;
+
+use serde_json::Value;
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::AppGuard;
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn disabling_a_script_makes_it_uninvocable_then_reenable() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("enabled");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/hello.rhai"), "\"hi\"").unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+    let manifest = |enabled_line: &str| {
+        format!(
+            "[app]\nslug = \"{slug}\"\nname = \"Enabled\"\n\n\
+             [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\n{enabled_line}"
+        )
+    };
+
+    // Active → invocable.
+    fs::write(&manifest_path, manifest("")).unwrap();
+    apply(&env, &manifest_path);
+    let id = script_id(&env, &slug, "hello");
+    assert_eq!(invoke_status(&env, &id), 200, "active script must invoke");
+
+    // Disabled → 404 (not invocable), but still deployed (re-pull would show it).
+    fs::write(&manifest_path, manifest("enabled = false\n")).unwrap();
+    apply(&env, &manifest_path);
+    assert_eq!(
+        invoke_status(&env, &id),
+        404,
+        "disabled script must 404 on execute-by-id"
+    );
+
+    // Re-enabled → invocable again.
+    fs::write(&manifest_path, manifest("enabled = true\n")).unwrap();
+    apply(&env, &manifest_path);
+    assert_eq!(
+        invoke_status(&env, &id),
+        200,
+        "re-enabled script must invoke"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn disabling_a_route_makes_it_404_then_reenable() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("enbl-route");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    // The app must claim a Host before its routes are reachable (two-phase
+    // dispatch: Host → app → route). A unique strict host avoids colliding
+    // with other tests' instance-global claims.
+    let host = format!("{slug}.test");
+    common::pic_as(&env)
+        .args(["apps", "domains", "add", &slug, &host])
+        .assert()
+        .success();
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/hello.rhai"), "\"hi\"").unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+    let manifest = |enabled_line: &str| {
+        format!(
+            "[app]\nslug = \"{slug}\"\nname = \"EnabledRoute\"\n\n\
+             [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\n\n\
+             [[routes]]\nscript = \"hello\"\nmethod = \"GET\"\n\
+             host_kind = \"any\"\npath_kind = \"exact\"\npath = \"/hello\"\n{enabled_line}"
+        )
+    };
+
+    // Active → the route serves.
+    fs::write(&manifest_path, manifest("")).unwrap();
+    apply(&env, &manifest_path);
+    assert_eq!(
+        route_status(&env, &host, "/hello"),
+        200,
+        "active route serves"
+    );
+
+    // Disabled → 404, indistinguishable from absent.
+    fs::write(&manifest_path, manifest("enabled = false\n")).unwrap();
+    apply(&env, &manifest_path);
+    assert_eq!(
+        route_status(&env, &host, "/hello"),
+        404,
+        "disabled route must 404"
+    );
+
+    // Re-enabled → serves again.
+    fs::write(&manifest_path, manifest("enabled = true\n")).unwrap();
+    apply(&env, &manifest_path);
+    assert_eq!(
+        route_status(&env, &host, "/hello"),
+        200,
+        "re-enabled route serves"
+    );
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn enabled_route_to_disabled_script_404s_flatly() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("enbl-os");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+    let host = format!("{slug}.test");
+    common::pic_as(&env)
+        .args(["apps", "domains", "add", &slug, &host])
+        .assert()
+        .success();
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/hello.rhai"), "\"hi\"").unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+    // Route stays enabled; the SCRIPT it binds is disabled (route-on/script-off).
+    fs::write(
+        &manifest_path,
+        format!(
+            "[app]\nslug = \"{slug}\"\nname = \"OS\"\n\n\
+             [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\nenabled = false\n\n\
+             [[routes]]\nscript = \"hello\"\nmethod = \"GET\"\n\
+             host_kind = \"any\"\npath_kind = \"exact\"\npath = \"/hello\"\n"
+        ),
+    )
+    .unwrap();
+    apply(&env, &manifest_path);
+
+    // 404, and the body must be the flat "no route matches" form — never the
+    // internal script id (no info leak; indistinguishable from absent).
+    let client = reqwest::blocking::Client::new();
+    let resp = client
+        .get(format!("{}/hello", env.url))
+        .header(reqwest::header::HOST, &host)
+        .send()
+        .unwrap();
+    assert_eq!(resp.status().as_u16(), 404);
+    let body = resp.text().unwrap();
+    assert!(body.contains("no route matches"), "flat 404 body: {body}");
+}
+
+fn apply(env: &common::TestEnv, manifest_path: &Path) {
+    common::pic_as(env)
+        .args(["apply", "--file"])
+        .arg(manifest_path)
+        .assert()
+        .success();
+}
+
+/// GET a user route under an explicit `Host` (the claimed domain) and return
+/// the HTTP status code.
+fn route_status(env: &common::TestEnv, host: &str, path: &str) -> u16 {
+    let client = reqwest::blocking::Client::new();
+    client
+        .get(format!("{}{path}", env.url))
+        .header(reqwest::header::HOST, host)
+        .send()
+        .unwrap()
+        .status()
+        .as_u16()
+}
+
+/// Resolve a script's id via the admin API (the manifest carries no ids).
+fn script_id(env: &common::TestEnv, slug: &str, name: &str) -> String {
+    let client = reqwest::blocking::Client::new();
+    let scripts: Vec<Value> = client
+        .get(format!("{}/api/v1/admin/scripts?app={slug}", env.url))
+        .bearer_auth(&env.token)
+        .send()
+        .unwrap()
+        .json()
+        .unwrap();
+    scripts
+        .into_iter()
+        .find(|s| s["name"] == name)
+        .and_then(|s| s["id"].as_str().map(String::from))
+        .expect("script id")
+}
+
+/// POST the execute-by-id bypass and return the HTTP status code.
+fn invoke_status(env: &common::TestEnv, id: &str) -> u16 {
+    let client = reqwest::blocking::Client::new();
+    client
+        .post(format!("{}/api/v1/execute/{id}", env.url))
+        .bearer_auth(&env.token)
+        .body("{}")
+        .send()
+        .unwrap()
+        .status()
+        .as_u16()
+}

crates/picloud-cli/tests/env_overlay.rs

@@ -0,0 +1,76 @@
+//! Env-scoped overlays (§4.1): `pic apply --env <E>` merges the sparse
+//! `picloud.<env>.toml` (per-env slug) onto the base and deploys to that
+//! environment's app — leaving the base app untouched.
+
+use std::fs;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::AppGuard;
+
+fn scripts_ls(env: &common::TestEnv, slug: &str) -> String {
+    String::from_utf8(
+        common::pic_as(env)
+            .args(["scripts", "ls", "--app", slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap()
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn apply_env_overlay_targets_the_env_app() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let base_slug = common::unique_slug("ovl");
+    let staging_slug = format!("{base_slug}-staging");
+    for s in [&base_slug, &staging_slug] {
+        common::pic_as(&env)
+            .args(["apps", "create", s])
+            .assert()
+            .success();
+    }
+    let _g1 = AppGuard::new(&env.url, &env.token, &base_slug);
+    let _g2 = AppGuard::new(&env.url, &env.token, &staging_slug);
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/hello.rhai"), "\"hi\"").unwrap();
+    let base = dir.path().join("picloud.toml");
+    fs::write(
+        &base,
+        format!(
+            "[app]\nslug = \"{base_slug}\"\nname = \"Ovl\"\n\n\
+             [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\n"
+        ),
+    )
+    .unwrap();
+    // Overlay redirects to the staging app.
+    fs::write(
+        dir.path().join("picloud.staging.toml"),
+        format!("[app]\nslug = \"{staging_slug}\"\n"),
+    )
+    .unwrap();
+
+    // Apply to staging only.
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&base)
+        .args(["--env", "staging"])
+        .assert()
+        .success();
+
+    assert!(
+        scripts_ls(&env, &staging_slug).contains("hello"),
+        "overlay apply must deploy to the staging app"
+    );
+    assert!(
+        !scripts_ls(&env, &base_slug).contains("hello"),
+        "the base app must be untouched by an --env apply"
+    );
+}

crates/picloud-cli/tests/init.rs

@@ -0,0 +1,99 @@
+//! `pic init` journey — offline scaffolding, no server/DB needed (so these
+//! tests are NOT gated on `DATABASE_URL` and never touch `common::fixture`).
+
+use std::fs;
+
+use assert_cmd::Command;
+use tempfile::TempDir;
+
+fn pic() -> Command {
+    Command::cargo_bin("pic").expect("pic binary")
+}
+
+#[test]
+fn init_scaffolds_a_deployable_project() {
+    let dir = TempDir::new().unwrap();
+    pic()
+        .current_dir(dir.path())
+        .args(["init", "demo-app"])
+        .assert()
+        .success();
+
+    let toml = fs::read_to_string(dir.path().join("picloud.toml")).unwrap();
+    assert!(toml.contains("slug = \"demo-app\""), "got:\n{toml}");
+    assert!(
+        toml.contains("name = \"Demo App\""),
+        "name defaults to title-cased slug:\n{toml}"
+    );
+    assert!(
+        dir.path().join("scripts/hello.rhai").exists(),
+        "scaffold writes the example script"
+    );
+    let gitignore = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
+    assert!(
+        gitignore.lines().any(|l| l.trim() == ".picloud/"),
+        "init must gitignore .picloud/:\n{gitignore}"
+    );
+}
+
+#[test]
+fn init_derives_slug_from_a_not_yet_created_dir() {
+    // The natural `pic init --dir new-project` flow: the target doesn't exist
+    // yet, and the slug is derived from its name (not via canonicalize, which
+    // would fail on a missing path).
+    let parent = TempDir::new().unwrap();
+    let target = parent.path().join("new-project");
+    pic()
+        .args(["init", "--dir"])
+        .arg(&target)
+        .assert()
+        .success();
+    let toml = fs::read_to_string(target.join("picloud.toml")).unwrap();
+    assert!(
+        toml.contains("slug = \"new-project\""),
+        "slug should derive from the dir name:\n{toml}"
+    );
+}
+
+#[test]
+fn init_refuses_to_overwrite_without_force() {
+    let dir = TempDir::new().unwrap();
+    pic()
+        .current_dir(dir.path())
+        .args(["init", "demo-app"])
+        .assert()
+        .success();
+    // A second run must refuse rather than clobber edits.
+    pic()
+        .current_dir(dir.path())
+        .args(["init", "demo-app"])
+        .assert()
+        .failure();
+    // `--force` overrides.
+    pic()
+        .current_dir(dir.path())
+        .args(["init", "demo-app", "--force"])
+        .assert()
+        .success();
+}
+
+#[test]
+fn init_appends_to_an_existing_gitignore_once() {
+    let dir = TempDir::new().unwrap();
+    fs::write(dir.path().join(".gitignore"), "target/\n").unwrap();
+    pic()
+        .current_dir(dir.path())
+        .args(["init", "demo-app"])
+        .assert()
+        .success();
+    let gitignore = fs::read_to_string(dir.path().join(".gitignore")).unwrap();
+    assert!(
+        gitignore.contains("target/"),
+        "must preserve existing rules"
+    );
+    assert_eq!(
+        gitignore.matches(".picloud/").count(),
+        1,
+        "must add the ignore exactly once:\n{gitignore}"
+    );
+}

crates/picloud-cli/tests/plan.rs

@@ -0,0 +1,81 @@
+//! `pic plan` journey: a freshly-pulled manifest must diff to all-no-op
+//! (pull→plan is idempotent), and editing a script source must surface
+//! as an update.
+
+use tempfile::TempDir;
+
+use crate::common;
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn plan_roundtrips_then_detects_change() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let (script_id, guard) = common::deploy_fixture(&env, "plan", "hello.rhai");
+    let app = guard.slug().to_string();
+
+    common::pic_as(&env)
+        .args([
+            "routes", "create", "--script", &script_id, "--path", "/p", "--method", "GET",
+        ])
+        .assert()
+        .success();
+
+    // Pull the live state, then plan it back — must be a clean no-op.
+    let dir = TempDir::new().expect("tempdir");
+    common::pic_as(&env)
+        .args(["pull", &app, "--dir"])
+        .arg(dir.path())
+        .assert()
+        .success();
+    let manifest = dir.path().join("picloud.toml");
+
+    let out = common::pic_as(&env)
+        .args(["plan", "--file"])
+        .arg(&manifest)
+        .output()
+        .expect("plan");
+    assert!(
+        out.status.success(),
+        "plan failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let stdout = String::from_utf8(out.stdout).unwrap();
+    let hello = stdout
+        .lines()
+        .find(|l| l.contains("hello"))
+        .unwrap_or_else(|| panic!("no hello row in plan:\n{stdout}"));
+    assert!(
+        hello.contains("noop"),
+        "expected hello no-op, got:\n{stdout}"
+    );
+    assert!(
+        !stdout.contains("create") && !stdout.contains("delete"),
+        "fresh pull should diff clean, got:\n{stdout}"
+    );
+
+    // Edit the script source on disk → plan must report an update.
+    std::fs::write(
+        dir.path().join("scripts/hello.rhai"),
+        "let body = #{ ok: false }; body",
+    )
+    .expect("rewrite source");
+    let out = common::pic_as(&env)
+        .args(["plan", "--file"])
+        .arg(&manifest)
+        .output()
+        .expect("plan after edit");
+    let stdout = String::from_utf8(out.stdout).unwrap();
+    let hello = stdout
+        .lines()
+        .find(|l| l.contains("hello"))
+        .unwrap_or_else(|| panic!("no hello row in plan:\n{stdout}"));
+    assert!(
+        hello.contains("update"),
+        "expected hello update after source edit, got:\n{stdout}"
+    );
+
+    drop(guard);
+}

crates/picloud-cli/tests/prune.rs

@@ -0,0 +1,177 @@
+//! `pic apply --prune` journey: a resource dropped from the manifest
+//! survives a plain (additive) apply but is deleted with `--prune`.
+
+use std::fs;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::AppGuard;
+
+fn scripts_ls(env: &common::TestEnv, slug: &str) -> String {
+    String::from_utf8(
+        common::pic_as(env)
+            .args(["scripts", "ls", "--app", slug])
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap()
+}
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn prune_deletes_stale_resources() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("prune");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/keep.rhai"), "let x = 1; x").unwrap();
+    fs::write(dir.path().join("scripts/drop.rhai"), "let y = 2; y").unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+
+    // v1: two scripts + a route on `drop`.
+    let v1 = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"Prune Test\"\n\n\
+         [[scripts]]\nname = \"keep\"\nfile = \"scripts/keep.rhai\"\n\n\
+         [[scripts]]\nname = \"drop\"\nfile = \"scripts/drop.rhai\"\n\n\
+         [[routes]]\nscript = \"drop\"\nmethod = \"GET\"\n\
+         host_kind = \"any\"\npath_kind = \"exact\"\npath = \"/drop\"\n"
+    );
+    fs::write(&manifest_path, &v1).unwrap();
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .assert()
+        .success();
+
+    // v2: drop `drop` and its route.
+    let v2 = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"Prune Test\"\n\n\
+         [[scripts]]\nname = \"keep\"\nfile = \"scripts/keep.rhai\"\n"
+    );
+    fs::write(&manifest_path, &v2).unwrap();
+
+    // Plain apply is additive — `drop` survives.
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .assert()
+        .success();
+    assert!(
+        scripts_ls(&env, &slug).contains("drop"),
+        "additive apply must not delete"
+    );
+
+    // Prune apply removes `drop` and its route. `--yes` skips the
+    // confirmation prompt (this test runs non-interactively).
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .args(["--prune", "--yes"])
+        .output()
+        .expect("apply --prune");
+    assert!(
+        out.status.success(),
+        "prune failed: {}",
+        String::from_utf8_lossy(&out.stderr)
+    );
+    let report = String::from_utf8(out.stdout).unwrap();
+    assert!(
+        report.contains("-1"),
+        "expected deletions in report:\n{report}"
+    );
+
+    let s = scripts_ls(&env, &slug);
+    assert!(!s.contains("drop"), "prune should delete `drop`:\n{s}");
+    assert!(s.contains("keep"), "prune must keep `keep`:\n{s}");
+
+    // Plan is clean after prune.
+    let p = String::from_utf8(
+        common::pic_as(&env)
+            .args(["plan", "--file"])
+            .arg(&manifest_path)
+            .output()
+            .unwrap()
+            .stdout,
+    )
+    .unwrap();
+    assert!(
+        !p.contains("delete"),
+        "plan should be clean after prune:\n{p}"
+    );
+}
+
+/// The prune confirmation gate: `--prune` without `--yes`, run
+/// non-interactively (no TTY, as in CI), must refuse and delete nothing.
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn prune_without_yes_refuses_noninteractively() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("prune-gate");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/keep.rhai"), "let x = 1; x").unwrap();
+    fs::write(dir.path().join("scripts/drop.rhai"), "let y = 2; y").unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+
+    // Deploy two scripts, then drop one from the manifest.
+    let v1 = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"Gate Test\"\n\n\
+         [[scripts]]\nname = \"keep\"\nfile = \"scripts/keep.rhai\"\n\n\
+         [[scripts]]\nname = \"drop\"\nfile = \"scripts/drop.rhai\"\n"
+    );
+    fs::write(&manifest_path, &v1).unwrap();
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .assert()
+        .success();
+    let v2 = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"Gate Test\"\n\n\
+         [[scripts]]\nname = \"keep\"\nfile = \"scripts/keep.rhai\"\n"
+    );
+    fs::write(&manifest_path, &v2).unwrap();
+
+    // `--prune` with no `--yes` and no TTY → refuse, non-zero exit.
+    let out = common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .arg("--prune")
+        .output()
+        .expect("apply --prune");
+    assert!(
+        !out.status.success(),
+        "prune without --yes must refuse non-interactively"
+    );
+    let err = String::from_utf8_lossy(&out.stderr);
+    assert!(
+        err.contains("--yes"),
+        "refusal should mention --yes:\n{err}"
+    );
+
+    // The dropped script must still be there — the gate blocked the delete.
+    let s = scripts_ls(&env, &slug);
+    assert!(
+        s.contains("drop"),
+        "refused prune must not delete anything:\n{s}"
+    );
+}

crates/picloud-cli/tests/pull.rs

@@ -0,0 +1,91 @@
+//! `pic pull` journey: stand up an app with a script, route, cron trigger,
+//! and a secret, then export it and assert the manifest + script file.
+
+use tempfile::TempDir;
+
+use crate::common;
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn pull_exports_manifest_and_sources() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+
+    // App + script "hello" (deploy derives the name from the file stem).
+    let (script_id, guard) = common::deploy_fixture(&env, "pull", "hello.rhai");
+    let app = guard.slug().to_string();
+
+    // Route → script.
+    common::pic_as(&env)
+        .args([
+            "routes", "create", "--script", &script_id, "--path", "/hook", "--method", "POST",
+        ])
+        .assert()
+        .success();
+
+    // Cron trigger → script.
+    common::pic_as(&env)
+        .args([
+            "triggers",
+            "create-cron",
+            "--app",
+            &app,
+            "--script",
+            &script_id,
+            "--schedule",
+            "0 0 * * * *",
+        ])
+        .assert()
+        .success();
+
+    // Secret (name only ends up in the manifest; value stays server-side).
+    common::pic_as(&env)
+        .args(["secrets", "set", "--app", &app, "api_key"])
+        .write_stdin("xyzzy")
+        .assert()
+        .success();
+
+    // Pull into a scratch dir.
+    let out_dir = TempDir::new().expect("pull tempdir");
+    common::pic_as(&env)
+        .args(["pull", &app, "--dir"])
+        .arg(out_dir.path())
+        .assert()
+        .success();
+
+    // Manifest exists and captures every resource.
+    let manifest = std::fs::read_to_string(out_dir.path().join("picloud.toml"))
+        .expect("picloud.toml should be written");
+    assert!(
+        manifest.contains(&format!("slug = \"{app}\"")),
+        "manifest missing app slug:\n{manifest}"
+    );
+    assert!(
+        manifest.contains("name = \"hello\"") && manifest.contains("scripts/hello.rhai"),
+        "manifest missing script entry:\n{manifest}"
+    );
+    assert!(
+        manifest.contains("[[routes]]") && manifest.contains("path = \"/hook\""),
+        "manifest missing route:\n{manifest}"
+    );
+    assert!(
+        manifest.contains("[[triggers.cron]]") && manifest.contains("schedule = \"0 0 * * * *\""),
+        "manifest missing cron trigger:\n{manifest}"
+    );
+    assert!(
+        manifest.contains("api_key"),
+        "manifest missing secret name:\n{manifest}"
+    );
+
+    // Script source was written out faithfully.
+    let src = std::fs::read_to_string(out_dir.path().join("scripts/hello.rhai"))
+        .expect("scripts/hello.rhai should be written");
+    assert!(
+        src.contains("hello from pic"),
+        "exported source mismatch:\n{src}"
+    );
+
+    drop(guard);
+}

crates/picloud-cli/tests/staleness.rs

@@ -0,0 +1,82 @@
+//! Bound-plan staleness: `pic plan` records a state token under `.picloud/`,
+//! and a later `pic apply` refuses (without `--force`) if the app changed
+//! out-of-band since the plan was reviewed.
+
+use std::fs;
+
+use tempfile::TempDir;
+
+use crate::common;
+use crate::common::cleanup::AppGuard;
+
+#[ignore = "needs DATABASE_URL pointing at a running Postgres"]
+#[test]
+fn apply_refuses_when_state_moved_since_plan() {
+    let Some(fx) = common::fixture_or_skip() else {
+        return;
+    };
+    let env = common::admin_env(fx);
+    let slug = common::unique_slug("stale");
+    common::pic_as(&env)
+        .args(["apps", "create", &slug])
+        .assert()
+        .success();
+    let _guard = AppGuard::new(&env.url, &env.token, &slug);
+
+    let dir = TempDir::new().unwrap();
+    fs::create_dir_all(dir.path().join("scripts")).unwrap();
+    fs::write(dir.path().join("scripts/hello.rhai"), "let x = 1; x").unwrap();
+    let manifest_path = dir.path().join("picloud.toml");
+    let manifest = format!(
+        "[app]\nslug = \"{slug}\"\nname = \"Stale\"\n\n\
+         [[scripts]]\nname = \"hello\"\nfile = \"scripts/hello.rhai\"\n"
+    );
+    fs::write(&manifest_path, &manifest).unwrap();
+
+    // Establish hello, then plan — the plan records the current state token.
+    apply(&env, &manifest_path).assert().success();
+    common::pic_as(&env)
+        .args(["plan", "--file"])
+        .arg(&manifest_path)
+        .assert()
+        .success();
+    assert!(
+        dir.path().join(".picloud/plan.json").exists(),
+        "plan must record the bound-plan token under .picloud/"
+    );
+
+    // Out-of-band change: deploy an extra script the manifest doesn't mention.
+    fs::write(dir.path().join("scripts/sneaky.rhai"), "let y = 2; y").unwrap();
+    common::pic_as(&env)
+        .args(["scripts", "deploy"])
+        .arg(dir.path().join("scripts/sneaky.rhai"))
+        .args(["--app", &slug])
+        .assert()
+        .success();
+
+    // Apply must now refuse — the live state no longer matches the plan.
+    let refused = apply(&env, &manifest_path).output().expect("apply");
+    assert!(
+        !refused.status.success(),
+        "apply must refuse after out-of-band change"
+    );
+    let err = String::from_utf8_lossy(&refused.stderr);
+    assert!(
+        err.contains("changed since") || err.contains("pic plan"),
+        "refusal should explain the staleness:\n{err}"
+    );
+
+    // `--force` bypasses the check and applies anyway.
+    common::pic_as(&env)
+        .args(["apply", "--file"])
+        .arg(&manifest_path)
+        .arg("--force")
+        .assert()
+        .success();
+}
+
+fn apply(env: &common::TestEnv, manifest_path: &std::path::Path) -> assert_cmd::Command {
+    let mut cmd = common::pic_as(env);
+    cmd.args(["apply", "--file"]).arg(manifest_path);
+    cmd
+}

crates/picloud/src/lib.rs

@@ -10,13 +10,13 @@ use axum::middleware::from_fn_with_state;
 use axum::{routing::get, Json, Router};
 use picloud_executor_core::{Engine, Limits};
 use picloud_manager_core::{
-    admin_router, admins_router, api_keys_router, app_members_router, apps_api, apps_router,
+    admin_router, admins_router, api_keys_router, app_members_router, apply_router, apps_api,
-    attach_principal_if_present, auth_router, compile_routes, dead_letters_router,
+    apps_router, attach_principal_if_present, auth_router, compile_routes, dead_letters_router,
     dev_emails_router, email_inbound_router, files_admin_router, kv_admin_router, migrations,
     require_authenticated, route_admin_router, secrets_router, topics_router, triggers_router,
     AbandonedRepo, AdminPrincipalResolver, AdminSessionRepository, AdminState, AdminUserRepository,
     AdminsState, ApiKeyRepository, ApiKeysState, AppDomainRepository, AppMembersRepository,
-    AppMembersState, AppRepository, AppsState, AuthState, AuthzRepo, DeadLetterRepo,
+    AppMembersState, AppRepository, ApplyService, AppsState, AuthState, AuthzRepo, DeadLetterRepo,
     DeadLettersState, DevEmailState, Dispatcher, DocsServiceImpl, EmailInboundState,
     EmailServiceImpl, FilesAdminState, FilesConfig, FilesServiceImpl, FsFilesRepo, HttpConfig,
     HttpServiceImpl, InboundNonceDedup, KvAdminState, KvServiceImpl, OutboxEventEmitter,
@@ -42,8 +42,8 @@ use picloud_orchestrator_core::{
 use picloud_shared::{
     DeadLetterService, DocsService, EmailService, ExecutionLogSink, FilesService, HttpService,
     InboxResolver, KvService, MasterKey, OutboxWriter, PubsubService, RealtimeAuthority,
-    RealtimeBroadcaster, ScriptValidator, SecretsService, ServiceEventEmitter, Services,
+    RealtimeBroadcaster, SecretsService, ServiceEventEmitter, Services, UsersService, API_VERSION,
-    UsersService, API_VERSION, PRODUCT_VERSION, SDK_VERSION, WIRE_VERSION,
+    PRODUCT_VERSION, SDK_VERSION, WIRE_VERSION,
 };
 use sqlx::postgres::PgPoolOptions;
 use sqlx::PgPool;
@@ -399,7 +399,7 @@ pub async fn build_app(
         logs: log_repo,
         apps: apps_repo.clone(),
         authz: authz.clone(),
-        validator: engine as Arc<dyn ScriptValidator>,
+        validator: engine.clone(),
         sandbox_ceiling: SandboxCeiling::from_env(),
     };
     let route_admin = RouteAdminState {
@@ -414,7 +414,7 @@ pub async fn build_app(
         resolver,
         log_sink,
         app_domains: app_domain_table.clone(),
-        routes: route_table,
+        routes: route_table.clone(),
         inbox: inbox_registry,
         outbox: outbox_writer,
     };
@@ -440,7 +440,7 @@ pub async fn build_app(
     // v1.1.4: cron scheduler. Polls cron_trigger_details on a tick and
     // enqueues due triggers into the outbox; the dispatcher above
     // delivers them like any other async trigger.
-    picloud_manager_core::spawn_cron_scheduler(pool, trigger_config.cron_tick_interval_ms);
+    picloud_manager_core::spawn_cron_scheduler(pool.clone(), trigger_config.cron_tick_interval_ms);
     // v1.1.6: GC empty realtime broadcast channels (one-shot subscribers)
     // and sweep orphaned `*.tmp.*` blobs left by crashed file writes.
     spawn_realtime_gc(broadcaster_concrete, DEFAULT_GC_INTERVAL_SECS);
@@ -453,6 +453,23 @@ pub async fn build_app(
         config: trigger_config,
         master_key: master_key.clone(),
     };
+    // Declarative reconcile engine (pic plan / apply). Trait-object repos
+    // for the read/diff path; shares the same handles as the CRUD routers.
+    let apply_service = ApplyService {
+        pool: pool.clone(),
+        scripts: script_repo.clone(),
+        routes: route_repo.clone(),
+        triggers: trigger_repo.clone(),
+        secrets: secrets_repo.clone(),
+        apps: apps_repo.clone(),
+        domains: domains_repo.clone(),
+        authz: authz.clone(),
+        validator: engine.clone(),
+        sandbox_ceiling: SandboxCeiling::from_env(),
+        trigger_config,
+        route_table: route_table.clone(),
+        master_key: master_key.clone(),
+    };
     // v1.1.9: keep a clone for the queues-api state (built later).
     let trigger_repo_for_queues = trigger_repo.clone();
     // v1.1.7 public inbound-email receiver. Outside the admin auth layer
@@ -562,6 +579,7 @@ pub async fn build_app(
         ))
         .merge(api_keys_router(api_keys_state))
         .merge(triggers_router(triggers_state))
+        .merge(apply_router(apply_service))
         .merge(picloud_manager_core::queues_api::queues_router(
             picloud_manager_core::queues_api::QueuesState {
                 queues: queue_repo.clone(),

crates/shared/src/lib.rs

@@ -39,6 +39,14 @@ pub mod users;
 pub mod validator;
 pub mod version;
 
+/// serde `default` for `enabled`-style boolean fields that should default to
+/// `true` when absent from the wire (bool's own `Default` is `false`). Shared
+/// by `Script`/`Route`, the apply `Bundle` types, and the CLI manifest.
+#[must_use]
+pub fn default_true() -> bool {
+    true
+}
+
 pub use app::{App, AppDomain, DomainShape};
 pub use auth::{AppRole, InstanceRole, Principal, Scope, UserId};
 pub use crypto::{decrypt, encrypt, CryptoError, EncryptResult, MasterKey, MasterKeyError};

crates/shared/src/route.rs

@@ -99,5 +99,11 @@ pub struct Route {
     #[serde(default)]
     pub dispatch_mode: DispatchMode,
 
+    /// Three-state lifecycle (§4.3): `false` means the route is deployed but
+    /// inert — it is dropped from the compiled match table, so a request 404s
+    /// indistinguishably from an absent route. Defaults `true`.
+    #[serde(default = "crate::default_true")]
+    pub enabled: bool,
+
     pub created_at: DateTime<Utc>,
 }

crates/shared/src/script.rs

@@ -122,6 +122,12 @@ pub struct Script {
     /// have to add it back when that's built.
     pub memory_limit_mb: u32,
 
+    /// Three-state lifecycle (§4.3): `false` means deployed-but-inert — the
+    /// script is not invocable (route 404s, trigger doesn't fire) but stays
+    /// as desired state (not pruned). Defaults `true`.
+    #[serde(default = "crate::default_true")]
+    pub enabled: bool,
+
     pub created_at: DateTime<Utc>,
     pub updated_at: DateTime<Utc>,
 }

docs/design/groups-and-project-tool.md

@@ -0,0 +1,877 @@
+# Groups, Projects & the Declarative Project Tool
+
+> **Status:** Draft — design discussion captured 2026-06-18, revised 2026-06-20 with resolutions
+> grounded in precedent (GitLab, Kustomize, Helm, Terraform, Kubernetes Server-Side Apply, Pulumi)
+> and corrected against the codebase after three independent review passes (consistency, gaps,
+> feasibility). Not yet scheduled into a phase.
+>
+> **Scope:** turns the imperative `pic` CLI into a declarative, file-based project tool, and
+> introduces a server-side **groups** hierarchy so apps can share and inherit config/scripts
+> without duplication.
+>
+> **Blueprint impact:** this reverses the §11.5 *snapshot-copy, not live-link* stance — top-down
+> hierarchical inheritance (group → app) is now in scope, implemented as a *materialized, auto-
+> invalidated* view (§5.1). Fold the outcome back into the blueprint before it drifts.
+
+---
+
+## 1. Motivation
+
+Today `pic` is **imperative**: every command (`pic deploy`, `pic routes create`, `pic triggers
+create-cron`, …) maps 1:1 to an admin API call. There is no memory of desired state, no project
+file, and no way to express "these N apps are the staging/prod/tenant variants of one thing."
+
+Two gaps follow:
+
+1. **No declarative project layer.** Developers re-run commands by hand; nothing reconciles a repo
+   to an instance.
+2. **No server-side notion of a project/group.** If we model environments as separate apps, the
+   *shared* base (config, scripts) is duplicated across every deployed app. The server sees N
+   unrelated apps.
+
+This document designs both halves: a **declarative manifest + project tool**, and a **groups
+hierarchy** on the server that the project tool projects onto the filesystem.
+
+`manager-core` is the single writer to Postgres. We lean on that for two concrete things — an
+**atomic desired-state write** (one DB transaction per apply, §4.2) and a **server-computed diff**
+(§4.2) — *not* for continuous reconciliation, which we deliberately decline (drift handling is
+detect-and-surface, §4.2). This is a narrower and more honest claim than "we can reconcile live
+state": most comparable tools cannot do an all-or-nothing apply because their effects are
+un-undoable cloud resources; ours are Postgres rows, so we can — within the boundary described in
+§4.2.
+
+---
+
+## 2. Vocabulary
+
+| Term | Meaning |
+|---|---|
+| **Group** | Server-side hierarchy node. Single-parent tree. Owns shared **code/config** definitions and members/roles. Nestable. |
+| **App** | A deployable unit. The data-isolation boundary (`app_id`). Lives under a group. |
+| **Environment** | A deployment variant (staging/production/…). **An environment *is* an app.** Also a *scope dimension* on config (§3). |
+| **Tenant** | Modeled like an environment — a scope dimension / overlay axis, **not** a second parent (§5.4). |
+| **Project** | *Not* a server entity. A CLI view over a repo-managed **subtree** of groups. |
+| **Definition** | Entities that can be group-owned: **scripts/modules, vars, secret-refs only**. Routes, triggers, collections, topics are **app-scoped** (§3, §5.1). |
+| **Data** | KV/docs/files collections + pub/sub topics. **Always app-owned** (`app_id`). The isolation boundary never moves. |
+| **Manifest** | TOML file describing desired state for one node (group base) + per-env overlays. |
+| **Effective view** | The materialized, per-app resolved set of definitions, served to the runtime and recomputed on any ancestor change (§5.1). |
+| **Attach point** | Where a local working tree's root binds into the server group tree. Its ceiling — you cannot apply above it. |
+
+Relationship: **the base + per-env-overlay "project" is the degenerate one-group subtree** of the
+general groups model. Nesting generalizes it to multi-group subtrees. Nothing about the
+single-project model is lost.
+
+---
+
+## 3. Configuration resolution
+
+This is the single rule the whole design hangs off; everything else (env vars, secrets, `enabled`,
+overrides) resolves through it. **All three precedent systems converge on the same shape, so we
+adopt it wholesale:**
+
+> ⚠️ **Net-new, not an extension (verified against the codebase).** PiCloud has **no env-agnostic
+> config / `vars` layer today** — this resolution engine and the `vars` table are greenfield. Only
+> `secrets` exists, and as a value-bearing table, not a ref. Read §3 as *new* infrastructure, not a
+> modification of something present.
+
+> **Resolution = sparse, per-field merge; environment is a *pre-filter*, not a precedence tier;
+> proximity wins across levels.**
+
+To resolve a key for an app in environment `E`:
+
+1. **Filter by environment first, per level.** A value scoped `@E` is eligible; a value scoped `*`
+   (env-agnostic) is the fallback. At a *single* level, `@E` beats `*`. Environment scope decides
+   *eligibility*, it is **not** a competing precedence rank.
+   *Evidence:* GitLab CI/CD variables use `environment_scope` exactly this way — a `production`-scoped
+   row simply isn't visible to a `staging` job ([GitLab CI/CD variables](https://docs.gitlab.com/ci/variables/)).
+2. **Then nearest level wins.** Walk up `acme → team-a → blog → app`; the closest level that defines
+   the (filtered) key wins. GitLab states this verbatim: *"if the same variable name exists in a
+   group and its subgroups, the job uses the value from the closest subgroup."*
+3. **Merge granularity:** maps/vars deep-merge **per key** (set `title`, still inherit `region`);
+   entities (scripts) replace **by identity** (name); deletion is an explicit tombstone.
+   *Evidence:* Kustomize strategic-merge patches are sparse and deep-merge maps but **replace lists
+   unless keyed** ([Kustomize patches](https://kubectl.docs.kubernetes.io/references/kustomize/kustomization/patches/));
+   Helm deep-merges nested maps and uses `null` to delete a defaulted key
+   ([Helm values](https://helm.sh/docs/chart_template_guide/values_files/)).
+
+This one rule resolves several issues at once: it makes **group config environment-scopable** (a
+group sets `db_url@production` and `db_url@staging`; descendants inherit the right one — no per-leaf
+duplication), it defines **merge granularity** (maps per-key, entities per-identity), and it makes
+**`enabled` just another sparse field** (§4.3).
+
+### 3.1 The env-scope manifest syntax
+
+```toml
+# group manifest (e.g. team-a/picloud.toml)
+[vars]
+region = "eu"                 # env-agnostic (scope *)
+
+[vars.production]             # scope @production
+db_url = "postgres://prod/..."
+[vars.staging]
+db_url = "postgres://staging/..."
+
+[secrets]
+names = ["stripe_key"]        # name-only; values pushed per env via CLI (§4.6)
+```
+
+### 3.2 The one deliberately-novel precedence call
+
+With `db_url@production` at the root **and** a plain `db_url` at the leaf, the **leaf's env-agnostic
+value wins** for a production app — proximity beats farther-level env-specificity. The research was
+explicit that **no major system lets env-specificity override proximity across levels**; GitLab
+structurally avoids the question by treating env as a per-level filter (above). So **proximity-first
+is the evidence-backed default**, and we document it as a chosen rule: *inner scope shadows outer,
+like lexical scoping.* To avoid this becoming a debugging nightmare at depth, `pic config
+--effective --explain` must show **why** a value resolved (which level/scope it came from).
+
+> **Residual risk (carried, not solved):** multi-level + environment scopes is genuinely novel
+> territory — GitLab is two-level (group→project). Proximity-first at arbitrary depth is consistent
+> and defensible but not battle-tested. The `--explain` tooling is a hard requirement, not a nicety.
+
+---
+
+## 4. Manifest & apply
+
+### 4.1 Manifest & layering
+
+- **TOML, data not code.** PiCloud runs untrusted scripts; the deployment descriptor stays inert,
+  diffable, server-validatable, and dashboard-renderable. Turing-completeness belongs in the Rhai
+  script, never the manifest.
+- **Base + overlays.** A node's `picloud.toml` holds the shared base; `picloud.<env>.toml` overlays
+  add per-env vars/secrets/slug. Shared config is written once; overlays only carry deltas (the
+  Kustomize base+overlay model — overlays are sparse patches, not replacements).
+- **Routes are top-level**, referencing scripts by name (one place to see all routing).
+- **`pull` is first-class** — see §4.6 for its inheritance/masking semantics.
+
+### 4.2 The apply engine
+
+The IaC research reshaped this section materially; the relevant precedents are cited inline.
+
+**Plan/apply split with a bound plan artifact.** `pic plan` computes a reviewable diff
+(`create / update / delete / replace`), the server persists it as a row, and `apply` executes
+*exactly that stored plan* — **detecting and refusing if state moved underneath it** (state
+version/serial check, cheap given the single writer). *Evidence:* Terraform's saved plan
+*"reliably perform[s] an exact set of pre-approved changes, even if the configuration or state has
+changed in the minutes since"* ([Terraform run](https://developer.hashicorp.com/terraform/cloud-docs/run)).
+The "state moved" check covers **both** the per-node **content** version **and** the **tree-structure**
+version (§6): a reparent or a new app created under the subtree between `plan` and `apply` changes the
+true blast radius, so the bound plan must refuse if *either* counter moved. Both are **net-new** — the
+existing `scripts.version` column is an unconditional write counter, **not** a compare-and-set guard,
+and must not be mistaken for one.
+
+**Atomicity — the corrected position.** No major IaC tool does transactional apply, *because their
+effects are un-undoable cloud resources* (you cannot un-create a VM). **That constraint does not
+bind us:** a PiCloud manifest apply is **almost entirely Postgres row writes** (script source,
+route/trigger defs, config). Therefore:
+
+- **The desired-state write is a single DB transaction** — all-or-nothing across the subtree. This
+  is achievable *because* our substrate is one transactional store (unlike Terraform's un-undoable
+  cloud resources). It removes the downstream-breakage hazard of an earlier draft (which proposed
+  per-node, stop-on-error apply — **now superseded**): there is no intermediate state where a parent
+  committed and a child did not.
+- **Propagation is forward-convergent, not transactional.** The post-commit effective-view refresh +
+  cache invalidation (§5.1) live *outside* the transaction. So "atomic" means
+  **atomic-for-desired-state, eventually-consistent-for-effect**, with a bounded window where the DB
+  says new and a cache says old.
+
+> **Invariant to enforce — and it does NOT hold today (verified).** The single-transaction model
+> requires apply to write **only Postgres**, but the *current* admin write paths violate that: route
+> and domain writes prime in-process caches **synchronously, interleaved with the write**
+> (`route_admin.rs:234`, `apps_api.rs:396`), and files fsync to disk. So moving to "commit the
+> transaction, *then* refresh views" is a deliberate **restructuring of manager-core**, not a
+> description of the status quo. Additionally, **domains** are a non-DB effect (Caddy/filesystem, out
+> of band) — they must be **excluded from the transactional core** and handled by the convergence
+> model below. Treat "apply writes only Postgres" as an invariant to *establish and guard*, not one
+> we already have.
+
+**Idempotent, convergent recovery (for the non-transactional propagation and any future side
+effects).** Operations are idempotent upserts keyed by stable identity; "re-apply converges" is the
+recovery story; "rollback" means revert the declaration and re-apply forward. Per-node status
+(applied/pending/failed/drifted) is recorded so a partial propagation failure is observable and
+re-runnable. *Evidence:* every surveyed tool (Terraform, ArgoCD, Flux, Pulumi) chose idempotent
+forward-only convergence over distributed rollback.
+
+**Drift model — upgraded from pure last-write-wins.** An earlier draft chose model "(c)":
+last-write-wins, just log. That is *exactly* the pre-SSA `kubectl apply` footgun — KEP-555 lists the
+scenario verbatim: *"User does an apply, then `kubectl edit`, then applies again: surprise!"*
+Kubernetes Server-Side Apply exists to convert that silent clobber into an explicit conflict
+([K8s SSA](https://kubernetes.io/docs/reference/using-api/server-side-apply/)). Concrete risk for us:
+CI runs `pic apply` and silently re-enables a route an operator killed in an emergency. Resolution:
+
+- **Keep (c)'s simplicity for ordinary fields** (last-write-wins, change reported).
+- **For the security-relevant subset only** — `enabled` and a secret's *reference/existence in
+  desired state* — track a "changed out-of-band" bit. If that subset was changed outside the manifest
+  (e.g. an operator disabled a route in the dashboard), `pic plan` surfaces it as a **labeled
+  conflict** and `apply` **refuses without `--force`**. This is a scoped version of SSA's field
+  ownership; we deliberately avoid full per-field managers (object bloat, complexity).
+- **Secret *values* are explicitly out of scope of the conflict/state-version machinery.** Values
+  are never in the manifest or the plan (§4.6), so a routine `pic secret set` between `plan` and
+  `apply` is the *supported* workflow and does **not** trip the state-version refusal — the
+  state-version check covers manifest-managed *desired state* (definitions, including secret
+  *references* and `enabled`), not value rotation.
+- Out-of-band changes render as a **distinct labeled diff** in `plan`, separate from intended
+  changes. *Evidence:* Terraform/Pulumi both make read-only drift detection default and label drift
+  distinctly ([Pulumi drift](https://www.pulumi.com/docs/iac/operations/stack-management/drift/)).
+
+> **Residual risk:** the conflict bit reintroduces some of the complexity (c) was chosen to avoid.
+> The cruder fallback, if even that is too much: keep pure (c) but add a non-revertible
+> **operational lock** flag an operator sets in an emergency that `apply` won't touch. Either closes
+> the silent-revert hole; neither is free.
+
+**Gating high-stakes applies: trigger ≠ authorization.** Any CI trigger may `plan`, but applying to
+a confirm-required env is a separate, default-off, explicitly-authorized step. A blanket `--yes`
+covers ordinary confirms; **confirm-required envs require an explicit per-env `--approve <env>`**, so
+CI must opt in per environment. "Override a gate" is its own audited capability (maps onto
+`manager-core::authz::can`). *Evidence:* Terraform Cloud parks runs in *Needs Confirmation* and
+separates the *apply runs* permission from *manage policy overrides*
+([TFC run states](https://developer.hashicorp.com/terraform/cloud-docs/run/states)).
+
+**Concurrency.** Start with a **coarse per-instance (or per-root-group) apply lock** — one apply at a
+time — which is trivially correct for the single-node MVP and makes "last-commit-wins" hold. Refine
+*later* to **per-blast-radius advisory locks** (lock the affected apps in `app_id` order so
+overlapping radii serialize while disjoint ones proceed; queue triggers already use this advisory-lock
+primitive) only if contention appears. Don't build the fine-grained version speculatively.
+
+**Blast radius — defined and bounded.** The blast radius is **the set of descendant apps whose
+materialized effective view would actually change** — a *diff*, not "all descendants." Per changed
+definition at node N it is `subtree(N)` minus apps that override that key nearer. `pic plan`
+enumerates it for small radii; for large ones it **summarizes (count + sample)** and a threshold
+triggers extra confirmation (`"this changes 4,213 apps — confirm"`). Because only scripts/vars/secret-
+refs inherit (§5.1), blast radius applies to *those* changes; an app-scoped change (a route or
+trigger) has blast radius = that single app. Root-level changes are accepted as expensive, rare, and
+high-confirmation; the computation is bounded by `subtree(N)` size, and the confirmed radius is
+re-validated at apply against the tree-structure version (above) so it can't go stale.
+
+**In-flight executions.** An apply that disables or replaces a script affects **new invocations
+immediately** (via the `enabled` re-check + view invalidation; pending trigger outbox rows are dropped
+at fire-time, §4.3 — so no separate outbox purge is needed). **In-flight *running* executions run to
+completion**, and note (verified) the executor has **no external-cancel path today**: executions are
+`spawn_blocking` Rhai calls interruptible only by their operation budget or a pre-set wall-clock
+deadline self-checked in `engine.on_progress`. A true must-stop-now **kill-switch** is therefore a
+*net-new* capability — buildable by having that same `on_progress` hook also poll a per-execution
+cancel flag — **gated by an admin capability (`authz::can`) and audited**, scheduled as a later item,
+not phase 1. Killing mid-run also risks partial side effects, so it stays the explicit extreme, never
+default apply behavior.
+
+**Apply flow (end to end):**
+
+```
+CLI builds bundle (manifests + script sources) for the subtree
+  → server computes plan + blast radius (incl. descendant apps in OTHER repos)
+  → server persists plan artifact; checks state version
+  → dev reviews; confirm/approve per env policy
+  → server applies desired state in ONE DB transaction (all-or-nothing)
+  → server refreshes effective views + bumps generation + invalidates caches (convergent)
+  → server returns change report; CLI logs created / updated / re-enabled / pruned / conflicts
+```
+
+### 4.3 The three-state `enabled` lifecycle
+
+`enabled` is a real platform feature (DB + server runtime + UI badge/toggle), not CLI sugar, and — by
+§3 — it is **just another sparse, proximity-resolved field**.
+
+| State | Meaning | Pruned? |
+|---|---|---|
+| declared, `enabled = true` (or omitted) | deployed, active | kept |
+| declared, `enabled = false` | deployed but **inert** (route short-circuits, trigger doesn't fire, script not invocable) | **kept** — still desired state |
+| absent from merged manifest | stale | **deleted** by prune / `--prune` |
+
+- Default `true`; last-write-wins on merge; a base `enabled = false` is inherited until an overlay
+  (env *or* a nearer group/app) explicitly sets `enabled = true`. Overriding a *different* field does
+  **not** implicitly re-enable.
+- **Across the group axis (resolved):** a descendant disables an inherited (group-owned) script via a
+  **sparse override** — an override row that sets only `enabled = false`, inheriting the source. A
+  descendant can re-enable a parent-disabled entity because nearest-level wins. This is the same
+  sparse-field mechanism as everything else (Kustomize patches are sparse — you specify only what
+  changes).
+- **Base = the superset across envs.** Per-env you may toggle `enabled` in *either* direction
+  (disable a base-active entity, or re-enable a base-disabled one — proximity wins, §3), but you
+  **cannot *remove* a base entity** in one env; true removal requires the entity not be in the base.
+  An entity unique to one env goes in that overlay; an entity present everywhere but off in staging
+  goes in base with `enabled = false` in the staging overlay.
+- **Disabled = invisible.** External callers hitting a disabled route get **404** (indistinguishable
+  from absent — no info leak).
+- **Schema note:** the `triggers` table **already** has `enabled` (+ `dispatch_mode`, retry columns)
+  and it is **honored at match/schedule time** (`trigger_repo.rs`, `cron_scheduler.rs` — verified).
+  New work is `enabled` on **scripts** and **routes** only, plus runtime honoring in the matcher /
+  invoker.
+- **Fire-time re-check (shipped, test-backed):** the dispatcher re-checks `enabled` at fire time on
+  every async path, so a pending event for a now-disabled trigger/script is dropped (not executed)
+  when it comes up — the §5.1 security-disable guarantee on the trigger path. **Outbox arm:**
+  `resolve_trigger` sets `active = trigger.enabled && script.enabled` and the async-HTTP/invoke
+  builders set `active = script.enabled`; `dispatch_one` drops the row via a single unified
+  `if !resolved.active` gate. **Queue arm:** `dispatch_one_queue` runs a separate path (it claims from
+  `queue_messages`, not the outbox), so in addition to `list_active_queue_consumers` filtering
+  `s.enabled`/`t.enabled` at list time, it re-checks the **freshly-read** `script.enabled` before
+  executing and **releases the claim (`nack`)** when disabled — closing the per-tick TOCTOU window
+  (a script disabled after the list snapshot but before its in-flight message runs). Both arms are
+  regression-locked: `dispatcher::tests::outbox_enabled_gate::disabled_http_outbox_row_is_dropped_without_executing`
+  covers the unified `!resolved.active` outbox gate, and
+  `dispatcher::tests::queue_enabled_gate::disabled_queue_consumer_releases_claim_without_executing`
+  covers the queue arm. **Same-app backstop:** every async build path (`resolve_trigger`,
+  `build_http_request`, `build_invoke_request`, `dispatch_one_queue`) rejects a row whose script's
+  `app_id` ≠ the row's `app_id`, so a hand-edited/restored row can't run one app's script under
+  another's `SdkCallCx` — the cross-app isolation boundary.
+- **Provenance caveat (accepted):** a single boolean carries no "manual vs manifest" provenance, so a
+  disabled entity looks the same however it got there — which is *why* the §4.2 conflict bit on the
+  `enabled`/secrets subset exists, to stop the next apply silently reverting an operational disable.
+
+### 4.4 Identity & naming
+
+- **Kebab everywhere.** One canonical identifier regex: `^[a-z0-9][a-z0-9-]{0,62}$` for project
+  names, env names, script names/slugs, trigger names — unified with the existing app-slug rule.
+- **Scripts:** unique `name`/`slug` per app = merge/upsert key.
+- **Routes:** identity = the triple **`<method> <host> <path>`**, e.g.
+  `ANY *.beta.example.com /hello/:name`. `dispatch_mode`, `host_param_name`, etc. are *attributes*
+  overridable without changing identity. The CLI infers `host_kind`/`path_kind` from the pattern
+  syntax (`*`, `{name}`, `:name`, exact), with an explicit `kind` key as override (mirrors the UI).
+  Needs a normalization rule (default method `ANY`, default host `*`, case-folding) so manifest ↔
+  server match exactly.
+- **Slugs are instance-global, derived from the path.** Two identifiers coexist:
+  - **path** = `acme/team-a/blog` — hierarchical, group-scoped, display/organization/RBAC.
+  - **slug** = flat, instance-global, the deployment key.
+  The derived default is **`{flattened-path}-{env}`** (e.g. `team-a-blog-staging`) — unique by
+  construction in the multi-group world, unlike a bare `{leaf}-{env}` which collides whenever two
+  groups reuse a leaf name. On >63-char overflow, truncate + short hash suffix. Explicit override
+  allowed; the path never *is* the slug, it only *seeds* it.
+
+### 4.5 Triggers
+
+Triggers are **app-scoped, not group-inherited** (§5.1 explains why). Two senses of "app" are in play
+and must not be conflated: a trigger is declared once in the **leaf (logical app)** base — an
+authoring convenience — and **materializes into per-`app_id` rows**, one per env-app, where `app_id`
+is the server-app isolation boundary (§5.2). It is never group-owned.
+
+Two distinct constraints:
+
+- **`name`** (explicit, kebab) = the merge/identity key + upsert target. Unique per app.
+  *(Triggers have no name column today — new.)*
+- **Semantic uniqueness** = a *post-merge validation*: no two triggers may share their kind-specific
+  semantic key. Checked after merging, so overriding a base trigger per-env (reuse `name`, change a
+  field) is fine; two differently-named triggers with identical effect is an error.
+
+| Kind | Semantic key | Note |
+|---|---|---|
+| kv / docs / files | `(script, collection_glob, ops)` | **canonicalize `ops`** (sort + dedupe) |
+| cron | `(script, schedule, timezone)` | exact TEXT match |
+| pubsub | `(script, topic_pattern)` | |
+| dead-letter | `(script, source_filter, trigger_id_filter, script_id_filter)` | NULL = literal value (not wildcard) for equality |
+| email | `(script)` | no filters — one per script |
+| **queue** | `(queue_name)` — **not** script-scoped | already server-enforced (advisory lock: one consumer per `(app_id, queue_name)`) |
+
+- **Matching vs identity:** `[]`/`NULL`/globs mean **"any/wildcard" at dispatch time** (unchanged
+  runtime matching) but are compared as **literal structural values for dedup**. We dedup on
+  **structural identity, never on overlap/subsumption** — `ops = []` and `ops = ["insert"]` are *not*
+  duplicates. Overlapping triggers coexist (multiple triggers firing on one event is already how
+  PiCloud works).
+- **Name backfill** for existing nameless rows: `{kind}-{entity}-{n}`, where `{entity}` is the kind's
+  identity token (collection / queue / topic; for cron/email/dead-letter fall back to `{kind}-{n}`),
+  sanitized to the kebab regex, with `-{n}` (discovery order) guaranteeing per-app uniqueness.
+
+> **Ergonomic debt (accepted, watch it):** because triggers don't inherit, 100 tenant apps each
+> needing the same 5 triggers = 500 declarations. The fix is group trigger/route **templates** that
+> fan out per descendant (a *template/instantiation* mechanism, not inheritance) — deferred, but it
+> bites early if tenant cardinality is high. Pressure-test against real tenant counts before
+> committing to the narrow-inheritance choice (§5.1).
+
+### 4.6 Secrets & `pull`
+
+- **Name-only in the manifest; value pushed via CLI** (`pic secret set`, reads stdin). Unanimous
+  across every comparable tool — never commit secret values.
+- **Env-scoped** like any var (`[secrets]` names declared once; values set per env).
+- **Warn (don't block) if a referenced secret is not set.** This requires an app dev to see that a
+  group secret **exists / is set** (a boolean) without reading its value — an accepted, explicit
+  authz boundary. GitLab surfaces inherited masked variable *keys* the same way.
+- **Email's `inbound_secret` is a reference**, not inline — same rule; the server already encrypts it
+  at rest.
+- **`pull` exports own-rows only** (this node's overrides), **never effective/inherited state.** A
+  separate read-only **`pic config --effective`** shows the inherited result with **masked secrets
+  rendered as `<set>` / `***`, never plaintext.** This makes pull-under-masking safe *by
+  construction* — you cannot pull a secret you cannot read, and you cannot accidentally duplicate
+  inherited config into a leaf. *Evidence:* GitLab shows inherited group variables in a project
+  read-only and separate from project-own variables.
+- Flat pull for a new project; "smart" delta-pull (own-vs-effective diff) is server-computed since an
+  app dev's checkout lacks ancestor manifests.
+
+### 4.7 Apply-time warnings
+
+- Enabled route/trigger pointing at a **disabled script**.
+- An **endpoint** script deployed with **no route and no trigger** (unreachable). Modules are exempt.
+- Sandbox override exceeding the admin **ceiling**.
+- Referenced secret not set.
+- An out-of-band change to the `enabled`/secrets subset (surfaced as a conflict, §4.2).
+
+---
+
+## 5. Groups & inheritance
+
+### 5.1 What inherits, and the runtime model
+
+- **GitLab-like, nested, single-parent tree.** Single parent keeps inheritance acyclic and
+  resolution deterministic (no diamond precedence).
+- **Inherit code/config — narrowly. Inherit data — no.** Group-inheritable = **scripts/modules,
+  vars, secret-refs only.** Routes, triggers, collections, topics, files are **app-scoped.**
+  - *Why narrow:* routes and triggers are **bindings**, not pure code/config. A group-owned trigger
+    has no app data to watch (triggers fire on app-owned collections/topics/queues); a group-owned
+    route has no host (routing is Host→app first). Inheriting them is incoherent. There are two
+    distinct sharing mechanisms and an earlier draft conflated them: the **leaf base+overlay** shares
+    routes/triggers across *one app's environments*; **group inheritance** shares *code/config across
+    many apps*. *Evidence:* Serverless/SAM keep `events:` per-function and share logic via *layers* —
+    bindings local, code shared.
+  - *Why data stays app-owned:* a group script executes in the *inheriting app's* context, so its
+    `cx.app_id` still scopes data to that app. Group-level *collections/topics* would break `app_id`
+    as the isolation boundary — that is the v1.3 cross-app data-sharing problem and stays **out**.
+- **Runtime model: a materialized effective view + versioned cache (not per-request live-resolve).**
+  An earlier draft said "live-resolve," which is underspecified and would fight the existing cache.
+  The real model:
+  - manager-core **resolves-at-write into a materialized per-app effective view** (§3 rule applied:
+    sparse merge, env filter, proximity, CoW, `enabled`).
+  - The orchestrator/executor serve from that view, **keyed by `app_id` + a generation/version**. The
+    app_id-keyed *shape* survives (today's route cache is already an `app_id`→routes map), **but the
+    substance is net-new (verified):** there is no generation counter anywhere today, the route cache
+    is rebuilt whole-table on every write rather than per-app, and script *bodies* are live-resolved
+    per request. Read "serve from it" as *extend the keying*, not *reuse the mechanism* — and note
+    that fanning out today's full-rescan invalidation to thousands of descendants would be a
+    regression, so per-app incremental invalidation is part of the build.
+  - On any write to a node, manager-core (single writer, knows the tree) **recomputes descendants'
+    views + bumps the generation + invalidates caches.**
+  - The materialized view is a **derived cache, not a second source of truth** — canonical config
+    still lives once at the owning node, so this does **not** reintroduce duplication. This dissolves
+    the long-running "snapshot vs. live" tension: no duplication *and* propagation *and* a fast hot
+    path.
+
+> **Residual risk (relocated, not solved):** cache invalidation is now a **correctness/security
+> requirement** — disabling a script for a security reason must stop it running *everywhere* within
+> bounded time. This is the same class as the existing PrincipalCache revocation lag. Require
+> **synchronous invalidation for the security-relevant subset** (`enabled=false`, secret rotation)
+> and accept bounded eventual staleness elsewhere; the hard SLA at fan-out to thousands of
+> descendants is genuinely unsolved.
+
+### 5.2 Schema impact
+
+- Inheritable definition kinds get a polymorphic owner (`owner_kind ∈ {group, app}` + `owner_id`):
+  **scripts** (modify the existing table — and note its `app_id` FK is `ON DELETE RESTRICT`, not
+  CASCADE, so a pruning apply needs explicit ordering), plus **`vars` and `secret-refs`, which are
+  net-new tables** (they do not exist today — §3). So this is *one table modified + two invented*,
+  not "three tables touched." **Routes, triggers, collections, topics, files stay strictly
+  `app_id`-owned**, so the runtime isolation boundary stays fixed. (The CLAUDE.md `app_id NOT NULL …
+  CASCADE` rule is itself not universal today — scripts already use RESTRICT.)
+- New: a `groups` table (single-parent, `parent_id`), group `membership`/roles, an `owner_project`
+  column on group nodes (§7), and the materialized effective-view store keyed by `app_id` +
+  generation (§5.1). Env-scoped values carry an `environment_scope` column (`*` or a specific env).
+
+### 5.3 RBAC
+
+- **Hierarchy-aware capabilities.** `authz::can(principal, cap, on=node)` resolves by walking
+  ancestors and taking the highest effective role. Instance → group(s) → app.
+- **Inherited membership** (GitLab-style): a group admin is implicitly admin of every subgroup/app
+  beneath it.
+- **Masked group secrets:** a group secret is *used by* an app at runtime but *not human-readable* by
+  the app's developers. Two orthogonal gates: **runtime resolution** (engine injects plaintext) vs
+  **human-read authz** (admin API returns a value only to a principal with rights at the *owning
+  group*). An app-scoped admin call never returns group secrets; runtime injection bypasses the human
+  gate. App devs may see a group secret **exists** (for the unset-warning, §4.6) but not its value.
+  **An app can run with config its own developers cannot see.**
+
+  > **Crypto caveat (verified):** secrets today are AES-GCM sealed with AAD `secret:{app_id}:{name}`,
+  > and the decrypt path hard-codes `cx.app_id` (migration 0042). A **group-owned** secret is *not
+  > expressible* under this scheme — there is no group identity in the AAD. It needs a new AAD
+  > identity (e.g. `secret:group:{group_id}:{name}`) **and** an owner-aware decrypt path that resolves
+  > whether the inherited secret is group- or app-owned. This is the single hardest correctness detail
+  > of group secrets and gates phasing step 3.
+
+### 5.4 Tenants & single-parent
+
+Single-parent forbids an app combining two *sibling* groups' configs — which seems to threaten the
+multi-tenant use case (shared-platform base + per-tenant overlay). It does not, because **the
+shared base is an *ancestor*, not a sibling:** model tenants as leaf apps under a `tenants/`
+subgroup that inherits the platform base up the chain (tenant leaf → `tenants` group → platform
+group). The "two parents" intuition is satisfied by the *chain*. Better still, **a tenant is a scope
+dimension like environment** (§3) — the same env-scope machinery generalizes to a `tenant` scope, so
+multi-tenant needs no new hierarchy primitive. *Evidence:* GitLab is single-parent and serves
+multi-tenant teams via subgroups; "combine two siblings" is handled by promoting shared config to a
+common ancestor.
+
+---
+
+### 5.5 Module & import resolution under inheritance
+
+A group-owned script may `import` modules, and inheritance makes "which module?" ambiguous (an
+inherited script importing a module a leaf has shadowed could bind to app-dev code it never saw — a
+trust inversion). The rule:
+
+- **Lexical by default.** An inherited script's imports resolve against the module set visible at the
+  **script's own defining node** (walking up from there), **not** the inheriting app's effective
+  view. A leaf cannot shadow a module an inherited group script depends on, and a group script
+  behaves identically across every app that inherits it — preserving both **determinism** and the
+  **trust boundary** (a security-authored shared script is tamper-proof from below). Ordinary lexical
+  scoping; "sealed by default", like non-`open` classes in Kotlin / `final` in Java.
+- **"Defining node" of a CoW-overridden script = the node that authored *that body*.** An
+  inherited-unchanged script's defining node is the ancestor that owns it (imports sealed to the
+  ancestor's modules). A script a leaf *overrides* (CoW, §3) has the **leaf** as its defining node, so
+  the override's imports resolve from the leaf — which is *not* a trust inversion, because the app
+  owner wrote that body and is trusted within their own app. Consequence (intended): the same module
+  name can resolve to **two different bodies** in one app, depending on the importing script's defining
+  node.
+- **Explicit extension points for opt-in polymorphism.** A module is marked an extension point in the
+  **manifest** (an `[extension_points]` declaration — *not* in Rhai source, keeping code inert and
+  giving the `plan` checker something to read). Such a module is one descendants are *expected* to
+  provide or override; **only** these resolve against the inheriting app's effective view. Controlled
+  template-method customization (a shared `render` whose `theme` module each tenant supplies) without
+  the blanket trust inversion of dynamic resolution. When a name is declared at more than one level —
+  or is concrete on one path and an extension point on another — **the nearest declaration's kind
+  wins** (proximity, §3); its default body (if any) is the inherited fallback.
+- **Apply-time checks:** a dangling import (inherited script → missing module) is a `plan` error; an
+  extension point with no provider in a given app is an error for that app (a hard failure, joining
+  §4.7).
+
+> **Residual (verified):** executor-core's `PicloudModuleResolver` is app-scoped today and ignores the
+> importing script's origin (`module_resolver.rs` passes `_source` unused). Rhai *does* expose that
+> origin, so the lexical-vs-dynamic split is expressible — but it requires re-keying the resolver cache
+> by owner identity and adding per-import policy (sealed vs. extension point), i.e. a real
+> resolver+cache redesign, not a parameter tweak. Lands with phasing step 4.
+
+### 5.6 Tree lifecycle: delete, reparent, rename
+
+Structural mutations of the group tree, which the rest of the design depends on staying acyclic and
+non-orphaning:
+
+- **Delete = RESTRICT, never implicit CASCADE.** Deleting a non-empty group is refused — an implicit
+  cascade would destroy descendant apps and their isolated data. CLI `--recursive` expands a delete
+  into *ordered, explicit, confirmed* child deletions; the DB FK stays RESTRICT. Corollary: a
+  referenced ancestor **cannot vanish while it has descendants**, so cross-repo read-only references
+  (§7) can't be orphaned by deletion — the RESTRICT protects them automatically. **App *data* is
+  destroyed only on explicit opt-in:** an app delete refuses unless `--purge-data`, which then removes
+  its KV/docs rows *and* its files blob tree under `PICLOUD_FILES_ROOT/<app_id>/` — a non-DB,
+  non-undoable effect run outside the transaction and logged. So `--recursive` group delete requires
+  `--purge-data` to touch any descendant app's data; without it, a non-empty app blocks the delete.
+- **Reparent / rename: the slug is frozen at creation.** The path only *seeds* the derived slug
+  (§4.4); a move or rename updates the **display path** but never rewrites the **instance-global
+  slug** — the deployment key stays stable, external references don't break. After a move the slug no
+  longer mirrors the path (cosmetic, accepted).
+- **Reparent recomputes descendant effective views** (it changes the resolution chain — the same
+  fan-out invalidation as a node write, §5.1) and is **doubly capability-gated**: group-admin at
+  *both* the source and destination parent (you remove from one ancestor's domain and add to
+  another's). Because it changes the resolution chain, a reparent is **validated like a plan and
+  refused (unless forced)** if the recompute would orphan a sparse `enabled`-override (now shadowing
+  nothing) or leave an extension point with no provider (§5.5) — a structural move must not silently
+  produce a state `apply` would have rejected.
+- **Cycle guard, under the apply lock.** Reparent runs an **ancestor-walk check in manager-core**
+  (walk from the destination up to root; reject if it reaches the node being moved). A Postgres
+  `CHECK` can't express this; the guard is what guarantees §9's "resolution always terminates."
+  Single-parent + this guard = acyclic. **All structural mutations (reparent/rename/delete) take the
+  same coarse apply-lock (§4.2)**, so the ancestor-walk + `parent_id` write run serialized — two
+  concurrent reparents can't race into a cycle, and a reparent's view-recompute can't collide with an
+  overlapping apply on the materialized-view store.
+
+## 6. The CLI ↔ server projection
+
+- **Directories = groups** (the hierarchy axis). Single-parent falls out of the filesystem for free.
+- **Overlay files = environments/apps** (the deployment-variant axis) — *not* subdirectories,
+  because envs share scripts/structure and only diverge on vars/secrets/slug; files structurally
+  prevent per-env script drift.
+- **`scripts/` at every level cascades** up the tree; nearer overrides farther by name (CoW).
+- **A leaf group = one logical app; its environments = the actual server apps.** Multiple distinct
+  apps = multiple sibling leaf groups. **Intermediate groups may also bear apps** (a dir may have
+  both subdirs and overlay files) — allowed, no special-casing.
+- **Environment registry lives at the app-bearing node**, but **confirm-policy is inheritable** (set
+  "production always confirms" once at root; it flows down via the §3 mechanism). Leaves may declare
+  independent environment sets; a tree-wide `pic apply --env production` simply skips a leaf that has
+  no `production`.
+- **Attach point:** the local root manifest declares where it binds into the server tree
+  (`parent_group = "acme"` or instance root). Ancestors above it are inherited/referenced but **not
+  present locally** — which makes the sparse checkout enforce the RBAC masking for free; effective
+  config needs a server round-trip (`pic config --effective`).
+- **Stable IDs in gitignored `.picloud/`** (group IDs, instance URL, token ref) so a directory
+  rename/move maps to a server **reparent**, not delete+create.
+- **Local/server structural divergence is detected, not silently fought.** Alongside the per-node
+  content version (§4.2), each node carries a **per-subtree structure version** (covering its own
+  parentage/subtree — **not one global counter**, so a structural edit in one team's subtree never
+  force-refuses an unrelated repo's plan). `pic plan` compares the local parent-by-ID (from
+  `.picloud/`) against the server's; on a structural mismatch (someone reparented server-side, or a
+  dir moved locally) it **refuses**, requiring an explicit `--adopt-server-structure` or
+  `--force-local-structure` (Terraform's detect-and-refuse on stale state). The content-version check
+  alone would miss a pure structural move; reparent/rename/delete each bump the affected subtree's
+  structure version.
+- **Mono-repo** = attach at instance root. **Per-team repo** = attach at a subgroup, contain only its
+  slice. Same model, different attach depth.
+
+---
+
+## 7. Ownership of shared nodes
+
+**Single-owner-per-node, ceilinged by the attach point.**
+
+1. Each group node is owned by exactly one **project-root** (the repo that *manages* it — contains
+   its manifest as something it applies, not merely references). The server records `owner_project`.
+2. **Your attach point is your ceiling** — you cannot apply to anything above your local root.
+3. **First apply claims; transfer is explicit and capability-gated.** A second repo applying to an
+   owned node is rejected (`owned by project X; use --takeover`); takeover needs group-admin
+   capability. This stops one team silently clobbering org-wide config — whose blast radius (via the
+   effective-view fan-out) is the whole subtree.
+4. **Ownership ⟂ RBAC.** Ownership = *which manifest is authoritative*; RBAC = *whether this
+   principal may*. The owner still needs group-admin capability to apply.
+5. A node with **no project claim is UI/API-owned** — the dashboard is its source of truth and no
+   manifest fights it. Every node is either manifest-owned (one repo) or UI-owned.
+
+**Corollary:** don't co-own a node — split config downward. Shared config lives *higher* (owned by a
+platform/shared repo attaching at root); team-specific bits go into subgroups each team owns.
+
+---
+
+## 8. Diagrams
+
+### 8.1 Server ownership & containment
+
+Only scripts/vars/secret-refs are group-ownable (polymorphic owner); routes/triggers/data are always
+app-owned via `app_id`.
+
+```mermaid
+graph TD
+    INST([Instance])
+    INST --> RG["Group: acme (root)"]
+    RG --> SGA["Group: team-a"]
+    RG --> SGB["Group: team-b"]
+    SGA --> LGA["Group: blog (leaf)"]
+    SGA --> LGB["Group: shop (leaf)"]
+    LGA --> APP1["App: blog-staging"]
+    LGA --> APP2["App: blog-production"]
+
+    RG -.->|"owns (shared)"| D0["scripts, vars, secret-refs"]
+    SGA -.->|owns| D1["team-a scripts, vars"]
+    APP2 -.->|"owns / overrides"| D2["app scripts, vars, secrets"]
+
+    APP1 ==>|app_id| C1[("routes, triggers, KV/Docs/Files, Topics")]
+    APP2 ==>|app_id| C2[("routes, triggers, KV/Docs/Files, Topics")]
+```
+
+### 8.2 Entity identity & cardinalities
+
+```mermaid
+erDiagram
+    GROUP ||--o{ GROUP : "parent-of (single parent)"
+    GROUP ||--o{ APP : contains
+    GROUP ||--o{ DEFINITION : "owns (scripts/vars/secret-refs)"
+    APP ||--o{ DEFINITION : "owns (override)"
+    APP ||--o{ APPSCOPED : "owns (routes/triggers/data, app_id)"
+    GROUP ||--o{ MEMBERSHIP : "role (inherited down)"
+    APP ||--o{ MEMBERSHIP : "role"
+```
+
+### 8.3 Config resolution (sparse merge, env filter, proximity-first)
+
+Effective value for one app+env, resolved by §3. Env-scope filters per level; nearest level wins;
+maps merge per key.
+
+```mermaid
+graph TB
+    I["Instance defaults"] --> RG["Group acme<br/>secret stripe_key (ref)<br/>script auth.rhai<br/>db_url@production"]
+    RG --> SG["Group team-a<br/>var region = eu"]
+    SG --> LG["Group blog<br/>script render.rhai<br/>var title = Blog"]
+    LG --> EV["Env overlay: production<br/>var title = Blog PROD (override)<br/>secret stripe_key = prod value"]
+    EV --> EFF[["Materialized effective view (app=blog-production):<br/>auth.rhai (acme), render.rhai (blog)<br/>title = Blog PROD (leaf overlay)<br/>region = eu (team-a)<br/>db_url = prod (acme @production filter)<br/>stripe_key = prod"]]
+```
+
+### 8.4 Filesystem ↔ server mapping
+
+```mermaid
+graph LR
+    subgraph FS["Git working tree"]
+      direction TB
+      R["acme/<br/>picloud.toml<br/>scripts/"]
+      R --> TA["team-a/<br/>picloud.toml<br/>scripts/"]
+      TA --> BL["blog/<br/>picloud.toml (base)<br/>picloud.staging.toml<br/>picloud.production.toml<br/>scripts/render.rhai"]
+      LINK[".picloud/ (gitignored)<br/>group IDs, instance URL, token ref"]
+    end
+    subgraph SRV["PiCloud server"]
+      direction TB
+      G0["Group acme"] --> G1["Group team-a"]
+      G1 --> G2["Group blog"]
+      G2 --> A1["App blog-staging"]
+      G2 --> A2["App blog-production"]
+    end
+    R -.->|defines| G0
+    TA -.->|defines| G1
+    BL -.->|"base defines"| G2
+    BL -.->|"staging.toml"| A1
+    BL -.->|"production.toml"| A2
+    LINK -.->|"stable IDs"| SRV
+```
+
+### 8.5 Multi-repo subtree views & single-owner ownership
+
+```mermaid
+graph TD
+    subgraph Server["Server group tree (authoritative)"]
+      acme["acme (root)"]
+      acme --> ta["team-a"]
+      acme --> tb["team-b"]
+      ta --> blog["blog"]
+      tb --> shop["shop"]
+    end
+    subgraph PR["Repo: platform"]
+      pr["manages acme"]
+    end
+    subgraph AR["Repo: team-a"]
+      ar["attaches at acme<br/>manages team-a + blog"]
+    end
+    pr ==>|owns| acme
+    ar ==>|owns| ta
+    ar ==>|owns| blog
+    ar -.->|"reference, read-only"| acme
+```
+
+### 8.6 Apply pipeline (bound plan → DB-atomic write → convergent propagation)
+
+```mermaid
+sequenceDiagram
+    actor Dev
+    participant CLI as pic CLI
+    participant Mgr as manager-core (single writer)
+    participant DB as Postgres
+    participant View as effective views + caches
+    Dev->>CLI: pic plan --env production
+    CLI->>CLI: read manifests + .rhai sources, build subtree bundle
+    CLI->>Mgr: send bundle (whole subtree)
+    Mgr->>DB: read state + version of affected nodes
+    Mgr->>Mgr: diff + blast radius + persist plan artifact
+    Mgr-->>CLI: PLAN (changes, N descendant apps incl. other repos, conflicts)
+    CLI-->>Dev: show plan; confirm/approve per env policy
+    Dev->>CLI: pic apply (executes stored plan)
+    CLI->>Mgr: apply (plan id)
+    Mgr->>DB: refuse if content or structure version moved; else ONE transaction (all-or-nothing)
+    Mgr->>View: recompute effective views + bump generation + invalidate
+    Mgr-->>CLI: change report (created/updated/re-enabled/pruned/conflicts)
+    CLI-->>Dev: log changes
+```
+
+### 8.7 RBAC: masked group secrets
+
+```mermaid
+graph TD
+    GA["Group admin (team-a)"] -->|"sets + can read"| GS["Group secret: stripe_key<br/>owned by team-a, encrypted at rest"]
+    AD["App developer (blog)"] -- "cannot read value (sees exists)" --x GS
+    AD -->|"can edit"| AS["App script source + app vars"]
+    GS ==>|"runtime injects plaintext"| EX["Executor: running app script"]
+    AS --> EX
+    GATE["Two gates:<br/>human-read authz  vs  runtime resolution"]
+    GATE -.-> GS
+    GATE -.-> EX
+```
+
+### 8.8 The three-state `enabled` lifecycle
+
+```mermaid
+stateDiagram-v2
+    [*] --> Active: declared (enabled=true/omitted)
+    Active --> Disabled: set enabled=false (manifest or UI toggle)
+    Disabled --> Active: set enabled=true (manifest or UI toggle)
+    Active --> Pruned: removed from manifest + prune/--prune
+    Disabled --> Pruned: removed from manifest + prune/--prune
+    Pruned --> [*]
+    note right of Disabled
+      Still desired state, NOT pruned.
+      Route 404s, trigger inert, script not invocable.
+      Out-of-band toggle on this field is conflict-guarded (4.2).
+    end note
+```
+
+---
+
+## 9. Adoption & backfill
+
+Groups land onto a live instance with existing flat apps, so a migration is a prerequisite, not an
+afterthought:
+
+- Create a **root group** (and/or a per-owner personal namespace, GitLab-style) and **reparent every
+  existing app** under it. Every app must have a parent from day one so resolution always terminates.
+- Existing apps have no group-owned definitions, so their effective view = their own rows — the
+  materialized-view store can be backfilled trivially (identity resolution).
+- The trigger `name` backfill (§4.5) runs in the same migration window.
+- Existing app slugs are already instance-global, so no slug rewrite is needed; the path is new
+  metadata layered on top.
+
+---
+
+## 10. Open questions & residual risks
+
+Resolved items now live inline next to their topic. What genuinely remains:
+
+- **Effective-view invalidation SLA (§5.1)** — the security-staleness guarantee at fan-out to many
+  descendants is unsolved; synchronous-for-security + eventual-elsewhere is the proposed shape, not a
+  proven one. Highest-risk open item.
+- **Conflict bit vs. operational lock (§4.2)** — *decided:* phase 1 ships the `enabled` / secret-
+  reference **conflict bit**; the operational-lock flag is the documented fallback if the bit proves
+  too heavy. (Was listed as undecided; resolved here to match §11 phase 1.)
+- **Multi-level env-scope precedence (§3.2)** — *decided default:* proximity-first with env as a
+  per-level filter. The open part is only *validation at depth*, which is why `pic config --effective
+  --explain` is a **phase-3 hard requirement** (when multi-level resolution first ships), not a
+  precondition to adopting the rule.
+- **Inherited-membership revocation lag (§5.3)** — revoking a group admin must drop implicit admin on
+  every descendant app, but §5.1's synchronous-invalidation subset covers only `enabled`/secrets, not
+  **role revocation** — leaving an unbounded window. New residual risk; should get the same
+  synchronous-for-security bound, lands with phase 3's authz.
+- **External execution cancel (§4.2 kill-switch)** — the executor has **no external-cancel path
+  today** (`spawn_blocking` + self-checked deadline, verified); the kill-switch is a net-new capability
+  (a cancel flag polled in `on_progress`), deferred past phase 1. Until it exists, the strongest stop
+  is op-budget/deadline + the dispatcher fire-time `enabled` re-check (§4.3) for the trigger path.
+- **Narrow-inheritance vs. trigger/route templates (§4.5, §5.1)** — the per-app binding tax bites
+  early at high tenant cardinality. Decide whether templates are truly deferrable for your target.
+- **`pull --factor`** — auto-extract a shared base by diffing two pulled envs (later nicety).
+
+---
+
+## 11. Suggested phasing
+
+> **Status (Phase 1): ✅ shipped.** `init`, `pull`, `config --effective`, manifest parse/validate,
+> `plan` + `apply` (single-transaction desired-state write, post-commit route refresh, domains/files
+> outside the tx), `prune`, secrets push, `.picloud/` link state, env-scoped overlays
+> (`picloud.<env>.toml` base+overlay), the three-state `enabled` lifecycle on scripts/routes (runtime
+> 404 + dispatcher fire-time re-check), the trigger `name` column + backfill, a coarse per-app apply
+> lock, and in-flight-finish-no-kill all landed.
+>
+> The **bound-plan staleness check** is the **content-fingerprint** form (a `state_token` hash of the
+> live state; apply 409s if it moved) rather than a persisted plan-artifact + serial — it delivers the
+> §4.2 guarantee without a migration or interactive-write-path changes, and the token covers
+> `enabled`/secret names so the **`enabled`/secrets conflict** is enforced on the plan→apply path.
+>
+> Deferred, with rationale: the **tree-structure version counter** is a no-op seam until groups exist
+> (Phase 2); **`vars`** + multi-level/`--explain` resolution and the richer per-env overlay are Phase 3;
+> trigger **upsert-by-name** (in-place Update; the diff still Create/Deletes by semantic identity) and
+> the **dashboard enabled toggle** are tracked follow-ups beyond the §11 bullets.
+
+1. **Declarative project tool, single-app (no groups yet).** `init`, `pull`/`config --effective`,
+   manifest parse/validate, `plan` (bound artifact), `apply` (**atomic desired-state write — requires
+   the manager-core post-commit-refresh restructuring of §4.2, domains/files excluded from the
+   transactional core**), `prune`, secrets push, link state, env-scoped config. Adds `enabled` to
+   scripts/routes + the three-state runtime + the dispatcher fire-time `enabled` re-check (§4.3) +
+   trigger `name` column/backfill + the `enabled`/secrets conflict bit + the net-new content +
+   tree-structure version counters + a coarse per-instance apply lock; in-flight executions finish (no
+   kill).
+2. **Groups as pure org/RBAC/UI container.** Nested groups (single-parent, `parent_id`, **delete =
+   RESTRICT**, reparent/rename with the **ancestor-walk cycle guard** + **slug-freeze** +
+   **tree-structure version**, §5.6), inherited membership, hierarchy-aware `can`, UI grouping, the §9
+   backfill. No shared resources yet — cheap, no data-plane schema change.
+3. **Group-inherited config** (vars, secret-refs, env-scoped). The net-new `vars`/`secret-refs`
+   tables + polymorphic owner; the group-secret AAD scheme (§5.3 caveat); masked group secrets; the
+   effective-view resolver + materialization + invalidation; **`config --effective --explain`** (hard
+   requirement, since multi-level resolution first ships here).
+4. **Group-inherited scripts/modules.** CoW overrides; the **scope-aware module/import resolver +
+   extension points** (§5.5); cache-invalidation fan-out hardening; versioning/pinning if needed.
+5. **Project tool maps onto groups.** Nested manifests, attach point, single-owner, server-computed
+   tree plan, per-env approval gating.
+6. **(Much later) group-level collections/topics** — the v1.3 cross-app data-sharing problem, with a
+   real shared-scope authz model. Optionally, trigger/route **templates** (§4.5) if cardinality
+   demands.
+
+---
+
+## 12. Contracts still to draft
+
+- The **apply bundle / plan artifact / change-report** wire contract (what the CLI ships, what the
+  server persists and returns), including the conflict and blast-radius shapes.
+- The **effective-view resolver** (the read primitive) — the §3 rule made executable, plus the
+  materialization + invalidation protocol (§5.1).
+- The **full manifest schema** spelling every block (scripts, routes, the 8 trigger kinds, storage
+  config, env-scoped vars, secret-refs, domains, `[project.environments]` + confirm policy).