IPCRecoveryClient

An IPC client with automatic connection recovery. Wraps IPCClient to provide automatic reconnection when the server becomes unavailable, with session reattachment, exponential backoff, and status callbacks for UI updates. Like IPCClient, it exposes the .session(...) + ask/complete/stream facade (recommended); the methods documented below are the low-level building blocks the facade uses.

from jaato_sdk.client import IPCRecoveryClient, ConnectionState
from jaato_sdk.client.config import RecoveryConfig, get_recovery_config
from jaato_sdk.events import ClientType

Same recovery over WebSocket — WSRecoveryClient

For a remote daemon, WSRecoveryClient offers identical recovery over ws:// / wss://: the same state machine, session reattachment, backoff, and on_status_change callback, just on a WebSocket transport (plus ssl= / ca= for self-signed wss:// TLS). Reach for it via jaato.session(mode="ws", recovery=True, …) or WSRecoveryClient.session(…); it needs the jaato-sdk[ws] extra. Everything documented below applies to both clients.

Guide

For a walkthrough of connection recovery concepts, state transitions, and UI patterns, see the Connection Recovery guide.

Quick example (facade)

import asyncio
from jaato_sdk import IPCRecoveryClient

async def main():
    # The .session() facade + ask/complete/stream, with auto-reconnect.
    # Connection, session creation, and reattachment are handled for you.
    async with IPCRecoveryClient.session(
        profile={"model": "...", "provider": "...", "plugins": []},
        on_status_change=lambda st: print(st.state),  # RECONNECTING / CONNECTED
    ) as s:
        print(await s.ask("Long task..."))

asyncio.run(main())

Low-level API

The methods below (connect(), create_session(), send_message(), events(), ...) are the building blocks the facade is built on. Use them directly for custom event handling; with the facade, iterating events() yourself is optional.

Constructor

init

IPCRecoveryClient(
  socket_path: str = DEFAULT_SOCKET_PATH,
  *,
  client_type: ClientType,
  config: Optional[RecoveryConfig] = None,
  auto_start: bool = True,
  env_file: str = ".env",
  workspace_path: Optional[Path] = None,
  on_status_change: Optional[StatusCallback] = None,
  min_protocol_version: Optional[str] = None
)

Creates a new recovery client instance.

socket_path str optional
Path to Unix domain socket or Windows named pipe. Defaults to DEFAULT_SOCKET_PATH.
client_type ClientType required
Required keyword argument. Identifies the kind of client for server-side presentation and lifecycle filters. Pass ClientType.TERMINAL for interactive TUI clients or ClientType.API for headless orchestrators, batch jobs, or cascade harnesses. Forwarded verbatim to the inner IPCClient. Import from jaato_sdk.events.
config RecoveryConfig optional
Recovery configuration controlling reconnection behavior. If None, loads automatically from config files and environment variables via get_recovery_config().
auto_start bool optional
Whether to auto-start the server daemon if it is not already running. Default: True.
env_file str optional
Path to .env file passed to the auto-started server. Default: ".env".
workspace_path Path optional
Workspace path for loading project-level config (.jaato/client.json).
on_status_change StatusCallback optional
Callback invoked on every connection state transition. Receives a ConnectionStatus object. Signature: (status: ConnectionStatus) -> None.
min_protocol_version str optional
Override the inner IPCClient.MIN_PROTOCOL_VERSION for this connection. When set, the client refuses to connect to servers whose wire-protocol version is older than this value and raises IncompatibleServerError. Use only when targeting an unreleased daemon; production deployments should rely on the class-level default.

Returns

IPCRecoveryClient

Create with defaults (auto-loads config)

from jaato_sdk.client import IPCRecoveryClient
from jaato_sdk.events import ClientType

# Config loaded from ~/.jaato/client.json,
# .jaato/client.json, and JAATO_IPC_* env vars
client = IPCRecoveryClient(
    socket_path="/tmp/jaato.sock",
    client_type=ClientType.TERMINAL,   # required
)

Create with explicit config

from jaato_sdk.client import IPCRecoveryClient
from jaato_sdk.client.config import RecoveryConfig
from jaato_sdk.events import ClientType

config = RecoveryConfig(
    max_attempts=5,
    base_delay=2.0,
    max_delay=30.0,
)

client = IPCRecoveryClient(
    socket_path="/tmp/jaato.sock",
    client_type=ClientType.API,        # required
    config=config,
    auto_start=False,
    on_status_change=on_status,
)

Connection

connect

async connect(timeout: float = 5.0) -> bool

Connect to the server. If auto_start is enabled and the server is not running, starts it automatically.

timeout float optional
Connection timeout in seconds. Default: 5.0.

Returns

bool — True if connected successfully.

Raises

ConnectionError if connection fails.
ConnectionClosedError if the client is already closed.

disconnect

async disconnect() -> None

Disconnect gracefully. Cancels any reconnection in progress and transitions to DISCONNECTED. The client can be reconnected later by calling connect() again.

close

async close() -> None

Permanently close the connection. Transitions to the terminal CLOSED state — no further reconnection attempts will be made.

close() Is Permanent

After calling close(), the client cannot be reconnected. Use disconnect() for temporary disconnection.

is_connected property

Returns True when the state is CONNECTED.

is_reconnecting property

Returns True when the state is RECONNECTING.

is_closed property

Returns True when the state is CLOSED.

state property

Returns the current ConnectionState.

client_id property

Returns the client ID assigned by the server, or None if not connected.

server_version property

Returns the server's package version string (from ConnectedEvent.server_info["server_version"]), or None if not connected or the server did not report one. Delegates to the inner IPCClient.server_version. This is a diagnostics-only field — protocol compatibility is checked against the wire-protocol version, not the package version.

Connection lifecycle

from jaato_sdk.events import ClientType

client = IPCRecoveryClient(
    "/tmp/jaato.sock",
    client_type=ClientType.TERMINAL,
)

# Connect (auto-starts server if needed)
await client.connect(timeout=10.0)

# Check state
print(client.is_connected)   # True
print(client.state)           # ConnectionState.CONNECTED
print(client.client_id)       # "client-abc123"

# ... use client ...

# Temporary disconnect (can reconnect later)
await client.disconnect()
print(client.is_connected)    # False

# Reconnect
await client.connect()

# Permanent close (no more reconnection)
await client.close()
print(client.is_closed)       # True

Sessions

create_session

async create_session(
  name: Optional[str] = None,
  profile: Optional[Union[str, Dict]] = None,
  agent: Optional[str] = None,
  agent_params: Optional[Dict[str, str]] = None,
  cascade_driver_id: Optional[str] = None,
  timeout: float = 60.0
) -> Optional[str]

Create a new session on the server. Automatically tracks the session ID for recovery. Optionally initialize the session with a predefined agent profile that configures the model, provider, tools, system instructions, and GC settings.

name str optional
Human-readable name for the session.
profile str | Dict optional
Either a profile name (str) referencing a file under .jaato/profiles/, or an inline spec dict with keys such as model, plugins, system_instructions, etc. The two forms are mutually exclusive — pass one or the other. Passing any other type raises TypeError.
agent str optional
Agent name. The agent's rendered markdown becomes the session's system instructions. Orthogonal to profile — they compose freely.
agent_params Dict[str, str] optional
Parameter values for the agent's {{param}} placeholders. Only used when agent is specified.
cascade_driver_id str optional
Phase-2 cascade-sharing tenant ID. Sessions sharing the same opaque ID reuse a warm pool slot (warm imports + plugin state + LSP connections) across cascade stages. Pass the same UUID across every session of one cascade; subagent sessions inherit automatically via the shared runner. None (default) = standalone session, no slot reuse.
timeout float optional
Seconds to wait for the server's SessionInfoEvent confirmation. Mirrors IPCClient.create_session(timeout=…) so a plain→recovery swap is drop-in (forwarded to the inner client). Default 60.0. The recovery client's own reconnect timing is governed separately (recovery config + connect(timeout=)).

Returns

str | None — Session ID if created, None otherwise.

Raises

ReconnectingError if currently reconnecting.
ConnectionClosedError if connection is closed.
TypeError if profile is not None, str, or dict.

attach_session

async attach_session(session_id: str) -> bool

Attach to an existing session. Tracks the session ID for recovery.

session_id str required
The session to attach to.

Returns

bool — True if attach command was sent.

set_session_id

set_session_id(session_id: str) -> None

Manually set the session ID for reattachment after reconnection. Call this when the session was created or attached through other means (e.g., from a SessionInfoEvent).

session_id str required
The session ID to track for recovery.

session_id property

Returns the current session ID, or None if no session is tracked.

list_profiles

async list_profiles() -> None

Request the server to list available agent profiles. The server responds with a SessionProfilesEvent containing profile summaries (name, description, model, provider, icon). Profiles are discovered from .jaato/profiles/*.json files in the workspace.

Raises

ReconnectingError if currently reconnecting.
ConnectionClosedError if connection is closed.

Creating a session with a profile

from jaato_sdk.events import ClientType

client = IPCRecoveryClient(
    "/tmp/jaato.sock",
    client_type=ClientType.TERMINAL,
)
await client.connect()

# List available profiles first
await client.list_profiles()
async for event in client.events():
    if isinstance(event, SessionProfilesEvent):
        for p in event.profiles:
            print(f"{p['name']}: {p['description']}")
        break

# Create a session with a profile
session_id = await client.create_session(
    name="research-task",
    profile="researcher-claude",
)
client.set_session_id(session_id)

Creating and tracking sessions

client = IPCRecoveryClient(
    "/tmp/jaato.sock",
    client_type=ClientType.TERMINAL,
)
await client.connect()

# Create a new session (auto-tracked)
session_id = await client.create_session("my-session")
print(client.session_id)  # "session-abc123"

# Or attach to existing session (auto-tracked)
await client.attach_session("session-xyz789")
print(client.session_id)  # "session-xyz789"

Manual session tracking

# If you get the session ID from an event
# instead of create_session/attach_session:
from jaato_sdk.events import SessionInfoEvent

async for event in client.events():
    if isinstance(event, SessionInfoEvent):
        client.set_session_id(event.session_id)
        # Now recovery will reattach to this session

Messaging

send_message

async send_message(
  text: str,
  attachments: Optional[list] = None,
  parallel_tools: Optional[bool] = None
) -> None

Send a message to the model via the server.

text str required
The message text.
attachments list optional
File attachments to include with the message.
parallel_tools bool optional
Per-call override of parallel tool execution. None (default) keeps the environment-configured behavior (JAATO_PARALLEL_TOOLS). True forces parallel tool execution; False forces sequential — for this turn only.

Raises

ReconnectingError if currently reconnecting.
ConnectionClosedError if connection is closed.

events

async events() -> AsyncIterator[Event]

Async iterator that yields events from the server. Automatically handles reconnection — if the connection is lost and recovery is enabled, the iterator waits for reconnection and then resumes yielding events from the new connection.

Yields

Event — Events from the server (AgentOutputEvent, PermissionRequestedEvent, etc.)

stop

async stop() -> None

Stop the current operation. Only effective when connected. Does not raise during reconnection.

execute_command

async execute_command(
command: str,
args: Optional[list] = None
) -> None

Execute a server-side command.

respond_to_permission

async respond_to_permission(
  request_id: str,
  response: str,
  edited_arguments: Optional[Dict[str, Any]] = None
) -> None

Respond to a permission request from the server.

respond_to_clarification

async respond_to_clarification(
request_id: str,
response: str
) -> None

Respond to a clarification question from the model.

Sending messages and receiving events

from jaato_sdk.events import (
    AgentOutputEvent,
    PermissionRequestedEvent,
)

await client.send_message("List files in /tmp")

async for event in client.events():
    if isinstance(event, AgentOutputEvent):
        print(event.text, end="")

    elif isinstance(event, PermissionRequestedEvent):
        await client.respond_to_permission(
            event.request_id,
            "y",  # approve
        )

Handling reconnection in the event loop

# events() handles reconnection automatically.
# No special try/except needed — the iterator
# waits during RECONNECTING and resumes when
# the connection is restored.

async for event in client.events():
    handle_event(event)
    # If server restarts, events() reconnects
    # and resumes yielding from the new connection.

Queue-and-retry pattern for send

from jaato_sdk.client.recovery import (
    ReconnectingError,
    ConnectionClosedError,
)

async def send_with_retry(client, message):
    while True:
        try:
            await client.send_message(message)
            return
        except ReconnectingError:
            # Wait for reconnection, then retry
            while client.is_reconnecting:
                await asyncio.sleep(0.5)
            if client.is_closed:
                raise ConnectionClosedError()

Status

get_status

get_status() -> ConnectionStatus

Get current connection status. Returns a snapshot of the connection state, reconnection progress, and session info.

Returns

ConnectionStatus — Current state and recovery details.

Polling status

status = client.get_status()

print(f"State: {status.state.value}")
print(f"Session: {status.session_id}")
print(f"Client: {status.client_id}")

if status.state == ConnectionState.RECONNECTING:
    print(f"Attempt: {status.attempt}/{status.max_attempts}")
    if status.next_retry_in is not None:
        print(f"Next retry in: {status.next_retry_in:.1f}s")

Configuration

RecoveryConfig

@dataclass
RecoveryConfig

Controls reconnection behavior. Uses exponential backoff with jitter for retry delays.

Field	Type	Default	Description
`enabled`	`bool`	`True`	Whether automatic reconnection is enabled
`max_attempts`	`int`	`10`	Maximum reconnection attempts before giving up (min: 1)
`base_delay`	`float`	`1.0`	Initial backoff delay in seconds (min: 0.1)
`max_delay`	`float`	`60.0`	Maximum backoff delay cap in seconds (must be ≥ base_delay)
`jitter_factor`	`float`	`0.3`	Random jitter range, 0.0–1.0 (0.3 = ±30%)
`connection_timeout`	`float`	`5.0`	Timeout per connection attempt in seconds (min: 1.0)
`reattach_session`	`bool`	`True`	Automatically reattach to previous session after reconnecting

get_recovery_config

get_recovery_config(workspace_path: Optional[Path] = None) -> RecoveryConfig

Load recovery config with layered precedence: built-in defaults → user config (~/.jaato/client.json) → project config (.jaato/client.json) → environment variables (highest).

load_client_config

load_client_config(workspace_path: Optional[Path] = None) -> ClientConfig

Load the full client configuration. Returns a ClientConfig object containing the recovery section and any future configuration sections.

Environment Variables

Variable	Config Field	Default
`JAATO_IPC_AUTO_RECONNECT`	`enabled`	`true`
`JAATO_IPC_RETRY_MAX_ATTEMPTS`	`max_attempts`	`10`
`JAATO_IPC_RETRY_BASE_DELAY`	`base_delay`	`1.0`
`JAATO_IPC_RETRY_MAX_DELAY`	`max_delay`	`60.0`
`JAATO_IPC_RETRY_JITTER`	`jitter_factor`	`0.3`
`JAATO_IPC_CONNECTION_TIMEOUT`	`connection_timeout`	`5.0`
`JAATO_IPC_REATTACH_SESSION`	`reattach_session`	`true`

RecoveryConfig

from jaato_sdk.client.config import RecoveryConfig
from jaato_sdk.events import ClientType

# Custom config
config = RecoveryConfig(
    enabled=True,
    max_attempts=5,
    base_delay=2.0,
    max_delay=30.0,
    jitter_factor=0.2,
    connection_timeout=10.0,
    reattach_session=True,
)

client = IPCRecoveryClient(
    socket_path="/tmp/jaato.sock",
    client_type=ClientType.TERMINAL,   # required
    config=config,
)

Load from config files

from pathlib import Path
from jaato_sdk.client.config import (
    get_recovery_config,
    load_client_config,
)

# Load just recovery config
config = get_recovery_config(
    workspace_path=Path.cwd()
)

# Or load full client config
full_config = load_client_config(
    workspace_path=Path.cwd()
)
config = full_config.recovery

.jaato/client.json

{
  "recovery": {
    "enabled": true,
    "max_attempts": 10,
    "base_delay": 1.0,
    "max_delay": 60.0,
    "jitter_factor": 0.3,
    "connection_timeout": 5.0,
    "reattach_session": true
  }
}

Environment variable overrides

# Override via environment (highest precedence)
export JAATO_IPC_AUTO_RECONNECT=true
export JAATO_IPC_RETRY_MAX_ATTEMPTS=20
export JAATO_IPC_RETRY_BASE_DELAY=0.5
export JAATO_IPC_CONNECTION_TIMEOUT=10.0

Types

ConnectionState

class ConnectionState(Enum)

Connection state enum. String values match the state names in lowercase.

Member	Value	Description
`DISCONNECTED`	`"disconnected"`	Not connected, no reconnection in progress
`CONNECTING`	`"connecting"`	Initial connection or retry attempt in progress
`CONNECTED`	`"connected"`	Successfully connected to server
`RECONNECTING`	`"reconnecting"`	Connection lost, waiting for next retry
`DISCONNECTING`	`"disconnecting"`	Graceful disconnect initiated
`CLOSED`	`"closed"`	Terminal state, no more reconnection attempts

ConnectionStatus

@dataclass
ConnectionStatus

Snapshot of connection state for UI display. Passed to the on_status_change callback.

Field	Type	Description
`state`	`ConnectionState`	Current connection state
`attempt`	`int`	Current reconnection attempt (0 if not reconnecting)
`max_attempts`	`int`	Maximum reconnection attempts configured
`next_retry_in`	`float \| None`	Seconds until next retry (None if not waiting)
`last_error`	`str \| None`	Description of the last error encountered
`session_id`	`str \| None`	ID of the attached session
`client_id`	`str \| None`	ID assigned by server

StatusCallback

StatusCallback = Callable[[ConnectionStatus], None]

Type alias for the status change callback function.

ReconnectingError

class ReconnectingError(Exception)

Raised when an operation (e.g., send_message) is attempted while the client is reconnecting. Catch this to queue messages for retry.

ConnectionClosedError

class ConnectionClosedError(Exception)

Raised when an operation is attempted after the connection has been permanently closed via close() or after max reconnection attempts are exhausted.

IncompatibleServerError

class IncompatibleServerError(Exception)

Raised during connect() when the server's wire-protocol version is incompatible with the client's minimum requirement (IPCClient.MIN_PROTOCOL_VERSION or the min_protocol_version constructor argument). Because an incompatible server will not become compatible on retry, IPCRecoveryClient classifies this error as permanent: it transitions immediately to CLOSED and makes no further reconnection attempts.

Attributes:

server_protocol (str) — wire-protocol version the daemon reported.
min_protocol (str) — the client's required minimum wire-protocol version.
server_version (str) — daemon package version from ConnectedEvent.server_info; "unknown" when absent. Diagnostics only.
min_version (str) — alias property for min_protocol (pre-1.0 compatibility).

Imported from jaato_sdk.client.ipc and also re-exported from jaato_sdk.client.recovery for convenience.

Import paths

# Main client and state enum
from jaato_sdk.client import (
    IPCRecoveryClient,
    ConnectionState,
)

# Required for client_type argument
from jaato_sdk.events import ClientType

# Recovery-specific types
from jaato_sdk.client.recovery import (
    ConnectionStatus,
    StatusCallback,
    ReconnectingError,
    ConnectionClosedError,
    IncompatibleServerError,
)

# Configuration
from jaato_sdk.client.config import (
    RecoveryConfig,
    ClientConfig,
    get_recovery_config,
    load_client_config,
)

Exception handling

from jaato_sdk.client.recovery import (
    ReconnectingError,
    ConnectionClosedError,
    IncompatibleServerError,
)

try:
    await client.connect()
    await client.send_message("Hello!")
except IncompatibleServerError as e:
    # Server too old; no retry will help.
    # Client is now in CLOSED state.
    print(f"Server protocol {e.server_protocol} "
          f"below minimum {e.min_protocol} "
          f"(daemon {e.server_version})")
except ReconnectingError:
    # Queue the message, retry after reconnect
    message_queue.append("Hello!")
except ConnectionClosedError:
    # Connection is permanently closed
    print("Connection lost. Please restart.")

IPCRecoveryClient

Constructor

__init__

Connection

connect

disconnect

close

is_connected property

is_reconnecting property

is_closed property

state property

client_id property

server_version property

Sessions

create_session

attach_session

set_session_id

session_id property

list_profiles

Messaging

send_message

events

stop

execute_command

respond_to_permission

respond_to_clarification

Status

get_status

Configuration

RecoveryConfig

get_recovery_config

load_client_config

Environment Variables

Types

ConnectionState

ConnectionStatus

StatusCallback

ReconnectingError

ConnectionClosedError

IncompatibleServerError

init