feat(project-tool): bound-plan staleness check (content fingerprint)

`pic plan` now records a fingerprint of the live state it diffed against;
`pic apply` replays it and the server refuses (HTTP 409) if the app
changed underneath the reviewed plan — the §4.2 "apply exactly what you
reviewed" guarantee, in its content-addressed form (no migration, no
changes to interactive write paths).

Server (manager-core):
- `state_token(CurrentState)`: deterministic FNV-1a fingerprint over what
  the diff keys on — script name+version (version bumps on any edit),
  route identity+binding/attrs, trigger membership+enabled, secret names.
  Order-independent; a collision can only yield a false "unchanged", never
  a false refusal.
- `plan` returns it (flattened onto the plan JSON, so the wire stays
  additive); `apply` takes an optional `expected_token` and, under the
  apply lock before any write, returns `StateMoved` (409) on mismatch.

CLI:
- `.picloud/` link state (`linkstate`): `pic plan` writes the token scoped
  to the app slug; `pic apply` replays it, then clears it on success (the
  token is single-use — the next apply re-plans). `--force` skips the
  check; apply with no recorded plan still works standalone (today's
  behavior). `.picloud/` is already gitignored by `pic init`.

The tree-structure version (the other half of §4.2's counter) stays a
deliberate no-op until groups exist — it only guards reparent/structural
moves, which don't exist single-app.

Tested: state_token unit test (stable/order-independent/sensitive) + a
staleness journey (plan → out-of-band deploy → apply refuses → --force
applies); manager-core lib 363 + cli bins 31 + 13 journeys green; clippy
-D warnings clean.

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

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

@@ -312,6 +312,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 +345,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}")]
@@ -377,12 +393,17 @@ 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. Creates
@@ -398,6 +419,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,6 +444,15 @@ 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 has changed since — computed
+        // under the apply lock so it reflects what we're about to write. Done
+        // before any mutation so a stale apply rolls back nothing.
+        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)?;
@@ -1564,6 +1595,62 @@ 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 membership +
+/// `enabled`, and secret names. Any out-of-band change to those flips the
+/// token, so a `plan`-then-`apply` can detect that the app moved underneath it.
+///
+/// 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));
+    }
+    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(""),
+        ));
+    }
+    for t in &current.triggers {
+        parts.push(format!("t|{:?}|{}", t.id, t.enabled));
+    }
+    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 {
@@ -1875,6 +1962,41 @@ mod tests {
         .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 trigger_diff_create_noop_delete() {
         let s = script("h", "x");