jsam · Oct 3, 2026
diff --git a/TESTING.md b/TESTING.md
@@ -106,7 +106,8 @@ added/retracted rows as `subscription_delta` pushes, with contiguous `seq` and
 increasing `revision`, and end equal to a fresh full query on another
 connection, without re-querying.
 Scenarios cover one subscriber, 64 subscribers, reconnect/resubscribe and
-crash-restart, unrelated writes, and write bursts.
+crash-restart, unrelated writes, write bursts, and deltas arriving while the
+agent's own long query runs on its connection.
 
 Every request the harness sends carries an `id`, and a reply that does not
 echo it fails the scenario (`Violation::Uncorrelated`); a `notice` is never

diff --git a/config.example.toml b/config.example.toml
@@ -197,6 +197,11 @@ ws_max_messages_per_sec = 100
 # Maximum standing-query subscriptions (.subscribe) per WebSocket connection (0 = unlimited)
 ws_max_subscriptions = 64
 
+# Maximum requests per WebSocket connection admitted and not yet answered (at least 1).
+# Queries on one connection overlap up to this bound; writes, KG switches and session
+# changes still run one at a time, in order. Replies always come back in request order.
+ws_max_in_flight_requests = 16
+
 # Maximum HTTP requests per second per IP address (0 = unlimited, not recommended)
 per_ip_max_rps = 100
 

diff --git a/docs/content/docs/guides/configuration.mdx b/docs/content/docs/guides/configuration.mdx
@@ -225,6 +225,10 @@ ws_max_messages_per_sec = 1000
 # Maximum WebSocket connection lifetime in seconds (0 = unlimited, default: 24h)
 ws_max_lifetime_secs = 86400
 
+# Maximum requests per WebSocket connection admitted and not yet answered
+# (at least 1, default: 16). Queries overlap up to this bound; replies stay in request order.
+ws_max_in_flight_requests = 16
+
 # Maximum HTTP requests per second per IP address (0 = unlimited)
 per_ip_max_rps = 0
 

diff --git a/docs/content/docs/guides/websocket-api.mdx b/docs/content/docs/guides/websocket-api.mdx
@@ -97,12 +97,23 @@ Any request (`login`, `authenticate`, `execute`, `ping`) may carry an `id`, a no
 {"type": "execute", "id": "42", "program": "?edge(X, Y)"}
 ```
 
-Every frame answering it echoes the `id`: `authenticated`, `auth_error`, `result`, `result_start`, `result_chunk`, `result_end`, `error` and `pong`. A request without an `id` gets replies without one. Keep an `id` unique among your requests still awaiting a reply; it can be reused once answered. The server answers requests one at a time, in the order received.
+Every frame answering it echoes the `id`: `authenticated`, `auth_error`, `result`, `result_start`, `result_chunk`, `result_end`, `error` and `pong`. A request without an `id` gets replies without one. An `id` must be unique among your requests still awaiting a reply; it can be reused once answered. A request reusing an `id` that is still awaiting its reply gets an `invalid_request` error and does not run.
 
 Frames you did not ask for never carry an `id` and have their own types, so they cannot be read as a reply: [notifications](#notifications), [standing-query pushes](#standing-queries) and [notices](#notices).
 
 A request that is not a valid frame (bad JSON, unknown `type`, a missing field, or an `id` that is not a 1-64 byte string or appears twice) gets exactly one `error` with code `invalid_request`, carrying the `id` when it could be read. Nothing runs and the connection stays usable.
 
+### Pipelining
+
+A client may send several requests without waiting for each reply:
+
+- **Replies come back in request order**, one reply (or one streamed `result_start` ... `result_end` sequence) per request, even when a later request finishes first.
+- **Queries overlap.** A program made only of queries (`?...`) runs concurrently with the connection's other queries.
+- **Everything else runs alone, in order**: inserts, deletes, rules, session rules and facts, meta commands such as `.kg use`, `.session clear`, `.subscribe` and `.unsubscribe`. Such a request starts only after every earlier request has replied, and no later request starts before it replies. A query sent after `.kg use other` therefore always reads `other`, and an insert sent after it always writes there.
+- **Pushes are not held up by requests.** Notifications and subscription deltas are delivered while the connection's own requests run, between replies (never inside a streamed result).
+- At most `http.rate_limit.ws_max_in_flight_requests` requests (default 16) are admitted and not yet answered; at that bound the server stops reading the connection until a reply goes out.
+- Closing the connection abandons its running queries; a running write finishes before the session is closed.
+
 ### Result Response
 
 ```json

diff --git a/docs/spec/asyncapi.yaml b/docs/spec/asyncapi.yaml
@@ -59,9 +59,10 @@ info:
     Any request (`login`, `authenticate`, `execute`, `ping`) may carry an `id`: a
     non-empty string of at most 64 bytes. Every frame answering that request echoes it:
     `authenticated`, `auth_error`, `result`, `result_start`, `result_chunk`, `result_end`,
-    `error` and `pong`. A request without an `id` gets replies without one. Keep an `id`
-    unique among your requests still awaiting a reply; it may be reused once answered.
-    The server answers requests one at a time, in the order received.
+    `error` and `pong`. A request without an `id` gets replies without one. An `id` must
+    be unique among your requests still awaiting a reply; it may be reused once answered.
+    A request reusing the `id` of an unanswered request gets an `invalid_request` error
+    and does not run. Replies come back in request order (see Pipelining).
 
     Frames the client did not ask for never carry an `id` and have their own types, so
     they can never be mistaken for a reply:
@@ -89,6 +90,16 @@ info:
     6. Server may push change notifications (`persistent_update`, `rule_change`, ...), for `.subscribe`d queries `subscription_delta` / `subscription_error`, and connection `notice`s
     7. Client disconnects → Server auto-closes session
 
+    ## Pipelining
+
+    Clients may send requests without waiting for replies. Replies come back in
+    request order. Programs made only of queries run concurrently; any other
+    request (writes, rules, session changes, `.kg use`, `.subscribe`, ...) starts
+    after every earlier request has replied and finishes before any later one
+    starts. Pushes are delivered while requests run, between replies. At most
+    `http.rate_limit.ws_max_in_flight_requests` (default 16) requests are
+    unanswered at once; beyond that the server stops reading the connection.
+
     ## Raw Text Protocol
 
     The `execute` message accepts any valid IQL statement or meta command as raw text.

diff --git a/src/config.rs b/src/config.rs
@@ -484,6 +484,12 @@ pub struct RateLimitConfig {
     #[serde(default = "default_ws_max_subscriptions")]
     pub ws_max_subscriptions: usize,
 
+    /// Maximum requests per WebSocket connection admitted and not yet answered
+    /// (at least 1). Queries overlap up to this bound; at the bound the server
+    /// stops reading the connection until a reply goes out.
+    #[serde(default = "default_ws_max_in_flight_requests")]
+    pub ws_max_in_flight_requests: usize,
+
     /// Notification broadcast channel buffer size (per-subscriber queue depth)
     #[serde(default = "default_notification_buffer_size")]
     pub notification_buffer_size: usize,
@@ -647,6 +653,9 @@ fn default_ws_max_lifetime_secs() -> u64 {
 fn default_ws_max_subscriptions() -> usize {
     64
 }
+fn default_ws_max_in_flight_requests() -> usize {
+    16
+}
 fn default_notification_buffer_size() -> usize {
     4096
 }
@@ -665,6 +674,7 @@ impl Default for RateLimitConfig {
             ws_max_messages_per_sec: default_ws_max_messages_per_sec(),
             ws_max_lifetime_secs: default_ws_max_lifetime_secs(),
             ws_max_subscriptions: default_ws_max_subscriptions(),
+            ws_max_in_flight_requests: default_ws_max_in_flight_requests(),
             notification_buffer_size: default_notification_buffer_size(),
             per_ip_max_rps: default_per_ip_max_rps(),
             ws_max_preauth_per_ip: default_ws_max_preauth_per_ip(),
@@ -815,6 +825,12 @@ impl Config {
             self.http.ws_auth_timeout_ms = default_ws_auth_timeout_ms();
         }
 
+        // ws_max_in_flight_requests=0 would never read a request
+        if self.http.rate_limit.ws_max_in_flight_requests == 0 {
+            tracing::warn!("ws_max_in_flight_requests = 0 is invalid, auto-correcting to 16");
+            self.http.rate_limit.ws_max_in_flight_requests = default_ws_max_in_flight_requests();
+        }
+
         // persist buffer_size=0 would cause infinite flush loops
         if self.storage.persist.buffer_size == 0 {
             tracing::warn!("persist.buffer_size = 0 is invalid, auto-correcting to 1000");
@@ -1252,6 +1268,7 @@ mod tests {
         assert_eq!(rl.ws_max_messages_per_sec, 100);
         assert_eq!(rl.ws_max_lifetime_secs, 86400);
         assert_eq!(rl.ws_max_subscriptions, 64);
+        assert_eq!(rl.ws_max_in_flight_requests, 16);
         assert_eq!(rl.per_ip_max_rps, 100);
     }
 

diff --git a/src/protocol/handler.rs b/src/protocol/handler.rs
@@ -5193,6 +5193,20 @@ fn parse_program(program: &str) -> Result<Vec<statement::Statement>, Vec<Validat
     }
 }
 
+/// Whether every statement of `program`, split the way it executes, is a
+/// query (`?...`): such a program reads its KG and session without changing
+/// either. Lexical, so it costs no parse: a `?` statement parses as a query
+/// or fails without effect.
+pub(crate) fn is_query_program(program: &str) -> bool {
+    let text = join_continuation_lines(&strip_comments(program));
+    let mut statements = text.lines().map(str::trim).filter(|line| !line.is_empty());
+    let mut any = false;
+    statements.all(|line| {
+        any = true;
+        line.starts_with('?')
+    }) && any
+}
+
 /// Whether a statement creates, switches to or drops the system KG.
 fn targets_internal_kg(stmt: &statement::Statement) -> bool {
     matches!(