Graph fixes, batch edge writes, graph metrics and read-path speed-ups

docs/ai/agent-memory.md

@@ -89,8 +89,16 @@ WHERE id = $fact_id;
 Current task state — plans, intermediate results, tool outputs. Uses the KV engine with TTL for automatic expiry when the task is done.
 
 ```sql
--- Create a KV collection for working memory
-CREATE COLLECTION working_memory TYPE kv;
+-- Create a KV collection for working memory. The primary key is the hash key.
+CREATE COLLECTION working_memory (
+    key TEXT PRIMARY KEY,
+    goal TEXT,
+    steps TEXT,
+    current_step INT,
+    status TEXT,
+    artifact TEXT,
+    ttl INT
+) WITH (engine='kv');
 
 -- Store current plan (expires in 1 hour)
 INSERT INTO working_memory {

docs/architecture.md

@@ -345,6 +345,15 @@ Counters are incremented at the sites where resources are consumed:
 
 All metrics are dimensionalized by database and tenant to enable per-customer tracking and alerting.
 
+**Graph Write Metrics** (unlabelled, per node):
+
+- `nodedb_graph_edges_written_total` — Edge versions that writes applied. A loader compares its rate with its own send rate to pace itself.
+- `nodedb_graph_edges_deleted_total` — Live edges that deletes removed. A delete of an absent edge adds nothing.
+- `nodedb_graph_edges` — Live edges this node stores.
+- `nodedb_graph_nodes` — Nodes with at least one live edge that this node stores.
+
+The two counters count an autocommit write when the Data Plane applies it. They count a committed transaction once its install settles, and a rolled-back install adds nothing. A dual-homed edge counts once, on its source home. Restart replay adds nothing. `TRUNCATE` cuts the edge collection without counting edges. The two gauges refresh every 10 seconds per core.
+
 ## Cross-Engine Identity
 
 All engines use a unified, distributed identity space called **surrogate identity**. Each row, cell, node, and document has a surrogate ID (u64) that is globally unique within a database. Surrogates enable fused queries that combine filtering across all engines in a single bitmap.

docs/graph.md

@@ -55,6 +55,20 @@ GRAPH INSERT EDGE IN 'edges' FROM 'users:alice' TO 'users:bob' TYPE 'KNOWS'
 
 The `IN '<collection>'` clause is **required** — edges are overlays on a named document collection, not a global namespace. Statements without `IN` fail to parse.
 
+#### Several edges in one statement
+
+A clause an edge already has starts the next edge. Each edge takes its own `PROPERTIES`, `VALID_TIME` and `UNTIL`.
+
+```sql
+GRAPH INSERT EDGE IN 'edges'
+  FROM 'users:alice' TO 'users:bob'   TYPE 'KNOWS' PROPERTIES { since: 2020 },
+  FROM 'users:alice' TO 'users:carol' TYPE 'KNOWS' PROPERTIES { since: 2021 };
+```
+
+The statement binds its endpoints in one exchange and writes every edge as one unit. When the edges home on more than one vShard, that unit is one Calvin transaction, so a bulk load costs one commit chain per statement instead of one per edge. The command tag reports the number of edges written. An edge that is missing `FROM`, `TO` or `TYPE` refuses the whole statement.
+
+A row-level-security write policy decides each edge on its own `PROPERTIES`, as for a single edge. One refused edge refuses the whole statement, and no edge lands.
+
 ### Delete Edge
 
 ```sql
@@ -63,6 +77,45 @@ GRAPH DELETE EDGE IN 'edges' FROM 'users:alice' TO 'users:bob' TYPE 'KNOWS';
 
 Removes both forward and reverse index entries atomically. Cascading delete is available for all edges touching a node.
 
+#### Several edges in one statement
+
+`GRAPH DELETE EDGE` takes a list of edges the same way the insert does. A clause an edge already has starts the next edge.
+
+```sql
+GRAPH DELETE EDGE IN 'edges'
+  FROM 'users:alice' TO 'users:bob'   TYPE 'KNOWS',
+  FROM 'users:alice' TO 'users:carol' TYPE 'KNOWS';
+```
+
+- **One unit** — every edge leaves both of its homes, or none does. When the edges home on more than one vShard, that unit is one Calvin transaction. Inside `BEGIN ... COMMIT` the deletes stage with the transaction.
+- **Command tag** — reports the number of edges that were live and are now deleted. An edge that was never stored is not counted.
+- **Missing clause** — an edge without `FROM`, `TO` or `TYPE` refuses the whole statement.
+- **Row-level security** — a write policy decides each edge against that edge's stored properties, as for a single edge. One refused edge refuses the whole statement, and every edge stays.
+
+To replay a load idempotently, delete the batch, then insert it.
+
+#### Statement limits
+
+Both edge-list statements share these caps. A statement over a cap is refused whole.
+
+| Limit | Value | Error |
+|---|---|---|
+| Edges per statement | 1000 | SQLSTATE `54000`, naming the edge count and the cap |
+| SQL text per statement | 1 MiB (1048576 bytes) | SQLSTATE `42601`, naming the text size and the cap |
+
+Split a larger load into several statements. Both statements need the collection's write grant.
+
+#### Native protocol
+
+Native clients send the same writes without SQL text:
+
+| Opcode | Payload (`payload` field, MessagePack) | Answer |
+|---|---|---|
+| `EdgePutBatch` (`0x9D`) | `EdgePutBatch { edges: [EdgeWrite] }`. Each edge carries `src`, `dst`, `label`, `properties`, `valid_from_ms`, `valid_until_ms`. | `rows_affected`: edges written |
+| `EdgeDeleteBatch` (`0x9E`) | `EdgeDeleteBatch { edges: [EdgeKey] }`. Each edge carries `src`, `dst`, `label`. | `rows_affected`: edges deleted |
+
+The request's `collection` field names the collection. A request runs the same handler as the SQL statement: one unit, the same edge cap, write grant and per-edge write policy. In `nodedb-client`, `NodeDbGraph::graph_insert_edges` and `NodeDbGraph::graph_delete_edges` send one request (native) or one statement (pgwire).
+
 ---
 
 ## Traversal Queries
@@ -315,7 +368,9 @@ GRAPH RAG FUSION ON entities
 | `RRF_K`           | Weighting constants `(vector_k, graph_k)` for RRF scoring                                                            |
 | `MAX_VISITED`     | Memory budget cap — BFS stops early if exceeded                                                                      |
 
-The response includes a truncation flag if the memory budget forced early termination.
+The statement returns one row per fused hit, in rank order, with the columns `node_id`, `rrf_score`, `vector_rank`, `vector_distance` and `hop_distance`. `vector_rank` and `vector_distance` are NULL for a node found only by expansion. `hop_distance` is 0 for a seed node.
+
+When the memory budget stops the walk early, or the walk reaches nodes with no identity binding, the server sends a pgwire NOTICE that says so.
 
 ### Step-by-Step Pattern (Application-Level)
 

nodedb-bridge/src/buffer.rs

@@ -58,8 +58,15 @@ struct Shared<T> {
 
     /// Shared metrics counters.
     metrics: BridgeMetrics,
+
+    /// Called by the producer after each push. A consumer that sleeps between
+    /// polls wakes from it, so a pushed value waits no longer than the wake.
+    push_hook: Option<PushHook>,
 }
 
+/// A wake the producer runs after each push.
+pub type PushHook = Arc<dyn Fn() + Send + Sync>;
+
 // SAFETY: The SPSC protocol ensures that producer and consumer never access
 // the same slot concurrently. The producer writes slots[tail & mask] and the
 // consumer reads slots[head & mask]. As long as the ring buffer is not full,
@@ -117,6 +124,19 @@ pub struct RingBuffer;
 
 impl RingBuffer {
     pub fn channel<T>(capacity: usize) -> (Producer<T>, Consumer<T>) {
+        Self::build(capacity, None)
+    }
+
+    /// Like [`Self::channel`], with `hook` run by the producer after every
+    /// successful push.
+    pub fn channel_with_push_hook<T>(
+        capacity: usize,
+        hook: PushHook,
+    ) -> (Producer<T>, Consumer<T>) {
+        Self::build(capacity, Some(hook))
+    }
+
+    fn build<T>(capacity: usize, push_hook: Option<PushHook>) -> (Producer<T>, Consumer<T>) {
         assert!(capacity > 0, "ring buffer capacity must be > 0");
 
         let capacity = capacity.next_power_of_two();
@@ -135,6 +155,7 @@ impl RingBuffer {
             mask,
             disconnected: AtomicBool::new(false),
             metrics: BridgeMetrics::new(),
+            push_hook,
         });
 
         let producer = Producer {
@@ -191,6 +212,9 @@ impl<T> Producer<T> {
             .store(tail.wrapping_add(1), Ordering::Release);
 
         self.shared.metrics.record_push();
+        if let Some(hook) = &self.shared.push_hook {
+            hook();
+        }
         Ok(())
     }
 

nodedb-client-tests/tests/graph_edge_batch_round_trip.rs

@@ -0,0 +1,152 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! End-to-end `graph_insert_edges` and `graph_delete_edges` over pgwire
+//! (`NodeDbRemote`, one `GRAPH INSERT EDGE` / `GRAPH DELETE EDGE` statement)
+//! and the native protocol (`NativeClient`, one `EdgePutBatch` /
+//! `EdgeDeleteBatch` request).
+//!
+//! Both answer with the edge count. Each edge keeps its own properties,
+//! a delete counts only the edges that were live, and a call over the edge
+//! cap is refused whole.
+
+use std::collections::BTreeSet;
+
+use nodedb_client::native::pool::PoolConfig;
+use nodedb_client::{NativeClient, NodeDb, NodeDbRemote, NodeId, Value};
+use nodedb_test_support::pgwire_harness::TestServer;
+use nodedb_types::filter::{EdgeFilter, MetadataFilter};
+use nodedb_types::graph::Direction;
+use nodedb_types::id::EdgeId;
+use nodedb_types::protocol::{EdgeWrite, MAX_EDGES_PER_STATEMENT};
+
+fn node(id: &str) -> NodeId {
+    NodeId::try_new(id).expect("valid node id")
+}
+
+fn scored(src: &str, dst: &str, score: i64) -> EdgeWrite {
+    let mut edge = EdgeWrite::new(node(src), node(dst), "L");
+    edge.properties
+        .insert("score".to_string(), Value::Integer(score));
+    edge
+}
+
+fn edge_id(src: &str, dst: &str) -> EdgeId {
+    EdgeId::try_first(node(src), node(dst), "L").expect("edge id")
+}
+
+async fn reachable<D: NodeDb>(
+    db: &D,
+    collection: &str,
+    filter: Option<&EdgeFilter>,
+) -> BTreeSet<String> {
+    db.graph_traverse(collection, &node("hub"), 1, Direction::Out, filter)
+        .await
+        .expect("traversal completes")
+        .nodes
+        .into_iter()
+        .map(|n| n.id.as_str().to_string())
+        .filter(|id| id != "hub")
+        .collect()
+}
+
+fn set(ids: &[&str]) -> BTreeSet<String> {
+    ids.iter().map(|id| id.to_string()).collect()
+}
+
+async fn exercise<D: NodeDb>(db: &D, collection: &str) {
+    db.execute_sql(&format!("CREATE COLLECTION {collection}"), &[])
+        .await
+        .expect("create collection");
+
+    let written = db
+        .graph_insert_edges(
+            collection,
+            &[
+                scored("hub", "a", 9),
+                scored("hub", "b", 1),
+                scored("hub", "c", 9),
+            ],
+        )
+        .await
+        .expect("insert a batch");
+    assert_eq!(written, 3, "the call counts every edge it wrote");
+    assert_eq!(reachable(db, collection, None).await, set(&["a", "b", "c"]));
+
+    let high = EdgeFilter {
+        labels: vec!["L".to_string()],
+        property_filters: vec![MetadataFilter::Gt {
+            field: "score".into(),
+            value: Value::Integer(5),
+        }],
+    };
+    assert_eq!(
+        reachable(db, collection, Some(&high)).await,
+        set(&["a", "c"]),
+        "each edge keeps its own properties"
+    );
+
+    let deleted = db
+        .graph_delete_edges(
+            collection,
+            &[
+                edge_id("hub", "a"),
+                edge_id("hub", "b"),
+                edge_id("hub", "nowhere"),
+            ],
+        )
+        .await
+        .expect("delete a batch");
+    assert_eq!(deleted, 2, "only edges that were live are counted");
+    assert_eq!(reachable(db, collection, None).await, set(&["c"]));
+
+    assert_eq!(
+        db.graph_insert_edges(collection, &[]).await.expect("empty"),
+        0
+    );
+    assert_eq!(
+        db.graph_delete_edges(collection, &[]).await.expect("empty"),
+        0
+    );
+
+    let over: Vec<EdgeWrite> = (0..=MAX_EDGES_PER_STATEMENT)
+        .map(|i| scored("hub", &format!("over{i}"), 1))
+        .collect();
+    let error = db
+        .graph_insert_edges(collection, &over)
+        .await
+        .expect_err("a call over the edge cap is refused");
+    assert!(error.to_string().contains("split the batch"), "{error}");
+    assert_eq!(
+        reachable(db, collection, None).await,
+        set(&["c"]),
+        "a refused call writes none of its edges"
+    );
+}
+
+#[tokio::test]
+async fn remote_edge_batch_round_trip() {
+    let server = TestServer::start().await;
+    let conn_str = format!(
+        "host=127.0.0.1 port={} user=nodedb dbname=default",
+        server.pg_port
+    );
+    let remote = NodeDbRemote::connect(&conn_str)
+        .await
+        .expect("pgwire connect to harness must succeed");
+    exercise(&remote, "eb_remote").await;
+    server.graceful_shutdown().await;
+}
+
+#[tokio::test]
+async fn native_edge_batch_round_trip() {
+    let server = TestServer::start().await;
+    let pool = PoolConfig::new(
+        format!("127.0.0.1:{}", server.native_port),
+        nodedb_types::protocol::AuthMethod::Trust {
+            username: "nodedb".into(),
+        },
+    );
+    let native = NativeClient::new(pool);
+    exercise(&native, "eb_native").await;
+    server.graceful_shutdown().await;
+}

nodedb-client-tests/tests/graph_shortest_path_default_round_trip.rs

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: BUSL-1.1
 
-//! End-to-end test that `NodeDb::graph_shortest_path` returns the
+//! End-to-end test that `NodeDbGraph::graph_shortest_path` returns the
 //! actual path between connected nodes.
 //!
 //! With edges A→B→C in place, `graph_shortest_path("social", A, C)`
@@ -10,7 +10,7 @@
 //! guards against), and not a wire-format mismatch between the client
 //! and the server's `GRAPH PATH` DSL.
 
-use nodedb_client::{NodeDb, NodeDbRemote, NodeId};
+use nodedb_client::{NodeDb, NodeDbGraph, NodeDbRemote, NodeId};
 use nodedb_test_support::pgwire_harness::TestServer;
 
 #[tokio::test]

nodedb-client-tests/tests/graph_traverse_remote_round_trip.rs

@@ -1,6 +1,6 @@
 // SPDX-License-Identifier: BUSL-1.1
 
-//! End-to-end test that `NodeDb::graph_traverse` returns the subgraph
+//! End-to-end test that `NodeDbGraph::graph_traverse` returns the subgraph
 //! reachable from edges inserted in the same session.
 //!
 //! After three `graph_insert_edge` calls fanning from a seed
@@ -13,7 +13,7 @@
 //! The test also checks In/Out/Both direction, edge orientation, reciprocal
 //! edges, a self-loop and discovery depths.
 
-use nodedb_client::{NodeDb, NodeDbRemote, NodeId};
+use nodedb_client::{NodeDb, NodeDbGraph, NodeDbRemote, NodeId};
 use nodedb_test_support::pgwire_harness::TestServer;
 
 #[tokio::test]

nodedb-client/src/lib.rs

@@ -52,7 +52,7 @@ mod remote_parse;
 pub mod native;
 
 pub use capabilities::Capabilities;
-pub use traits::NodeDb;
+pub use traits::{NodeDb, NodeDbGraph};
 
 #[cfg(feature = "remote")]
 pub use remote::NodeDbRemote;