Skip to main content
AgentSocketServer is the server half of the agent socket protocol — it drives every WebSocket an agent holds. Construct one per actor instance and delegate the actor’s webSocket(ws, req) to attach(). The server sends each admitted connection a state snapshot followed by hello, dispatches inbound RPC frames to the actor’s @rpc()-decorated methods, and pushes incremental updates through the broadcast* methods as the agent’s data changes.
With an authorize hook configured, every connection becomes an authorized session: a client that opens with an attach frame presents its token, authorize(token) maps it to claims, and per-stream subscriptions are negotiated; a client that never attaches is authorized as anonymous — authorize(undefined) — so admit it with limited claims, or throw to reject. RPC call frames require the "rpc" claim, which is what lets read-only watchers and fully privileged operators share one socket endpoint.

​
attach()

attach(ws, req?): Promise<void>
Attach a socket (call from the actor’s webSocket(ws, req)). If onConnect is configured it runs first as a gate: throw or ctx.close() rejects the connection (closed, never admitted as a watcher, no hello/state). On accept, sends the state snapshot then hello, and handles inbound call/ping frames. Returns a Promise so the caller may await it. If the initial snapshot can’t be read the connection is REJECTED (closed, no hello) rather than admitted half-initialized — see sendSnapshot. Parameters Returns Promise<void>

​
broadcastPatch()

broadcastPatch(patch): void
Push an incremental state patch to every socket subscribed to state. Parameters Returns void

​
broadcastSnapshot()

broadcastSnapshot(state?): Promise<void>
Push a full-state snapshot to every socket subscribed to state (defaults to getState()). Parameters Returns Promise<void>

​
broadcastMessages()

broadcastMessages(appended): void
Push newly-appended messages to every socket subscribed to messages (after this.messages.add/append). Parameters Returns void

​
broadcastEvent()

broadcastEvent(event): void
Push one persistent event to every attached session subscribed to events (call after this.events.emit(...)). Only attached sessions receive event frames — the pre-attach protocol has no event vocabulary, so plain v1 sockets are never sent one. Parameters Returns void

​
watcherCount

get watcherCount(): number
The number of currently attached (admitted) sockets. Returns number

​
close()

close(code?, reason?): void
Detach all sockets (call on actor deactivation if you want a clean close). Parameters Returns void

​
AgentSocketServerOptions

Options for AgentSocketServer — the callbacks that connect the server to one agent’s state, message log, and event log, plus the connection-authorization policy. Type Parameters Properties attachGraceMs?
optional attachGraceMs?: number
How long (ms) an attach-capable bootstrap waits for the client’s first frame before authorizing the connection as ANONYMOUS and running the v1 bootstrap — clients that never attach send nothing until hello, so without this fallback they would wait forever. Default 300. The fallback is never a permanent downgrade: an attach arriving AFTER the window elapsed (a slow link losing the race) still upgrades the session with the token’s grants. But if authorize REJECTS anonymous connections, an expired window closes the socket before a late attach can arrive — clients of such a server must attach within this window, so give slow links a generous grace, or grant anonymous a minimal (e.g. read-only) claim set instead of rejecting. Only used when authorize is configured.
authorize?
optional authorize?: (token) => readonly Claim[] | Promise<readonly Claim[]>
Resolves a credential into this connection’s claims (grants). Throw to reject: the socket gets error { code: "unauthorized" } and is closed before any snapshot. Claims are application-defined strings echoed back verbatim in attached.grants; the server itself only interprets "rpc" (required for call frames once this hook is configured). Providing this hook makes EVERY connection an authorized session:
  • A connection opening with a v2 attach is authorized with the frame’s token and negotiates grants + per-stream subscriptions.
  • Any other connection (a v1 first frame, or silence until the attachGraceMs window expires) is authorized as ANONYMOUS — this hook is called with undefined — before the v1 bootstrap (snapshot(s) + hello). Grant claims to admit anonymous (e.g. read-only) clients; throw to reject them.
  • An attach arriving later upgrades an anonymous session in place, replacing its grants and subscriptions with the token’s.
When this hook is absent, the server behaves exactly as before (no sessions, no gating). Parameters Returns readonly Claim[] | Promise<readonly Claim[]>
getEvents?
optional getEvents?: (afterSeq) => StoredEvent[] | Promise<StoredEvent[]>
Optional: returns the actor’s persistent events after an exclusive cursor (afterSeq), in seq order. Wire it to the actor’s event log (e.g. (after) => this.events.read(after)). When provided, attached sessions can subscribe to the events stream: the bootstrap replays events past the client’s cursor (or all retained events without one), and broadcastEvent pushes live events to those subscribers. Parameters Returns StoredEvent[] | Promise<StoredEvent[]>
getMessages?
optional getMessages?: () => unknown[] | Promise<unknown[]>
Optional: returns the actor’s conversation log (this.messages.all()). When provided, a messages snapshot is sent on connect and broadcastMessages pushes appends to every watcher. Returns unknown[] | Promise<unknown[]>
getState
getState: () => TState | Promise<TState>
Returns the actor’s current persistent state (snapshot on connect). Returns TState | Promise<TState>
onConnect?
optional onConnect?: (ctx) => void | Promise<void>
Called once per new connection BEFORE the socket is admitted as a watcher or sent hello/state. Receives the upgrade request (the front door’s headers — auth, identity — ride here) and a close to reject the connection. Throw, or call ctx.close(), to reject: the socket is closed and never joins the watcher set. Resolve normally to accept. Wire it from the actor’s onConnect via a closure: new AgentSocketServer(this, { getState: () => this.getState(), onConnect: (c) => this.onConnect(c) }) Parameters Returns void | Promise<void>

​
AgentServerSocket

The ws-style socket surface the server drives, modeled structurally so this module carries no ws dependency — anything with this shape works, including the socket the platform hands an actor’s webSocket(ws, req). Properties CLOSED
readonly CLOSED: number
The readyState value for a fully closed socket.
OPEN
readonly OPEN: number
The readyState value for an open, writable socket.
readyState
readonly readyState: number
The socket’s current state, compared against OPEN / CLOSED. Methods close()
close(code?, reason?): void
Close the socket with an optional close code and reason. Parameters Returns void
on() Call Signature
on(event, listener): unknown
Register a listener for inbound frames; it may return a promise a host can await. Parameters Returns unknown Call Signature
on(event, listener): unknown
Register a listener for the socket closing. Parameters Returns unknown Call Signature
on(event, listener): unknown
Register a listener for socket errors. Parameters Returns unknown
send()
send(data): void
Send one text frame. Parameters Returns void

​
AgentConnectContext

Context passed to onConnect for per-connection auth/gating. Properties req?
optional req?: Request
The upgrade request (front-door headers, e.g. x-user, ride here). Methods close()
close(code?, reason?): void
Close this connection now to reject it (it won’t receive hello/state). Parameters Returns void