fix(kg): bound the fence list before the writer transaction opens

crates/khive-pack-kg/src/handler_defs.rs

@@ -24,7 +24,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [
             ParamDef { name: "stream", param_type: "string", required: true, description: "Namespace-scoped stream name, at most 512 UTF-8 bytes and no U+0000.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "record", param_type: "JSON value", required: true, description: "Required JSON value, including scalar or null; stored as note content.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "expected_seq", param_type: "integer", required: false, description: "Append only if this entry would receive this sequence; conflict details include next_seq.", resolution_mode: IdResolutionMode::NotApplicable },
-            ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "A fence object {key, kind, expected_version} or a non-empty list of distinct (kind, key) objects, checked in order in the same writer transaction. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable },
+            ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "A fence object {key, kind, expected_version} or a non-empty list of at most 100 distinct (kind, key) objects, checked in order in the same writer transaction. Oversized lists return invalid_input naming the cap and count sent before entry interpretation or writer admission. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "note_kind", param_type: "string", required: false, description: "Registered note kind; defaults to observation.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "embed", param_type: "boolean", required: false, description: "Defaults false: no embedding rows or vector-index work, while lexical indexing and stream reads remain available. True embeds under every registered model unless embedding_model selects one.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "embedding_model", param_type: "string", required: false, description: "Select one registered embedding model. Requires embed=true; otherwise invalid_input before any write.", resolution_mode: IdResolutionMode::NotApplicable },
@@ -60,7 +60,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [
         visibility: Visibility::Verb,
         category: VerbCategory::Commissive,
         params: &[
-            ParamDef { name: "ops", param_type: "array of object", required: true, description: "Members in order. {op: \"append\", stream, record, expected_seq?, fence?, embed?, embedding_model?}: record is any JSON value, including null. Append embed defaults false; true embeds under all registered models or the one embedding_model selects. embedding_model requires embed=true or the entire list is invalid_input before any write; atomic errors name details.member. Each append fence is an object or non-empty ordered list of {key, kind, expected_version}; all append-member fences are checked in order before any member writes in their transaction. A stale or missing list entry reports a string index; atomic refusal also names the string member position. Null, empty lists and duplicate (kind,key) entries are invalid before any member writes. {op: \"write\", key, kind, doc, tags?, embed?, expected_version?}: doc is any JSON value; omitted/null expected_version creates only if absent, while a positive version updates only an existing key at that version. Duplicate write (kind, key) pairs refuse the whole batch. An unknown op is that member's unknown_op refusal, placed by the mode.", resolution_mode: IdResolutionMode::NotApplicable },
+            ParamDef { name: "ops", param_type: "array of object", required: true, description: "Members in order. {op: \"append\", stream, record, expected_seq?, fence?, embed?, embedding_model?}: record is any JSON value, including null. Append embed defaults false; true embeds under all registered models or the one embedding_model selects. embedding_model requires embed=true or the entire list is invalid_input before any write; atomic errors name details.member. Each append fence is an object or non-empty ordered list of at most 100 {key, kind, expected_version} entries; oversized lists return invalid_input naming the cap and count sent before fence entry interpretation or any writer admission, in both modes. All append-member fences are checked in order before any member writes in their transaction. A stale or missing list entry reports a string index; atomic refusal also names the string member position. Null, empty lists and duplicate (kind,key) entries are invalid before any member writes. {op: \"write\", key, kind, doc, tags?, embed?, expected_version?}: doc is any JSON value; omitted/null expected_version creates only if absent, while a positive version updates only an existing key at that version. Duplicate write (kind, key) pairs refuse the whole batch. An unknown op is that member's unknown_op refusal, placed by the mode.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "fence", param_type: "object", required: false, description: "{key, kind, expected_version}: a keyed note whose positive version must match, checked once inside the atomic transaction before the first write. A missing or stale note refuses the whole batch with fence_conflict. Selects atomic mode; atomic=false beside a fence is refused. List-valued fences are not supported.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "observed", param_type: "array of object", required: false, description: "Requires atomic mode; supplying observed alone does not select it. Set atomic=true or supply fence. [{key, kind, version, id?, live_until?}], checked inside the transaction before the first write. A positive version asserts the live key's exact version; explicit null asserts no live holder. A mismatch refuses the batch with version_conflict naming the key and the entry index. The version field is required. Optional live_until is a dotted document path (for example lease.expires_at), requires a positive version, and must hold an RFC 3339 timestamp strictly later than one writer-transaction clock reading. Expired or unreadable fields refuse with expired or live_until_unreadable; an expired refusal names the deadline it read and the clock it compared, an unreadable one names value_type (absent, string, number, boolean, array, object, null) and never the value itself. Optional id pins the note UUID and requires a positive version; a replacement refuses with identity_conflict before the version check. The refusal includes current_id only when the caller may learn the holder, under the same disclosure rule as key_conflict. Without id, the entry asserts version equality on the note holding the key at commit time. Refused with invalid_input in per-member mode.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "atomic", param_type: "boolean", required: false, description: "true: one transaction, all or nothing. false: one transaction per member, refusals as member values. Defaults to whether fence is present. observed alone does not select atomic mode; observed without fence requires atomic=true, otherwise invalid_input.", resolution_mode: IdResolutionMode::NotApplicable },
@@ -76,7 +76,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [
         params: &[
             ParamDef { name: "key", param_type: "string", required: false, description: "Singleton notes only: immutable live namespace/kind identity, at most 512 UTF-8 bytes without U+0000. An occupied key fails with key_conflict; existing_id is disclosed only when list is allowed.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "embed", param_type: "bool", required: false, description: "Singleton notes only: defaults false for head and true otherwise. False skips inference and vector insertion while retaining lexical indexing.", resolution_mode: IdResolutionMode::NotApplicable },
-            ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of distinct (kind, key) objects, checked in order in the same writer transaction. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable },
+            ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of at most 100 distinct (kind, key) objects, checked in order in the same writer transaction. Oversized lists return invalid_input naming the cap and count sent before entry interpretation or writer admission. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef {
                 name: "kind",
                 param_type: "string",
@@ -462,7 +462,7 @@ pub(crate) static KG_HANDLERS: [HandlerDef; 24] = [
         category: VerbCategory::Declaration,
         params: &[
             ParamDef { name: "expected_version", param_type: "integer", required: false, description: "Notes only: positive persisted version required inside the writer transaction. A stale version fails without mutation, with reason=version_conflict and expected_version/current_version details. Omission preserves unconditional caller semantics.", resolution_mode: IdResolutionMode::NotApplicable },
-            ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of distinct (kind, key) objects, checked in order in the same writer transaction. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable },
+            ParamDef { name: "fence", param_type: "object or array of object", required: false, description: "Singleton notes only. A fence object {key, kind, expected_version} or a non-empty list of at most 100 distinct (kind, key) objects, checked in order in the same writer transaction. Oversized lists return invalid_input naming the cap and count sent before entry interpretation or writer admission. Missing or stale returns fence_conflict; list refusals also carry string index (zero-based). Explicit null is invalid.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef { name: "embed", param_type: "bool", required: false, description: "Notes only: omission retains embedding state. True enables reindexing; false performs no inference, removes existing vector rows transactionally and retains lexical indexing. Delayed reindex work cannot restore a stale revision.", resolution_mode: IdResolutionMode::NotApplicable },
             ParamDef {
                 name: "id",

crates/khive-pack-kg/src/handlers/stream_fence_batch_tests.rs

@@ -193,3 +193,118 @@ async fn ordered_fences_batch_validates_all_members_before_writing() {
         }
     }
 }
+
+#[tokio::test]
+async fn ordered_fences_cap_accepts_exactly_100_on_every_write_surface() {
+    let (_, registry) = batch_surface();
+    let mut fences = Vec::new();
+    for index in 0..100 {
+        let key = format!("fence-cap/{index}");
+        lease(&registry, &key).await;
+        fences.push(fence(&key, 1));
+    }
+    let target = registry
+        .dispatch(
+            "create",
+            json!({"kind":"head","content":"{}","fence":fences}),
+        )
+        .await
+        .unwrap();
+    registry
+        .dispatch(
+            "update",
+            json!({"id":target["id"],"content":"{\"updated\":true}","fence":fences}),
+        )
+        .await
+        .unwrap();
+    registry
+        .dispatch(
+            "stream.append",
+            json!({"stream":"fence-cap","record":1,"fence":fences}),
+        )
+        .await
+        .unwrap();
+    for atomic in [true, false] {
+        let result = registry
+            .dispatch(
+                "stream.batch",
+                json!({"atomic":atomic,"ops":[
+                    {"op":"append","stream":"fence-cap","record":2,"fence":fences}
+                ]}),
+            )
+            .await
+            .unwrap();
+        assert_eq!(result["committed"], true);
+    }
+    assert_eq!(heads(&registry, &["fence-cap"]).await, vec![3]);
+    let target = registry
+        .dispatch("get", json!({"id":target["id"]}))
+        .await
+        .unwrap();
+    assert_eq!(target["version"], 2);
+}
+
+#[tokio::test]
+async fn ordered_fences_cap_refuses_before_entry_interpretation_and_writer_admission() {
+    let (rt, registry) = batch_surface();
+    let target = lease(&registry, "target").await;
+    let distinct: Vec<Value> = (0..101)
+        .map(|index| fence(&format!("fence-cap/{index}"), 1))
+        .collect();
+    for fences in [
+        json!(distinct),
+        json!(vec![Value::Null; 101]),
+        json!(vec![fence("duplicate", 0); 101]),
+    ] {
+        let before = population(&rt).await;
+        let before_writers = rt.backend().pool().writer_acquisition_snapshot();
+        for (verb, args) in [
+            (
+                "create",
+                json!({"kind":"head","content":"{}","fence":fences}),
+            ),
+            (
+                "update",
+                json!({"id":target["id"],"content":"{}","fence":fences}),
+            ),
+            (
+                "stream.append",
+                json!({"stream":"fence-cap","record":1,"fence":fences}),
+            ),
+            (
+                "stream.batch",
+                json!({"atomic":true,"ops":[
+                    {"op":"append","stream":"fence-cap","record":1},
+                    {"op":"append","stream":"fence-cap","record":2,"fence":fences}
+                ]}),
+            ),
+            (
+                "stream.batch",
+                json!({"atomic":false,"ops":[
+                    {"op":"append","stream":"fence-cap","record":1},
+                    {"op":"append","stream":"fence-cap","record":2,"fence":fences}
+                ]}),
+            ),
+        ] {
+            let error = registry.dispatch(verb, args).await.unwrap_err();
+            let RuntimeError::InvalidInput(message) = error else {
+                panic!("{verb} must refuse invalid_input: {error}");
+            };
+            assert!(message.contains("at most 100 entries"), "{verb}: {message}");
+            assert!(message.contains("sent 101"), "{verb}: {message}");
+            assert_eq!(population(&rt).await, before, "{verb}");
+            assert_eq!(
+                rt.backend().pool().writer_acquisition_snapshot(),
+                before_writers,
+                "{verb} must refuse before writer admission"
+            );
+        }
+    }
+    assert_eq!(heads(&registry, &["fence-cap"]).await, vec![0]);
+    let after = registry
+        .dispatch("get", json!({"id":target["id"]}))
+        .await
+        .unwrap();
+    assert_eq!(after["version"], target["version"]);
+    assert_eq!(after["content"], target["content"]);
+}

crates/khive-runtime/src/note_write.rs

@@ -7,6 +7,9 @@ use uuid::Uuid;
 
 use crate::{KhiveRuntime, NamespaceToken, RuntimeError, RuntimeResult};
 
+/// Each fence adds a keyed read while holding the writer, like a batch observation.
+pub const MAX_NOTE_FENCES: usize = 100;
+
 #[derive(Clone, Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct NoteFence {
@@ -42,6 +45,15 @@ impl From<NoteFence> for NoteFences {
 }
 
 impl NoteFences {
+    fn validate_count(count: usize) -> RuntimeResult<()> {
+        if count > MAX_NOTE_FENCES {
+            return Err(RuntimeError::InvalidInput(format!(
+                "fence list admits at most {MAX_NOTE_FENCES} entries; this call sent {count}: each fence is a read taken while holding the writer"
+            )));
+        }
+        Ok(())
+    }
+
     pub fn entries(&self) -> &[NoteFence] {
         match self {
             Self::One(fence) => std::slice::from_ref(fence),
@@ -50,6 +62,7 @@ impl NoteFences {
     }
 
     pub fn validate(&self) -> RuntimeResult<()> {
+        Self::validate_count(self.entries().len())?;
         if self.entries().is_empty() {
             return Err(RuntimeError::InvalidInput(
                 "fence list must not be empty".into(),
@@ -76,7 +89,11 @@ impl<'de> Deserialize<'de> for NoteFences {
             One(NoteFence),
             Many(Vec<NoteFence>),
         }
-        let fences = match Shape::deserialize(deserializer)? {
+        let value = serde_json::Value::deserialize(deserializer)?;
+        if let Some(entries) = value.as_array() {
+            Self::validate_count(entries.len()).map_err(serde::de::Error::custom)?;
+        }
+        let fences = match Shape::deserialize(value).map_err(serde::de::Error::custom)? {
             Shape::One(fence) => Self::One(fence),
             Shape::Many(fences) => Self::Many(fences),
         };

crates/khive-runtime/src/note_write_tests.rs

@@ -18,6 +18,53 @@ use crate::{KhiveRuntime, NamespaceToken, RuntimeError, RuntimeResult};
 
 const MODEL: &str = "note-version-test";
 
+#[test]
+fn ordered_fences_cap_precedes_entry_validation_for_json_and_typed_inputs() {
+    use crate::note_write::NoteFences;
+
+    let entries: Vec<NoteFence> = (0..100)
+        .map(|index| NoteFence {
+            key: format!("fence-cap/{index}"),
+            kind: "head".into(),
+            expected_version: 1,
+        })
+        .collect();
+    let at_cap = NoteFences::Many(entries.clone());
+    at_cap.validate().unwrap();
+    serde_json::from_value::<NoteFences>(serde_json::to_value(&at_cap).unwrap()).unwrap();
+
+    let mut over_cap = entries;
+    over_cap.push(NoteFence {
+        key: "fence-cap/100".into(),
+        kind: "head".into(),
+        expected_version: 1,
+    });
+    for malformed in [false, true] {
+        let mut entries = over_cap.clone();
+        if malformed {
+            entries[0].expected_version = 0;
+        }
+        let fences = NoteFences::Many(entries);
+        let error = fences.validate().unwrap_err();
+        assert!(matches!(error, RuntimeError::InvalidInput(_)), "{error}");
+        for message in [
+            error.to_string(),
+            serde_json::from_value::<NoteFences>(serde_json::to_value(&fences).unwrap())
+                .unwrap_err()
+                .to_string(),
+        ] {
+            assert!(message.contains("at most 100 entries"), "{message}");
+            assert!(message.contains("sent 101"), "{message}");
+        }
+    }
+    let malformed = json!(vec![serde_json::Value::Null; 101]);
+    let message = serde_json::from_value::<NoteFences>(malformed)
+        .unwrap_err()
+        .to_string();
+    assert!(message.contains("at most 100 entries"), "{message}");
+    assert!(message.contains("sent 101"), "{message}");
+}
+
 #[tokio::test]
 async fn ordered_fences_observe_prior_write_in_same_transaction() {
     let (runtime, token, _) = fixture();