Skip to content

feat(protocol): subscription groups and snapshot reads at one revision - #278

Merged
jsam merged 4 commits into
mainfrom
fm/en-en15
Oct 5, 2026
Merged

feat(protocol): subscription groups and snapshot reads at one revision#278
jsam merged 4 commits into
mainfrom
fm/en-en15

Conversation

@jsam

@jsam jsam commented Oct 4, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor

Intent

I want you to really hyper focus on making the symbolic engine, you know, crisp and really truly production ready. Like I really need you to, you know, put insane amount of effort into doing that. And then once you have that, I will give you the endpoints, but only then.

(Context: approved plan ~/firstmate/data/nsdl-plan/report.md §6.1, GitHub milestone 7 'Engine production readiness'. This task is item EN-15: subscription groups: several queries in one subscribe frame, an atomic snapshot, one publication per commit touching any member, all members exact at one revision, unchanged members marked — why: coherent windows with one revision_seen (challenge-1 B2). M1 critical path)

Substance of the referenced item (plan §6.1 EN-15, GitHub issue #265 "Engine: snapshot multi-query reads and subscription groups at one revision"): a caller that renders a state window from several views must see them all at the same revision; today an unchanged view publishes nothing, so a subscriber cannot tell a view is still current at a revision, and a program with several ? queries replies with the last one only. Deliver (1) a snapshot read of several queries at one revision in one request, and (2) subscription groups: several queries in one subscribe frame, an atomic initial snapshot, one publication per commit that touches any member, all members exact at one revision, unchanged members marked. Done when both are in the protocol and both SDKs (Python and JS), with tests for atomicity under concurrent writes. Every engine PR must not regress millisecond query and update latency.

What Changed

  • The WebSocket protocol (ws-protocol, src/protocol) gets two new client frames. read runs several named ? queries on one pinned knowledge-graph snapshot and replies with a snapshot frame: one result per query, all exact at a single revision. Large replies stream as snapshot_start/snapshot_chunk/snapshot_end. A read sees persistent data only, uses the request's deadline and cancellation, and fails as a whole while naming the failing query. subscribe opens a subscription group of several queries. It starts with one atomic initial snapshot, then sends one subscription_group_delta (or a chunked start/chunk/end) per commit that touches any member. Every member is exact at that delta's revision, and members that didn't change are marked unchanged. A group ends with .unsubscribe.
  • Both SDKs support the new frames. The JS SDK adds kg.read() and kg.subscribeGroup() (GroupSubscription, GroupChange, MemberChange, ReadResult). The Python SDK adds the same set with matching connection, protocol and exception handling. Each SDK checks streamed snapshots and group deltas (chunk order, row counts, an error arriving mid-stream) before applying them.
  • Docs are updated: the AsyncAPI spec, the WebSocket/JS/Python guides and both SDK READMEs. New shared conformance fixtures and Rust tests cover group atomicity, reads under concurrent writes, cancellation and delivery (standing-query, differential-oracle, ws_cancel and ws_delivery tests), along with new JS and Python SDK tests. The subscription benchmarks are adjusted to the new code.

Risk Assessment

⚠️ Medium: The server core and both SDKs look correct: atomic group refresh, unchanged-member marking and chunked reassembly were all checked against the frame definitions. What remains is minor SDK option validation and error typing, but this is a large protocol change (version 5).

Testing

I built a release head server and a release base-commit server outside the worktree, each on its own throwaway data dir and port, and drove EN-15 end to end against them. Every live scenario passed. With raw /ws frames I checked: - the multi-query read; - a subscription group: unchanged members are marked, one push per commit, and no push when no member changes; - atomicity across 5 runs of 4 concurrent writers, with zero disagreements. Early runs of my own driver failed because it exceeded the engine's 100 msgs/s per-connection rate limit. Once I paced the driver, every run passed. The repo's live suites passed: Python subscriptions_live.py (17 tests) and JS subscriptions.integration.test.ts (15 tests). Scripts I wrote confirmed review fixes 1, 2 and 6 live in both SDKs; review-1 used a proxy that rewrites protocol_version to 4. Review-3 is a compile-time type shape with no runtime surface, so it is untested live. An offline tsc check with a negative control supports it. The latency probe showed base and head within noise. I tore down the servers, the proxy, the data dirs and the build dirs, and the worktree is clean. There is no UI, so the evidence is transcripts.

  • Live validation: ✅ go - 9 of 10 scenarios driven live against the product
Scenario Result Live Evidence
Raw /ws read frame with 3 named queries returns one snapshot at one revision, results in request order; limit honoured with truncated flag; a bad query fails the whole read naming it ('Query 'brok… ✅ pass live raw_protocol_transcript.txt (Scenario A)
Raw /ws subscribe group: atomic snapshot naming the subscription at one revision; a commit touching one member gives ONE subscription_group_delta listing every member in order, with the untouched me… ✅ pass live raw_protocol_transcript.txt (Scenario B)
Atomicity under concurrent writes (raw frames): 4 writers commit left(k)+right(k) together with a rule both(K); every group push and every read saw left == right == both, and the group converged to th… ✅ pass live atomicity_5_runs.txt: 5/5 runs, 1600 commits each, ~590-613 group pushes and ~180 reads checked per run, 0 disagreements
Python SDK kg.read / kg.subscribe_group live suite (reads at one revision, group deltas with unchanged members, atomicity under concurrent writers, streamed snapshot and group deltas, seq gap and rese… ✅ pass live py_subscriptions_live.txt: pytest tests/subscriptions_live.py, 17 passed
JS SDK kg.read / kg.subscribeGroup live integration suite (atomicity, unchanged members, streamed, gap and reset, refusals) ✅ pass live js_subscriptions_integration.txt: vitest run tests/subscriptions.integration.test.ts, 15 passed
review-1: SDK meets an engine reporting protocol_version 4 (pass-through proxy rewriting only the authenticated frame); read and subscribe_group / subscribeGroup fail with a plain ConnectionError ('ne… ✅ pass live py_sdk_review_fixes.txt, js_sdk_review_fixes.txt (v4_proxy.py in front of the real server)
review-2: Python kg.read refuses limit, offset or order_by given beside iql= with SubscriptionRejected('limit_offset'); an iql= target alone returns every row ✅ pass live py_sdk_review_fixes.txt
review-6: a CompileError raised while compiling one member of read or subscribe_group is still a CompileError, prefixed "Query 'broken': "; extra dict keys stay computed columns, as in kg.query ✅ pass live py_sdk_review_fixes.txt
review-3: JS kg.read accepts timeoutMs and signal; params is a type error ⏸️ untested no The earlier payload did not show a live result for this scenario. It is a compile-time TypeScript API shape with no runtime surface to drive against the running server. Only an offline tsc check was r…
No query or update latency regression: base (c550f35) and head release servers measured alternately, 600 queries and 600 updates each, plus head with a 3-member group live ✅ pass live latency_base_vs_head.txt: query p50 1.79-1.82ms on both; update p50 2.03-2.08ms on both; with a group live, query p50 2.06ms and update p50 2.24ms
Evidence: Raw /ws protocol transcript (read, group, atomicity) 

== Scenario A: one `read` frame answers several queries at one revision ==
  authenticated: {"type": "authenticated", "session_id": "1a47ee15-4406-469c-a6cb-643c25578009", "knowledge_graph": "default", "version": "0.1.0", "role": "admin", "protocol_version": 5, "stream_epoch": "e74637ad1fedb2e0"}
  PASS engine reports protocol_version 5
  <- {"type": "snapshot", "id": "w1", "knowledge_graph": "en15_h1", "revision": 3699, "results": [{"name": "orders", "columns": ["S", "O"], "rows": [["s1", 1042], ["s1", 1043]], "total_count": 2, "truncated": false}, {"name": "eta", "columns": ["O", "D"], "rows": [[1042, 3]], "total_count": 1, "truncated": false}, {"name": "first", "columns": ["S", "O"], "rows": [["s1", 1042]], "total_count": 2, "truncated": true}], "execution_time_ms": 0}
  PASS one snapshot frame, results in request order
  PASS snapshot carries one revision (3699)
  PASS each result exact (2 orders, 1 eta, limit 1 honoured)
  <- {"type": "error", "id": "w2", "message": "Query 'broken': Invalid atom: eta(", "code": "validation"}
  PASS a failing query fails the read, naming it
  <- {"type": "error", "id": "w3", "message": "A read names two queries 'a'; names must be unique.", "code": "validation"}
  PASS duplicate query names refused
  <- {"type": "snapshot", "id": "w4", "knowledge_graph": "en15_h1", "revision": 3699, "results": [{"name": "s", "columns": [], "rows": [], "total_count": 0, "truncated": false}], "execution_time_ms": 0}
  PASS session rule is not visible to a read (no rows; persistent data only)

== Scenario B: subscription group: atomic snapshot, one push per commit, unchanged marked ==
  <- {"type": "snapshot", "id": "s1", "knowledge_graph": "en15_h1", "revision": 3699, "results": [{"name": "orders", "columns": ["S", "O"], "rows": [["s1", 1042], ["s1", 1043]], "total_count": 2, "truncated": false}, {"name": "eta", "columns": ["O", "D"], "rows": [[1042, 3]], "total_count": 1, "truncated": false}, {"name": "other", "columns": [], "rows": [], "total_count": 0, "truncated": false}], "execution_time_ms": 0, "subscribed": {"subscription": "window", "generation": 1, "revision": 3699}}
  PASS subscribe answered by snapshot naming the subscription
  PASS all members at the snapshot's one revision
  push: {"type": "subscription_group_delta", "subscription": "window", "generation": 1, "knowledge_graph": "en15_h1", "seq": 1, "revision": 3700, "members": [{"name": "orders", "unchanged": true, "columns": ["S", "O"], "inserted": [], "retracted": []}, {"name": "eta", "unchanged": false, "columns": ["O", "D"], "inserted": [[1043, 5]], "retracted": []}, {"name": "other", "unchanged": true, "columns": [], "inserted": [], "retracted": []}]}
  PASS one push for a commit touching one member
  PASS push lists every member in group order
  PASS changed member carries its delta
  PASS untouched members marked unchanged
  PASS seq 1, revision advanced
  push: {"type": "subscription_group_delta", "subscription": "window", "generation": 1, "knowledge_graph": "en15_h1", "seq": 2, "revision": 3701, "members": [{"name": "orders", "unchanged": false, "columns": ["S", "O"], "inserted": [["s2", 2000]], "retracted": []}, {"name": "eta", "unchanged": false, "columns": ["O", "D"], "inserted": [], "retracted": [[1042, 3]]}, {"name": "other", "unchanged": true, "columns": [], "inserted": [], "retracted": []}]}
  PASS a commit touching two members is ONE push
  PASS both deltas in that push, the third unchanged
  PASS seq has no gap
  PASS a commit touching no member pushes nothing
  PASS snapshot+deltas match a fresh read
  <- {"type": "result", "id": "x11", "columns": ["message"], "rows": [["Unsubscribed 'window'."]], "row_count": 1, "total_count": 1, "truncated": false, "execution_time_ms": 0, "errors": []}
  PASS no pushes after .unsubscribe
  <- {"type": "error", "id": "s2", "message": "Subscriptions track the whole result set; remove limit/offset from the query."}
  PASS group member with limit refused

== Scenario C: atomicity under concurrent writers (raw frames) ==
  4 writers, 1600 committed two-relation programs in 7.3s; 594 group pushes, 180 reads checked
  PASS no group push and no read ever saw left/right/both disagree
  PASS checks actually ran under load
  PASS group state converged to the final read

RESULT: ALL PASS
Evidence: Atomicity under 4 concurrent writers, 5 runs 
== Scenario C: atomicity under concurrent writers (raw frames) ==
  4 writers, 1600 committed two-relation programs in 7.3s; 594 group pushes, 180 reads checked
RESULT: ALL PASS
== Scenario C: atomicity under concurrent writers (raw frames) ==
  4 writers, 1600 committed two-relation programs in 7.3s; 613 group pushes, 180 reads checked
RESULT: ALL PASS
== Scenario C: atomicity under concurrent writers (raw frames) ==
  4 writers, 1600 committed two-relation programs in 7.2s; 609 group pushes, 179 reads checked
RESULT: ALL PASS
== Scenario C: atomicity under concurrent writers (raw frames) ==
  4 writers, 1600 committed two-relation programs in 7.2s; 589 group pushes, 179 reads checked
RESULT: ALL PASS
== Scenario C: atomicity under concurrent writers (raw frames) ==
  4 writers, 1600 committed two-relation programs in 7.3s; 613 group pushes, 181 reads checked
RESULT: ALL PASS
Evidence: Python SDK live suite output 
============================= test session starts ==============================
platform linux -- Python 3.12.3, pytest-9.1.1, pluggy-1.6.0 -- ~/.no-mistakes/worktrees/9eea230c3aae/01M4666NEWPP35AW24S71PED4H/packages/inputlayer-py/.venv/bin/python
rootdir: ~/.no-mistakes/worktrees/9eea230c3aae/01M4666NEWPP35AW24S71PED4H/packages/inputlayer-py
configfile: pyproject.toml
plugins: langsmith-0.12.1, asyncio-1.3.0, anyio-4.13.0
asyncio: mode=Mode.AUTO, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collecting ... collected 17 items

tests/subscriptions_live.py::test_history_matches_the_oracle_generator PASSED [  5%]
tests/subscriptions_live.py::test_snapshot_plus_deltas_equals_a_fresh_query_over_random_histories PASSED [ 11%]
tests/subscriptions_live.py::test_projection_two_supporting_tuples PASSED [ 17%]
tests/subscriptions_live.py::test_streamed_snapshot_and_streamed_deltas_are_assembled PASSED [ 23%]
tests/subscriptions_live.py::test_a_seq_gap_and_a_reset_end_in_unverified_then_an_exact_resync PASSED [ 29%]
tests/subscriptions_live.py::test_a_slow_consumer_gets_unverified_then_a_resync_once_it_has_read PASSED [ 35%]
tests/subscriptions_live.py::test_reconnect_gives_unverified_at_once_then_only_what_changed PASSED [ 41%]
tests/subscriptions_live.py::test_an_acl_revoke_ends_the_subscription_with_access_denied PASSED [ 47%]
tests/subscriptions_live.py::test_refused_before_or_at_the_engine PASSED [ 52%]
tests/subscriptions_live.py::test_watch_yields_the_whole_result_and_on_calls_back PASSED [ 58%]
tests/subscriptions_live.py::test_read_answers_several_queries_at_one_revision PASSED [ 64%]
tests/subscriptions_live.py::test_group_snapshot_then_deltas_with_unchanged_members PASSED [ 70%]
tests/subscriptions_live.py::test_groups_and_reads_are_atomic_under_concurrent_writes PASSED [ 76%]
tests/subscriptions_live.py::test_streamed_read_group_snapshot_and_group_deltas_are_assembled PASSED [ 82%]
tests/subscriptions_live.py::test_a_group_seq_gap_and_reset_end_in_unverified_then_an_exact_resync PASSED [ 88%]
tests/subscriptions_live.py::test_a_group_reconnect_gives_unverified_then_only_what_changed PASSED [ 94%]
tests/subscriptions_live.py::test_groups_refused_at_the_engine PASSED    [100%]

============================= 17 passed in 14.95s ==============================
Evidence: JS SDK live integration output 

 RUN  v3.2.7 ~/.no-mistakes/worktrees/9eea230c3aae/01M4666NEWPP35AW24S71PED4H/packages/inputlayer-js

(node:2297498) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(Use `node --trace-warnings ...` to show where the warning was created)
(node:2297498) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:2297498) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:2297498) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:2297498) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
(node:2297498) MaxListenersExceededWarning: Possible EventTarget memory leak detected. 11 disconnected listeners added to EventTarget. MaxListeners is 10. Use events.setMaxListeners() to increase limit
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > snapshot plus deltas equals a fresh query at every checkpoint of random histories 7540ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > projection_two_supporting_tuples: a projected row leaves with its last support 146ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > assembles a streamed snapshot and streamed deltas 382ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > a seq gap and a reset end in unverified, then an exact resync 100ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > a slow consumer gets unverified, then a resync once it has read the queue 582ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > reconnects: unverified at once, then a resync with only what changed meanwhile 171ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > an ACL revoke ends the subscription: unverified, then SubscriptionRejectedError 120ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > refuses what a standing query cannot track, before or at the engine 47ms
 ✓ tests/subscriptions.integration.test.ts > Live: subscriptions > watch yields the whole result; on calls back with each change 54ms
 ✓ tests/subscriptions.integration.test.ts > Live: snapshot reads and subscription groups > atomicity: members a program changes together are equal in every event and every read 2536ms
 ✓ tests/subscriptions.integration.test.ts > Live: snapshot reads and subscription groups > a group yields every member at one revision, marks unchanged members, and converges 138ms
 ✓ tests/subscriptions.integration.test.ts > Live: snapshot reads and subscription groups > assembles a streamed read, a streamed group snapshot and streamed group deltas 342ms
 ✓ tests/subscriptions.integration.test.ts > Live: snapshot reads and subscription groups > a seq gap and a reset end in unverified, then an exact resync of every member 82ms
 ✓ tests/subscriptions.integration.test.ts > Live: snapshot reads and subscription groups > a read takes limit and orderBy, sees persistent data only, and fails naming the query 51ms
 ✓ tests/subscriptions.integration.test.ts > Live: snapshot reads and subscription groups > refuses a group the engine cannot track, typed 46ms

 Test Files  1 passed (1)
      Tests  15 passed (15)
   Start at  14:25:44
   Duration  14.53s (transform 213ms, setup 0ms, collect 263ms, tests 14.02s, environment 0ms, prepare 69ms)
Evidence: Python SDK review-fix checks (review-1, -2, -6) 
  read -> revision 7768, sizes {'e': 20, 'raw': 20}
  PASS iql= target alone reads every row
review-2: limit/offset/order_by beside iql= are refused, typed
    limit: SubscriptionRejected('limit_offset'): Query 'a': iql= takes the whole query as written: drop limit, offset and order_by.
  PASS limit beside iql= -> SubscriptionRejected limit_offset
    offset: SubscriptionRejected('limit_offset'): Query 'a': iql= takes the whole query as written: drop limit, offset and order_by.
  PASS offset beside iql= -> SubscriptionRejected limit_offset
    order_by: SubscriptionRejected('limit_offset'): Query 'a': iql= takes the whole query as written: drop limit, offset and order_by.
  PASS order_by beside iql= -> SubscriptionRejected limit_offset
review-6: a CompileError from one member names that member, type kept
    read: CompileError: Query 'broken': read takes a query or iql=, not both (put the whole query in iql=, or drop iql=)
  PASS read: CompileError prefixed with the member name
    subscribe_group: CompileError: Query 'broken': subscribe takes a query or iql=, not both (put the whole query in iql=, or drop iql=)
  PASS subscribe_group: CompileError prefixed with the member name
review-6: extra dict keys stay computed columns (unchanged, as kg.query)
    read rows[:3]: [Row(a=0, dbl=0), Row(a=1, dbl=2), Row(a=2, dbl=4)]  kg.query rows[:3]: [namespace(a=0, dbl=0), namespace(a=1, dbl=2), namespace(a=2, dbl=4)]
  PASS extra key 'dbl' is a computed column in a read target
review-1: engine speaking protocol 4 (via rewriting proxy): refusal reaches the user unchanged
    v4 query rows: 20 E
  PASS plain query still works against the v4 engine
    read: InputLayerConnectionError: The engine speaks /ws protocol 4; reads and subscription groups need version 5. Upgrade the engine.
  PASS read: plain connection error, not ConnectionLost
    subscribe_group: InputLayerConnectionError: The engine speaks /ws protocol 4; reads and subscription groups need version 5. Upgrade the engine.
  PASS subscribe_group: plain connection error, not ConnectionLost
RESULT: ALL PASS
Evidence: JS SDK review-1 and group smoke 
  read -> {"revision":7773,"sizes":{"e":2,"f":1}}
  PASS kg.read: both queries answered in one snapshot
  group event 1: snapshot revision 7773
  group event 2: delta revision 7774 {"e":{"unchanged":false,"inserted":1},"f":{"unchanged":true,"inserted":0}}
  PASS subscribeGroup: one delta, e changed, f marked unchanged, revision advanced
review-1: engine reporting protocol_version 4 (rewriting proxy)
  PASS plain query still works against the v4 engine
    read: ConnectionError: The engine speaks /ws protocol 4; reads and subscription groups need version 5. Upgrade the engine.
  PASS read: plain ConnectionError, not ConnectionLostError
    subscribeGroup: ConnectionError: The engine speaks /ws protocol 4; reads and subscription groups need version 5. Upgrade the engine.
  PASS subscribeGroup: plain ConnectionError, not ConnectionLostError
RESULT: ALL PASS
Evidence: review-3 ReadOptions type check (offline) 
tsc exit 0 (0 = @ts-expect-error on {params} is satisfied, i.e. params is rejected; timeoutMs/signal accepted)
../../../../../../../../tmp/nm-en15-neg.ts(5,41): error TS2353: Object literal may only specify known properties, and 'params' does not exist in type 'ReadOptions'.
Evidence: Latency base vs head

base query p50 1.788/1.823ms, update p50 2.081/2.047ms; head query p50 1.805/1.791ms, update p50 2.025/2.038ms; head with a group live: query 2.061ms, update 2.244ms

[base run 1]
query   n=600 p50=1.788ms p95=1.905ms p99=2.002ms
update  n=600 p50=2.081ms p95=4.831ms p99=9.319ms
[head run 1]
query   n=600 p50=1.805ms p95=1.994ms p99=2.253ms
update  n=600 p50=2.025ms p95=3.900ms p99=9.524ms
[base run 2]
query   n=600 p50=1.823ms p95=2.000ms p99=2.159ms
update  n=600 p50=2.047ms p95=3.802ms p99=8.596ms
[head run 2]
query   n=600 p50=1.791ms p95=2.042ms p99=2.274ms
update  n=600 p50=2.038ms p95=3.191ms p99=9.399ms
[head run 3, with a 3-member subscription group live on the KG]
query   n=600 p50=2.061ms p95=2.252ms p99=2.350ms
update  n=600 p50=2.244ms p95=3.628ms p99=8.439ms
Evidence: Driver and proxy scripts 
"""Drive EN-15 at the wire level against a live inputlayer-server.

Prints a transcript of real frames: a multi-query `read`, a `subscribe`
group with unchanged-member marking, and an atomicity check under concurrent
writers (raw frames, no SDK).
"""
import asyncio, json, os, random, sys, time
import websockets

URL = os.environ["IL_URL"]
PW = os.environ["IL_PW"]
KG = os.environ.get("IL_KG", "en15_raw")


def show(tag, frame):
    s = json.dumps(frame)
    print(f"  {tag} {s if len(s) < 900 else s[:900] + '...'}")


async def connect(kg=KG):
    ws = await websockets.connect(URL, max_size=None)
    await ws.send(json.dumps({"type": "login", "username": "admin", "password": PW}))
    auth = json.loads(await ws.recv())
    assert auth["type"] == "authenticated", auth
    await ex(ws, f".kg use {kg}")
    return ws, auth


_n = 0


async def ex(ws, program, pushes=None):
    global _n
    _n += 1
    rid = f"x{_n}"
    await ws.send(json.dumps({"type": "execute", "id": rid, "program": program}))
    while True:
        f = json.loads(await ws.recv())
        if f.get("id") == rid:
            return f
        if pushes is not None:
            pushes.append(f)


async def req(ws, frame, pushes=None):
    await ws.send(json.dumps(frame))
    while True:
        f = json.loads(await ws.recv())
        if f.get("id") == frame["id"]:
            return f
        if pushes is not None:
            pushes.append(f)


async def drain(ws, secs=0.6):
    out = []
    try:
        while True:
            out.append(json.loads(await asyncio.wait_for(ws.recv(), secs)))
    except asyncio.TimeoutError:
        return out


def check(cond, msg):
    print(("  PASS " if cond else "  FAIL ") + msg)
    if not cond:
        check.failed = True


check.failed = False


async def scenario_read():
    print("\n== Scenario A: one `read` frame answers several queries at one revision ==")
    ws, auth = await connect("default")
    show("authenticated:", auth)
    check(auth["protocol_version"] == 5, "engine reports protocol_version 5")
    await ex(ws, f".kg create {KG}")
    await ex(ws, f".kg use {KG}")
    await ex(ws, "+order[(\"s1\", 1042), (\"s1\", 1043)]\n+eta[(1042, 3)]")
    r = await req(ws, {"type": "read", "id": "w1", "timeout_ms": 2000, "queries": [
        {"name": "orders", "query": "?order(S, O)"},
        {"name": "eta", "query": "?eta(O, D)"},
        {"name": "first", "query": "?order(S, O), limit(1)"}]})
    show("<-", r)
    check(r["type"] == "snapshot" and [x["name"] for x in r["results"]] == ["orders", "eta", "first"],
          "one snapshot frame, results in request order")
    check(isinstance(r.get("revision"), int), f"snapshot carries one revision ({r.get('revision')})")
    check(len(r["results"][0]["rows"]) == 2 and len(r["results"][1]["rows"]) == 1
          and len(r["results"][2]["rows"]) == 1, "each result exact (2 orders, 1 eta, limit 1 honoured)")
    bad = await req(ws, {"type": "read", "id": "w2", "queries": [
        {"name": "ok", "query": "?order(S, O)"}, {"name": "broken", "query": "?eta("}]})
    show("<-", bad)
    check(bad["type"] == "error" and "Query 'broken'" in json.dumps(bad), "a failing query fails the read, naming it")
    dup = await req(ws, {"type": "read", "id": "w3", "queries": [
        {"name": "a", "query": "?eta(O, D)"}, {"name": "a", "query": "?eta(O, D)"}]})
    show("<-", dup)
    check(dup["type"] == "error", "duplicate query names refused")
    await ex(ws, "sess(O) <- order(_, O)")
    sv = await req(ws, {"type": "read", "id": "w4", "queries": [{"name": "s", "query": "?sess(O)"}]})
    show("<-", sv)
    check(sv["type"] == "error" or sv["results"][0]["rows"] == [],
          "session rule is not visible to a read (no rows; persistent data only)")
    await ws.close()


async def scenario_group():
    print("\n== Scenario B: subscription group: atomic snapshot, one push per commit, unchanged marked ==")
    ws, _ = await connect()
    sub = await req(ws, {"type": "subscribe", "id": "s1", "subscription": "window", "queries": [
        {"name": "orders", "query": "?order(S, O)"},
        {"name": "eta", "query": "?eta(O, D)"},
        {"name": "other", "query": "?unrelated(X)"}]})
    show("<-", sub)
    check(sub["type"] == "snapshot" and sub.get("subscribed", {}).get("subscription") == "window",
          "subscribe answered by snapshot naming the subscription")
    check(sub["subscribed"]["revision"] == sub["revision"], "all members at the snapshot's one revision")
    w, _ = await connect()
    pushes = []
    # Commit touching only eta.
    await ex(w, "+eta[(1043, 5)]")
    pushes += await drain(ws)
    gd = [p for p in pushes if p["type"] == "subscription_group_delta"]
    for p in gd:
        show("push:", p)
    check(len(gd) == 1, "one push for a commit touching one member")
    if gd:
        m = {x["name"]: x for x in gd[0]["members"]}
        check([x["name"] for x in gd[0]["members"]] == ["orders", "eta", "other"], "push lists every member in group order")
        check(m["eta"]["unchanged"] is False and m["eta"]["inserted"] == [[1043, 5]], "changed member carries its delta")
        check(m["orders"]["unchanged"] and m["other"]["unchanged"], "untouched members marked unchanged")
        check(gd[0]["revision"] > sub["revision"] and gd[0]["seq"] == 1, "seq 1, revision advanced")
    # Commit touching two members in one program.
    await ex(w, "+order[(\"s2\", 2000)]\n-eta[(1042, 3)]")
    p2 = [p for p in await drain(ws) if p["type"] == "subscription_group_delta"]
    for p in p2:
        show("push:", p)
    check(len(p2) == 1, "a commit touching two members is ONE push")
    if p2:
        m = {x["name"]: x for x in p2[0]["members"]}
        check(m["orders"]["inserted"] == [["s2", 2000]] and m["eta"]["retracted"] == [[1042, 3]]
              and m["other"]["unchanged"], "both deltas in that push, the third unchanged")
        check(p2[0]["seq"] == 2, "seq has no gap")
    # Commit touching no member.
    await ex(w, "+noise(1)")
    p3 = [p for p in await drain(ws) if p["type"].startswith("subscription")]
    check(p3 == [], "a commit touching no member pushes nothing")
    # Delta equals a fresh read.
    r = await req(w, {"type": "read", "id": "chk", "queries": [
        {"name": "orders", "query": "?order(S, O)"}, {"name": "eta", "query": "?eta(O, D)"}]})
    check(sorted(map(tuple, r["results"][0]["rows"])) == [("s1", 1042), ("s1", 1043), ("s2", 2000)]
          and r["results"][1]["rows"] == [[1043, 5]], "snapshot+deltas match a fresh read")
    un = await ex(ws, ".unsubscribe window")
    show("<-", un)
    await ex(w, "+eta[(9, 9)]")
    check([p for p in await drain(ws) if p["type"].startswith("subscription")] == [], "no pushes after .unsubscribe")
    bad = await req(ws, {"type": "subscribe", "id": "s2", "subscription": "pg", "queries": [
        {"name": "p", "query": "?order(S, O), limit(1)"}]})
    show("<-", bad)
    check(bad["type"] == "error", "group member with limit refused")
    await ws.close(); await w.close()


async def scenario_atomic():
    print("\n== Scenario C: atomicity under concurrent writers (raw frames) ==")
    setup, _ = await connect()
    await ex(setup, "+left(0)\n+right(0)\n+both(K) <- left(K), right(K)")
    sub, _ = await connect()
    snap = await req(sub, {"type": "subscribe", "id": "g", "subscription": "atom", "queries": [
        {"name": "left", "query": "?left(K)"}, {"name": "right", "query": "?right(K)"},
        {"name": "both", "query": "?both(K)"}]})
    state = {r["name"]: set(map(tuple, r["rows"])) for r in snap["results"]}
    stats = {"pushes": 0, "reads": 0, "bad": 0, "writes": 0, "last_seq": 0, "last_rev": snap["revision"]}
    done = asyncio.Event()

    async def listen():
        while not done.is_set() or True:
            try:
                f = json.loads(await asyncio.wait_for(sub.recv(), 1.5))
            except asyncio.TimeoutError:
                if done.is_set():
                    return
                continue
            if f["type"] in ("persistent_update", "rule_change", "kg_change", "schema_change"):
                continue  # data-change notifications, not group pushes
            if f["type"] != "subscription_group_delta":
                print("  unexpected", f); stats["bad"] += 1; continue
            assert f["seq"] == stats["last_seq"] + 1 and f["revision"] > stats["last_rev"]
            stats["last_seq"], stats["last_rev"] = f["seq"], f["revision"]
            for m in f["members"]:
                s = state[m["name"]]
                s -= set(map(tuple, m["retracted"])); s |= set(map(tuple, m["inserted"]))
            stats["pushes"] += 1
            if not (state["left"] == state["right"] == state["both"]):
                stats["bad"] += 1

    async def reader():
        ws, _ = await connect()
        i = 0
        while not done.is_set():
            i += 1
            r = await req(ws, {"type": "read", "id": f"r{i}", "queries": [
                {"name": "left", "query": "?left(K)"}, {"name": "right", "query": "?right(K)"},
                {"name": "both", "query": "?both(K)"}]})
            if r.get("code") == "rate_limited":  # engine's per-connection message rate cap
                await asyncio.sleep(0.05); continue
            if r["type"] != "snapshot":
                print("  read reply", r); stats["bad"] += 1; continue
            v = [sorted(map(tuple, x["rows"])) for x in r["results"]]
            stats["reads"] += 1
            await asyncio.sleep(0.03)  # stay well under 100 msgs/sec
            if not (v[0] == v[1] == v[2]):
                stats["bad"] += 1
        await ws.close()

    async def writer(seed):
        ws, _ = await connect()
        rng = random.Random(seed)
        for _ in range(400):
            k = rng.randrange(30)
            op = "+" if rng.random() < 0.5 else "-"
            r = await ex(ws, f"{op}left({k})\n{op}right({k})")
            if r["type"] != "result":
                print("  write reply", r); stats["bad"] += 1
            stats["writes"] += 1
            await asyncio.sleep(0.012)  # stay under the 100 msgs/sec per-connection cap
        await ws.close()

    lt = asyncio.ensure_future(listen()); rt = asyncio.ensure_future(reader())
    t0 = time.time()
    await asyncio.gather(*(writer(s) for s in range(4)))
    done.set(); await rt; await lt
    final = await req(setup, {"type": "read", "id": "fin", "queries": [
        {"name": "left", "query": "?left(K)"}, {"name": "both", "query": "?both(K)"}]})
    fin = set(map(tuple, final["results"][0]["rows"]))
    print(f"  4 writers, {stats['writes']} committed two-relation programs in {time.time()-t0:.1f}s; "
          f"{stats['pushes']} group pushes, {stats['reads']} reads checked")
    check(stats["bad"] == 0, "no group push and no read ever saw left/right/both disagree")
    check(stats["pushes"] > 20 and stats["reads"] > 20, "checks actually ran under load")
    check(state["left"] == fin and state["both"] == fin, "group state converged to the final read")
    await setup.close(); await sub.close()


async def main():
    await scenario_read()
    await scenario_group()
    await scenario_atomic()
    print("\nRESULT:", "FAIL" if check.failed else "ALL PASS")
    sys.exit(1 if check.failed else 0)

asyncio.run(main())

Pipeline

Updates from git push no-mistakes

✅ **intent** - passed

✅ No issues found.

✅ **Rebase** - passed

✅ No issues found.

⚠️ **Review** - 2 infos
  • ⚠️ packages/inputlayer-js/src/subscription.ts:721 - A protocol-version mismatch reaches the user as a retryable ConnectionLostError. When kg.subscribeGroup() runs against an engine reporting protocol_version 4, the pump in connection.ts:642-651 rejects the call with a plain ConnectionError ('need version 5'). openWithBackoff then rewraps every non-lost ConnectionError as ConnectionLostError('Connection lost: ...', 'closed'). The same thing happens on a resync after reconnecting to an older engine. kg.read returns the plain ConnectionError correctly. Fix: let non-lost ConnectionErrors through unchanged.
  • ⚠️ packages/inputlayer-py/src/inputlayer/knowledge_graph.py:883 - _shape returns early for a target given as iql= and silently drops limit, offset and order_by, which now mean something for read. For example, await kg.read({'a': {'iql': '?a(X)', 'limit': 10}}) returns every row instead of 10 and raises no error. Fix: refuse limit, offset and order_by alongside iql=.
  • ℹ️ packages/inputlayer-js/src/knowledge-graph.ts:593 - kg.read accepts ExecuteOptions, which includes params. The params are stored on the call but only execute sends them, so kg.read(..., {params: {x: 1}}) drops them silently. Fix: use a narrower options type for read, or refuse params.
  • ℹ️ packages/inputlayer-js/src/subscription.ts:1108 - Unverified group events mark every member unchanged: true. That contradicts the documented meaning 'exact at the event's revision', because an unverified event carries the last verified revision.
  • ℹ️ packages/inputlayer-py/src/inputlayer/subscription.py:470 - read's session-rule check calls list_rules() with the connection's default timeout, not the read's timeout. With timeout=None and a short default, a slow '.session list' can fail a read that succeeded. Fix: pass the read's timeout to list_rules().
  • ℹ️ packages/inputlayer-py/src/inputlayer/knowledge_graph.py:848 - Only SubscriptionRejected errors get the 'Query <name>:' prefix, so a CompileError from one member doesn't say which member failed. Also, misspelled keys in a dict target (e.g. 'limt') are taken as computed columns instead of being rejected.

🔧 Fix applied.
2 infos still open:

  • ℹ️ packages/inputlayer-js/src/subscription.ts:1108 - Unverified group events mark every member unchanged: true. That contradicts the documented meaning 'exact at the event's revision', because an unverified event carries the last verified revision.
  • ℹ️ packages/inputlayer-py/src/inputlayer/subscription.py:470 - read's session-rule check calls list_rules() with the connection's default timeout, not the read's timeout. With timeout=None and a short default, a slow '.session list' can fail a read that succeeded. Fix: pass the read's timeout to list_rules().
✅ **Test** - passed

✅ No issues found.

  • Live validation: ✅ go - 9 of 10 scenarios driven live against the product
Scenario Result Live Evidence
Raw /ws read frame with 3 named queries returns one snapshot at one revision, results in request order; limit honoured with truncated flag; a bad query fails the whole read naming it ('Query 'brok… ✅ pass live raw_protocol_transcript.txt (Scenario A)
Raw /ws subscribe group: atomic snapshot naming the subscription at one revision; a commit touching one member gives ONE subscription_group_delta listing every member in order, with the untouched me… ✅ pass live raw_protocol_transcript.txt (Scenario B)
Atomicity under concurrent writes (raw frames): 4 writers commit left(k)+right(k) together with a rule both(K); every group push and every read saw left == right == both, and the group converged to th… ✅ pass live atomicity_5_runs.txt: 5/5 runs, 1600 commits each, ~590-613 group pushes and ~180 reads checked per run, 0 disagreements
Python SDK kg.read / kg.subscribe_group live suite (reads at one revision, group deltas with unchanged members, atomicity under concurrent writers, streamed snapshot and group deltas, seq gap and rese… ✅ pass live py_subscriptions_live.txt: pytest tests/subscriptions_live.py, 17 passed
JS SDK kg.read / kg.subscribeGroup live integration suite (atomicity, unchanged members, streamed, gap and reset, refusals) ✅ pass live js_subscriptions_integration.txt: vitest run tests/subscriptions.integration.test.ts, 15 passed
review-1: SDK meets an engine reporting protocol_version 4 (pass-through proxy rewriting only the authenticated frame); read and subscribe_group / subscribeGroup fail with a plain ConnectionError ('ne… ✅ pass live py_sdk_review_fixes.txt, js_sdk_review_fixes.txt (v4_proxy.py in front of the real server)
review-2: Python kg.read refuses limit, offset or order_by given beside iql= with SubscriptionRejected('limit_offset'); an iql= target alone returns every row ✅ pass live py_sdk_review_fixes.txt
review-6: a CompileError raised while compiling one member of read or subscribe_group is still a CompileError, prefixed "Query 'broken': "; extra dict keys stay computed columns, as in kg.query ✅ pass live py_sdk_review_fixes.txt
review-3: JS kg.read accepts timeoutMs and signal; params is a type error ⏸️ untested no The earlier payload did not show a live result for this scenario. It is a compile-time TypeScript API shape with no runtime surface to drive against the running server. Only an offline tsc check was r…
No query or update latency regression: base (c550f35) and head release servers measured alternately, 600 queries and 600 updates each, plus head with a 3-member group live ✅ pass live latency_base_vs_head.txt: query p50 1.79-1.82ms on both; update p50 2.03-2.08ms on both; with a group live, query p50 2.06ms and update p50 2.24ms
  • cargo build --release --bin inputlayer-server (head, CARGO_TARGET_DIR in /tmp), server on a throwaway --data-dir, port 18734
  • python raw_protocol_driver.py: raw /ws read, subscribe group and concurrent-writer atomicity (5 repeated runs)
  • pytest tests/subscriptions_live.py -v against the live server (17 passed)
  • vitest run tests/subscriptions.integration.test.ts against the live server (15 passed)
  • python py_sdk_review_fixes.py: review-1 (via v4_proxy.py), review-2 and review-6 in the Python SDK, live
  • node js_sdk_review_fixes.cjs (esbuild bundle of the worktree src): JS read and subscribeGroup smoke plus review-1 via v4_proxy.py, live
  • tsc --noEmit read_options_typecheck.ts plus a negative control (review-3, offline compile-time check)
  • git archive c550f359 plus a release build of the base server; latency_probe.py alternating base and head, and head with a group live
✅ **Document** - passed

✅ No issues found.

✅ **Lint** - passed

✅ No issues found.

✅ **Push** - passed

✅ No issues found.

@jsam
Merge remote-tracking branch 'origin/main' into fm/en-en15
d63f22b 
# Conflicts:
#	docs/content/docs/guides/websocket-api.mdx
#	docs/spec/asyncapi.yaml
#	gui/lib/docs-bundle.ts
#	packages/inputlayer-js/src/connection.ts
#	packages/inputlayer-js/src/protocol.ts
#	packages/inputlayer-py/src/inputlayer/connection.py
#	src/protocol/rest/handlers/ws.rs
#	src/protocol/rest/handlers/ws/execute.rs
#	src/protocol/rest/handlers/ws/request.rs
#	src/protocol/subscription/connection.rs
#	src/protocol/subscription/connection/tests.rs
#	src/protocol/subscription/dependencies.rs
#	src/protocol/subscription/subscriber/tests.rs
#	ws-protocol/src/client.rs
#	ws-protocol/src/lib.rs
@jsam
jsam marked this pull request as ready for review October 5, 2026 14:05
@jsam jsam changed the title feat(protocol): subscription groups and snapshot reads at one revision (EN-15) feat(protocol): subscription groups and snapshot reads at one revision Oct 5, 2026
@jsam
no-mistakes(ci): I fixed the failing Python SDK check by updating one…
b5ea494 
… stale test assertion. The failing test was `tests/test_meta.py::TestSessionCommands::test_list_rules_sends_session_and_parses_the_clauses`. **Cause:** the round-1 review-5 fix (which the user chose) made `Session.list_rules` accept a keyword-only `timeout` and always pass it on as `execute('.session', timeout=timeout)`. That way `read`'s session-rule check uses the read's own timeout. The test still expected `execute('.session')` with no `timeout` argument, so it failed against the intended behaviour. **Rule that must hold:** `list_rules` sends `.session` with the caller's timeout, and the default is `None`. I checked the other places that call `execute('.session')`: - The drop-clause test only checks positional args (`.args`), so it is unaffected. - The subscription and group tests' fake connections match on the program string, so they are unaffected. - The `*_live.py` tests call `kg.execute` directly, not `list_rules`. **Fix:** the assertion now expects `execute('.session', timeout=None)`. I made no change to SDK source code, and the review-5 behaviour stays as the user chose it. **Verified:** `make python-test` gives 1339 passed, 35 skipped, 0 failed
@jsam
jsam merged commit a6c20e3 into main Oct 5, 2026
21 checks passed
@jsam
jsam deleted the fm/en-en15 branch October 5, 2026 14:52
This was referenced Oct 5, 2026
Closed
Closed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

No reviews

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

See More

Successfully merging this pull request may close these issues.

1 participant

@jsam