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

@@ -16,7 +16,9 @@ use serde::Deserialize;
 use serde_json::json;
 
 use crate::app_repo::AppRepository;
-use crate::apply_service::{ApplyError, ApplyReport, ApplyService, Bundle, BundleTrigger, Plan};
+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`.
@@ -32,6 +34,10 @@ 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(
@@ -89,7 +95,13 @@ async fn apply_handler(
         .map_err(map_authz)?;
     }
     let report = svc
-        .apply(app_id, &req.bundle, req.prune, principal.user_id)
+        .apply(
+            app_id,
+            &req.bundle,
+            req.prune,
+            principal.user_id,
+            req.expected_token.as_deref(),
+        )
         .await?;
     Ok(Json(report))
 }
@@ -99,7 +111,7 @@ async fn plan_handler(
     Extension(principal): Extension<Principal>,
     Path(id_or_slug): Path<String>,
     Json(bundle): Json<Bundle>,
-) -> Result<Json<Plan>, ApplyError> {
+) -> 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
@@ -139,6 +151,7 @@ impl IntoResponse for ApplyError {
                 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");

crates/manager-core/src/apply_service.rs

@@ -90,6 +90,10 @@ pub struct BundleScript {
     pub memory_limit_mb: Option<i32>,
     #[serde(default)]
     pub sandbox: Option<ScriptSandbox>,
+    /// Three-state lifecycle (§4.3); omitted ⇒ active. Declarative (not
+    /// leave-as-is): a script absent the key is `enabled = true`.
+    #[serde(default = "picloud_shared::default_true")]
+    pub enabled: bool,
 }
 
 #[derive(Debug, Clone, Deserialize)]
@@ -107,6 +111,9 @@ pub struct BundleRoute {
     pub path: String,
     #[serde(default)]
     pub dispatch_mode: DispatchMode,
+    /// Three-state lifecycle (§4.3); omitted ⇒ active.
+    #[serde(default = "picloud_shared::default_true")]
+    pub enabled: bool,
 }
 
 /// Desired trigger, tagged by kind on the wire (`{ "kind": "cron", … }`).
@@ -312,6 +319,16 @@ impl Plan {
     }
 }
 
+/// What `plan` returns: the diff plus a fingerprint of the live state it was
+/// computed against. The token is flattened onto the plan JSON, so the wire
+/// shape stays `{ scripts, routes, triggers, secrets, state_token }`.
+#[derive(Debug, Clone, Serialize)]
+pub struct PlanResult {
+    #[serde(flatten)]
+    pub plan: Plan,
+    pub state_token: String,
+}
+
 // ----------------------------------------------------------------------------
 // Current state snapshot
 // ----------------------------------------------------------------------------
@@ -335,6 +352,12 @@ pub enum ApplyError {
     AppNotFound(String),
     #[error("invalid manifest: {0}")]
     Invalid(String),
+    #[error(
+        "live state changed since `pic plan` (someone edited this app's scripts, \
+         routes, triggers, or secrets); re-run `pic plan` to review, then apply — \
+         or `pic apply --force` to skip the check"
+    )]
+    StateMoved,
     #[error("forbidden")]
     Forbidden,
     #[error("authorization repo error: {0}")]
@@ -347,8 +370,8 @@ pub enum ApplyError {
 // Service
 // ----------------------------------------------------------------------------
 
-/// Reconcile engine. Holds trait-object repos for the read/diff path;
+/// Reconcile engine. Holds trait-object repos for the read/diff path; the
-/// the transactional write path (next milestone) reuses the same handles.
+/// transactional write path (`apply`) reuses the same handles plus `pool`.
 #[derive(Clone)]
 pub struct ApplyService {
     /// Pool for the transactional write path (`apply`).
@@ -377,17 +400,22 @@ impl ApplyService {
     ///
     /// # Errors
     /// `Invalid` for a malformed bundle; `Backend` for repo failures.
-    pub async fn plan(&self, app_id: AppId, bundle: &Bundle) -> Result<Plan, ApplyError> {
+    pub async fn plan(&self, app_id: AppId, bundle: &Bundle) -> Result<PlanResult, ApplyError> {
         self.validate_bundle(bundle)?;
         self.validate_route_hosts(app_id, bundle).await?;
         let current = self.load_current(app_id).await?;
         validate_email_secrets_present(bundle, &current.secret_names)?;
-        Ok(compute_diff(&current, bundle))
+        Ok(PlanResult {
+            plan: compute_diff(&current, bundle),
+            // Fingerprint of the live state this plan was computed against, so
+            // `apply` can refuse if the app changed underneath it (§4.2).
+            state_token: state_token(&current),
+        })
     }
 
-    /// Reconcile app `app_id` to `bundle` in a single transaction.
+    /// Reconcile app `app_id` to `bundle` in a single transaction. Creates
-    /// Additive: creates and updates are applied; deletions are surfaced
+    /// and updates are always applied; deletions of resources absent from the
-    /// in the diff but only executed when prune is set (next milestone).
+    /// bundle are executed only when `prune` is set (secrets are never pruned).
     ///
     /// # Errors
     /// `Invalid` for a bad bundle; `Backend` for repo/transaction failures.
@@ -398,6 +426,7 @@ impl ApplyService {
         bundle: &Bundle,
         prune: bool,
         actor: AdminUserId,
+        expected_token: Option<&str>,
     ) -> Result<ApplyReport, ApplyError> {
         self.validate_bundle(bundle)?;
         self.validate_route_hosts(app_id, bundle).await?;
@@ -422,11 +451,31 @@ impl ApplyService {
             .map_err(|e| ApplyError::Backend(e.to_string()))?;
 
         let current = self.load_current(app_id).await?;
+        // Bound-plan check (§4.2): if the caller passed the token from a prior
+        // `pic plan`, refuse when the live state changed since. Computed from
+        // the same `load_current` snapshot the diff uses, before any mutation
+        // (so a stale apply rolls back nothing). NOTE: like the diff, this read
+        // is on the pool, not `&mut *tx` (see the lock comment above), so it
+        // catches drift between plan and apply but shares that same narrow
+        // apply-vs-interactive-write window — it is not a substitute for the
+        // tx-scoped read the follow-up will add.
+        if let Some(expected) = expected_token {
+            if state_token(&current) != expected {
+                return Err(ApplyError::StateMoved);
+            }
+        }
         // Surface a missing email-secret reference here so `plan` and `apply`
         // agree, rather than only failing deep in `resolve_and_seal` below.
         validate_email_secrets_present(bundle, &current.secret_names)?;
         let plan = compute_diff(&current, bundle);
         let mut report = ApplyReport::default();
+        // §4.7 warning: an enabled binding pointing at a disabled script is
+        // deployed-but-unreachable. Not an error (it's valid desired state),
+        // but surfaced so the operator isn't surprised by a silent 404.
+        report.warnings.extend(disabled_target_warnings(bundle));
+        report
+            .warnings
+            .extend(unreachable_endpoint_warnings(bundle));
 
         let bundle_scripts: HashMap<String, &BundleScript> = bundle
             .scripts
@@ -460,6 +509,7 @@ impl ApplyService {
                         timeout_seconds: bs.timeout_seconds,
                         memory_limit_mb: bs.memory_limit_mb,
                         sandbox: bs.sandbox,
+                        enabled: bs.enabled,
                         imports: self.script_imports(bs)?,
                     };
                     let created = insert_script_tx(&mut tx, &new).await.map_err(map_repo)?;
@@ -471,12 +521,17 @@ impl ApplyService {
                     let cur = current_scripts[&key];
                     let patch = ScriptPatch {
                         name: None,
-                        description: Some(bs.description.clone()),
+                        // Sparse: `None` (omitted in the manifest) leaves the
+                        // stored description untouched; `Some(text)` sets it.
+                        // Mirrors `script_update_reason`'s leave-as-is rule.
+                        description: bs.description.clone().map(Some),
                         source: Some(bs.source.clone()),
                         timeout_seconds: bs.timeout_seconds,
                         memory_limit_mb: bs.memory_limit_mb,
                         sandbox: bs.sandbox,
                         kind: Some(bs.kind),
+                        // Declarative: always reconcile to the manifest's value.
+                        enabled: Some(bs.enabled),
                         imports: Some(self.script_imports(bs)?),
                     };
                     update_script_tx(&mut tx, cur.id, &patch)
@@ -831,6 +886,15 @@ impl ApplyService {
                 )));
             }
             let res = if s.kind == ScriptKind::Module {
+                // Parity with the interactive create path (`api.rs`): a module
+                // may not shadow a built-in SDK namespace, or `import "kv"`
+                // could resolve to a user module instead of the real bridge.
+                if crate::api::RESERVED_MODULE_NAMES.contains(&s.name.as_str()) {
+                    return Err(ApplyError::Invalid(format!(
+                        "script `{}`: reserved module name (shadows a built-in SDK namespace)",
+                        s.name
+                    )));
+                }
                 self.validator.validate_module(&s.source)
             } else {
                 self.validator.validate(&s.source)
@@ -909,49 +973,7 @@ impl ApplyService {
                     t.script()
                 )));
             }
-            match t {
+            validate_trigger_shape(t)?;
-                BundleTrigger::Email {
-                    inbound_secret_ref, ..
-                } if inbound_secret_ref.trim().is_empty() => {
-                    return Err(ApplyError::Invalid(format!(
-                        "email trigger on `{}` needs inbound_secret_ref (a secret name)",
-                        t.script()
-                    )));
-                }
-                BundleTrigger::Queue {
-                    queue_name,
-                    visibility_timeout_secs,
-                    ..
-                } => {
-                    if queue_name.trim().is_empty() {
-                        return Err(ApplyError::Invalid("queue trigger needs queue_name".into()));
-                    }
-                    if let Some(v) = visibility_timeout_secs {
-                        if !(5..=3600).contains(v) {
-                            return Err(ApplyError::Invalid(format!(
-                                "queue `{queue_name}`: visibility_timeout_secs must be in [5, 3600]"
-                            )));
-                        }
-                    }
-                }
-                BundleTrigger::Cron {
-                    schedule, timezone, ..
-                } => {
-                    crate::cron_scheduler::validate_schedule(schedule).map_err(|e| {
-                        ApplyError::Invalid(format!(
-                            "cron trigger on `{}`: invalid schedule: {e}",
-                            t.script()
-                        ))
-                    })?;
-                    crate::cron_scheduler::validate_timezone(timezone).map_err(|e| {
-                        ApplyError::Invalid(format!(
-                            "cron trigger on `{}`: invalid timezone: {e}",
-                            t.script()
-                        ))
-                    })?;
-                }
-                _ => {}
-            }
             let id = t.identity();
             if !trig_keys.insert(id.clone()) {
                 return Err(ApplyError::Invalid(format!("duplicate trigger `{id}`")));
@@ -1084,8 +1106,21 @@ fn script_update_reason(cur: &Script, desired: &BundleScript) -> Option<String>
     if cur.kind != desired.kind {
         return Some("kind changed".into());
     }
-    if cur.description != desired.description {
+    // Declarative (default true): always reconcile to the manifest's value.
-        return Some("description changed".into());
+    if cur.enabled != desired.enabled {
+        return Some(if desired.enabled {
+            "enabled".into()
+        } else {
+            "disabled".into()
+        });
+    }
+    // Sparse like the other optional fields: an omitted (`None`) description
+    // means "leave as-is", so only an explicitly-set value that differs
+    // forces an update. (Clearing a description is done in the dashboard.)
+    if let Some(d) = &desired.description {
+        if cur.description.as_ref() != Some(d) {
+            return Some("description changed".into());
+        }
     }
     if let Some(t) = desired.timeout_seconds {
         if i64::from(cur.timeout_seconds) != i64::from(t) {
@@ -1160,6 +1195,7 @@ fn diff_routes(
                 if cur_script != Some(r.script.as_str())
                     || cur.dispatch_mode != r.dispatch_mode
                     || cur.host_param_name != r.host_param_name
+                    || cur.enabled != r.enabled
                 {
                     out.push(ResourceChange {
                         op: Op::Update,
@@ -1286,6 +1322,12 @@ fn diff_secrets(current: &CurrentState, bundle: &Bundle) -> Vec<ResourceChange>
 /// Reject any email trigger whose referenced secret isn't set, so `plan` and
 /// `apply` give the same answer (apply also fails in `resolve_and_seal`, but
 /// only after taking the lock — surfacing it here keeps plan honest).
+///
+/// This checks the secret *reference exists*. `resolve_and_seal` additionally
+/// requires the resolved value to be a non-empty string; `plan` deliberately
+/// does not decrypt secrets, so a reference to a set-but-empty secret passes
+/// `plan` and is caught at `apply` — the one residual plan/apply divergence,
+/// and a deliberate one (plan stays read-only and crypto-free).
 fn validate_email_secrets_present(
     bundle: &Bundle,
     secret_names: &[String],
@@ -1396,6 +1438,156 @@ fn reject_reserved_path(path: &str) -> Result<(), ApplyError> {
     Ok(())
 }
 
+/// §4.7 reachability warning: an *enabled* endpoint script with no route and
+/// no trigger has no event surface — it's only reachable via the
+/// execute-by-id bypass / `invoke()`. Modules are exempt (never invoked
+/// directly); disabled endpoints are intentionally inert, so skip them.
+fn unreachable_endpoint_warnings(bundle: &Bundle) -> Vec<String> {
+    let bound: HashSet<&str> = bundle
+        .routes
+        .iter()
+        .map(|r| r.script.as_str())
+        .chain(bundle.triggers.iter().map(BundleTrigger::script))
+        .collect();
+    bundle
+        .scripts
+        .iter()
+        .filter(|s| s.kind == ScriptKind::Endpoint && s.enabled && !bound.contains(s.name.as_str()))
+        .map(|s| {
+            format!(
+                "endpoint `{}` has no route or trigger — only reachable via the \
+                 execute-by-id bypass / invoke()",
+                s.name
+            )
+        })
+        .collect()
+}
+
+/// §4.7 reachability warnings: an *enabled* route or trigger bound to a
+/// script the manifest marks *disabled* is deployed but unreachable (the
+/// route 404s, the trigger won't fire). Valid desired state, so a warning —
+/// not an error — keyed on the bundle alone.
+fn disabled_target_warnings(bundle: &Bundle) -> Vec<String> {
+    let disabled: HashSet<&str> = bundle
+        .scripts
+        .iter()
+        .filter(|s| !s.enabled)
+        .map(|s| s.name.as_str())
+        .collect();
+    if disabled.is_empty() {
+        return Vec::new();
+    }
+    let mut out = Vec::new();
+    for r in &bundle.routes {
+        if r.enabled && disabled.contains(r.script.as_str()) {
+            out.push(format!(
+                "route `{}` is enabled but its script `{}` is disabled — it will 404",
+                route_key(
+                    r.method.as_deref(),
+                    r.host_kind,
+                    &r.host,
+                    r.path_kind,
+                    &r.path
+                ),
+                r.script
+            ));
+        }
+    }
+    for t in &bundle.triggers {
+        if disabled.contains(t.script()) {
+            out.push(format!(
+                "{} trigger targets disabled script `{}` — it will not fire",
+                t.kind_str(),
+                t.script()
+            ));
+        }
+    }
+    out
+}
+
+/// Per-kind structural validation for one desired trigger — kept at parity
+/// with the interactive trigger API so `apply` can't write a trigger the
+/// dashboard would have rejected. Pure (no live state), so it's unit-tested
+/// directly. The script-binding checks live in `validate_bundle`.
+fn validate_trigger_shape(t: &BundleTrigger) -> Result<(), ApplyError> {
+    match t {
+        BundleTrigger::Email {
+            inbound_secret_ref, ..
+        } if inbound_secret_ref.trim().is_empty() => {
+            return Err(ApplyError::Invalid(format!(
+                "email trigger on `{}` needs inbound_secret_ref (a secret name)",
+                t.script()
+            )));
+        }
+        BundleTrigger::Queue {
+            queue_name,
+            visibility_timeout_secs,
+            ..
+        } => {
+            if queue_name.trim().is_empty() {
+                return Err(ApplyError::Invalid("queue trigger needs queue_name".into()));
+            }
+            if let Some(v) = visibility_timeout_secs {
+                // Parity with the interactive API: the floor is the dispatcher
+                // tick/reclaim-cadence minimum (`triggers_api`), and the ceiling
+                // matches `trigger_repo`'s queue check. Apply previously allowed
+                // a [5, 29] floor the dashboard rejects.
+                let min = crate::triggers_api::MIN_QUEUE_VISIBILITY_TIMEOUT_SECS;
+                if !(min..=3600).contains(v) {
+                    return Err(ApplyError::Invalid(format!(
+                        "queue `{queue_name}`: visibility_timeout_secs must be in [{min}, 3600]"
+                    )));
+                }
+            }
+        }
+        BundleTrigger::Cron {
+            schedule, timezone, ..
+        } => {
+            crate::cron_scheduler::validate_schedule(schedule).map_err(|e| {
+                ApplyError::Invalid(format!(
+                    "cron trigger on `{}`: invalid schedule: {e}",
+                    t.script()
+                ))
+            })?;
+            crate::cron_scheduler::validate_timezone(timezone).map_err(|e| {
+                ApplyError::Invalid(format!(
+                    "cron trigger on `{}`: invalid timezone: {e}",
+                    t.script()
+                ))
+            })?;
+        }
+        // Parity with the interactive trigger API, which rejects an empty
+        // collection_glob (`create_{kv,docs,files}_trigger`).
+        BundleTrigger::Kv {
+            collection_glob, ..
+        }
+        | BundleTrigger::Docs {
+            collection_glob, ..
+        }
+        | BundleTrigger::Files {
+            collection_glob, ..
+        } if collection_glob.trim().is_empty() => {
+            return Err(ApplyError::Invalid(format!(
+                "{} trigger on `{}` needs a non-empty collection_glob",
+                t.kind_str(),
+                t.script()
+            )));
+        }
+        // Parity with `create_pubsub_trigger`'s topic-pattern check — otherwise
+        // a malformed pattern is written and silently never matches at dispatch.
+        BundleTrigger::Pubsub { topic_pattern, .. } => {
+            picloud_shared::validate_topic_pattern(topic_pattern).map_err(|e| {
+                ApplyError::Invalid(format!(
+                    "pubsub trigger on `{}`: invalid topic_pattern: {e}",
+                    t.script()
+                ))
+            })?;
+        }
+        _ => {}
+    }
+    Ok(())
+}
+
 /// Summary of what an `apply` changed.
 #[derive(Debug, Default, Serialize)]
 pub struct ApplyReport {
@@ -1437,6 +1629,7 @@ async fn insert_bundle_route(
         path: br.path.clone(),
         method: br.method.clone(),
         dispatch_mode: br.dispatch_mode,
+        enabled: br.enabled,
     };
     insert_route_tx(tx, &new).await.map_err(map_repo)?;
     Ok(())
@@ -1509,6 +1702,84 @@ fn map_trig(e: TriggerRepoError) -> ApplyError {
     }
 }
 
+/// A stable fingerprint of an app's live state, covering exactly what the
+/// reconcile diff keys on: script identity + write-counter (`version` bumps on
+/// any script edit), route identity + binding/attrs, trigger semantic identity
+/// (via `current_trigger_identity`, so dead-letter triggers — which the diff
+/// ignores — are excluded), and secret names. Any out-of-band change to those
+/// flips the token, so `plan`-then-`apply` can detect the app moving underneath.
+///
+/// Deliberately mirrors the diff's inputs so a mismatch means the *reviewed
+/// plan* would now differ — not merely that some unrelated row changed.
+///
+/// Uses FNV-1a (deterministic across process restarts, unlike `DefaultHasher`'s
+/// per-process seed) so a token stored by `pic plan` still matches on a later
+/// `pic apply`. A hash collision can only ever yield a false "unchanged", which
+/// is the same risk class as not checking at all — never a false refusal.
+#[must_use]
+pub fn state_token(current: &CurrentState) -> String {
+    let mut parts: Vec<String> = Vec::with_capacity(
+        current.scripts.len()
+            + current.routes.len()
+            + current.triggers.len()
+            + current.secret_names.len(),
+    );
+    for s in &current.scripts {
+        parts.push(format!(
+            "s|{}|{}|{}",
+            s.name.to_lowercase(),
+            s.version,
+            s.enabled
+        ));
+    }
+    for r in &current.routes {
+        parts.push(format!(
+            "r|{}|{:?}|{:?}|{}|{}",
+            route_key(
+                r.method.as_deref(),
+                r.host_kind,
+                &r.host,
+                r.path_kind,
+                &r.path
+            ),
+            r.script_id,
+            r.dispatch_mode,
+            r.host_param_name.as_deref().unwrap_or(""),
+            r.enabled,
+        ));
+    }
+    // Mirror exactly what the trigger diff keys on: `current_trigger_identity`
+    // excludes dead-letter triggers (the diff ignores them) and keys on the
+    // semantic identity — NOT raw id/enabled. Hashing dead-letter rows or the
+    // (currently inert) `enabled` flag would force a false refusal over a
+    // change the manifest can't represent.
+    let script_name_by_id: HashMap<ScriptId, String> = current
+        .scripts
+        .iter()
+        .map(|s| (s.id, s.name.clone()))
+        .collect();
+    for t in &current.triggers {
+        if let Some(id) = current_trigger_identity(t, &script_name_by_id) {
+            parts.push(format!("t|{id}"));
+        }
+    }
+    for n in &current.secret_names {
+        parts.push(format!("k|{n}"));
+    }
+    // Order-independent: sort the per-resource tokens before hashing.
+    parts.sort_unstable();
+    let mut h: u64 = 0xcbf2_9ce4_8422_2325;
+    for p in &parts {
+        for b in p.as_bytes() {
+            h ^= u64::from(*b);
+            h = h.wrapping_mul(0x0000_0100_0000_01b3);
+        }
+        h ^= u64::from(b'\n');
+        h = h.wrapping_mul(0x0000_0100_0000_01b3);
+    }
+    format!("{h:016x}")
+}
+
 /// Per-app advisory lock key, namespaced so it can't collide with the
 /// queue-trigger lock space.
 fn apply_lock_key(app_id: AppId) -> i64 {
@@ -1536,6 +1807,7 @@ mod tests {
             timeout_seconds: 30,
             sandbox: ScriptSandbox::empty(),
             memory_limit_mb: 256,
+            enabled: true,
             created_at: Utc::now(),
             updated_at: Utc::now(),
         }
@@ -1550,6 +1822,7 @@ mod tests {
             timeout_seconds: None,
             memory_limit_mb: None,
             sandbox: None,
+            enabled: true,
         }
     }
 
@@ -1636,6 +1909,7 @@ mod tests {
             path: "/p".into(),
             method: Some("POST".into()),
             dispatch_mode: DispatchMode::Sync,
+            enabled: true,
             created_at: Utc::now(),
         };
         let current = CurrentState {
@@ -1655,6 +1929,7 @@ mod tests {
                 path_kind: PathKind::Exact,
                 path: "/p".into(),
                 dispatch_mode: DispatchMode::Sync,
+                enabled: true,
             }],
             triggers: vec![],
             secrets: vec![],
@@ -1687,6 +1962,7 @@ mod tests {
             id: TriggerId::from(uuid::Uuid::new_v4()),
             app_id: AppId::from(uuid::Uuid::nil()),
             script_id,
+            name: "t".into(),
             kind,
             enabled: true,
             dispatch_mode: TriggerDispatchMode::Async,
@@ -1731,6 +2007,246 @@ mod tests {
         assert_eq!(compute_diff(&current, &chg2).scripts[0].op, Op::Update);
     }
 
+    #[test]
+    fn description_is_sparse() {
+        // An omitted (`None`) description must mean "leave as-is", matching
+        // the other optional fields — not "clear it".
+        let mut cur = script("a", "let x = 1;");
+        cur.description = Some("kept".into());
+        let current = CurrentState {
+            scripts: vec![cur],
+            ..CurrentState::default()
+        };
+        // Omitted → NoOp (leave the stored description alone).
+        let mut omit = empty_bundle();
+        omit.scripts = vec![bundle_script("a", "let x = 1;")];
+        assert_eq!(compute_diff(&current, &omit).scripts[0].op, Op::NoOp);
+        // Same explicit value → NoOp.
+        let mut same = empty_bundle();
+        let mut s = bundle_script("a", "let x = 1;");
+        s.description = Some("kept".into());
+        same.scripts = vec![s];
+        assert_eq!(compute_diff(&current, &same).scripts[0].op, Op::NoOp);
+        // Different explicit value → Update.
+        let mut chg = empty_bundle();
+        let mut s2 = bundle_script("a", "let x = 1;");
+        s2.description = Some("changed".into());
+        chg.scripts = vec![s2];
+        assert_eq!(compute_diff(&current, &chg).scripts[0].op, Op::Update);
+    }
+
+    #[test]
+    fn trigger_shape_validation_matches_interactive_api() {
+        // Empty collection_glob is rejected (kv/docs/files).
+        assert!(validate_trigger_shape(&BundleTrigger::Kv {
+            script: "h".into(),
+            collection_glob: "  ".into(),
+            ops: vec![],
+            dispatch_mode: None,
+            retry_max_attempts: None,
+        })
+        .is_err());
+        // A non-empty glob passes.
+        assert!(validate_trigger_shape(&BundleTrigger::Docs {
+            script: "h".into(),
+            collection_glob: "users".into(),
+            ops: vec![],
+            dispatch_mode: None,
+            retry_max_attempts: None,
+        })
+        .is_ok());
+        // A malformed pubsub topic_pattern (mid-pattern wildcard) is rejected;
+        // a valid one passes.
+        assert!(validate_trigger_shape(&BundleTrigger::Pubsub {
+            script: "h".into(),
+            topic_pattern: "user.*.created".into(),
+            dispatch_mode: None,
+            retry_max_attempts: None,
+        })
+        .is_err());
+        assert!(validate_trigger_shape(&BundleTrigger::Pubsub {
+            script: "h".into(),
+            topic_pattern: "orders.created".into(),
+            dispatch_mode: None,
+            retry_max_attempts: None,
+        })
+        .is_ok());
+        // Queue visibility floor matches the interactive API (>= 30): a value
+        // below the floor is rejected; the floor itself and `None` are ok.
+        let queue = |vis| BundleTrigger::Queue {
+            script: "h".into(),
+            queue_name: "jobs".into(),
+            visibility_timeout_secs: vis,
+            dispatch_mode: None,
+            retry_max_attempts: None,
+        };
+        assert!(validate_trigger_shape(&queue(Some(10))).is_err());
+        assert!(validate_trigger_shape(&queue(Some(
+            crate::triggers_api::MIN_QUEUE_VISIBILITY_TIMEOUT_SECS
+        )))
+        .is_ok());
+        assert!(validate_trigger_shape(&queue(None)).is_ok());
+        // Empty email inbound_secret_ref is rejected.
+        assert!(validate_trigger_shape(&BundleTrigger::Email {
+            script: "h".into(),
+            inbound_secret_ref: "  ".into(),
+            dispatch_mode: None,
+            retry_max_attempts: None,
+        })
+        .is_err());
+    }
+
+    #[test]
+    fn state_token_is_stable_order_independent_and_sensitive() {
+        let a = script("a", "x"); // version 1
+        let b = script("b", "y");
+        let base = CurrentState {
+            scripts: vec![a.clone(), b.clone()],
+            secret_names: vec!["S".into()],
+            ..CurrentState::default()
+        };
+        // Deterministic + order-independent.
+        let reordered = CurrentState {
+            scripts: vec![b.clone(), a.clone()],
+            secret_names: vec!["S".into()],
+            ..CurrentState::default()
+        };
+        assert_eq!(state_token(&base), state_token(&reordered));
+        // A script edit bumps `version`, which must flip the token even if the
+        // source-by-name set is otherwise unchanged (out-of-band redeploy).
+        let mut a2 = a.clone();
+        a2.version += 1;
+        let bumped = CurrentState {
+            scripts: vec![a2, b.clone()],
+            secret_names: vec!["S".into()],
+            ..CurrentState::default()
+        };
+        assert_ne!(state_token(&base), state_token(&bumped));
+        // Adding a secret name flips it too.
+        let with_secret = CurrentState {
+            scripts: vec![a, b],
+            secret_names: vec!["S".into(), "T".into()],
+            ..CurrentState::default()
+        };
+        assert_ne!(state_token(&base), state_token(&with_secret));
+    }
+
+    #[test]
+    fn state_token_ignores_dead_letter_triggers() {
+        // The diff ignores dead-letter triggers (not manifest-representable),
+        // so the token must too — otherwise an out-of-band dead-letter change
+        // would force a spurious `--force`.
+        let s = script("h", "x");
+        let sid = s.id;
+        let base = CurrentState {
+            scripts: vec![s.clone()],
+            ..CurrentState::default()
+        };
+        let with_dl = CurrentState {
+            scripts: vec![s],
+            triggers: vec![trig(
+                sid,
+                TriggerDetails::DeadLetter {
+                    source_filter: None,
+                    trigger_id_filter: None,
+                    script_id_filter: None,
+                },
+            )],
+            ..CurrentState::default()
+        };
+        assert_eq!(
+            state_token(&base),
+            state_token(&with_dl),
+            "dead-letter triggers must not affect the token"
+        );
+    }
+
+    #[test]
+    fn enabled_is_declarative_and_diffed() {
+        // A live-active script the manifest marks disabled → Update; the token
+        // is sensitive to the flag so an out-of-band toggle is detectable.
+        let cur = script("a", "let x = 1;"); // enabled: true
+        let base = CurrentState {
+            scripts: vec![cur.clone()],
+            ..CurrentState::default()
+        };
+        let mut disable = empty_bundle();
+        let mut bs = bundle_script("a", "let x = 1;");
+        bs.enabled = false;
+        disable.scripts = vec![bs];
+        assert_eq!(compute_diff(&base, &disable).scripts[0].op, Op::Update);
+
+        let mut flipped = cur;
+        flipped.enabled = false;
+        let toggled = CurrentState {
+            scripts: vec![flipped],
+            ..CurrentState::default()
+        };
+        assert_ne!(
+            state_token(&base),
+            state_token(&toggled),
+            "token must flip on an enabled change"
+        );
+    }
+
+    #[test]
+    fn warns_on_enabled_binding_to_disabled_script() {
+        let mut b = empty_bundle();
+        let mut s = bundle_script("h", "x");
+        s.enabled = false;
+        b.scripts = vec![s];
+        b.routes = vec![BundleRoute {
+            script: "h".into(),
+            method: None,
+            host_kind: HostKind::Any,
+            host: String::new(),
+            host_param_name: None,
+            path_kind: PathKind::Exact,
+            path: "/h".into(),
+            dispatch_mode: DispatchMode::Sync,
+            enabled: true,
+        }];
+        let w = disabled_target_warnings(&b);
+        assert!(
+            w.iter().any(|m| m.contains("will 404")),
+            "expected a 404 reachability warning: {w:?}"
+        );
+        // No warning when the script is active.
+        b.scripts[0].enabled = true;
+        assert!(disabled_target_warnings(&b).is_empty());
+    }
+
+    #[test]
+    fn warns_on_unreachable_endpoint() {
+        // §4.7: an enabled endpoint with no route and no trigger is flagged.
+        let mut b = empty_bundle();
+        b.scripts = vec![bundle_script("orphan", "x")];
+        let w = unreachable_endpoint_warnings(&b);
+        assert!(
+            w.iter().any(|m| m.contains("no route or trigger")),
+            "expected an unreachable-endpoint warning: {w:?}"
+        );
+        // Bound by a route → no warning.
+        b.routes = vec![BundleRoute {
+            script: "orphan".into(),
+            method: None,
+            host_kind: HostKind::Any,
+            host: String::new(),
+            host_param_name: None,
+            path_kind: PathKind::Exact,
+            path: "/o".into(),
+            dispatch_mode: DispatchMode::Sync,
+            enabled: true,
+        }];
+        assert!(unreachable_endpoint_warnings(&b).is_empty());
+        // A module is exempt even with no binding.
+        let mut m = empty_bundle();
+        let mut lib = bundle_script("lib", "x");
+        lib.kind = ScriptKind::Module;
+        m.scripts = vec![lib];
+        assert!(unreachable_endpoint_warnings(&m).is_empty());
+    }
+
     #[test]
     fn trigger_diff_create_noop_delete() {
         let s = script("h", "x");

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/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 {
@@ -393,8 +398,8 @@ pub(crate) async fn insert_script_tx(
     let res = sqlx::query_as::<_, ScriptRow>(&format!(
         "INSERT INTO scripts ( \
             app_id, name, description, source, kind, \
-            timeout_seconds, memory_limit_mb, sandbox \
+            timeout_seconds, memory_limit_mb, sandbox, enabled \
-         ) VALUES ($1, $2, $3, $4, $5, COALESCE($6, 30), COALESCE($7, 256), $8) \
+         ) VALUES ($1, $2, $3, $4, $5, COALESCE($6, 30), COALESCE($7, 256), $8, $9) \
          RETURNING {SCRIPT_SELECT_COLS}"
     ))
     .bind(input.app_id.into_inner())
@@ -405,6 +410,7 @@ pub(crate) async fn insert_script_tx(
     .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 {
@@ -441,6 +447,7 @@ pub(crate) async fn update_script_tx(
             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 \
@@ -455,6 +462,7 @@ pub(crate) async fn update_script_tx(
     .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 {
@@ -505,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>,
 }
@@ -531,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) => {
@@ -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())
@@ -173,10 +175,10 @@ pub(crate) async fn insert_route_tx(
     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 \
+            path_kind, path, method, dispatch_mode, enabled \
-         ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) \
+         ) 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, created_at",
+                   path_kind, path, method, dispatch_mode, enabled, created_at",
     )
     .bind(input.app_id.into_inner())
     .bind(input.script_id.into_inner())
@@ -187,6 +189,7 @@ pub(crate) async fn insert_route_tx(
     .bind(&input.path)
     .bind(input.method.as_deref())
     .bind(input.dispatch_mode.as_str())
+    .bind(input.enabled)
     .fetch_one(&mut **tx)
     .await;
     match res {
@@ -202,6 +205,12 @@ pub(crate) async fn insert_route_tx(
 }
 
 /// 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,
@@ -225,6 +234,7 @@ struct RouteRow {
     path: String,
     method: Option<String>,
     dispatch_mode: String,
+    enabled: bool,
     created_at: chrono::DateTime<chrono::Utc>,
 }
 
@@ -249,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,
@@ -735,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",
         )
@@ -766,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),
@@ -800,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",
         )
@@ -831,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),
@@ -863,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",
         )
@@ -891,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),
@@ -927,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",
         )
@@ -957,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),
@@ -992,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",
         )
@@ -1023,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),
@@ -1056,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",
         )
@@ -1084,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),
@@ -1116,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",
         )
@@ -1143,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),
@@ -1185,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",
@@ -1213,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",
@@ -1463,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",
         )
@@ -1494,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),
@@ -1522,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?;
@@ -1700,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),
@@ -1742,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

@@ -920,14 +920,32 @@ impl Client {
         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 });
+        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
     }
 }
@@ -965,6 +983,10 @@ pub struct PlanDto {
     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)]

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

@@ -1,11 +1,12 @@
 //! `pic apply [--file picloud.toml]` — reconcile the live app to the
-//! manifest's desired state in one server-side transaction. Additive in
+//! manifest's desired state in one server-side transaction. Creates and
-//! this milestone (creates + updates); pruning of stale resources lands
+//! updates are applied; `--prune` additionally deletes live scripts/routes/
-//! next.
+//! triggers absent from the manifest (secrets are never pruned).
 
+use std::io::{IsTerminal, Write};
 use std::path::Path;
 
-use anyhow::Result;
+use anyhow::{Context, Result};
 
 use crate::client::Client;
 use crate::cmds::plan::build_bundle;
@@ -13,15 +14,46 @@ use crate::config;
 use crate::manifest::Manifest;
 use crate::output::{KvBlock, OutputMode};
 
-pub async fn run(manifest_path: &Path, prune: bool, mode: OutputMode) -> Result<()> {
+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(manifest_path)?;
+    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 report = client.apply(&manifest.app.slug, &bundle, prune).await?;
+    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
@@ -50,3 +82,30 @@ pub async fn run(manifest_path: &Path, prune: bool, mode: OutputMode) -> Result<
     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

@@ -3,8 +3,10 @@ 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;

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

@@ -14,15 +14,25 @@ use crate::config;
 use crate::manifest::Manifest;
 use crate::output::{OutputMode, Table};
 
-pub async fn run(manifest_path: &Path, mode: OutputMode) -> Result<()> {
+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(manifest_path)?;
+    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(())
 }
@@ -51,6 +61,7 @@ pub fn build_bundle(manifest: &Manifest, base_dir: &Path) -> Result<Value> {
         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));
     }
 

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

@@ -23,10 +23,24 @@ use crate::manifest::{
 };
 use crate::output::{KvBlock, OutputMode};
 
-pub async fn run(app_ident: &str, dir: &Path, mode: OutputMode) -> Result<()> {
+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?;
@@ -49,6 +63,7 @@ pub async fn run(app_ident: &str, dir: &Path, mode: OutputMode) -> Result<()> {
                 path_kind: r.path_kind,
                 path: r.path,
                 dispatch_mode: r.dispatch_mode,
+                enabled: r.enabled,
             });
         }
     }
@@ -62,8 +77,9 @@ pub async fn run(app_ident: &str, dir: &Path, mode: OutputMode) -> Result<()> {
     for s in &scripts {
         if !is_safe_filename(&s.name) {
             anyhow::bail!(
-                "script name {:?} is not filesystem-safe (contains a path \
+                "script name {:?} is not filesystem-safe (a path separator, \
-                 separator, `..`, or a leading dot); cannot pull",
+                 `..`, a leading dot, a control character, or longer than 200 \
+                 bytes); cannot pull",
                 s.name
             );
         }
@@ -89,6 +105,7 @@ pub async fn run(app_ident: &str, dir: &Path, mode: OutputMode) -> Result<()> {
             } else {
                 Some(s.sandbox)
             },
+            enabled: s.enabled,
         });
     }
 
@@ -186,7 +203,6 @@ pub async fn run(app_ident: &str, dir: &Path, mode: OutputMode) -> Result<()> {
         },
     };
 
-    let manifest_path = dir.join(MANIFEST_FILE);
     std::fs::write(&manifest_path, manifest.to_toml()?)
         .with_context(|| format!("writing {}", manifest_path.display()))?;
 
@@ -207,14 +223,35 @@ fn trigger_count(t: &ManifestTriggers) -> usize {
 }
 
 /// True if `name` is safe to use as a single path component in `scripts/`.
-/// Rejects empty names, path separators, `.`/`..`, and leading dots.
+/// 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.contains('\0')
+        && !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.
@@ -276,6 +313,29 @@ mod tests {
         }
     }
 
+    #[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"] {

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,7 @@ use clap::{Args, Parser, Subcommand, ValueEnum};
 mod client;
 mod cmds;
 mod config;
+mod linkstate;
 mod manifest;
 mod output;
 
@@ -159,7 +160,8 @@ enum Cmd {
     },
 
     /// Reconcile the live app to a `picloud.toml` manifest in one
-    /// transaction (additive: creates + updates).
+    /// 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
@@ -170,6 +172,15 @@ enum Cmd {
     /// (+ `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)]
@@ -181,6 +192,18 @@ struct ApplyArgs {
     /// 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)]
@@ -188,6 +211,10 @@ 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)]
@@ -197,6 +224,39 @@ struct PullArgs {
     /// 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)]
@@ -1086,9 +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.prune, mode).await,
+        Cmd::Apply(args) => {
-        Cmd::Plan(args) => cmds::plan::run(&args.file, mode).await,
+            cmds::apply::run(
-        Cmd::Pull(args) => cmds::pull::run(&args.app, &args.dir, mode).await,
+                &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

@@ -58,6 +58,81 @@ impl Manifest {
     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)]
@@ -84,6 +159,13 @@ pub struct ManifestScript {
     /// 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)]
@@ -102,6 +184,13 @@ pub struct ManifestRoute {
     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]]`, …).
@@ -244,6 +333,11 @@ 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()
 }
@@ -268,6 +362,7 @@ mod tests {
                     timeout_seconds: Some(10),
                     memory_limit_mb: Some(256),
                     sandbox: None,
+                    enabled: true,
                 },
                 ManifestScript {
                     name: "lib".into(),
@@ -280,6 +375,7 @@ mod tests {
                         max_operations: Some(5_000_000),
                         ..ScriptSandbox::empty()
                     }),
+                    enabled: false,
                 },
             ],
             routes: vec![ManifestRoute {
@@ -291,6 +387,7 @@ mod tests {
                 path_kind: PathKind::Exact,
                 path: "/posts".into(),
                 dispatch_mode: DispatchMode::Sync,
+                enabled: true,
             }],
             triggers: ManifestTriggers {
                 cron: vec![CronTriggerSpec {
@@ -337,6 +434,70 @@ mod tests {
         );
         // 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]

crates/picloud-cli/tests/cli.rs

@@ -18,8 +18,12 @@ 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;
@@ -30,4 +34,5 @@ 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

@@ -135,7 +135,7 @@ fn prune_refuses_to_orphan_email_trigger() {
     let out = common::pic_as(&env)
         .args(["apply", "--file"])
         .arg(&manifest_path)
-        .arg("--prune")
+        .args(["--prune", "--yes"])
         .output()
         .expect("apply --prune");
     assert!(

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/prune.rs

@@ -72,11 +72,12 @@ fn prune_deletes_stale_resources() {
         "additive apply must not delete"
     );
 
-    // Prune apply removes `drop` and its route.
+    // 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)
-        .arg("--prune")
+        .args(["--prune", "--yes"])
         .output()
         .expect("apply --prune");
     assert!(
@@ -109,3 +110,68 @@ fn prune_deletes_stale_resources() {
         "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/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/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

@@ -296,12 +296,23 @@ CLI builds bundle (manifests + script sources) for the subtree
   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.
-- **Outbox fire-time gap (verified):** the dispatcher does **not** re-check `enabled` on an
+- **Fire-time re-check (shipped, test-backed):** the dispatcher re-checks `enabled` at fire time on
-  already-enqueued outbox row (`dispatcher.rs:699`), so disabling a trigger stops *new* matches while
+  every async path, so a pending event for a now-disabled trigger/script is dropped (not executed)
-  a pending item still fires. **Fix:** add an `enabled` re-check (trigger *and* script) at fire time
+  when it comes up — the §5.1 security-disable guarantee on the trigger path. **Outbox arm:**
-  in `resolve_trigger`, so a pending outbox row for a now-disabled trigger/script is dropped when it
+  `resolve_trigger` sets `active = trigger.enabled && script.enabled` and the async-HTTP/invoke
-  comes up — closing the gap cheaply, with no cache or kill-switch involvement. This is the home for
+  builders set `active = script.enabled`; `dispatch_one` drops the row via a single unified
-  the §5.1 security-disable guarantee on the trigger path.
+  `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.
@@ -813,6 +824,23 @@ Resolved items now live inline next to their topic. What genuinely remains:
 
 ## 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