Skip to content

feat(inputlayer-py): routed connection core with request ids, deadlines and reconnect - #236

Merged
jsam merged 4 commits into
mainfrom
fm/sdk-p1-d
Oct 4, 2026
Merged

feat(inputlayer-py): routed connection core with request ids, deadlines and reconnect#236
jsam merged 4 commits into
mainfrom
fm/sdk-p1-d

Conversation

@jsam

@jsam jsam commented Oct 4, 2026

Copy link
Copy Markdown
Contributor

Intent

So fix the SDKs by improving them. (Context: the captain asked to rework the SDKs so agents and agentic loops use InputLayer without writing Datalog, and to fix the SDK release blockers by improving the SDKs rather than patching. The agreed design, signed off by an independent challenger after three rounds, is ~/firstmate/data/sdk-design/report.md; this item is its phase-1 PR P1-D P1-D (Py).)

The referenced item, P1-D of the agreed design (Python): the connection core. Replace the Python SDK connection (a lock around send-and-receive, with the receive loop never started, so no notifications or subscription pushes arrive while idle and subscriptions cannot be hosted) with: one reader task that reads every frame and routes it - frames with an id to the call registered under that id, with result_start/chunk/end assembled there and checked against the announced counts; subscription pushes routed by subscription and generation, a stale generation dropped and counted; notifications to the notification dispatcher; notices logged, and closing ones fail every pending call with ConnectionLost(code). One connection per knowledge graph handle, bound at connect time with ?kg=, never sending .kg use on a connection that could hold subscriptions. Ids on every request (r per connection), at most ws_max_in_flight_requests - 1 in flight (default 15) so a cancel can always be sent, queries pipelined and replies matched by id so the execute lock disappears. Deadlines and cancel: timeout= sets timeout_ms; a local timer at timeout + grace sends cancel and raises DeadlineExceeded if no reply arrives; a cancel_ack of too_late means the write committed and the reply is awaited. Idle delivery: the reader always runs, and an application ping is sent every keepalive seconds (default 20) when nothing else was sent so a subscription-only client never hits idle_timeout. Pending calls fail on close with ConnectionLost, flagging writes that may have committed (may_have_committed) so the caller re-reads before retrying. Reconnect with exponential backoff 1 s to 30 s with jitter, re-authenticating to the same knowledge graph, resuming the notification cursor with last_seq/epoch, reporting a notification gap on replay_gap and a session_reset event (plus reconnected) on il.events; max_reconnect_attempts exhausted ends iterators with ConnectionLost. Errors per the design: DeadlineExceeded, Cancelled, ConnectionLost(code, may_have_committed), RateLimited (retried once after the window), ProtocolError for invalid_request (an SDK bug, and it says so). Tests: mock-server unit tests replaying shared conformance frame fixtures (chunked results, missing chunk, stale generation, replay_gap, close mid-stream, duplicate id rejection), plus live tests against a real engine that run in CI. Millisecond latency must not regress. Other phase-1 items (compile path, subscribe/watch/on, claim, README) are separate PRs and out of scope.

What Changed

  • Swapped the Python SDK's lock-around-send/receive connection for a connection core with a single reader task that always runs. Replies are routed to their caller by request id (r<n>), and chunked results are assembled and checked against the announced counts. Subscription pushes are routed by subscription and generation, with stale generations dropped. Notifications go to a shared dispatcher, and a closing notice fails every pending call. Queries are now pipelined, with at most ws_max_in_flight_requests - 1 in flight. Each knowledge graph handle gets its own connection, bound with ?kg=<name>. The client exposes il.events (disconnected, reconnected, session_reset, notification_gap, closed) plus new default_timeout, keepalive and epoch options.
  • Added deadlines with cancel: a call sends cancel and raises DeadlineExceeded when no reply arrives in time; a too_late ack means the call keeps waiting for its reply. An application ping is sent when the connection has been idle for the keepalive interval. Reconnect uses exponential backoff with jitter, re-authenticates to the same graph and resumes the notification cursor (last_seq/epoch). New typed errors: DeadlineExceeded, Cancelled, ConnectionLost(code, may_have_committed), RateLimited (retried once after the window) and ProtocolError. All are exported, and the Python SDK guide documents them.
  • Added shared conformance frame fixtures under packages/conformance/connection/. They cover chunked results, missing chunks, count mismatch, stale generation, replay_gap, close mid-query/mid-stream, duplicate id rejection, rate limiting and typed error codes. Mock-server unit tests replay them, and the connection tests were rewritten against the mock server. Also added live engine tests (test_connection_live.py), wired into make python-test-live. A new path-filtered Python SDK CI job runs make python-test.

Risk Assessment

⚠️ Medium: This round's drop/close fixes are correct. They close every connection to a dropped graph for good, move the client's own connection to "default" (the server's WS default is literally "default"), and keep stale handles from re-creating the graph. close() now finishes before any reopen. The mock tests assert the right outcomes. Medium risk remains because the change rewrites the whole Python connection core (routing, reconnect, deadlines), and several earlier findings were deliberately left unfixed.

Testing

I ran the targeted mock-server and conformance tests: 118 passed. The one failure is test_sync.py::test_loop_thread_custom_timeout, which this change does not touch and which fails on Python 3.10. Then I built the engine from this branch and ran the 7 live connection tests plus a custom live script against isolated local servers. All of these passed: pipelining past the 15-request in-flight limit, idle notifications, subscriptions surviving queries on another handle, deadline, cancel, reconnect with replay and session_reset/reconnected, notification_gap after a reconnect with no cursor, keepalive under a 3 s idle timeout (the control run without keepalive was closed with idle_timeout), and ConnectionLost when the server was killed (may_have_committed=True for the write, False for the query, iterators ended). Dropping a graph left no connection on it and a stale handle could not recreate it; close then reopen worked again. Latency was compared with the base SDK. One latency run on the rate-limited server ended in RateLimited and then ConnectionLost when 500 requests were fired at once; that is the server's 100 msg/s limit, so the comparison was re-run with the limit off. The conformance-fixture scenarios ran only through the mock server, not live. This is an SDK change with no UI, so there are CLI transcripts instead of screenshots. All servers, temp data and build output were removed afterwards.

  • Live validation: ✅ go - 13 of 14 scenarios driven live against the product
Scenario Result Live Evidence
40 queries pipelined at once on one graph handle all get their own reply, matched by id, and nothing is left in flight ✅ pass live test_connection_live.py::test_pipelined_queries_beyond_the_in_flight_bound (live_tests.txt)
A handle on a missing graph creates it and binds its connection to it with ?kg= ✅ pass live test_connection_live.py::test_missing_graph_is_created_and_bound
A subscription on graph A still gets pushes after many queries and writes on graph B ✅ pass live test_connection_live.py::test_two_handles_subscription_survives_queries_on_the_other
A client that sends nothing still receives persistent_update notifications ✅ pass live test_connection_live.py::test_notifications_arrive_while_idle
A long query with a 1 ms timeout raises DeadlineExceeded and the connection stays usable ✅ pass live test_connection_live.py::test_deadline_stops_a_long_query_and_the_connection_stays_usable
Cancelling an in-flight call cancels it on the server and leaves nothing in flight ✅ pass live test_connection_live.py::test_cancelling_a_call_cancels_it_on_the_server
After the socket is aborted, the client reconnects to the same graph, replays notifications missed during the outage and emits disconnected/session_reset/reconnected ✅ pass live test_connection_live.py::test_reconnect_restores_graph_and_replays_missed_notifications
A reconnect with no resume cursor emits notification_gap ✅ pass live live_scenarios.py cursorless_gap -> events ['disconnected','session_reset','reconnected','notification_gap'] (live_scenarios_head.txt)
A subscription-only client survives a 3 s server idle timeout through keepalive pings; without keepalive the server closes it with idle_timeout ✅ pass live keepalive.txt
Killing the server mid-call raises ConnectionLost (write may_have_committed=True, query False), and notification iterators end with ConnectionLost once the reconnect attempts run out ✅ pass live server_kill.txt
Dropping a graph leaves no connection bound to it: a stale handle raises ConnectionLost and does not recreate the graph, and the client's own connection moves to default ✅ pass live live_scenarios.py drop_graph (live_scenarios_head.txt)
close() and then connecting again: earlier handles and new handles work again ✅ pass live live_scenarios.py close_reopen (live_scenarios_head.txt)
Millisecond query latency does not regress compared with the base SDK ✅ pass live latency_head_vs_base.txt: sequential p50 0.59-0.72 ms vs 0.55-0.57 ms base (still under 1 ms); 500 concurrent queries about 4x faster
Conformance fixtures (chunked results, missing chunk, count mismatch, stale generation, replay_gap, close mid-stream, duplicate id, typed errors, RateLimited retried once, ProtocolError) replayed thro… ⏸️ untested no The previous payload did not establish a live result for this scenario. It ran only through the mock server replaying the shared fixtures (118 passed in pytest), not against the live engine. These mal…
Evidence: Live connection tests against the real engine 
rootdir: ~/.no-mistakes/worktrees/9eea230c3aae/01M43BC6FTWYNST8AA5J0MGZAY/packages/inputlayer-py
configfile: pyproject.toml
plugins: langsmith-0.12.1, asyncio-1.3.0, anyio-4.13.0
asyncio: mode=auto, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function
collecting ... collected 7 items

tests/test_connection_live.py::test_pipelined_queries_beyond_the_in_flight_bound PASSED [ 14%]
tests/test_connection_live.py::test_missing_graph_is_created_and_bound PASSED [ 28%]
tests/test_connection_live.py::test_two_handles_subscription_survives_queries_on_the_other PASSED [ 42%]
tests/test_connection_live.py::test_notifications_arrive_while_idle PASSED [ 57%]
tests/test_connection_live.py::test_deadline_stops_a_long_query_and_the_connection_stays_usable PASSED [ 71%]
tests/test_connection_live.py::test_cancelling_a_call_cancels_it_on_the_server PASSED [ 85%]
tests/test_connection_live.py::test_reconnect_restores_graph_and_replays_missed_notifications PASSED [100%]

============================== 7 passed in 1.70s ===============================
Evidence: Custom live scenario driver script 
"""Extra live scenarios for the P1-D Python connection core, driven against a
real inputlayer-server. Usage: python live_scenarios.py <scenario> (URL from IL_URL)."""

from __future__ import annotations

import asyncio
import os
import sys
import time
import uuid

from inputlayer import InputLayer
from inputlayer.exceptions import InputLayerError
try:
    from inputlayer.exceptions import ConnectionLost
except ImportError:  # base SDK, for the latency baseline
    ConnectionLost = InputLayerError

URL = os.environ.get("IL_URL", "ws://127.0.0.1:18089/ws")


def client(**kw):
    return InputLayer(URL, username="admin", password="admin", **kw)


def name(p):
    return f"{p}_{uuid.uuid4().hex[:6]}"


async def until(cond, timeout=10.0):
    end = time.monotonic() + timeout
    while not cond():
        if time.monotonic() > end:
            raise AssertionError("condition not met")
        await asyncio.sleep(0.02)


async def drop_graph():
    g = name("drop")
    async with client(initial_kg="default") as il:
        kg = il.knowledge_graph(g)
        await kg.execute("+x(1)")
        print("created", g, "lists:", g in await il.list_knowledge_graphs())
        await il.drop_knowledge_graph(g)
        print("after drop listed:", g in await il.list_knowledge_graphs())
        try:
            await kg.execute("+x(2)")
            print("STALE HANDLE WROTE (bad)")
        except ConnectionLost as e:
            print("stale handle raised ConnectionLost:", e)
        print("graph re-created by stale handle:", g in await il.list_knowledge_graphs())
    # Client whose own connection is bound to the graph being dropped.
    g2 = name("dropmain")
    async with client(initial_kg="default") as admin:
        await admin.knowledge_graph(g2).execute("+y(1)")
        async with client(initial_kg=g2) as il:
            print("main bound to:", il._conn.current_kg)
            await il.drop_knowledge_graph(g2)
            print("main after drop bound to:", il._conn.current_kg)
            fresh = il.knowledge_graph(g2)
            print("fresh handle uses main conn:", fresh._conn is il._conn)
            await fresh.execute("+y(5)")
            print("fresh handle recreated graph rows:", (await fresh.execute("?y(X)")).rows)
            await il.drop_knowledge_graph(g2)


async def close_reopen():
    g = name("reopen")
    il = client()
    async with il:
        kg = il.knowledge_graph(g)
        await kg.execute("+v(1)")
    print("closed; connected =", il.connected)
    async with il:
        print("reopened; old handle query:", (await kg.execute("?v(X)")).rows)
        print("new handle same name:", (await il.knowledge_graph(g).execute("?v(X)")).rows)
        await il.drop_knowledge_graph(g)


async def kill_inflight():
    """Server is killed by the harness while a write and a query are in flight."""
    g = name("kill")
    il = client(max_reconnect_attempts=2, reconnect_delay=0.2)
    await il.connect()
    kg = il.knowledge_graph(g)
    await kg.execute("+e[" + ", ".join(f"({i}, {i+1})" for i in range(1500)) + "]")
    await kg.execute("+r(X, Y) <- e(X, Y)\n+r(X, Z) <- r(X, Y), e(Y, Z)")
    events = []
    il.events.on(callback=lambda e: events.append(e.type))
    end_err = []

    async def iterate():
        try:
            async for _ in il.notifications():
                pass
            end_err.append("ended cleanly")
        except BaseException as e:  # noqa: BLE001
            end_err.append(f"{type(e).__name__}: {e}")

    it = asyncio.ensure_future(iterate())
    q = asyncio.ensure_future(kg.execute("?r(X, Y)", timeout=60))
    w = asyncio.ensure_future(kg.execute("+big[" + ", ".join(f"({i})" for i in range(200000)) + "]", timeout=60))
    await asyncio.sleep(0.3)
    print("READY_TO_KILL", flush=True)
    for label, fut in (("query", q), ("write", w)):
        try:
            await fut
            print(label, "completed before kill")
        except ConnectionLost as e:
            print(label, "ConnectionLost code=", e.code, "may_have_committed=", e.may_have_committed)
        except InputLayerError as e:
            print(label, type(e).__name__, e)
    await asyncio.wait_for(it, 30)
    print("notifications iterator:", end_err)
    print("events:", events)
    await il.close()


async def keepalive_idle():
    """Server runs with ws_idle_timeout_ms=3000; client only listens for 10 s."""
    g = name("keep")
    async with client(keepalive=1.0) as il:
        kg = il.knowledge_graph(g)
        await kg.execute("+s(0)")
        pushes = []
        kg._conn.add_route("sub1", pushes.append)
        await kg.execute(".subscribe sub1 ?s(X)")
        events = []
        il.events.on(callback=lambda e: events.append(e.type))
        await asyncio.sleep(10)
        async with client() as w:
            await w.knowledge_graph(g).execute("+s(1)")
        await until(lambda: pushes, 5)
        print("after 10s idle: events=", events, "push inserted=", pushes[0].inserted,
              "state=", kg._conn.state)
        await il.drop_knowledge_graph(g)


async def no_keepalive_idle():
    g = name("nokeep")
    async with client(keepalive=None, auto_reconnect=False) as il:
        kg = il.knowledge_graph(g)
        await kg.execute("+s(0)")
        events = []
        il.events.on(callback=lambda e: events.append((e.type, getattr(e, "code", None))))
        await asyncio.sleep(6)
        print("control without keepalive after 6s: events=", events, "state=", kg._conn.state)


async def cursorless_gap():
    g = name("gap")
    async with client() as w:
        kg = w.knowledge_graph(g)
        await kg.execute("+s(0)")
        async with client(initial_kg=g, reconnect_delay=0.3) as il:
            events = []
            il.events.on(callback=lambda e: events.append(e.type))
            il._conn._ws.transport.abort()
            await until(lambda: "reconnected" in events)
            await until(lambda: "notification_gap" in events, 3)
            print("cursor-less reconnect events:", events)
        await w.drop_knowledge_graph(g)


async def latency():
    g = name("lat")
    async with client() as il:
        kg = il.knowledge_graph(g)
        await kg.execute("+p[" + ", ".join(f"({i})" for i in range(100)) + "]")
        for _ in range(50):
            await kg.execute("?p(1)")
        ts = []
        for i in range(500):
            t = time.perf_counter()
            await kg.execute(f"?p({i % 100})")
            ts.append((time.perf_counter() - t) * 1000)
        ts.sort()
        print(f"sequential ?p(i) x500: p50={ts[250]:.3f}ms p99={ts[495]:.3f}ms")
        t = time.perf_counter()
        await asyncio.gather(*(kg.execute(f"?p({i % 100})") for i in range(500)))
        print(f"500 concurrent: total={(time.perf_counter()-t)*1000:.1f}ms")
        await il.drop_knowledge_graph(g)


if __name__ == "__main__":
    asyncio.run(globals()[sys.argv[1]]())
Evidence: Drop graph, close and reopen, no-cursor gap transcripts (plus the rate-limited latency run) 
== drop_graph
created drop_4a30f3 lists: True
after drop listed: False
stale handle raised ConnectionLost: Not connected: the connection is closed
graph re-created by stale handle: False
main bound to: dropmain_df96af
main after drop bound to: default
fresh handle uses main conn: False
fresh handle recreated graph rows: [[5]]
== close_reopen
closed; connected = False
reopened; old handle query: [[1]]
new handle same name: [[1]]
== cursorless_gap
cursor-less reconnect events: ['disconnected', 'session_reset', 'reconnected', 'notification_gap']
== latency
  File "~/.no-mistakes/worktrees/9eea230c3aae/01M43BC6FTWYNST8AA5J0MGZAY/packages/inputlayer-py/src/inputlayer/knowledge_graph.py", line 965, in _naming_query
    return await pending
  File "~/.no-mistakes/worktrees/9eea230c3aae/01M43BC6FTWYNST8AA5J0MGZAY/packages/inputlayer-py/src/inputlayer/connection.py", line 518, in execute
    return await self._execute_once(program, deadline)
  File "~/.no-mistakes/worktrees/9eea230c3aae/01M43BC6FTWYNST8AA5J0MGZAY/packages/inputlayer-py/src/inputlayer/connection.py", line 534, in _execute_once
    result: ResultResponse = await self._await_reply(pending, deadline)
  File "~/.no-mistakes/worktrees/9eea230c3aae/01M43BC6FTWYNST8AA5J0MGZAY/packages/inputlayer-py/src/inputlayer/connection.py", line 637, in _await_reply
    return await asyncio.wait_for(asyncio.shield(pending.future), backstop)
  File "~/.local/share/uv/python/cpython-3.10-linux-x86_64-gnu/lib/python3.10/asyncio/tasks.py", line 445, in wait_for
    return fut.result()
inputlayer.exceptions.RateLimited: Rate limit exceeded (100 msgs/sec)
  query: ?p(48)
Evidence: Keepalive vs no-keepalive under a 3 s server idle timeout

no keepalive: events=[('disconnected','idle_timeout'),('closed','idle_timeout'),...] state=closed keepalive=1s: after 10s idle events=[] push inserted=[[1]] state=open

== no_keepalive_idle (server ws_idle_timeout_ms=3000)
server notice (idle_timeout): Idle timeout
server notice (idle_timeout): Idle timeout
control without keepalive after 6s: events= [('disconnected', 'idle_timeout'), ('closed', 'idle_timeout'), ('disconnected', 'idle_timeout'), ('closed', 'idle_timeout')] state= closed
== keepalive_idle (server ws_idle_timeout_ms=3000)
after 10s idle: events= [] push inserted= [[1]] state= open
Evidence: Server SIGKILL with calls in flight

query ConnectionLost code= None may_have_committed= False write ConnectionLost code= None may_have_committed= True notifications iterator: ["ConnectionLost: Connection to 'default' closed and not reconnected"]

READY_TO_KILL
query ConnectionLost code= None may_have_committed= False
write ConnectionLost code= None may_have_committed= True
notifications iterator: ["ConnectionLost: Connection to 'default' closed and not reconnected"]
events: ['disconnected', 'disconnected', 'closed', 'closed']
Evidence: Latency: head SDK vs base SDK

head: sequential p50 0.59-0.72ms p99 0.84-1.02ms; 500 concurrent 73-81ms base: sequential p50 0.55-0.57ms p99 0.73-0.81ms; 500 concurrent 290-313ms

== head SDK run 1
sequential ?p(i) x500: p50=0.593ms p99=0.844ms
500 concurrent: total=73.2ms
== base SDK (ea9d5d3c) run 1
sequential ?p(i) x500: p50=0.554ms p99=0.731ms
500 concurrent: total=296.6ms
== head SDK run 2
sequential ?p(i) x500: p50=0.716ms p99=1.015ms
500 concurrent: total=81.4ms
== base SDK (ea9d5d3c) run 2
sequential ?p(i) x500: p50=0.563ms p99=0.809ms
500 concurrent: total=312.6ms
== head SDK run 3
sequential ?p(i) x500: p50=0.611ms p99=0.942ms
500 concurrent: total=76.7ms
== base SDK (ea9d5d3c) run 3
sequential ?p(i) x500: p50=0.568ms p99=0.808ms
500 concurrent: total=290.0ms
- Outcome: ⚠️ 2 infos across 1 run (5m37s)

Pipeline

Updates from git push no-mistakes

✅ **intent** - passed

✅ No issues found.

✅ **Rebase** - passed

✅ No issues found.

🔧 **Review** - 8 issues found → auto-fixed (2) ✅
  • 🚨 packages/inputlayer-py/src/inputlayer/client.py:155 - _create_kg runs .kg create &lt;name&gt; on the client's own connection. The engine's KgCreate switches the session to the new graph (src/protocol/handler.rs:2614-2627 sets switched_kg_result = Some(name)), and Connection._accept (connection.py:845-846) records that as the main connection's _current_kg. Failing sequence: InputLayer(url, initial_kg=&#34;a&#34;), connect, a = il.knowledge_graph(&#34;a&#34;) (bound to the main connection via _connection_for), b = il.knowledge_graph(&#34;b&#34;), then await b.execute(...) on a missing "b" calls _create_kg(&#34;b&#34;) and moves the main session to "b". After that, a.execute(...)/a.query(...) silently read and write graph "b", reconnects of the main connection re-bind "b", and the switch drops the main session's state and subscriptions. This breaks the intent's invariant: "never sending .kg use on a connection that could hold subscriptions". The live test test_connection_live.py:72 passes only because it never touches a main-connection handle after a create. Fix: run the create on a short-lived connection that no handle uses (for example a temporary Connection bound to the server default, closed after the create), not on self._conn. drop_knowledge_graph's .kg use default on self._conn (client.py:208-209) is another switch of a handle-bearing connection; route it through the same throwaway admin connection.
  • 🚨 packages/inputlayer-py/src/inputlayer/connection.py:470 - When authenticated reports a new stream_epoch (the engine restarted), _authenticate replaces _epoch but keeps _last_seq from the old run. _deliver_notification then computes max(self._last_seq, frame.seq), so the cursor stays at the old high value. Failing sequence: the client has seen seq 100, the engine restarts, the reconnect gets replay_gap, and new-run notifications 1..60 arrive while _last_seq stays 100. On the next drop at new seq 60, the reconnect sends last_seq=100&amp;epoch=&lt;new&gt;, so the server skips everything up to 100 and notifications 61..100 of the outage are lost with no gap reported. NotificationDispatcher._last_seq (notifications.py, dispatch) and InputLayer.last_seq report the same stale max. Fix: reset _last_seq to 0 (and the dispatcher's last_seq) when the epoch changes, so the cursor counts in the new run.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:512 - timeout_ms is sent as the full timeout, but the local deadline (and its backstop, _await_reply line 611) started before _ensure_open and the slot wait. Failing sequence: 15 calls are in flight and a 16th waits 10 s for a slot (default timeout 30 s). The server deadline then falls at ~40 s, while the local backstop fires at ~31 s and sends cancel. The engine answers error code=cancelled, so the caller gets Cancelled although it never cancelled and only its deadline passed. The same happens when _ensure_open waits for a reconnect. Fix: send timeout_ms from the remaining time (self._remaining(deadline)), so the server's deadline_exceeded arrives before the local backstop. Related: execute's RateLimited retry (line 494-495) restarts a full timeout, so one call can take about 2×timeout+1 s.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:944 - Every pooled per-KG connection shares the client's single NotificationDispatcher (client.py _options[&#39;dispatcher&#39;]), and _give_up calls self._dispatcher.fail(...) on it. When one graph's connection gives up, every il.notifications() iterator of the whole client ends with ConnectionLost, even though the other connections still deliver notifications. Concrete triggers: a pooled graph is dropped by another client (the reconnect gets auth_error, _reconnect breaks and calls _give_up), or auto_reconnect=False and a single pooled socket drops (_on_lost calls _give_up). Fix: fail the shared dispatcher only when the client's own connection (or every connection) has given up, and report a single graph's loss via its closed event.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:888 - The keepalive pings whenever nothing was sent for keepalive seconds, even with requests in flight. The server's idle timer cannot fire while requests are pending (ws.rs: if ... requests.is_idle()), so these pings are unnecessary. They also hurt: the server admits pings into the same in-flight pipeline and releases replies in order (ws/pipeline.rs). With 15 slow executes outstanding for more than 20 s, the ping takes the 16th server slot until all 15 are answered, so the deadline-backstop cancel (or a task-cancellation _fire_cancel) is not read by the server. This breaks the intent's "at most ws_max_in_flight_requests - 1 in flight so a cancel can always be sent". _fire_cancel (line 657) has the same gap when several callers cancel at once: each cancel uses a server slot the client never accounts for. Fix: skip the keepalive ping while _pending is non-empty. Also, when _send fails, the loop never updates _loop_last_send (line 889 recomputes wait <= 0), so if the socket is already closed but the reader has not torn down yet, the loop re-registers and re-sends pings back to back. Advance _loop_last_send (or break) after a failed send.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:652 - _cancel_overdue raises DeadlineExceeded when neither the reply nor the cancel_ack arrives within the grace period. DeadlineExceeded documents "nothing it would have changed is applied" (exceptions.py, docs python-sdk.mdx). For a write, that claim is unproven here: no cancelled reply or ack arrived, and the server may still commit the write (or ack too_late after the client stopped waiting). The caller then retries without reading back first. The intent asks for DeadlineExceeded in this case, so changing the error type or adding a may_have_committed flag to this path is a product/API decision: should a write that hits the local backstop surface as uncertain (as ConnectionLost does)?
  • ⚠️ packages/inputlayer-py/src/inputlayer/client.py:94 - InputLayer.close() closes every pooled connection but keeps _pool and the cached _kgs handles. A lazy Connection in state closed never reopens: _ensure_open raises ConnectionLost (connection.py:543-546). Before this change, after await il.close(); await il.connect() (or two consecutive async with il: blocks) the same client and its handles worked again. Now every graph handle obtained earlier, and every new il.knowledge_graph(name) for a name already in _pool, fails with "the connection is closed". Fix: clear _pool/_kgs on close, or let a lazy connection reopen from closed on the next call.
  • ℹ️ packages/inputlayer-py/src/inputlayer/connection.py:393 - The reconnect sends a resume cursor only when _last_seq is not None, i.e. only after at least one notification was received. Sequence: a connection that has seen no notification drops, the graph changes during the outage, and the reconnect sends no last_seq, so the change is never replayed and no notification_gap is emitted. authenticated carries no current seq, so the fix needs a decision: send last_seq=0&amp;epoch=... (which replays older events) or emit notification_gap after any cursor-less reconnect.

🔧 Fix applied.
10 issues (2 errors, 6 warnings, 2 infos) still open:

  • 🚨 packages/inputlayer-py/src/inputlayer/client.py:155 - _create_kg runs .kg create &lt;name&gt; on the client's own connection. The engine's KgCreate switches the session to the new graph (src/protocol/handler.rs:2614-2627 sets switched_kg_result = Some(name)), and Connection._accept (connection.py:845-846) records that as the main connection's _current_kg. Failing sequence: InputLayer(url, initial_kg=&#34;a&#34;), connect, a = il.knowledge_graph(&#34;a&#34;) (bound to the main connection via _connection_for), b = il.knowledge_graph(&#34;b&#34;), then await b.execute(...) on a missing "b" calls _create_kg(&#34;b&#34;) and moves the main session to "b". After that, a.execute(...)/a.query(...) silently read and write graph "b", reconnects of the main connection re-bind "b", and the switch drops the main session's state and subscriptions. This breaks the intent's invariant: "never sending .kg use on a connection that could hold subscriptions". The live test test_connection_live.py:72 passes only because it never touches a main-connection handle after a create. Fix: run the create on a short-lived connection that no handle uses (for example a temporary Connection bound to the server default, closed after the create), not on self._conn. drop_knowledge_graph's .kg use default on self._conn (client.py:208-209) is another switch of a handle-bearing connection; route it through the same throwaway admin connection.
  • 🚨 packages/inputlayer-py/src/inputlayer/connection.py:470 - When authenticated reports a new stream_epoch (the engine restarted), _authenticate replaces _epoch but keeps _last_seq from the old run. _deliver_notification then computes max(self._last_seq, frame.seq), so the cursor stays at the old high value. Failing sequence: the client has seen seq 100, the engine restarts, the reconnect gets replay_gap, and new-run notifications 1..60 arrive while _last_seq stays 100. On the next drop at new seq 60, the reconnect sends last_seq=100&amp;epoch=&lt;new&gt;, so the server skips everything up to 100 and notifications 61..100 of the outage are lost with no gap reported. NotificationDispatcher._last_seq (notifications.py, dispatch) and InputLayer.last_seq report the same stale max. Fix: reset _last_seq to 0 (and the dispatcher's last_seq) when the epoch changes, so the cursor counts in the new run.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:512 - timeout_ms is sent as the full timeout, but the local deadline (and its backstop, _await_reply line 611) started before _ensure_open and the slot wait. Failing sequence: 15 calls are in flight and a 16th waits 10 s for a slot (default timeout 30 s). The server deadline then falls at ~40 s, while the local backstop fires at ~31 s and sends cancel. The engine answers error code=cancelled, so the caller gets Cancelled although it never cancelled and only its deadline passed. The same happens when _ensure_open waits for a reconnect. Fix: send timeout_ms from the remaining time (self._remaining(deadline)), so the server's deadline_exceeded arrives before the local backstop. Related: execute's RateLimited retry (line 494-495) restarts a full timeout, so one call can take about 2×timeout+1 s.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:944 - Every pooled per-KG connection shares the client's single NotificationDispatcher (client.py _options[&#39;dispatcher&#39;]), and _give_up calls self._dispatcher.fail(...) on it. When one graph's connection gives up, every il.notifications() iterator of the whole client ends with ConnectionLost, even though the other connections still deliver notifications. Concrete triggers: a pooled graph is dropped by another client (the reconnect gets auth_error, _reconnect breaks and calls _give_up), or auto_reconnect=False and a single pooled socket drops (_on_lost calls _give_up). Fix: fail the shared dispatcher only when the client's own connection (or every connection) has given up, and report a single graph's loss via its closed event.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:888 - The keepalive pings whenever nothing was sent for keepalive seconds, even with requests in flight. The server's idle timer cannot fire while requests are pending (ws.rs: if ... requests.is_idle()), so these pings are unnecessary. They also hurt: the server admits pings into the same in-flight pipeline and releases replies in order (ws/pipeline.rs). With 15 slow executes outstanding for more than 20 s, the ping takes the 16th server slot until all 15 are answered, so the deadline-backstop cancel (or a task-cancellation _fire_cancel) is not read by the server. This breaks the intent's "at most ws_max_in_flight_requests - 1 in flight so a cancel can always be sent". _fire_cancel (line 657) has the same gap when several callers cancel at once: each cancel uses a server slot the client never accounts for. Fix: skip the keepalive ping while _pending is non-empty. Also, when _send fails, the loop never updates _loop_last_send (line 889 recomputes wait <= 0), so if the socket is already closed but the reader has not torn down yet, the loop re-registers and re-sends pings back to back. Advance _loop_last_send (or break) after a failed send.
  • ⚠️ packages/inputlayer-py/src/inputlayer/connection.py:652 - _cancel_overdue raises DeadlineExceeded when neither the reply nor the cancel_ack arrives within the grace period. DeadlineExceeded documents "nothing it would have changed is applied" (exceptions.py, docs python-sdk.mdx). For a write, that claim is unproven here: no cancelled reply or ack arrived, and the server may still commit the write (or ack too_late after the client stopped waiting). The caller then retries without reading back first. The intent asks for DeadlineExceeded in this case, so changing the error type or adding a may_have_committed flag to this path is a product/API decision: should a write that hits the local backstop surface as uncertain (as ConnectionLost does)?
  • ⚠️ packages/inputlayer-py/src/inputlayer/client.py:94 - InputLayer.close() closes every pooled connection but keeps _pool and the cached _kgs handles. A lazy Connection in state closed never reopens: _ensure_open raises ConnectionLost (connection.py:543-546). Before this change, after await il.close(); await il.connect() (or two consecutive async with il: blocks) the same client and its handles worked again. Now every graph handle obtained earlier, and every new il.knowledge_graph(name) for a name already in _pool, fails with "the connection is closed". Fix: clear _pool/_kgs on close, or let a lazy connection reopen from closed on the next call.
  • ℹ️ packages/inputlayer-py/src/inputlayer/connection.py:393 - The reconnect sends a resume cursor only when _last_seq is not None, i.e. only after at least one notification was received. Sequence: a connection that has seen no notification drops, the graph changes during the outage, and the reconnect sends no last_seq, so the change is never replayed and no notification_gap is emitted. authenticated carries no current seq, so the fix needs a decision: send last_seq=0&amp;epoch=... (which replays older events) or emit notification_gap after any cursor-less reconnect.
  • ⚠️ packages/inputlayer-py/src/inputlayer/client.py:221 - Round 1's fixes (admin connection for .kg drop, and lazy reopen after close() in connection.py:555) left drop_knowledge_graph with two wrong outcomes that don't raise errors. (1) The round removed the .kg use default step instead of replacing it. When name is the client's own connection's graph (InputLayer(url, initial_kg=&#34;a&#34;) followed by drop_knowledge_graph(&#34;a&#34;)), the server allows the drop from the admin session, because the KgDrop check in handler.rs:2661 only compares against the dropping session's own graph. The main connection is then still bound to the dropped graph. il.knowledge_graph(&#34;a&#34;) then returns a handle on self._conn (_connection_for line 140 matches current_kg) that never recreates the graph. The main connection's next reconnect binds ?kg=a, gets auth_error and gives up, and because the main connection has ends_notifications=True this ends every il.notifications() iterator. (2) The pooled connection closed at line 223 has _closing=True and lazy=True. Any KnowledgeGraph handle the user kept from before the drop therefore reopens it on its next call (connection.py:555-558). _open_bound(first=True) then gets auth_error for the missing graph and calls create_kg, so kg.execute(&#34;+x(1)&#34;) silently recreates the dropped graph and writes into it. Before this round, that call raised ConnectionLost. Fix: make a connection closed by a drop terminal: close it in a way the lazy reopen does not apply to, e.g. clear _lazy/_create_kg or use a dedicated flag. When the dropped graph is self._conn.current_kg, move the main connection off it: either reconnect it bound to the default graph or run .kg use default there, which is acceptable because that graph's handle and subscriptions are gone. Also make _connection_for stop returning self._conn for that name.
  • ℹ️ packages/inputlayer-py/src/inputlayer/connection.py:555 - The lazy reopen added in round 1 can start while close() is still running. close() sets state=&#39;closed&#39; and _closing=True first, then awaits the reconnect-task cancel, ws.close() and the reader. A call made in that window reaches connect() and opens a new socket. close() then runs _teardown(old_ws, ...), which cancels the new socket's _keepalive_task and fails the new socket's pending calls with ConnectionLost. Concurrent close-and-call is rare, but the fix is small: only allow the reopen after close() has finished, for example by setting the reopenable flag at the end of close().

🔧 Fix applied.
✅ Re-checked - no issues remain.

⚠️ **Test** - 2 infos
  • ℹ️ packages/inputlayer-py/tests/test_sync.py:641 - tests/test_sync.py::test_loop_thread_custom_timeout fails on the local Python 3.10. Before 3.11, concurrent.futures.TimeoutError is a different class from the builtin TimeoutError. This change does not modify _LoopThread or that test.
  • ℹ️ Sequential single-query latency went up slightly (p50 about 0.04-0.15 ms) compared with the base SDK but stays under 1 ms. Concurrent throughput is about 4x better.
  • Live validation: ✅ go - 13 of 14 scenarios driven live against the product
Scenario Result Live Evidence
40 queries pipelined at once on one graph handle all get their own reply, matched by id, and nothing is left in flight ✅ pass live test_connection_live.py::test_pipelined_queries_beyond_the_in_flight_bound (live_tests.txt)
A handle on a missing graph creates it and binds its connection to it with ?kg= ✅ pass live test_connection_live.py::test_missing_graph_is_created_and_bound
A subscription on graph A still gets pushes after many queries and writes on graph B ✅ pass live test_connection_live.py::test_two_handles_subscription_survives_queries_on_the_other
A client that sends nothing still receives persistent_update notifications ✅ pass live test_connection_live.py::test_notifications_arrive_while_idle
A long query with a 1 ms timeout raises DeadlineExceeded and the connection stays usable ✅ pass live test_connection_live.py::test_deadline_stops_a_long_query_and_the_connection_stays_usable
Cancelling an in-flight call cancels it on the server and leaves nothing in flight ✅ pass live test_connection_live.py::test_cancelling_a_call_cancels_it_on_the_server
After the socket is aborted, the client reconnects to the same graph, replays notifications missed during the outage and emits disconnected/session_reset/reconnected ✅ pass live test_connection_live.py::test_reconnect_restores_graph_and_replays_missed_notifications
A reconnect with no resume cursor emits notification_gap ✅ pass live live_scenarios.py cursorless_gap -> events ['disconnected','session_reset','reconnected','notification_gap'] (live_scenarios_head.txt)
A subscription-only client survives a 3 s server idle timeout through keepalive pings; without keepalive the server closes it with idle_timeout ✅ pass live keepalive.txt
Killing the server mid-call raises ConnectionLost (write may_have_committed=True, query False), and notification iterators end with ConnectionLost once the reconnect attempts run out ✅ pass live server_kill.txt
Dropping a graph leaves no connection bound to it: a stale handle raises ConnectionLost and does not recreate the graph, and the client's own connection moves to default ✅ pass live live_scenarios.py drop_graph (live_scenarios_head.txt)
close() and then connecting again: earlier handles and new handles work again ✅ pass live live_scenarios.py close_reopen (live_scenarios_head.txt)
Millisecond query latency does not regress compared with the base SDK ✅ pass live latency_head_vs_base.txt: sequential p50 0.59-0.72 ms vs 0.55-0.57 ms base (still under 1 ms); 500 concurrent queries about 4x faster
Conformance fixtures (chunked results, missing chunk, count mismatch, stale generation, replay_gap, close mid-stream, duplicate id, typed errors, RateLimited retried once, ProtocolError) replayed thro… ⏸️ untested no The previous payload did not establish a live result for this scenario. It ran only through the mock server replaying the shared fixtures (118 passed in pytest), not against the live engine. These mal…
  • uv run --extra dev pytest tests/test_connection.py tests/test_connection_conformance.py tests/test_statement_errors.py tests/test_sync.py -q (mock server replaying the conformance fixtures: 118 passed, 1 failed, test_sync.py::test_loop_thread_custom_timeout, which this change does not touch; it fails on Python 3.10)
  • cargo build --release --bin inputlayer-server, then an isolated server per scenario (temp --data-dir, ports 18089/18090/18091)
  • INPUTLAYER_TEST_SERVER=ws://127.0.0.1:18089/ws uv run --extra dev pytest tests/test_connection_live.py -v (7/7 passed against the live engine)
  • live_scenarios.py drop_graph against the live engine
  • live_scenarios.py close_reopen against the live engine
  • live_scenarios.py cursorless_gap against the live engine
  • live_scenarios.py no_keepalive_idle and keepalive_idle against a server started with INPUTLAYER_HTTP__WS_IDLE_TIMEOUT_MS=3000
  • live_scenarios.py kill_inflight, with the harness SIGKILLing inputlayer-server while a write and a query were in flight
  • live_scenarios.py latency, 3 runs each for the head SDK and the base SDK (ea9d5d3c, extracted with git archive), against a server with the rate limit off
✅ **Document** - passed

✅ No issues found.

⚠️ **Lint** - 1 info
  • ℹ️ packages/inputlayer-py/src/inputlayer/connection.py - ruff format --check would reformat 11 of the changed Python SDK files (connection.py, client.py, the tests and others). This is not new to this change: at the base commit, 58 of the package's files were already unformatted, and neither CI nor the Makefile runs ruff format. Reformatting only these files would add a large whitespace-only diff to an unrelated change, so I left them alone. If formatting is wanted, a one-time ruff format of the whole package in a separate commit, with a CI check added, is the follow-up. ruff check passes. mypy reports 54 errors, all of which already existed (56 at the base commit), so this change adds no new ones.
✅ **Push** - passed

✅ No issues found.

jsam added 4 commits October 4, 2026 11:39
@jsam
feat(inputlayer-py): routed connection core with ids, deadlines and r…
65ca0a2 
…econnect

Replace the Python SDK's lock-around-send-and-receive connection with one
reader task per connection that routes every frame: replies by request id
(streamed results assembled and checked against their end counts),
standing-query pushes by subscription and generation (stale ones dropped
and counted), notifications to the dispatcher, notices logged.

- Every request carries an id; requests are pipelined with at most 15
  unanswered, leaving one server slot free for a cancel.
- Each call has a deadline sent as timeout_ms, with a local backstop that
  cancels the request on the server; cancelling the awaiting task cancels
  it too. New typed errors: DeadlineExceeded, Cancelled, ConnectionLost
  (with may_have_committed), RateLimited (retried once), ProtocolError.
- Notifications and deltas arrive while idle; a keepalive ping keeps a
  listening client under the server's idle timeout.
- A dropped connection fails pending calls and reconnects with jittered
  exponential backoff to the same knowledge graph, resuming notifications
  from the last seq and epoch; il.events reports disconnected,
  session_reset, reconnected, notification_gap and closed.
- One connection per knowledge graph handle, bound with ?kg= (a missing
  graph is created on first use); the SDK no longer sends .kg use.

Tests: shared frame fixtures in packages/conformance/connection replayed
through a real-socket mock server, timing and pool tests, live tests in
make python-test-live, and a fast python-sdk CI job for the unit suite.
@jsam
no-mistakes(review): Fix KG switching, cursor, deadline, keepalive an…
b85135d 
…d pool lifecycle bugs
@jsam
no-mistakes(review): Make dropped graphs terminal and close() finish …
7292fe2 
…before reopen
@jsam
no-mistakes(document): Document RateLimited and ProtocolError in Pyth…
26fd560 
…on SDK guide
@jsam
jsam merged commit b260bf7 into main Oct 4, 2026
19 checks passed
@jsam
jsam deleted the fm/sdk-p1-d branch October 4, 2026 12:07
This was referenced Oct 4, 2026
Merged
Merged
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