jsam · Oct 4, 2026 · Oct 4, 2026
diff --git a/docs/content/docs/guides/websocket-api.mdx b/docs/content/docs/guides/websocket-api.mdx
@@ -196,6 +196,25 @@ A command that cannot join the transaction (`.kg use`, `.rel drop`, `.index`, `.
 
 An empty `errors` means every statement succeeded. Do not match message text to detect failures.
 
+### Statement Counts
+
+A program that commits fact statements lists them in `statements`, in program order, with the tuples each one changed. The list rides on the reply even when the program ends with a query, whose rows are the `rows`:
+
+```json
+{
+  "type": "result",
+  "columns": ["message"],
+  "rows": [["Inserted 1 fact(s) into 'task'."], ["Update: 0 deleted, 1 inserted."]],
+  "errors": [],
+  "statements": [
+    {"index": 0, "kind": "insert", "inserted": 1, "deleted": 0},
+    {"index": 1, "kind": "update", "inserted": 1, "deleted": 0}
+  ]
+}
+```
+
+`kind` is `insert`, `delete` or `update`. Counts are effective, as of the statements before it: inserting a fact that is present or deleting one that is absent is listed with zero counts. A conditional write that matched nothing is listed with zero counts too, so `inserted` of a guarded update says whether it landed. Schema and rule statements are not listed. `statements` is absent when the program committed no fact statement, including a program that failed and rolled back. Read counts from `statements`, not from the message rows.
+
 ## Knowledge Graph Operations
 
 Use meta commands via execute:

diff --git a/docs/spec/asyncapi.yaml b/docs/spec/asyncapi.yaml
@@ -715,6 +715,27 @@ components:
         message:
           type: string
 
+    StatementCounts:
+      type: object
+      required: [index, kind, inserted, deleted]
+      description: |
+        What a committed fact statement changed. Counts are effective, as of
+        the statements before it in the same program: `inserted` counts tuples
+        that were absent and are now present, `deleted` tuples that were
+        present and are now absent. A statement that changed nothing is listed
+        with zero counts.
+      properties:
+        index:
+          type: integer
+          description: 0-based statement index in the program
+        kind:
+          type: string
+          enum: [insert, delete, update]
+        inserted:
+          type: integer
+        deleted:
+          type: integer
+
     NoticeResponse:
       type: object
       required: [type, code, message]
@@ -1212,6 +1233,14 @@ components:
           items:
             $ref: '#/components/schemas/StatementError'
           description: Failed statements of the program; empty when every statement succeeded
+        statements:
+          type: array
+          items:
+            $ref: '#/components/schemas/StatementCounts'
+          description: |
+            Fact statements the program committed, with their counts, in
+            program order. Absent when the program committed no fact statement
+            (a read, a failed program, or only schema and rule changes).
         subscribed:
           $ref: '#/components/schemas/Subscribed'
           description: Set on the reply to `.subscribe`, whose `rows` are the snapshot
@@ -1312,6 +1341,10 @@ components:
           type: array
           items:
             $ref: '#/components/schemas/StatementError'
+        statements:
+          type: array
+          items:
+            $ref: '#/components/schemas/StatementCounts'
         subscribed:
           $ref: '#/components/schemas/Subscribed'
       examples:

diff --git a/src/bin/client.rs b/src/bin/client.rs
@@ -285,6 +285,7 @@ impl WsClient {
                         proof_trees: start.proof_trees,
                         timing_breakdown: start.timing_breakdown,
                         errors: start.errors,
+                        statements: start.statements,
                         subscribed: None,
                     });
                 }

diff --git a/src/protocol/handler.rs b/src/protocol/handler.rs
@@ -722,6 +722,7 @@ impl ProofSnapshot {
                 proof_us,
             ),
             errors: Vec::new(),
+            statements: Vec::new(),
         })
     }
 
@@ -772,6 +773,7 @@ impl ProofSnapshot {
             proof_trees: Some(vec![graph]),
             timing_breakdown: proof_timing(timing_mode, start, query_us, "explanation", explain_us),
             errors: Vec::new(),
+            statements: Vec::new(),
         })
     }
 }
@@ -1463,6 +1465,7 @@ impl Handler {
             proof_trees: None,
             timing_breakdown: None,
             errors: Vec::new(),
+            statements: Vec::new(),
         })
     }
 
@@ -2218,6 +2221,7 @@ impl Handler {
                     proof_trees: None,
                     timing_breakdown: None,
                     errors: Vec::new(),
+                    statements: Vec::new(),
                 });
             }
 
@@ -2282,6 +2286,7 @@ impl Handler {
                 proof_trees: None,
                 timing_breakdown: None,
                 errors: Vec::new(),
+                statements: Vec::new(),
             });
         }
 
@@ -3027,6 +3032,7 @@ impl QueryJob {
                                             proof_trees: None,
                                             timing_breakdown: None,
                                             errors,
+                                            statements: Vec::new(),
                                         });
                                     }
                                     MetaCommand::AgentStart(_)
@@ -3254,13 +3260,16 @@ impl QueryJob {
                 current_stmt.clear();
             }
         }
+        // Counts of the fact statements committed below.
+        let mut statement_counts = Vec::new();
         if !write_run.is_empty() {
             if errors.is_empty() {
-                if let Err(failure) =
-                    self.commit_write_run(&storage, &kg_name, &mut write_run, &mut messages)
-                {
-                    stmt_index = failure.index;
-                    fail!(failure.code, failure.message);
+                match self.commit_write_run(&storage, &kg_name, &mut write_run, &mut messages) {
+                    Ok(counts) => statement_counts = counts,
+                    Err(failure) => {
+                        stmt_index = failure.index;
+                        fail!(failure.code, failure.message);
+                    }
                 }
             } else if let Some(note) = write_run.abandon(errors[0].index) {
                 // The loop stopped at the first failure, so it is the last row.
@@ -3296,6 +3305,7 @@ impl QueryJob {
             return Ok(QueryResult {
                 switched_kg: switched_kg_result,
                 errors,
+                statements: statement_counts,
                 execution_time_ms: start.elapsed().as_millis() as u64,
                 ..Handler::messages_result(messages)
             });
@@ -3317,6 +3327,7 @@ impl QueryJob {
                         return Ok(QueryResult {
                             switched_kg: switched_kg_result,
                             errors,
+                            statements: statement_counts,
                             execution_time_ms: start.elapsed().as_millis() as u64,
                             ..Handler::messages_result(messages)
                         });
@@ -3511,6 +3522,7 @@ impl QueryJob {
             proof_trees: None,
             timing_breakdown,
             errors,
+            statements: statement_counts,
         })
     }
 }
@@ -3799,6 +3811,7 @@ impl Handler {
             proof_trees: None,
             timing_breakdown,
             errors: Vec::new(),
+            statements: Vec::new(),
         })
     }
 
@@ -4438,6 +4451,7 @@ impl Handler {
             proof_trees: None,
             timing_breakdown: None,
             errors: Vec::new(),
+            statements: Vec::new(),
         }
     }
 
@@ -4530,6 +4544,7 @@ impl Handler {
             proof_trees: None,
             timing_breakdown: None,
             errors: Vec::new(),
+            statements: Vec::new(),
         }
     }
 
@@ -5158,6 +5173,7 @@ impl Handler {
             proof_trees: None,
             timing_breakdown: None,
             errors: Vec::new(),
+            statements: Vec::new(),
         })
     }
 

diff --git a/src/protocol/handler/fact_staging.rs b/src/protocol/handler/fact_staging.rs
@@ -8,7 +8,7 @@
 
 use super::{format_body_pred, format_term, term_to_value};
 use crate::ast::Term;
-use crate::protocol::wire::ErrorCode;
+use crate::protocol::wire::{ErrorCode, StatementKind};
 use crate::statement::{DeleteOp, DeletePattern, InsertOp, UpdateOp};
 use crate::storage_engine::{FactChange, KnowledgeGraphSnapshot};
 use crate::value::{Tuple, Value};
@@ -105,6 +105,15 @@ impl FactStatement {
         }
     }
 
+    /// The form of this statement, as reported in a result's counts.
+    pub fn kind(&self) -> StatementKind {
+        match self {
+            Self::Insert(_) => StatementKind::Insert,
+            Self::Delete(_) => StatementKind::Delete,
+            Self::Update(_) => StatementKind::Update,
+        }
+    }
+
     /// Whether staging this statement reads the knowledge graph.
     pub fn reads_state(&self) -> bool {
         matches!(

diff --git a/src/protocol/handler/write_run.rs b/src/protocol/handler/write_run.rs
@@ -12,7 +12,7 @@
 use super::catalog_staging::CatalogStatement;
 use super::fact_staging::{FactStatement, InsertLimits, StageError};
 use super::{storage_error_code, QueryJob};
-use crate::protocol::wire::ErrorCode;
+use crate::protocol::wire::{ErrorCode, StatementCounts};
 use crate::rule_catalog::RuleCatalog;
 use crate::storage_engine::{
     CommitError, KnowledgeGraphSnapshot, ProgramCommit, StagedChanges, StatementEffect,
@@ -117,8 +117,9 @@ impl WriteRun {
 }
 
 impl QueryJob {
-    /// Commit the statements queued in `run` to `kg` as one transaction and
-    /// fill their message rows in `messages`.
+    /// Commit the statements queued in `run` to `kg` as one transaction, fill
+    /// their message rows in `messages` and return the counts of its fact
+    /// statements.
     ///
     /// A program whose staging read the KG is staged again, up to
     /// [`MAX_STAGE_ATTEMPTS`] times with a short random pause, if a concurrent
@@ -129,7 +130,7 @@ impl QueryJob {
         kg: &str,
         run: &mut WriteRun,
         messages: &mut [String],
-    ) -> Result<(), RunFailure> {
+    ) -> Result<Vec<StatementCounts>, RunFailure> {
         let commit = match self.commit_queued(storage, kg, &run.queued) {
             Ok(commit) => commit,
             Err(mut failure) => {
@@ -145,10 +146,17 @@ impl QueryJob {
 
         let queued = std::mem::take(&mut run.queued);
         let mut inserted_total = 0;
+        let mut counts = Vec::new();
         for (queued, outcome) in queued.iter().zip(&commit.statements) {
             let message = match (&queued.statement, &outcome.effect) {
                 (WriteStatement::Facts(statement), StatementEffect::Facts(count)) => {
                     inserted_total += count.inserted;
+                    counts.push(StatementCounts {
+                        index: queued.index,
+                        kind: statement.kind(),
+                        inserted: count.inserted,
+                        deleted: count.deleted,
+                    });
                     statement.success_message(count.inserted, count.deleted)
                 }
                 (WriteStatement::Catalog(statement), StatementEffect::Catalog(outcome)) => {
@@ -179,7 +187,7 @@ impl QueryJob {
                 change.inserted + change.deleted,
             );
         }
-        Ok(())
+        Ok(counts)
     }
 
     /// Stage `queued` and commit it, re-staging while it goes stale.

diff --git a/src/protocol/mod.rs b/src/protocol/mod.rs
@@ -39,7 +39,8 @@ pub use error::{InputLayerError, InputLayerResult};
 
 // Re-export wire types
 pub use wire::{
-    ColumnDef, ErrorCode, QueryResult, StatementError, WireDataType, WireTuple, WireValue,
+    ColumnDef, ErrorCode, QueryResult, StatementCounts, StatementError, StatementKind,
+    WireDataType, WireTuple, WireValue,
 };
 
 // Re-export handler

diff --git a/src/protocol/rest/handlers/ws/execute.rs b/src/protocol/rest/handlers/ws/execute.rs
@@ -144,6 +144,7 @@ fn result_frame(id: Option<RequestId>, response: QueryResult) -> ServerFrame {
         proof_trees,
         timing_breakdown: response.timing_breakdown,
         errors: response.errors,
+        statements: response.statements,
         subscribed: None,
     })
 }
@@ -173,6 +174,7 @@ pub(super) fn subscription_reply(
         proof_trees: None,
         timing_breakdown: None,
         errors: Vec::new(),
+        statements: Vec::new(),
         subscribed,
     }
 }

diff --git a/src/protocol/rest/handlers/ws/stream.rs b/src/protocol/rest/handlers/ws/stream.rs
@@ -75,6 +75,7 @@ fn stream_result(result: ResultFrame) -> Result<Vec<String>, Undeliverable> {
         proof_trees,
         timing_breakdown,
         errors,
+        statements,
         subscribed,
     } = result;
     let empty_chunk = ServerFrame::ResultChunk {
@@ -116,6 +117,7 @@ fn stream_result(result: ResultFrame) -> Result<Vec<String>, Undeliverable> {
         proof_trees,
         timing_breakdown,
         errors,
+        statements,
         subscribed,
     }))?);
     let mut rows = rows.into_iter();

diff --git a/src/protocol/rest/handlers/ws/stream/tests.rs b/src/protocol/rest/handlers/ws/stream/tests.rs
@@ -1,4 +1,4 @@
-use inputlayer_ws_protocol::{RequestId, Subscribed};
+use inputlayer_ws_protocol::{RequestId, StatementCounts, StatementKind, Subscribed};
 use serde_json::{json, Value};
 
 use super::*;
@@ -30,6 +30,7 @@ fn result(rows: Vec<Row>) -> ResultFrame {
         proof_trees: None,
         timing_breakdown: None,
         errors: Vec::new(),
+        statements: Vec::new(),
         subscribed: Some(Subscribed {
             subscription: "s".into(),
             generation: 2,
@@ -61,12 +62,25 @@ fn a_small_result_is_one_frame_serialized_as_before() {
 #[test]
 fn a_large_result_streams_as_one_bounded_ordered_result() {
     let rows: Vec<Row> = (0..3_000).map(|i| row(i, 1_000)).collect();
-    let frames = parsed(&result_frames(result(rows.clone())).unwrap());
+    let counts = vec![StatementCounts {
+        index: 0,
+        kind: StatementKind::Insert,
+        inserted: 3,
+        deleted: 0,
+    }];
+    let frames = parsed(
+        &result_frames(ResultFrame {
+            statements: counts.clone(),
+            ..result(rows.clone())
+        })
+        .unwrap(),
+    );
 
     let (_, ServerFrame::ResultStart(start)) = &frames[0] else {
         panic!("{:?}", frames[0]);
     };
     assert_eq!(start.id.as_ref().map(RequestId::as_str), Some("q"));
+    assert_eq!(start.statements, counts, "counts ride on the header");
     assert_eq!(start.subscribed.as_ref().map(|s| s.generation), Some(2));
     let (mut got, mut provenance) = (Vec::new(), Vec::new());
     for (index, (bytes, frame)) in frames[1..frames.len() - 1].iter().enumerate() {