feat(kv): set_if compare-and-swap for KV (per-app + shared + SDK) (Track A M5)

No atomic CAS existed, so concurrent writers to the same KV key raced. Add
set_if(key, expected, new): writes only when the current value equals expected,
or (expected absent) only when the key is missing — the primitive for lock-free
counters / optimistic concurrency.

- KvService + GroupKvService traits gain set_if with a defaulted "unsupported"
  impl (Noop + other impls untouched); the real services override it.
- kv_repo/group_kv_repo: atomic set_if — a single UPDATE ... WHERE value =
  $expected (JSONB semantic equality) or INSERT ... ON CONFLICT DO NOTHING.
  Atomic without a transaction; returns whether the write happened.
- Services enforce the same value-size cap + (group) writes-authed + row/byte
  quotas as set; the event fires only on a successful swap.
- Executor Rhai SDK: kv::collection(c).set_if(key, expected, new) and the
  shared_collection variant; Rhai () for expected means "only if absent".

Pinned by kv_service/group_kv_service set_if unit tests (CAS + fail-closed) +
sdk_kv::kv_set_if_compare_and_swap (through a real script). No migration.
docs/sdk-shape.md documents the primitive.

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

crates/manager-core/src/group_kv_service.rs

@@ -263,6 +263,66 @@ impl GroupKvService for GroupKvServiceImpl {
         Ok(())
     }
 
+    async fn set_if(
+        &self,
+        cx: &SdkCallCx,
+        collection: &str,
+        key: &str,
+        expected: Option<serde_json::Value>,
+        new: serde_json::Value,
+    ) -> Result<bool, GroupKvError> {
+        let group_id = self.owning_group(cx, collection).await?;
+        let encoded_len = serde_json::to_vec(&new)
+            .map(|v| v.len())
+            .map_err(|e| GroupKvError::Backend(format!("encode value: {e}")))?;
+        if encoded_len > self.max_value_bytes {
+            return Err(GroupKvError::ValueTooLarge {
+                limit: self.max_value_bytes,
+                actual: encoded_len,
+            });
+        }
+        self.check_write(cx, group_id).await?;
+        // Same quota rules as `set` (best-effort pre-check; the swap itself is
+        // atomic in the repo). A swap that inserts a NEW key (expected absent)
+        // must fit the row ceiling; the byte ceiling uses the projected total.
+        let existing = self.repo.get(group_id, collection, key).await?;
+        if expected.is_none() && existing.is_none() {
+            let count = self.repo.count_rows(group_id).await?;
+            if count >= self.max_rows {
+                return Err(GroupKvError::QuotaExceeded {
+                    limit: usize::try_from(self.max_rows).unwrap_or(usize::MAX),
+                    actual: usize::try_from(count).unwrap_or(usize::MAX),
+                });
+            }
+        }
+        let old_len = existing
+            .as_ref()
+            .and_then(|v| serde_json::to_vec(v).ok())
+            .map_or(0u64, |b| b.len() as u64);
+        let used = self.repo.total_bytes(group_id).await?;
+        let projected = used.saturating_sub(old_len) + encoded_len as u64;
+        if projected > self.max_total_bytes {
+            return Err(GroupKvError::TotalBytesQuotaExceeded {
+                limit: self.max_total_bytes,
+                actual: projected,
+            });
+        }
+        let op = if expected.is_some() {
+            "update"
+        } else {
+            "insert"
+        };
+        let swapped = self
+            .repo
+            .set_if(group_id, collection, key, expected, new.clone())
+            .await?;
+        if swapped {
+            self.emit_shared(cx, group_id, op, collection, key, Some(new))
+                .await;
+        }
+        Ok(swapped)
+    }
+
     async fn delete(
         &self,
         cx: &SdkCallCx,
@@ -348,6 +408,27 @@ mod tests {
                 .insert((group_id, collection.to_string(), key.to_string()), value))
         }
 
+        async fn set_if(
+            &self,
+            group_id: GroupId,
+            collection: &str,
+            key: &str,
+            expected: Option<serde_json::Value>,
+            new: serde_json::Value,
+        ) -> Result<bool, GroupKvRepoError> {
+            let mut data = self.data.lock().await;
+            let k = (group_id, collection.to_string(), key.to_string());
+            let matches = match (&expected, data.get(&k)) {
+                (None, None) => true,
+                (Some(exp), Some(cur)) => exp == cur,
+                _ => false,
+            };
+            if matches {
+                data.insert(k, new);
+            }
+            Ok(matches)
+        }
+
         async fn delete(
             &self,
             group_id: GroupId,
@@ -643,4 +724,56 @@ mod tests {
         let err = kv.get(&cx, "", "k").await.unwrap_err();
         assert!(matches!(err, GroupKvError::InvalidCollection));
     }
+
+    #[tokio::test]
+    async fn set_if_is_compare_and_swap_and_writes_fail_closed() {
+        let app = AppId::new();
+        let group = GroupId::new();
+        let mut resolver = FakeResolver::default();
+        resolver.map.insert((app, "catalog".into()), group);
+        let kv = svc(resolver);
+        let cx = cx_with(app, Some(owner()));
+
+        // Insert-if-absent, then swap-if-equal (as the per-app path).
+        assert!(kv
+            .set_if(&cx, "catalog", "k", None, serde_json::json!(1))
+            .await
+            .unwrap());
+        assert!(!kv
+            .set_if(&cx, "catalog", "k", None, serde_json::json!(2))
+            .await
+            .unwrap());
+        assert!(!kv
+            .set_if(
+                &cx,
+                "catalog",
+                "k",
+                Some(serde_json::json!(9)),
+                serde_json::json!(3)
+            )
+            .await
+            .unwrap());
+        assert!(kv
+            .set_if(
+                &cx,
+                "catalog",
+                "k",
+                Some(serde_json::json!(1)),
+                serde_json::json!(3)
+            )
+            .await
+            .unwrap());
+        assert_eq!(
+            kv.get(&cx, "catalog", "k").await.unwrap(),
+            Some(serde_json::json!(3))
+        );
+
+        // A shared-collection CAS still fails closed for an anonymous writer.
+        let anon = cx_with(app, None);
+        let err = kv
+            .set_if(&anon, "catalog", "k", None, serde_json::json!(4))
+            .await
+            .unwrap_err();
+        assert!(matches!(err, GroupKvError::Forbidden));
+    }
 }