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.

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

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

The Contract

TurnExecutorProtocol is a runtime-checkable protocol with three async methods.

TurnPlacement is a frozen dataclass — mutating any field raises dataclasses.FrozenInstanceError.

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.

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.

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

Related