Ops Bridge

Rust ↔ Deno JSON-RPC reference — stream_append, state_cas, dispatch_process, and conversation ops.

The ops bridge connects user TypeScript (via @yoke/sdk) to the Rust kernel. The embedded Deno worker communicates over a unix socket using newline-delimited JSON-RPC.

TypeScript → Rust (host ops)

User code calls SDK methods; the worker forwards to Rust:

Op	Purpose
`stream_append`	Publish event to global stream (Zod-validated in worker)
`instances_create`	Create actor instance with queue policy + filter
`instances_destroy`	Tear down instance (`onDestroy`, delete rows)
`sqlite_exec`	Actor SQLite query with bound params
`state_get`	Read typed state document + version
`state_set`	Overwrite typed state
`state_cas`	Compare-and-swap with expected version
`portal_call`	Invoke portal interface method
`schedule_upsert`	Create/replace keyed schedule
`schedule_cancel`	Cancel schedule by instance + key
`schedule_list`	List schedules for instance
`conversation_create_actor`	Create chat identity
`conversation_create`	Create conversation
`conversation_add_participant`	Add participant stint
`conversation_leave_participant`	End stint
`conversation_post_message`	Insert message + stream append
`conversation_create_branch`	Soft fork
`conversation_list_messages`	Transcript read
`conversation_list_participants`	Participant list
`conversation_create_blob`	Register blob metadata
`conversation_get_blob`	Fetch blob metadata

The atomic state op is state_cas, not state_update. ctx.state.update() in the SDK wraps state_cas with retry logic.

Rust → TypeScript (worker RPC)

RPC	Purpose
`eval_filter`	Function filter check on fan-out slow path
`eval_queue_select`	Custom queue selector over pending events
`dispatch_process`	Run actor `process(ctx, event)`; dequeue on success
`get_actor_specs`	Queue policy, default state, filter hints for boot
`portal_invoke`	Execute portal interface method from `portal_call`

Execution flow

Rust scheduler
  → dispatch_process(instanceId, event)
  → Deno: await definition.process(ctx, event)
  → ctx.append() → stream_append → Postgres → fan-out
  → ctx.portals.* → portal_call → portal_invoke
  → return
  → Rust: dequeue on success, requeue on throw

state_cas detail

// Worker RPC (simplified)
const updated = await callHost("state_cas", {
  storeId: sqliteStoreId,
  expectedVersion: record.version,
  state: newData,
});

The runtime implements version check in a SQLite transaction. Conflict → worker retries ctx.state.update() up to MAX_STATE_RETRIES (32).

Custom queue selection

Rust: list_pending(instance, group)
  → eval_queue_select { events, config, definitionName }
  → Deno: queueSelectFromConfig(config)(events) or inline select
  → return event_id
  → Rust: claim_next(instance, group, policy, event_id)

Validation boundary

Validated in worker (Deno)	Validated in Rust
Event payload (Zod)	Instance config schema
Append input shape	SQL params binding
Tool parameters	Tenant scoping

Rust serializes payloads; it does not interpret business logic.

System endpoints (Rust HTTP, not ops)

Endpoint	Purpose
`GET /health`	`ok`
`GET /metrics`	Prometheus text

Type flow

User TS (defineRuntime)
  → loaded by DenoHost on start
  → definitions registered in kernel
  → instances via instances_create
  → fan-out → queues
  → dispatch_process → process(ctx, event)
  → stream_append → Fanout

Deno worker

Embedding and transport.

Context API

SDK wrappers over ops.

Deno worker

Context API

On this page