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.

Quick Start

1. Default — start an agent, the gateway calls the in-process executor

   Set no executor and the gateway drives every turn through InProcessTurnExecutor, running on the current event loop exactly as today.

   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()
   
2. Confirm the active executor from to_dict()

   Read the active executor off the config — the same string gateway doctor reports.

   from praisonaiagents import GatewayConfig
   
   GatewayConfig().to_dict()["executor"]   # → "inprocess"
   
3. Plug in an isolated executor via GatewayConfig

   Pass any object satisfying TurnExecutorProtocol to GatewayConfig.executor; the gateway resolves it once at start-up.

   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.

---

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.

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().

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.

# 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()

# 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)

# 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

Fence with epoch, always

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.

Keep teardown scoped to one session

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.

Reach for isolation only when you need blast-radius containment

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.

Never cross a boundary with a live callable

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.

Leave executor=None unless you need isolation

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

Gateway Turn Lock

Sibling primitive — serialises turns per session across replicas.

Gateway Loop Watchdog

Detects a wedged event loop.