jsam · Oct 4, 2026 · Oct 4, 2026 · Oct 4, 2026 · Oct 4, 2026 · Oct 4, 2026
diff --git a/Makefile b/Makefile
@@ -463,8 +463,10 @@ js-test-live:
 # python-sdk-live runs every query example in the Python SDK guide
 # (docs/content/docs/guides/python-sdk.mdx) against a live server, with the
 # pins of the SDK's compile rules (the shared meta-command fixture in
-# packages/conformance and a literal round-trip fuzz included), so a form the
-# engine rejects fails here rather than in a user's hands.
+# packages/conformance and a literal round-trip fuzz included) and the
+# subscription tests (snapshot plus deltas over random histories, gaps,
+# resets, slow consumers, reconnect, ACL revoke), so a form the engine
+# rejects fails here rather than in a user's hands.
 PY_LIVE_PORT ?= 8092
 python-sdk-live:
 	@cargo build --release --bin inputlayer-server
@@ -482,7 +484,8 @@ python-sdk-live:
 		INPUTLAYER_TEST_SERVER=ws://127.0.0.1:$(PY_LIVE_PORT)/ws \
 		INPUTLAYER_TEST_USER=admin \
 		INPUTLAYER_TEST_PASSWORD=admin \
-		uv run --extra dev pytest tests/docs_queries_live.py tests/compile_rules_live.py -v; \
+		uv run --extra dev pytest tests/docs_queries_live.py tests/compile_rules_live.py \
+			tests/subscriptions_live.py -v; \
 	TEST_EXIT=$$?; \
 	cd - > /dev/null; \
 	kill $$SERVER_PID 2>/dev/null || true; \

diff --git a/docs/content/docs/guides/python-sdk.mdx b/docs/content/docs/guides/python-sdk.mdx
@@ -582,9 +582,34 @@ session_rules = await kg.session.list_rules()
 await kg.session.clear()
 ```
 
+## Subscriptions
+
+`kg.subscribe()` takes the arguments of `query()` (or raw IQL as `iql="?..."`) and returns an async iterator of `Change` events that keeps the result exact on the client. Every event has the same fields (`kind`, `inserted`, `retracted`, `revision`, `seq`, `verified`), so a loop that applies `retracted` and `inserted` is always correct:
+
+```python
+async for change in kg.subscribe(Late):
+    for row in change.retracted:
+        cancel(row)
+    for row in change.inserted:
+        start(row)
+    if not change.verified:
+        pause()
+```
+
+- `snapshot`: the first event, the whole result with every row in `inserted`.
+- `delta`: the set difference between consecutive results. An unchanged result delivers nothing, and commits that coalesce arrive as one delta, so a row that appears and disappears between two evaluations is never seen.
+- `unverified`: the result may no longer be current (`reason` is `connection_lost`, `seq_gap`, `broken_stream`, `subscription_reset`, `subscription_error` or `slow_consumer`). Both lists are empty; act on nothing new until the resync arrives. The SDK subscribes again with backoff.
+- `resync`: the exact difference between the rows you hold and the fresh result.
+
+Rows are instances of the relation when one whole relation is selected, otherwise immutable `Row` records (`row.order` or `row["order"]`). A projected subscription reports a row only when the first engine tuple behind it appears or the last one goes away. `queue` (default 1024) bounds the events held for a slow reader: a reader that falls further behind gets `unverified` (`slow_consumer`) and then a `resync`, and the connection never blocks. Leaving the loop or calling `close()` ends the subscription, and `sub.stats` counts dropped stale pushes and resubscribes.
+
+A standing query tracks the whole result of persistent rules, so `subscribe()` raises `SubscriptionRejected` with a `reason`: `limit_offset`, `or_branches` or `session_view` (a session rule, an aggregate or a negated constant) before anything is sent, or the engine's `result_cap`, `access_denied`, `id_taken`, `subscription_limit` or `rejected`. To subscribe to such a query, declare a persistent rule and subscribe to that rule instead.
+
+`kg.watch()` yields the whole current result (`Live`: `rows`, `revision`, `verified`) each time it changes. `kg.on(Late, callback)` calls `callback` with every `Change` and returns a handle; `await handle.close()` stops it. On `InputLayerSync`, `kg.subscribe()` and `kg.watch()` are blocking iterators, and the `on()` callback runs on the client's background thread.
+
 ## Notifications
 
-Subscribe to live events as data changes in the knowledge graph. This is useful for event-driven pipelines, dashboards, or audit logs. To be told when a rule-derived answer changes, use a [standing query](/docs/guides/websocket-api#standing-queries) over the WebSocket API; `subscribe()` arrives in the SDK with the upcoming release.
+Subscribe to live events as data changes in the knowledge graph. This is useful for event-driven pipelines, dashboards, or audit logs. To be told when a rule-derived answer changes, use [subscriptions](#subscriptions).
 
 ```python
 # Register a callback for a specific relation

diff --git a/packages/inputlayer-py/README.md b/packages/inputlayer-py/README.md
@@ -4,7 +4,7 @@ Python SDK for [InputLayer](https://github.com/inputlayer/inputlayer), the live
 
 Take the rules out of your prompts: declare facts and rules as typed Python classes, and InputLayer keeps every conclusion current as facts change. Write Python, no query syntax required: the SDK compiles your classes into IQL, InputLayer's rule language, and sends it over WebSocket. A knowledge graph (`il.knowledge_graph("support")`) is the facts and rule-derived conclusions of one domain.
 
-**Today and next.** This package declares relations and rules (`Relation`, `Derived`, `From`), writes and deletes facts, and queries the derived views. Phase 1 of the new SDK adds `kg.subscribe()` (the engine pushes each change to the agent) and `kg.claim()` (an agent's action is recorded only while the rules allow it): in progress, not merged yet. Until then, standing queries run over the [WebSocket API](https://inputlayer.ai/docs/guides/websocket-api/). The [main README](../../README.md) shows both forms side by side.
+**Today and next.** This package declares relations and rules (`Relation`, `Derived`, `From`), writes and deletes facts, and queries the derived views. It also subscribes to them: `kg.subscribe()` (the engine pushes each change to the agent), `kg.watch()` and `kg.on()`, described in the [Python SDK guide](../../docs/content/docs/guides/python-sdk.mdx#subscriptions). Phase 1 of the new SDK also adds `kg.claim()` (an agent's action is recorded only while the rules allow it): in progress, not merged yet. The [main README](../../README.md) shows both forms side by side.
 
 ## Installation
 
@@ -458,6 +458,9 @@ The autodetector diffs your current Python models against the last migration's s
 | `list_acl()` | List access control entries |
 | `debug(*select, ...)` | Show query plan without executing (same arguments as `query`) |
 | `execute(iql)` | Execute raw IQL |
+| `subscribe(*select, ..., queue=1024)` | Async iterator of `Change` events (snapshot, then deltas) |
+| `watch(*select, ...)` | Async iterator of the whole current result (`Live`) |
+| `on(*select, callback)` | Call `callback` with every `Change`; returns a handle with `close()` |
 | `status()` | Get server status |
 | `compact()` | Trigger storage compaction |
 

diff --git a/packages/inputlayer-py/src/inputlayer/__init__.py b/packages/inputlayer-py/src/inputlayer/__init__.py
@@ -52,6 +52,7 @@
     SchemaConflictError,
     StatementFailedError,
     StoreReadOnlyError,
+    SubscriptionRejected,
     ValidationError,
 )
 
@@ -89,6 +90,16 @@
 
 # Session
 from inputlayer.session import Session
+
+# Subscriptions
+from inputlayer.subscription import (
+    Change,
+    Live,
+    Row,
+    Subscription,
+    SubscriptionHandle,
+    SubscriptionStats,
+)
 from inputlayer.types import Timestamp, Vector, VectorInt8
 
 __version__ = "0.1.0"
@@ -99,6 +110,7 @@
     "AuthenticationError",
     "Cancelled",
     "CannotDropError",
+    "Change",
     "ClearResult",
     "ColumnInfo",
     "CompileError",
@@ -128,6 +140,7 @@
     "KnowledgeGraph",
     "KnowledgeGraphExistsError",
     "KnowledgeGraphNotFoundError",
+    "Live",
     # Notifications
     "NotificationEvent",
     "OutcomeUnknownError",
@@ -145,6 +158,7 @@
     "RelationNotFoundError",
     # Result
     "ResultSet",
+    "Row",
     "RuleClause",
     "RuleInfo",
     "RuleNotFoundError",
@@ -155,6 +169,10 @@
     "StatementError",
     "StatementFailedError",
     "StoreReadOnlyError",
+    "Subscription",
+    "SubscriptionHandle",
+    "SubscriptionRejected",
+    "SubscriptionStats",
     "Timestamp",
     # Auth
     "UserInfo",

diff --git a/packages/inputlayer-py/src/inputlayer/client_sync.py b/packages/inputlayer-py/src/inputlayer/client_sync.py
@@ -6,8 +6,8 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
-from typing import Any
+from collections.abc import AsyncIterator, Callable, Iterator
+from typing import Any, TypeVar
 
 from inputlayer._sync import run_sync
 from inputlayer.auth import AclEntry, ApiKeyInfo, UserInfo
@@ -30,6 +30,15 @@
 )
 from inputlayer.relation import Relation
 from inputlayer.result import ResultSet
+from inputlayer.subscription import (
+    Change,
+    Live,
+    Subscription,
+    SubscriptionHandle,
+    SubscriptionStats,
+)
+
+T = TypeVar("T")
 
 
 class KnowledgeGraphSync:
@@ -169,6 +178,102 @@ def clear_prefix(self, prefix: str) -> ClearResult:
     def execute(self, iql: str, *, timeout: float | None = None) -> ResultSet:
         return run_sync(self._kg.execute(iql, timeout=timeout))
 
+    # ── Subscriptions ─────────────────────────────────────────────────
+
+    def subscribe(self, *select: Any, **kwargs: Any) -> SubscriptionSync:
+        """``KnowledgeGraph.subscribe`` as a blocking iterator of ``Change``
+        events. The subscription runs on the background event loop, so events
+        keep arriving (up to ``queue``) between reads; leaving the loop or
+        ``close()`` ends it."""
+        return SubscriptionSync(self._kg.subscribe(*select, **kwargs))
+
+    def watch(self, *select: Any, **kwargs: Any) -> Iterator[Live[Any]]:
+        """``KnowledgeGraph.watch`` as a blocking iterator of ``Live`` results."""
+        levels = self._kg.watch(*select, **kwargs)
+        try:
+            while True:
+                try:
+                    yield run_sync(_next(levels))
+                except StopAsyncIteration:
+                    return
+        finally:
+            run_sync(levels.aclose())  # type: ignore[attr-defined]
+
+    def on(self, *args: Any, **kwargs: Any) -> SubscriptionHandleSync:
+        """``KnowledgeGraph.on``. The callback runs on the background event
+        loop's thread, one change at a time; keep it short, or hand the change
+        to your own thread."""
+
+        async def start() -> SubscriptionHandle:
+            return self._kg.on(*args, **kwargs)
+
+        return SubscriptionHandleSync(run_sync(start()))
+
+
+async def _next(iterator: AsyncIterator[T]) -> T:
+    return await iterator.__anext__()
+
+
+class SubscriptionSync:
+    """A subscription read by blocking: ``for change in kg.subscribe(...)``."""
+
+    def __init__(self, sub: Subscription[Any]) -> None:
+        self._sub = sub
+
+    @property
+    def id(self) -> str:
+        return self._sub.id
+
+    @property
+    def query(self) -> str:
+        return self._sub.query
+
+    @property
+    def stats(self) -> SubscriptionStats:
+        return self._sub.stats
+
+    def __iter__(self) -> Iterator[Change[Any]]:
+        # A generator, so that leaving the loop closes the subscription.
+        try:
+            while True:
+                try:
+                    change = self.next()
+                except StopIteration:
+                    return
+                yield change
+        finally:
+            self.close()
+
+    def next(self) -> Change[Any]:
+        """The next event, blocking until it arrives."""
+        try:
+            return run_sync(_next(self._sub))
+        except StopAsyncIteration:
+            raise StopIteration from None
+
+    def close(self) -> None:
+        run_sync(self._sub.close())
+
+    def __enter__(self) -> SubscriptionSync:
+        return self
+
+    def __exit__(self, *exc: Any) -> None:
+        self.close()
+
+
+class SubscriptionHandleSync:
+    """A callback subscription; see ``KnowledgeGraphSync.on``."""
+
+    def __init__(self, handle: SubscriptionHandle) -> None:
+        self._handle = handle
+
+    @property
+    def stats(self) -> SubscriptionStats:
+        return self._handle.stats
+
+    def close(self) -> None:
+        run_sync(self._handle.close())
+
 
 class InputLayerSync:
     """Synchronous wrapper around InputLayer."""

diff --git a/packages/inputlayer-py/src/inputlayer/compiler.py b/packages/inputlayer-py/src/inputlayer/compiler.py
@@ -655,6 +655,14 @@ def labels(self) -> list[str]:
 
     def shape(self, rows: list[list[Any]]) -> list[list[Any]]:
         """Pick the selected columns out of the engine's rows, by position."""
+        rows = self.pick(rows)
+        if self.dedupe:
+            rows = _distinct(rows)
+        return [self.from_engine(row) for row in rows]
+
+    def pick(self, rows: list[list[Any]]) -> list[list[Any]]:
+        """The selected columns of the engine's rows, by position, as the
+        engine sent them: a projection keeps its repeats."""
         if rows and len(rows[0]) != len(self.columns):
             raise InternalError(
                 f"The engine returned {len(rows[0])} columns for a query binding "
@@ -667,11 +675,10 @@ def shape(self, rows: list[list[Any]]) -> list[list[Any]]:
             else:
                 pick = operator.itemgetter(*idx)
                 rows = [list(pick(row)) for row in rows]
-        if self.dedupe:
-            rows = _distinct(rows)
-        return [self._from_engine(row) for row in rows]
+        return rows
 
-    def _from_engine(self, row: list[Any]) -> list[Any]:
+    def from_engine(self, row: list[Any]) -> list[Any]:
+        """A picked row with engine values made Python values (R-TYPE)."""
         return [
             ms_to_datetime(v)
             if o.is_datetime and isinstance(v, int) and not isinstance(v, bool)
@@ -696,7 +703,7 @@ def shape_why(self, rows: list[list[Any]]) -> list[int]:
         return picked[start:end]
 
     def project_why(self, row: list[Any]) -> list[Any]:
-        return self._from_engine([row[self.why_columns.index(o.variable)] for o in self.outputs])
+        return self.from_engine([row[self.why_columns.index(o.variable)] for o in self.outputs])
 
 
 def compile_query(