> ## Documentation Index
> Fetch the complete documentation index at: https://praison.ai/docs/llms.txt
> Use this file to discover all available pages before exploring further.
# Gateway Turn Executor
> Place each session's turn on its own worker so a wedged turn is torn down scoped to that session
The turn executor decides *where* a session's agent turn runs, so a wedged, runaway, or crashed turn is contained to its own worker instead of taking down every session on the gateway.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
graph LR
S1[๐ค Session A] --> P1[๐ place]
S2[๐ค Session B] --> P2[๐ place]
P1 --> W1[๐ฅ๏ธ Worker A]
P2 --> W2[๐ฅ๏ธ Worker B
wedges ๐ฅ]
W1 --> R1[โ
result]
W2 --> WW[โ ๏ธ WorkerWedgedError]
WW --> T[๐งน teardown]
T --> P2b[๐ re-place
epoch+1]
classDef session fill:#8B0000,stroke:#7C90A0,color:#fff
classDef process fill:#189AB4,stroke:#7C90A0,color:#fff
classDef result fill:#10B981,stroke:#7C90A0,color:#fff
classDef warn fill:#F59E0B,stroke:#7C90A0,color:#fff
class S1,S2 session
class P1,P2,P2b,T process
class W1,R1 result
class W2,WW warn
```
## Quick Start
Set no executor and the gateway drives every turn through `InProcessTurnExecutor`, running on the current event loop exactly as today.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent
from praisonai_bot.gateway import WebSocketGateway
# No executor set โ gateway uses InProcessTurnExecutor by default.
# Every turn runs on the current loop, byte-for-byte compatible.
agent = Agent(name="Support", instructions="Help users")
WebSocketGateway(agent=agent).start()
```
Read the active executor off the config โ the same string `gateway doctor` reports.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import GatewayConfig
GatewayConfig().to_dict()["executor"] # โ "inprocess"
```
Pass any object satisfying `TurnExecutorProtocol` to `GatewayConfig.executor`; the gateway resolves it once at start-up.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
from praisonaiagents import Agent, GatewayConfig
from praisonaiagents.gateway import InProcessTurnExecutor
from praisonai_bot.gateway import WebSocketGateway
# Any object satisfying TurnExecutorProtocol works here.
config = GatewayConfig(executor=InProcessTurnExecutor())
gateway = WebSocketGateway(agent=Agent(name="Support", instructions="Help"),
config=config)
gateway.start()
config.to_dict()["executor"] # โ "InProcessTurnExecutor"
```
Reach for a concrete isolated executor (subprocess / container / remote) when you need blast-radius containment โ see [Common Patterns](#common-patterns).
***
## How It Works
The gateway asks the executor to `place` a session, runs the turn on that placement, and tears the placement down only when its worker wedges โ never the whole process.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant G as Gateway
participant E as TurnExecutor
participant W as Worker
G->>E: place(session_id)
E-->>G: TurnPlacement(worker_id, epoch)
G->>E: execute_turn(placement, turn, cancel_token, limits)
E->>W: run turn
alt worker makes progress
W-->>E: result
E-->>G: result
else worker wedges
W-->>E: no progress
E-->>G: WorkerWedgedError
G->>E: teardown(placement, reason)
Note over G,E: re-place session with epoch+1
end
```
| Piece | Owner | Role |
| ----------------------- | ----------- | ---------------------------------------------------------------------------- |
| `TurnExecutorProtocol` | Core (pure) | Async `place` / `execute_turn` / `teardown` contract for *where* a turn runs |
| `TurnPlacement` | Core (pure) | Frozen handle: which worker owns a session's turns, and at what `epoch` |
| `InProcessTurnExecutor` | Core (pure) | Default โ runs each turn on the current loop, byte-for-byte compatible |
| `WorkerWedgedError` | Core (pure) | Scoped fault signal โ tear down this placement, re-place the session |
`InProcessTurnExecutor` is the default and reproduces today's on-loop behaviour exactly: leaving the executor unset changes nothing and adds no dependency.
***
## How the gateway drives it
`WebSocketGateway` resolves the executor once at start-up, then routes every turn through `place()` โ `execute_turn()`.
```mermaid theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
sequenceDiagram
participant User
participant GW as WebSocketGateway
participant EX as executor
participant W as Worker
User->>GW: message
GW->>EX: place(session_id)
EX-->>GW: TurnPlacement
GW->>EX: execute_turn(placement, turn, cancel_token, limits)
EX->>W: run turn
alt worker replies
W-->>EX: result
EX-->>GW: result
GW-->>User: reply (outcome=ok)
else worker wedges
EX-->>GW: WorkerWedgedError
GW->>EX: teardown(placement, reason="wedged")
GW-->>User: worker wedged; session re-placed (outcome=error)
end
```
| Where | What it does |
| ------------------------------------- | ----------------------------------------------------------------------- |
| `GatewayConfig.executor` | Selects the executor. `None` โ `InProcessTurnExecutor`. |
| `WebSocketGateway.__init__` | Resolves `self._executor = config.executor or InProcessTurnExecutor()`. |
| `WebSocketGateway._drive_turn` | Every turn goes through `executor.place()` โ `executor.execute_turn()`. |
| `WorkerWedgedError` handler | Tears down **only that session's** worker (scoped, not process-wide). |
| Session queue outcome | Wedged turn reports `outcome_status="error"`, not the default `"ok"`. |
| `GatewayConfig.to_dict()["executor"]` | `"inprocess"` by default; the executor class name when explicit. |
Before the seam was wired, the gateway ran turns directly on the event loop and a wedged turn triggered a process-wide `os._exit`. Now `WebSocketGateway._drive_turn` routes through the configured executor, and a `WorkerWedgedError` tears down only the offending session's worker.
***
## The Contract
`TurnExecutorProtocol` is a runtime-checkable protocol with three async methods.
| Method | Signature | Does |
| -------------- | ------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------- |
| `place` | `place(session_id: str) -> TurnPlacement` | Returns the placement that owns this session's turns; may bump `epoch` when the worker is replaced |
| `execute_turn` | `execute_turn(placement, turn, *, cancel_token=None, limits=None) -> Any` | Runs `turn` on `placement` and returns its result; raises `WorkerWedgedError` if the worker can't make progress |
| `teardown` | `teardown(placement, *, reason: str) -> None` | Reclaims the placement's worker, scoped to the owning session only |
`TurnPlacement` is a frozen dataclass โ mutating any field raises `dataclasses.FrozenInstanceError`.
| Field | Type | Default | Description |
| ------------ | ----- | ------- | ------------------------------------------------------------------------------------------------------------------------------- |
| `session_id` | `str` | โ | The resolved session whose turns this placement serves |
| `worker_id` | `str` | โ | Opaque id of the worker that owns the session's turns; a constant (`"inprocess"`) for the in-process default |
| `epoch` | `int` | `0` | Monotonic generation bumped when the backing worker is replaced โ the fencing token; a turn carrying a stale epoch must not run |
`WorkerWedgedError` is a plain `Exception` subclass. It is **scoped**: the gateway tears down only the offending placement and re-places its session โ never a process-wide `os._exit`. Ordinary turn errors surface as the turn's own exception and do not condemn the worker.
An isolated executor does **not** ship the live `turn` callable (and its captured loop/agent state) across a process boundary. The worker owns the session via `place()` and rebuilds the turn from serialisable inputs on its own side; the passed `turn` is the gateway-side await point for the worker's result.
***
## Configuration Options
The seam is configured through `GatewayConfig.executor` (defaults to `None`, which resolves to `InProcessTurnExecutor` at runtime). `GatewayConfig.to_dict()["executor"]` reports the active executor for `gateway doctor` / observability.
`GatewayConfig` โ the `executor` field + full gateway config.
`TurnExecutorProtocol` โ auto-generated SDK reference.
`TurnPlacement` โ auto-generated SDK reference.
`InProcessTurnExecutor` โ auto-generated SDK reference.
`WorkerWedgedError` โ auto-generated SDK reference.
`limits` on `execute_turn` is a typed `Any` today, accepted for protocol symmetry. It is inert in-process and honoured only by isolated executors โ there is no `limits=` config class in core.
***
## Common Patterns
Three shapes cover almost every use.
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 1. Default โ no code needed. Leaving the executor unset runs turns on the
# current loop exactly as today, byte-for-byte backward compatible.
from praisonaiagents.gateway import InProcessTurnExecutor
executor = InProcessTurnExecutor()
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 2. A wrapper's isolated executor (sketch โ user code, not shipped by core).
from praisonaiagents.gateway import TurnPlacement
class SubprocessTurnExecutor: # illustrative โ your wrapper implements this
async def place(self, session_id: str) -> TurnPlacement:
worker_id = await self._spawn(session_id)
return TurnPlacement(session_id=session_id, worker_id=worker_id, epoch=0)
async def execute_turn(self, placement, turn, *, cancel_token=None, limits=None):
return await self._dispatch(placement, cancel_token, limits)
async def teardown(self, placement, *, reason: str) -> None:
await self._kill(placement.worker_id)
```
```python theme={"theme":{"light":"vitesse-light","dark":"vitesse-dark"}}
# 3. Handle a wedged worker โ tear down, then re-place with a bumped epoch.
from praisonaiagents.gateway import WorkerWedgedError, TurnPlacement
async def run_with_recovery(executor, session_id, turn):
placement = await executor.place(session_id)
try:
return await executor.execute_turn(placement, turn)
except WorkerWedgedError:
await executor.teardown(placement, reason="wedged")
fresh = await executor.place(session_id) # new worker, epoch+1
return await executor.execute_turn(fresh, turn)
```
***
## Best Practices
`epoch` is the fencing token. Bump it whenever a session's worker is replaced, and refuse any turn carrying a stale epoch โ that is what stops a reclaimed worker from executing against a session it no longer owns.
`teardown` reclaims a single placement's worker. A worker fault must fail scoped: tear down the offending placement and re-place its session, never the whole gateway. That is the whole point of the seam versus a process-wide `os._exit`.
The in-process default is zero-cost and dependency-free but offers no isolation โ a turn that wedges the loop still affects the process. Switch to an isolated executor (subprocess / container / remote) when one bad session must not degrade the rest.
An isolated worker owns the session via `place()` and rebuilds the turn from serialisable inputs on its own side. Do not try to send the live `turn` callable across a process boundary โ treat it as the gateway-side await point for the worker's result.
The default `GatewayConfig(executor=None)` resolves to `InProcessTurnExecutor` โ today's on-loop behaviour, no dependency, no cost. Only set `executor=` when you need blast-radius containment (subprocess / container / remote) or per-session limits. Verify what actually resolved with `config.to_dict()["executor"]` โ that same string is what `gateway doctor` prints.
***
## Related
Sibling primitive โ serialises turns per session across replicas.
Detects a wedged event loop.