Add Remote session foundations and feasibility checks

Add opt-in Tentacle protocol 1.1, explicit remote grants, lifecycle IPC and expiring core leases in both SDKs. Revoke access before update preparation, including failed updates.

Add transient secret entries, shared Braille cell export and isolated Xorg recovery experiments with focused tests and documented limitations. Production consent, desktop services and live interoperability remain pending.

sdk/go/tentacle/session.go

@@ -0,0 +1,301 @@
+// Copyright (c) 2026 Storm Dragon
+// SPDX-License-Identifier: GPL-3.0-or-later
+
+package tentacle
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+	"strconv"
+	"sync"
+	"syscall"
+	"time"
+
+	"git.stormux.org/storm/cthulhu/internal/tentaclewire"
+)
+
+type SessionEvent = tentaclewire.SessionEvent
+type RemoteOpen = tentaclewire.RemoteOpen
+type RemoteLease = tentaclewire.RemoteLease
+type RemoteOperation = tentaclewire.RemoteOperation
+
+type SessionError struct {
+	Code    int    `json:"code"`
+	Message string `json:"message"`
+}
+
+func (err *SessionError) Error() string { return "tentacle session error " + strconv.Itoa(err.Code) }
+
+type sessionReply struct {
+	value json.RawMessage
+	err   error
+}
+
+// SessionClient is single-use. ServeSession owns its bounded writer and receive
+// loop; command handlers and background workers may Call concurrently. Events
+// must be drained without blocking on network I/O. Overflow retires the process
+// session rather than dropping lifecycle events. No desktop access is implicit.
+type SessionClient struct {
+	mu            sync.Mutex
+	ctx           context.Context
+	abort         func()
+	active        bool
+	started       bool
+	sequence      uint64
+	eventSequence uint64
+	pending       map[string]chan sessionReply
+	events        chan SessionEvent
+	outbound      chan []byte
+	queuedBytes   int
+}
+
+func NewSessionClient() *SessionClient {
+	return &SessionClient{events: make(chan SessionEvent, 32), outbound: make(chan []byte, 32), pending: make(map[string]chan sessionReply)}
+}
+
+func (client *SessionClient) Events() <-chan SessionEvent { return client.events }
+
+// ServeSession opts into protocol 1.1; a protocol 1.0 host is rejected.
+func ServeSession(input io.ReadCloser, output io.WriteCloser, commands []Command, extensions *Extensions, client *SessionClient) error {
+	if client == nil {
+		return ErrProtocol
+	}
+	if extensions != nil {
+		if err := tentaclewire.ValidateExtensions(*extensions); err != nil {
+			return err
+		}
+	}
+	// os.Stdin may be created before the runtime marks a pipe pollable. A
+	// blocking read on it cannot reliably be interrupted by Close from another
+	// goroutine. Adopt owned nonblocking duplicates so Go's poller can wake both
+	// pipe directions on timeout. Non-file closers (e.g. io.Pipe) retain their
+	// documented Close behavior.
+	if file, ok := input.(*os.File); ok {
+		pollable, err := sessionPipe(file)
+		if err != nil {
+			input.Close()
+			output.Close()
+			return err
+		}
+		input = pollable
+	}
+	if file, ok := output.(*os.File); ok {
+		pollable, err := sessionPipe(file)
+		if err != nil {
+			input.Close()
+			output.Close()
+			return err
+		}
+		output = pollable
+	}
+	return serve(input, output, commands, extensions, client)
+}
+
+func sessionPipe(file *os.File) (*os.File, error) {
+	fd, err := syscall.Dup(int(file.Fd()))
+	if err != nil {
+		return nil, err
+	}
+	syscall.CloseOnExec(fd)
+	if err := syscall.SetNonblock(fd, true); err != nil {
+		syscall.Close(fd)
+		return nil, err
+	}
+	owned := os.NewFile(uintptr(fd), file.Name())
+	file.Close()
+	return owned, nil
+}
+
+func (client *SessionClient) start(ctx context.Context, abort func()) bool {
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	if client.started || client.events == nil {
+		return false
+	}
+	client.started, client.ctx, client.abort = true, ctx, abort
+	return true
+}
+
+func (client *SessionClient) enqueue(frame []byte) error {
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	return client.enqueueLocked(frame)
+}
+
+func (client *SessionClient) enqueueLocked(frame []byte) error {
+	if client.ctx == nil || client.ctx.Err() != nil {
+		return io.ErrClosedPipe
+	}
+	if client.queuedBytes+len(frame) > tentaclewire.SessionQueueBytes {
+		client.abort()
+		return tentaclewire.ErrLimit
+	}
+	select {
+	case client.outbound <- frame:
+		client.queuedBytes += len(frame)
+		return nil
+	default:
+		client.abort()
+		return tentaclewire.ErrLimit
+	}
+}
+
+func (client *SessionClient) write(output io.Writer) {
+	for {
+		select {
+		case <-client.ctx.Done():
+			return
+		case frame := <-client.outbound:
+			if client.ctx.Err() != nil {
+				return
+			}
+			n, err := output.Write(frame)
+			if err == nil && n != len(frame) {
+				err = io.ErrShortWrite
+			}
+			client.mu.Lock()
+			client.queuedBytes -= len(frame)
+			client.mu.Unlock()
+			if err != nil {
+				client.abort()
+				return
+			}
+		}
+	}
+}
+
+// Call has a two-second maximum deadline. Timeout/cancellation retires the
+// session, so an abandoned effect cannot leave unbounded pending calls behind.
+// The host currently implements remote.open, remote.close and remote.heartbeat.
+func (client *SessionClient) Call(ctx context.Context, method string, params any, result any) error {
+	if ctx == nil {
+		return errors.New("nil session call context")
+	}
+	ctx, cancel := context.WithTimeout(ctx, 2*time.Second)
+	defer cancel()
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	data, err := json.Marshal(params)
+	if err != nil || len(data) > 8192 || tentaclewire.ValidateJSON(data) != nil || len(method) == 0 || len(method) > 128 {
+		return ErrProtocol
+	}
+	client.mu.Lock()
+	if !client.active || client.ctx.Err() != nil {
+		client.mu.Unlock()
+		return io.ErrClosedPipe
+	}
+	if len(client.pending) >= 8 {
+		client.mu.Unlock()
+		return tentaclewire.ErrLimit
+	}
+	client.sequence++
+	id := "c:" + strconv.FormatUint(client.sequence, 10)
+	frame, err := json.Marshal(request{JSONRPC: "2.0", ID: id, Method: method, Params: data})
+	if err != nil {
+		client.mu.Unlock()
+		return ErrProtocol
+	}
+	response := make(chan sessionReply, 1)
+	client.pending[id] = response
+	err = client.enqueueLocked(append(frame, '\n'))
+	client.mu.Unlock()
+	defer func() { client.mu.Lock(); delete(client.pending, id); client.mu.Unlock() }()
+	if err != nil {
+		return err
+	}
+	select {
+	case <-ctx.Done():
+		client.abort()
+		return ctx.Err()
+	case <-client.ctx.Done():
+		return io.ErrClosedPipe
+	case reply := <-response:
+		if ctx.Err() != nil {
+			client.abort()
+			return ctx.Err()
+		}
+		if client.ctx.Err() != nil {
+			return io.ErrClosedPipe
+		}
+		if reply.err != nil {
+			return reply.err
+		}
+		if result != nil {
+			return tentaclewire.DecodeStrict(reply.value, result)
+		}
+		return nil
+	}
+}
+
+func (client *SessionClient) acceptReply(frame []byte) error {
+	var reply struct {
+		JSONRPC string          `json:"jsonrpc"`
+		ID      string          `json:"id"`
+		Result  json.RawMessage `json:"result"`
+		Error   *SessionError   `json:"error"`
+	}
+	if tentaclewire.DecodeStrict(frame, &reply) != nil || reply.JSONRPC != "2.0" {
+		return ErrProtocol
+	}
+	var fields map[string]json.RawMessage
+	_ = json.Unmarshal(frame, &fields)
+	_, hasResult := fields["result"]
+	_, hasError := fields["error"]
+	if hasResult == hasError || hasError && (reply.Error == nil || len(reply.Error.Message) > 1024) {
+		return ErrProtocol
+	}
+	if hasError {
+		var fields map[string]json.RawMessage
+		_ = json.Unmarshal(frame, &fields)
+		var detail map[string]json.RawMessage
+		_ = json.Unmarshal(fields["error"], &detail)
+		if len(detail["code"]) == 0 || string(detail["code"]) == "null" || len(detail["message"]) == 0 || string(detail["message"]) == "null" {
+			return ErrProtocol
+		}
+	}
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	waiter := client.pending[reply.ID]
+	if !client.active || waiter == nil {
+		return ErrProtocol
+	}
+	delete(client.pending, reply.ID)
+	var err error
+	if reply.Error != nil {
+		err = reply.Error
+	}
+	waiter <- sessionReply{value: reply.Result, err: err}
+	return nil
+}
+
+func (client *SessionClient) acceptEvent(params json.RawMessage) error {
+	var event SessionEvent
+	var fields map[string]json.RawMessage
+	if json.Unmarshal(params, &fields) != nil {
+		return ErrProtocol
+	}
+	for _, value := range fields {
+		if string(value) == "null" {
+			return ErrProtocol
+		}
+	}
+	if tentaclewire.DecodeStrict(params, &event) != nil || !tentaclewire.ValidSessionEvent(event) {
+		return ErrProtocol
+	}
+	client.mu.Lock()
+	defer client.mu.Unlock()
+	if !client.active || event.Sequence != client.eventSequence+1 {
+		return ErrProtocol
+	}
+	client.eventSequence = event.Sequence
+	select {
+	case client.events <- event:
+		return nil
+	default:
+		return tentaclewire.ErrLimit
+	}
+}

sdk/go/tentacle/tentacle.go

@@ -12,6 +12,7 @@ import (
 	"errors"
 	"git.stormux.org/storm/cthulhu/internal/config"
 	"io"
+	"slices"
 	"strconv"
 	"sync"
 
@@ -46,6 +47,9 @@ type Interaction = tentaclewire.Interaction
 type Submission = tentaclewire.Submission
 type Item = tentaclewire.Item
 type Form = interaction.Form
+
+// Element.Secret masks transient Entry controls; it is not supported in
+// persistent PreferencePage controls. Submitted Value.Text remains sensitive.
 type Element = interaction.Element
 type ListItem = interaction.Item
 type Value = interaction.Value
@@ -90,7 +94,7 @@ type request struct {
 // handlers and waits for them; the host kills a child that ignores cancellation.
 // Never print to stdout. Command declarations must remain unchanged during Serve.
 func Serve(input io.ReadCloser, output io.WriteCloser, commands []Command) error {
-	return serve(input, output, commands, nil)
+	return serve(input, output, commands, nil, nil)
 }
 
 // ServeWithExtensions declares session-owned preferences pages and quick settings.
@@ -98,14 +102,40 @@ func ServeWithExtensions(input io.ReadCloser, output io.WriteCloser, commands []
 	if err := tentaclewire.ValidateExtensions(extensions); err != nil {
 		return err
 	}
-	return serve(input, output, commands, &extensions)
+	return serve(input, output, commands, &extensions, nil)
 }
 
-func serve(input io.ReadCloser, output io.WriteCloser, commands []Command, extensions *Extensions) error {
+func serve(input io.ReadCloser, output io.WriteCloser, commands []Command, extensions *Extensions, client *SessionClient) error {
 	ctx, cancel := context.WithCancel(context.Background())
 	var workers sync.WaitGroup
 	var writeMutex sync.Mutex
-	defer func() { cancel(); input.Close(); output.Close(); workers.Wait() }()
+	clientStarted := false
+	defer func() {
+		cancel()
+		input.Close()
+		output.Close()
+		if clientStarted {
+			client.mu.Lock()
+			client.active = false
+			client.pending = nil
+			close(client.events)
+			client.mu.Unlock()
+		}
+		workers.Wait()
+		if clientStarted {
+			for len(client.outbound) != 0 {
+				<-client.outbound
+			}
+		}
+	}()
+	if client != nil {
+		if !client.start(ctx, func() { cancel(); input.Close(); output.Close() }) {
+			return ErrProtocol
+		}
+		clientStarted = true
+		workers.Add(1)
+		go func() { defer workers.Done(); client.write(output) }()
+	}
 	handlers := make(map[string]Handler)
 	readerCommands := make(map[string]bool)
 	if len(commands) == 0 || len(commands) > 32 {
@@ -131,6 +161,12 @@ func serve(input io.ReadCloser, output io.WriteCloser, commands []Command, exten
 			frame, _ = json.Marshal(map[string]any{"jsonrpc": "2.0", "id": id,
 				"error": map[string]any{"code": -32603, "message": "Invalid or oversized result"}})
 		}
+		if client != nil {
+			if release != nil {
+				release()
+			}
+			return client.enqueue(append(frame, '\n'))
+		}
 		writeMutex.Lock()
 		defer writeMutex.Unlock()
 		// A completed reply may reach the host before Write returns. Release
@@ -157,6 +193,18 @@ func serve(input io.ReadCloser, output io.WriteCloser, commands []Command, exten
 		if err != nil {
 			return err
 		}
+		if client != nil {
+			var fields map[string]json.RawMessage
+			if tentaclewire.ValidateJSON(frame) != nil || json.Unmarshal(frame, &fields) != nil {
+				return ErrProtocol
+			}
+			if _, reply := fields["method"]; !reply {
+				if err := client.acceptReply(frame); err != nil {
+					return err
+				}
+				continue
+			}
+		}
 		var message request
 		if tentaclewire.DecodeStrict(frame, &message) != nil || message.JSONRPC != "2.0" {
 			return ErrProtocol
@@ -190,16 +238,16 @@ func serve(input io.ReadCloser, output io.WriteCloser, commands []Command, exten
 				Capabilities []string `json:"capabilities"`
 			}
 			if session != "" || message.ID == "" || tentaclewire.DecodeStrict(message.Params, &params) != nil ||
-				params.Protocol.Major == nil || params.Protocol.Minor == nil || *params.Protocol.Major != 1 || *params.Protocol.Minor != 0 || params.Session == "" || len(params.Session) > 128 ||
-				!tentaclewire.ValidAPIs(params.APIs, params.Capabilities) {
+				params.Protocol.Major == nil || params.Protocol.Minor == nil || *params.Protocol.Major != 1 || (*params.Protocol.Minor != 0 && *params.Protocol.Minor != 1) || params.Session == "" || len(params.Session) > 128 ||
+				!tentaclewire.ValidNegotiation(*params.Protocol.Minor, params.APIs, params.Capabilities) || *params.Protocol.Minor == 1 && params.Capabilities == nil || (client != nil) != (*params.Protocol.Minor == 1) {
 				return ErrProtocol
 			}
 			for _, command := range commands {
-				if command.Presentation && len(params.APIs) < 2 || command.Interaction && (!command.Presentation || len(params.APIs) < 3) || command.Reader && len(params.APIs) != 4 {
+				if command.Presentation && !slices.Contains(params.APIs, "presentation") || command.Interaction && (!command.Presentation || !slices.Contains(params.APIs, "interaction")) || command.Reader && !slices.Contains(params.APIs, "reader-settings") {
 					return ErrProtocol
 				}
 			}
-			if extensions != nil && len(params.APIs) != 4 {
+			if extensions != nil && !slices.Contains(params.APIs, "reader-settings") {
 				return ErrProtocol
 			}
 			session = params.Session
@@ -218,6 +266,18 @@ func serve(input io.ReadCloser, output io.WriteCloser, commands []Command, exten
 				return ErrProtocol
 			}
 			active = true
+			if client != nil {
+				client.mu.Lock()
+				client.active = true
+				client.mu.Unlock()
+			}
+		case "session.event":
+			if !active || message.ID != "" || client == nil {
+				return ErrProtocol
+			}
+			if err := client.acceptEvent(message.Params); err != nil {
+				return err
+			}
 		case "command.invoke":
 			var params struct {
 				Command   string          `json:"command"`

sdk/python/cthulhu_tentacle/__init__.py

@@ -1,6 +1,12 @@
 # Copyright (c) 2026 Storm Dragon
 # SPDX-License-Identifier: GPL-3.0-or-later
-"""Experimental command/presentation SDK. stdout is reserved for the protocol."""
+"""Experimental command/presentation SDK. stdout is reserved for the protocol.
+
+Interaction form dictionaries may set secret=True on entries (kind=2). The host
+masks editing output and disallows copy/cut; successful submissions still carry
+plaintext values to the continuation. Never log or present those values. Secret
+entries cannot be used in persistent preference pages; older hosts reject them.
+"""
 
 import json
 import sys
@@ -80,7 +86,7 @@ def _fields(value, required, optional=()):
         raise ProtocolError("Invalid fields")
 
 
-def serve(commands, inputStream=None, outputStream=None, extensions=None):
+def serve(commands, inputStream=None, outputStream=None, extensions=None, session_client=None):
     """Run in a dedicated child, then exit. Handlers receive (cancelled, arguments).
 
     cancelled is a threading.Event shared by the session. Up to eight handlers
@@ -94,6 +100,9 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
     writeLock = threading.Lock()
     workers = []
     slots = threading.BoundedSemaphore(8)
+    client_started = False
+    if session_client is not None:
+        from .session import valid_negotiation
 
     def write_reply(requestId, value=None, failed=False, release=None):
         reply = {"jsonrpc": "2.0", "id": requestId}
@@ -108,6 +117,11 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
         except (ValueError, TypeError, UnicodeError, RecursionError, ProtocolError):
             frame = _encode({"jsonrpc": "2.0", "id": requestId, "error": {
                 "code": -32603, "message": "Invalid or oversized result"}})
+        if session_client is not None:
+            if release is not None:
+                release()
+            session_client._enqueue(frame + b"\n")
+            return
         with writeLock:
             # Bound eight handlers/waiting replies plus one serialized writer.
             if release is not None:
@@ -133,12 +147,17 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
             failed = False
         try:
             write_reply(requestId, value, failed, release_slot)
-        except (OSError, ValueError):
+        except (OSError, ValueError, ProtocolError, SessionError):
             cancelled.set()
+            if session_client is not None:
+                session_client._abort()
         finally:
             release_slot()
 
     try:
+        if session_client is not None:
+            session_client._start(cancelled, inputStream, outputStream)
+            client_started = True
         commands = list(commands)
         handlers = {command.id: command.handle for command in commands}
         if (not 1 <= len(commands) <= 32 or len(handlers) != len(commands)
@@ -151,7 +170,7 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
             raise ProtocolError("Invalid command capabilities")
         session, active, sequence = None, False, 0
         while not cancelled.is_set():
-            frame = inputStream.readline(MAX_FRAME + 2)
+            frame = session_client._readline() if session_client is not None else inputStream.readline(MAX_FRAME + 2)
             if not frame:
                 return
             if not frame.endswith(b"\n") or len(frame) > MAX_FRAME + 1:
@@ -162,6 +181,9 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
                 _depth(message)
             except (ValueError, UnicodeError, RecursionError) as error:
                 raise ProtocolError("Invalid JSON") from error
+            if session_client is not None and isinstance(message, dict) and "method" not in message:
+                session_client._reply(message)
+                continue
             _fields(message, ("jsonrpc", "method"), ("id", "params"))
             if message["jsonrpc"] != "2.0":
                 raise ProtocolError("Invalid JSON-RPC version")
@@ -179,19 +201,23 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
             if method == "initialize":
                 _fields(params, ("protocol", "session", "apis", "capabilities"))
                 version = params["protocol"]
-                if (session is not None or not requestId or version != {"major": 1, "minor": 0}
+                expected_minor = 1 if session_client is not None else 0
+                apis, capabilities = params["apis"], params["capabilities"]
+                valid_apis = (valid_negotiation(apis, capabilities) if session_client is not None else
+                              apis in (["commands"], ["commands", "presentation"],
+                                       ["commands", "presentation", "interaction"],
+                                       ["commands", "presentation", "interaction", "reader-settings"])
+                              and capabilities == apis)
+                if (session is not None or not requestId or version != {"major": 1, "minor": expected_minor}
                         or any(type(value) is not int for value in version.values())
                         or not isinstance(params["session"], str) or not params["session"]
                         or len(params["session"].encode("utf-8")) > 128
-                        or params["apis"] not in (["commands"], ["commands", "presentation"],
-                            ["commands", "presentation", "interaction"],
-                            ["commands", "presentation", "interaction", "reader-settings"])
-                        or params["capabilities"] != params["apis"]
+                        or not valid_apis
                         or (any(command.presentation for command in commands)
-                            and len(params["apis"]) < 2)
-                        or (any(command.interaction for command in commands) and len(params["apis"]) < 3)
+                            and "presentation" not in apis)
+                        or (any(command.interaction for command in commands) and "interaction" not in apis)
                         or ((extensions is not None or any(command.reader for command in commands))
-                            and len(params["apis"]) != 4)):
+                            and "reader-settings" not in apis)):
                     raise ProtocolError("Invalid initialization")
                 session = params["session"]
                 registration = {"protocol": version, "session": session,
@@ -204,6 +230,13 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
                 if session is None or active or requestId or params["session"] != session:
                     raise ProtocolError("Invalid activation")
                 active = True
+                if session_client is not None:
+                    with session_client._lock:
+                        session_client._active = True
+            elif method == "session.event":
+                if not active or requestId or session_client is None:
+                    raise ProtocolError("Unnegotiated event")
+                session_client._event(params)
             elif method == "command.invoke":
                 _fields(params, ("command", "arguments"))
                 if (not active or not requestId or not isinstance(params["command"], str)
@@ -226,9 +259,16 @@ def serve(commands, inputStream=None, outputStream=None, extensions=None):
                 raise ProtocolError("Unknown method")
     finally:
         cancelled.set()
+        if client_started:
+            session_client._finish()
         # Join before closing buffered stdout: a blocked writer must remain
         # subject to the host deadline rather than race a Python buffer close.
         for worker in workers:
             worker.join()
+        if client_started:
+            session_client._close()
         inputStream.close()
         outputStream.close()
+
+# Imported after helpers to keep the protocol 1.0 API compatible.
+from .session import SessionClient, SessionError

sdk/python/cthulhu_tentacle/session.py

@@ -0,0 +1,256 @@
+# Copyright (c) 2026 Storm Dragon
+# SPDX-License-Identifier: GPL-3.0-or-later
+"""Opt-in protocol 1.1 lifecycle transport. No desktop effects are provided."""
+
+import os
+import queue
+import select
+import threading
+
+from . import MAX_FRAME, ProtocolError, _encode, _fields
+
+REMOTE_GRANTS = frozenset(("output-observe", "remote-present", "keyboard-capture",
+                          "keyboard-inject", "braille-stream", "clipboard-exchange"))
+BASE_APIS = frozenset(("commands", "presentation", "interaction", "reader-settings"))
+
+
+def valid_negotiation(apis, capabilities):
+    if (not isinstance(apis, list) or not isinstance(capabilities, list)
+            or any(not isinstance(value, str) for value in apis + capabilities)):
+        return False
+    a, c = set(apis), set(capabilities)
+    return (len(a) == len(apis) and len(c) == len(capabilities)
+            and {"commands", "session-events"} <= a
+            and a <= BASE_APIS | {"session-events", "remote-access"}
+            and c <= (BASE_APIS - {"commands"}) | REMOTE_GRANTS
+            and a & (BASE_APIS - {"commands"}) == c & (BASE_APIS - {"commands"})
+            and ("interaction" not in a or "presentation" in a)
+            and ("reader-settings" not in a or "interaction" in a)
+            and ("remote-access" in a) == bool(c & REMOTE_GRANTS))
+
+
+def _positive(value):
+    return type(value) is int and 0 < value < 2**64
+
+
+def _token(value):
+    return isinstance(value, str) and len(value.encode("utf-8")) == 32
+
+
+class SessionError(Exception):
+    def __init__(self, code, message):
+        super().__init__("Tentacle session error " + str(code))
+        self.code, self.message = code, message
+
+
+class SessionClient:
+    """Single-use client passed as session_client= to serve.
+
+    Call from handlers or owned background workers, never from the receive loop.
+    next_event blocks until an event or session shutdown. Call timeout retires
+    the session; the host revokes its lease. Binary streams must have pipe FDs.
+    """
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._events = queue.Queue(32)
+        self._outbound = queue.Queue(32)
+        self._pending = {}
+        self._bytes = self._sequence = self._event_sequence = 0
+        self._started = self._active = False
+        self._stopped = threading.Event()
+        self._wake_read = self._wake_write = None
+        self._buffer = bytearray()
+
+    def _start(self, cancelled, input_stream, output_stream):
+        with self._lock:
+            if self._started:
+                raise ProtocolError("Session client already used")
+            self._started = True
+            self._cancelled = cancelled
+            self._input, self._output = input_stream.fileno(), output_stream.fileno()
+            self._wake_read, self._wake_write = os.pipe()
+            os.set_blocking(self._wake_write, False)
+            os.set_blocking(self._input, False)
+            os.set_blocking(self._output, False)
+        self._writer = threading.Thread(target=self._write_loop)
+        self._writer.start()
+
+    def _abort(self):
+        # Called under _lock too; never acquire it here.
+        self._stopped.set()
+        self._cancelled.set()
+        if self._wake_write is not None:
+            try:
+                os.write(self._wake_write, b"x")
+            except OSError:
+                pass
+        try:
+            self._outbound.put_nowait(None)
+        except queue.Full:
+            pass
+
+    def _finish(self):
+        self._abort()
+        self._writer.join()
+        with self._lock:
+            self._active = False
+            for waiter in self._pending.values():
+                waiter.put_nowait(SessionError(-32000, "Session closed"))
+            self._pending.clear()
+            while not self._events.empty():
+                self._events.get_nowait()
+            self._events.put_nowait(None)
+    def _close(self):
+        os.close(self._wake_read)
+        os.close(self._wake_write)
+        self._wake_read = self._wake_write = None
+
+    def _readline(self):
+        while not self._stopped.is_set():
+            end = self._buffer.find(b"\n")
+            if end >= 0:
+                frame = bytes(self._buffer[:end + 1])
+                del self._buffer[:end + 1]
+                if len(frame) > MAX_FRAME + 1:
+                    raise ProtocolError("Frame limit")
+                return frame
+            if len(self._buffer) > MAX_FRAME:
+                raise ProtocolError("Frame limit")
+            ready, _, _ = select.select([self._input, self._wake_read], [], [])
+            if self._wake_read in ready:
+                return b""
+            try:
+                data = os.read(self._input, 4096)
+            except BlockingIOError:
+                continue
+            if not data:
+                if self._buffer:
+                    raise ProtocolError("Unterminated frame")
+                return b""
+            self._buffer.extend(data)
+        return b""
+
+    def _enqueue_locked(self, frame):
+        if self._stopped.is_set():
+            raise SessionError(-32000, "Session closed")
+        if self._bytes + len(frame) > 256 * 1024:
+            self._abort()
+            raise ProtocolError("Session queue limit")
+        try:
+            self._outbound.put_nowait(frame)
+        except queue.Full:
+            self._abort()
+            raise ProtocolError("Session queue limit") from None
+        self._bytes += len(frame)
+
+    def _enqueue(self, frame):
+        with self._lock:
+            self._enqueue_locked(frame)
+
+    def _write_loop(self):
+        try:
+            while not self._stopped.is_set():
+                frame = self._outbound.get()
+                if frame is None:
+                    return
+                position = 0
+                while position < len(frame) and not self._stopped.is_set():
+                    ready, writable, _ = select.select([self._wake_read], [self._output], [])
+                    if ready:
+                        return
+                    if writable:
+                        try:
+                            position += os.write(self._output, frame[position:])
+                        except BlockingIOError:
+                            continue
+                with self._lock:
+                    self._bytes -= len(frame)
+        except OSError:
+            self._abort()
+
+    def call(self, method, params, timeout=2):
+        if not isinstance(method, str) or not 1 <= len(method.encode("utf-8")) <= 128 or len(_encode(params)) > 8192:
+            raise ProtocolError("Invalid session request")
+        if not 0 < timeout <= 2:
+            raise ValueError("Session call timeout must be in (0, 2]")
+        with self._lock:
+            if not self._active or self._stopped.is_set():
+                raise SessionError(-32000, "Session inactive")
+            if len(self._pending) >= 8:
+                raise ProtocolError("Pending call limit")
+            self._sequence += 1
+            request_id = "c:" + str(self._sequence)
+            waiter = queue.Queue(1)
+            self._pending[request_id] = waiter
+            try:
+                self._enqueue_locked(_encode({"jsonrpc": "2.0", "id": request_id,
+                                              "method": method, "params": params}) + b"\n")
+            except BaseException:
+                del self._pending[request_id]
+                raise
+        try:
+            value = waiter.get(timeout=timeout)
+            if isinstance(value, SessionError):
+                raise value
+            if self._stopped.is_set():
+                raise SessionError(-32000, "Session closed")
+            return value
+        except queue.Empty:
+            self._abort()
+            raise TimeoutError("Session call expired") from None
+        finally:
+            with self._lock:
+                self._pending.pop(request_id, None)
+
+    def next_event(self, timeout=None):
+        if self._stopped.is_set():
+            return None
+        return self._events.get(timeout=timeout)
+
+    def _reply(self, message):
+        _fields(message, ("jsonrpc", "id"), ("result", "error"))
+        if (message["jsonrpc"] != "2.0" or not isinstance(message["id"], str)
+                or ("result" in message) == ("error" in message)):
+            raise ProtocolError("Invalid session reply")
+        value = message.get("result")
+        if "error" in message:
+            error = message["error"]
+            _fields(error, ("code", "message"))
+            if type(error["code"]) is not int or not isinstance(error["message"], str) or len(error["message"].encode("utf-8")) > 1024:
+                raise ProtocolError("Invalid session error")
+            value = SessionError(error["code"], error["message"])
+        with self._lock:
+            waiter = self._pending.pop(message["id"], None)
+            if not self._active or waiter is None:
+                raise ProtocolError("Unexpected session reply")
+            waiter.put_nowait(value)
+
+    def _event(self, event):
+        _fields(event, ("sequence", "kind", "epoch"), ("action", "lease", "role", "grants", "reason"))
+        if not _positive(event["sequence"]) or not _positive(event["epoch"]):
+            raise ProtocolError("Invalid event sequence")
+        if event["kind"] == "remote.action":
+            grants = event.get("grants")
+            role = event.get("role")
+            if (not _token(event.get("action")) or event.get("lease", "") != "" or event.get("reason", "") != ""
+                    or role not in ("controller", "controlled") or not isinstance(grants, list)
+                    or not grants or any(not isinstance(grant, str) for grant in grants)
+                    or len(set(grants)) != len(grants) or not set(grants) <= REMOTE_GRANTS
+                    or role == "controller" and set(grants) & {"output-observe", "keyboard-inject"}
+                    or role == "controlled" and set(grants) & {"remote-present", "keyboard-capture"}):
+                raise ProtocolError("Invalid action event")
+        elif event["kind"] == "remote.revoked":
+            if (not _token(event.get("lease")) or event.get("action", "") != "" or event.get("role", "") != ""
+                    or event.get("grants", []) != [] or event.get("reason") not in ("closed", "expired", "local", "replaced")):
+                raise ProtocolError("Invalid revoke event")
+        else:
+            raise ProtocolError("Unknown session event")
+        with self._lock:
+            if not self._active or event["sequence"] != self._event_sequence + 1:
+                raise ProtocolError("Out-of-order session event")
+            self._event_sequence = event["sequence"]
+            try:
+                self._events.put_nowait(event)
+            except queue.Full:
+                self._abort()
+                raise ProtocolError("Event queue limit") from None