# Yoke Documentation (/docs)
Yoke is an **always-on runtime you embed in your own product**. Coding harnesses, personal assistants, research pipelines — those are products you build. Yoke is the event-driven kernel that runs **inside your server**. It is **not** an LLM framework and **not** the thing your users log into. Agents are ordinary actors that react to events, call models when needed, and persist state in SQLite.
## What makes Yoke different [#what-makes-yoke-different]
| Principle | What it means |
| ---------------------------------------- | ------------------------------------------------------------------------------ |
| **Stream is the spine** | All communication goes through one global event log. No direct actor channels. |
| **No targets** | Publishers emit typed events. Consumers self-select via filters. |
| **Definitions fixed, instances dynamic** | Actor types are registered at build time. Instances are created at runtime. |
| **Agents are actors** | LLM loops are event-driven actor instances, not a special runtime mode. |
| **Forum + monologue** | Public chat and private agent reasoning share one conversation store. |
## Stack at a glance [#stack-at-a-glance]
```
Your product (CLI / web / API)
↓
Your server
embeds yoke-runtime (Rust)
→ Postgres (stream, queues, instances, conversations)
→ SQLite per actor (state, indexes, custom tables)
→ Deno worker
→ your package (yoke.ts, @yoke/sdk)
```
Portal HTTP and the control-plane port are **private** to your process. Public ingress is whatever your server exposes.
## Who this documentation is for [#who-this-documentation-is-for]
* **Product builders** embedding Yoke in their own server (coding harnesses, assistants, ops systems)
* **Package authors** writing the TypeScript actors your server loads
* **Platform engineers** operating single-tenant or multi-tenant deployments
## Where to start [#where-to-start]
Read [Embed Yoke in your product](/docs/getting-started/embed) — how your server hosts the runtime and loads your package.
Complete [Installation](/docs/getting-started/installation) — Postgres, Rust toolchain, Deno 2, and environment variables.
Follow [Quick start](/docs/getting-started/quick-start) to run a minimal package locally and watch stream fan-out.
Read [Your first package](/docs/getting-started/first-package) to structure `yoke.ts`, actors, and portals.
# Agent bootstrap (/docs/agents/bootstrap)
Agent bootstrap follows a strict convention: **forums are created separately**, agents are **actor + egress**, and each forum join creates a **monologue stint** with ingestion and processor instances.
## 1. Create a chat room (forum) [#1-create-a-chat-room-forum]
Agents do **not** create forums. Use the Chat SDK or HTTP client:
```typescript
import { createChatClient } from "@yoke/sdk";
const chat = createChatClient(ctx.conversations);
const room = await chat.createRoom({ title: "support" });
const human = await chat.createActor({ name: "alice" });
const humanP = await chat.addMember(room.id, human.id);
await chat.send(room.id, "Hello!", { authorParticipantId: humanP.id });
```
### Headless HTTP [#headless-http]
For apps without a running actor context, use HTTP clients against the control plane:
```typescript
import { createChatHttpClient, createAgentBootstrapHttpClient } from "@yoke/sdk";
const chat = createChatHttpClient({ baseUrl: controlPlaneUrl, tenantId });
const agents = createAgentBootstrapHttpClient({
baseUrl: `${controlPlaneUrl}/v1/tenants/${tenantId}/ingress`,
});
const room = await chat.createRoom({ title: "support" });
const agent = await agents.createAgent({ name: "support-bot" });
await agents.joinConversation(agent, room.id, { model: "gpt-4o" });
```
## 2. Create an agent [#2-create-an-agent]
```typescript
import { createAgent } from "@yoke/sdk";
const agent = await createAgent(ctx, { name: "support-bot" });
// → { actorId, egressInstanceId }
```
This creates:
1. One `actors` row with `role: "agent"` (the persona)
2. One **egress** actor instance with config `{ agentActorId }` — routes replies from any monologue stint back to forums
`createAgent` does **not** create a monologue. Monologues are created per forum join.
## 3. Join a chat room [#3-join-a-chat-room]
```typescript
import { joinConversation } from "@yoke/sdk";
const join = await joinConversation(ctx, agent, room.id, {
model: "gpt-4o", // optional, default "fixture-echo"
});
```
Returns a `JoinHandle`:
```typescript
type JoinHandle = {
forumId: string;
forumParticipantId: string;
monologueId: string;
monologueParticipantId: string;
systemActorId: string;
systemParticipantId: string;
ingestionInstanceId: string;
processorInstanceId?: string;
followUpInstanceId?: string;
guardianSubscriptionInstanceId?: string;
};
```
### What joinConversation does [#what-joinconversation-does]
```mermaid
sequenceDiagram
participant App
participant Store as Conversation store
participant RT as Runtime
App->>Store: createMonologueForStint (system + agent participants)
App->>Store: addParticipant(forum, agent, metadata.monologueId)
App->>RT: create ingestion instance
App->>RT: create processor instance
App->>RT: create followUp instance (optional)
App->>RT: create guardianSubscription instance (optional)
App->>RT: append agent.forum.joined
Note over RT: Egress updates ctx.state.joins
```
1. Creates a **monologue** for this stint (`system` + agent participants)
2. Adds the agent as a forum participant with `metadata.monologueId`
3. Spins up **ingestion** and **processor** instances for this monologue
4. Optionally creates **follow-up** and **guardian subscription** instances
5. Emits `agent.forum.joined` so egress learns `forumId → { monologueId, forumParticipantId }`
Ingested forum messages are copied into the monologue by **external shadow actors** — one per forum author. No mirror relations. Ingested copies carry `body.meta.sourceConversationId` for egress routing.
## Join options [#join-options]
```typescript
await joinConversation(ctx, agent, forumId, {
metadata: { team: "support" }, // extra forum participant metadata
model: "claude-sonnet-4", // processor model id
createProcessor: true, // default true
followUp: false, // disable follow-up timer
followUp: { delayMs: 60_000, text: "Still need help?" },
guardianAngel: false, // disable guardian subscription
guardianAngel: { model: "guardian-steer" },
});
```
## 4. Register actor definitions [#4-register-actor-definitions]
Register SDK-provided actors in your `yoke.ts` package:
```typescript
import {
defineAgentStack,
defineEgressActor,
defineFollowUpActor,
defineGuardianAngel,
defineIngestionActor,
defineProcessorActor,
agentEvents,
conversationEvents,
defineEvents,
defineFor,
defineRuntime,
egressConfigSchema,
egressStateSchema,
followUpConfigSchema,
followUpStateSchema,
ingestionConfigSchema,
ingestionStateSchema,
processorConfigSchema,
processorStateSchema,
} from "@yoke/sdk";
const events = defineEvents({ ...conversationEvents, ...agentEvents });
const agentStack = defineAgentStack({
compaction: async (messages) => {
if (messages.length <= 20) return messages;
return messages.slice(-10);
},
});
const guardianAngel = defineGuardianAngel({
filter: agentSaidV1Filter,
model: "guardian-steer",
});
export default defineRuntime()
.events(events)
.actors(({ events }) => {
const { actor } = defineFor(events);
return {
ingestion: actor(defineIngestionActor({
configSchema: ingestionConfigSchema,
stateSchema: ingestionStateSchema,
})),
egress: actor(defineEgressActor({
configSchema: egressConfigSchema,
stateSchema: egressStateSchema,
})),
processor: actor(defineProcessorActor({
configSchema: processorConfigSchema,
stateSchema: processorStateSchema,
compaction: agentStack.compaction,
})),
followUp: actor(defineFollowUpActor({
configSchema: followUpConfigSchema,
stateSchema: followUpStateSchema,
})),
guardianSubscription: actor(guardianAngel.subscription),
guardianAngel: actor(guardianAngel.runner),
};
});
```
## Optional coordinator registry [#optional-coordinator-registry]
Store join handles in a coordinator actor's durable SQLite state:
```typescript
type AgentRegistryState = {
agents: Record }>;
};
```
Postgres `participants.metadata.monologueId` is **canonical** — the registry is optional convenience for portal handlers.
## Checklist [#checklist]
* [ ] Chat room created separately (`createChatClient` / HTTP)
* [ ] Agent created with `createAgent` (actor + egress only)
* [ ] Each forum joined via `joinConversation` (monologue + ingestion + processor per stint)
* [ ] `monologueId` on forum participant metadata (automatic)
* [ ] No mirror relations anywhere
* [ ] Processor posts `type: "reply"` when ready for forum (egress copies it out)
Expose agent bootstrap via portal routes (`POST /agents`, `POST /agents/:agentId/join`) — see [Portals](/docs/concepts/portals).
# Compaction (/docs/agents/compaction)
**Compaction** reduces monologue transcript size before LLM inference. When compaction changes the message list, the processor swaps to a fresh monologue conversation and repoints the forum participant stint.
## Interface [#interface]
```typescript
import type { InferenceMessage, MessageCompactor } from "@yoke/sdk";
const compaction: MessageCompactor = async (messages: InferenceMessage[]) => {
if (messages.length <= 20) return messages;
return messages.slice(-10);
};
const agentStack = defineAgentStack({ compaction });
defineProcessorActor({
configSchema: processorConfigSchema,
stateSchema: processorStateSchema,
compaction: agentStack.compaction,
});
```
`MessageCompactor` receives and returns `InferenceMessage[]` — the same shape as `ctx.llm.infer` input.
Preserve `body.meta` (especially `sourceActorId`) on user messages so hydration can reuse external shadow actors ingestion created.
## Flow [#flow]
```
monologue message posted
│
▼
processor reads transcript → InferenceMessage[]
│
▼
compaction(messages) [optional]
│
├─ unchanged ──► ctx.llm.infer(messages)
│
└─ changed ──► create new monologue
hydrate with compacted messages
update forum participant.metadata.monologueId
emit agent.monologue.compacted (egress updates routing)
ctx.llm.infer(compacted)
│
▼
post type: "reply" to active monologue
```
## Storage updates on swap [#storage-updates-on-swap]
| Location | What changes |
| ----------------------------------- | ------------------------------------------------------------------------- |
| `participants.metadata.monologueId` | Forum stint repointed to new monologue |
| Processor `ctx.state` | `monologueId`, `agentParticipantId` |
| Ingestion | Resolves monologue from participant metadata; resets external-actor cache |
| Egress `ctx.state.joins` | Updated via `agent.monologue.compacted` event |
Canonical monologue id lives in Postgres `participants.metadata.monologueId` — one monologue per forum stint.
## Helpers [#helpers]
| Export | Purpose |
| ------------------------------ | ----------------------------------------------------- |
| `defineAgentStack` | Typed holder for compaction and tools |
| `inferenceMessagesEqual` | Compare before/after compaction |
| `swapMonologueAfterCompaction` | Create monologue, hydrate, update participant |
| `hydrateMonologueWithMessages` | Post compacted inference messages |
| `collectExternalActors` | Gather shadow actors before swap |
| `resolveMonologueStint` | Read active monologue from forum participant metadata |
## Example compactor [#example-compactor]
Example compactor in your package:
```typescript
const agentStack = defineAgentStack({
compaction: async (messages) => {
if (messages.length <= 15) return messages;
// Keep system context + recent tail
return [messages[0], ...messages.slice(-8)];
},
});
```
## When to compact [#when-to-compact]
| Scenario | Strategy |
| ----------------------- | ------------------------------------------ |
| Context window pressure | Truncate/summarize old turns |
| Long support threads | Keep last N messages + summary line |
| Tool-heavy sessions | Drop intermediate tool rounds, keep finals |
Compaction runs **before** each infer call when configured. If the compactor returns an identical list (`inferenceMessagesEqual`), no swap occurs.
## Events [#events]
| Event | When |
| --------------------------- | ----------------------------------------------------------- |
| `agent.monologue.compacted` | After swap — egress updates `forumId → monologueId` routing |
## Related [#related]
# Egress actor (/docs/agents/egress)
The **egress** actor is the only place that decides what the user sees on the forum. The processor and inference loop do not think about visibility — they persist monologue messages and emit `conversation.message.posted` events. Egress reads those events and forwards the ones that should reach the user.
## Rule [#rule]
**If the LLM output includes text, send that text to the user. Everything else stays internal.**
| Monologue message | Forward to forum? |
| ---------------------------------------- | --------------------- |
| Agent-authored, has `text` parts | Yes — text parts only |
| Agent-authored, `tool_call` only | No |
| System-authored `tool_result` | No |
| Ingested user copy | No (already on forum) |
| Agent text + `tool_call` in same message | Yes — text parts only |
This is **not** tied to `type: reply`. Monologue traffic from the processor is always `type: message`. Egress does not filter on message type — it filters on **author** (agent monologue participant) and **content** (non-empty text parts).
`type: reply` remains available for other conversation semantics (threading, explicit reply relations). It is not the signal for user-visible agent output.
## Flow [#flow]
```mermaid
flowchart LR
P[Processor / inference loop] -->|post type: message| M[Monologue]
M -->|conversation.message.posted| E[Egress]
E -->|agent text only| F[Forum]
B[Chat platform bridge] -->|agent text on forum| U[User]
F --> B
```
The processor posts tool calls, tool results, and agent text to the monologue without deciding what is user-facing. Egress copies agent text to the forum. A chat platform bridge (if configured) watches forum traffic from the agent participant and delivers it to Slack, North chat, etc.
## Filter [#filter]
Egress subscribes to:
* `agent.forum.joined` — record routing for a new stint
* `agent.monologue.compacted` — repoint routing after compaction swap
* `conversation.message.posted` — **all** message posts (not just one message type)
On each `conversation.message.posted`:
1. Is `conversationId` a tracked monologue for this agent?
2. Is `authorParticipantId` the agent's monologue participant?
3. Does the message body contain non-empty `text` parts?
If all yes → post text parts to the forum. Routing uses join state keyed by monologue id, not `body.meta.sourceConversationId`.
## Join state [#join-state]
```typescript
joins[forumId] = {
forumParticipantId, // agent on the forum
monologueId, // active internal transcript
monologueParticipantId, // agent on the monologue — author filter
};
```
Updated on `agent.forum.joined` and `agent.monologue.compacted`.
## Contrast with processor [#contrast-with-processor]
| | Processor | Egress |
| --------------- | -------------------------------------------------- | --------------------------- |
| **Job** | LLM inference + tool execution | User-visible copy-out |
| **Visibility** | Does not decide | Decides from text content |
| **Output type** | Always `type: message` on monologue | `type: message` on forum |
| **Triggers on** | Monologue events (excludes forum + legacy `reply`) | All monologue message posts |
## Related [#related]
* [Processor](/docs/agents/processor) — event-driven inference; does not handle visibility
* [Mesh pattern](/docs/agents/mesh-pattern) — ingestion → processor → egress pipeline
* [Message types](/docs/conversations/message-types) — part types and body schema
# Follow-up (/docs/agents/follow-up)
The **follow-up** actor schedules a delayed forum message when the agent speaks last, and cancels when a human replies. It prevents conversations from going stale without blocking the processor.
## Behavior [#behavior]
```
Agent posts to forum
│
▼
followUp actor: schedule assistant.follow_up_due (delayMs)
│
├─ Human posts before fire → cancel schedule
│
└─ Timer fires → post follow-up message (meta.followUp: true)
```
Follow-up messages carry `meta.followUp: true` so they do not re-arm another timer.
## defineFollowUpActor [#definefollowupactor]
```typescript
import {
defineFollowUpActor,
followUpConfigSchema,
followUpStateSchema,
} from "@yoke/sdk";
followUp: actor(
defineFollowUpActor({
configSchema: followUpConfigSchema,
stateSchema: followUpStateSchema,
}),
),
```
### Config schema [#config-schema]
```typescript
const followUpConfigSchema = z.object({
sourceConversationId: z.string(), // forum id
forumParticipantId: z.string(),
agentActorId: z.string(),
delayMs: z.number().int().positive().default(30_000),
followUpText: z.string().default("Following up — still there?"),
});
```
### State schema [#state-schema]
```typescript
const followUpStateSchema = z.object({
lastScheduledAt: z.number().optional(),
followUpsSent: z.number().int().default(0),
});
```
## Filter logic [#filter-logic]
The follow-up actor subscribes to:
1. `conversation.message.posted` where `payload.conversationId === sourceConversationId` (forum)
2. `assistant.follow_up_due` where `payload.forumId === sourceConversationId`
On forum message:
* If author is the agent (and not `meta.followUp`) → schedule timer
* If author is human → cancel timer
On `assistant.follow_up_due`:
* Post `followUpText` as forum message from `forumParticipantId`
## Schedule key [#schedule-key]
```typescript
function scheduleKey(forumId: string) {
return `follow-up:${forumId}`;
}
```
Uses `ctx.schedules.schedule` / `cancel` — see [Schedules](/docs/sdk/schedules).
## Join with follow-up [#join-with-follow-up]
```typescript
await joinConversation(ctx, agent, forumId, {
model: "gpt-4o",
followUp: true,
// or:
followUp: { delayMs: 60_000, text: "Still need help?" },
});
```
`joinConversation` creates a follow-up instance alongside ingestion and processor.
## Queue policy [#queue-policy]
```typescript
queue: { mode: "fifo", concurrency: 1 },
```
FIFO preserves forum message ordering for schedule arm/cancel decisions.
## E2E verification [#e2e-verification]
Test the full timer flow via portal `/agents` routes and a follow-up schedule.
## Related [#related]
# Guardian Angel (/docs/agents/guardian)
The **guardian angel** pattern adds oversight to agent monologues without modifying the primary processor. A subscription actor watches monologue traffic; when a filter matches, it spawns an ephemeral runner that can inject steering messages.
The primary processor has **no knowledge** of guardians.
## Architecture [#architecture]
```
Monologue message posted
│
▼
guardianSubscription (filter matches)
│
▼
spawn guardianAngel runner instance
│
▼
runner: ctx.llm.infer with inject_message tool
│
▼
post steering message to monologue (meta.guardianInject)
│
▼
processor handles it like any other message
```
## defineGuardianAngel [#defineguardianangel]
```typescript
import {
defineGuardianAngel,
agentSaidV1Filter,
} from "@yoke/sdk";
const guardianAngel = defineGuardianAngel({
filter: agentSaidV1Filter,
model: "guardian-steer",
});
// In yoke.ts actors:
guardianSubscription: actor(guardianAngel.subscription),
guardianAngel: actor(guardianAngel.runner),
```
### Subscription actor [#subscription-actor]
* Filters `conversation.message.posted` on the monologue
* Runs user-defined filter (e.g. `agentSaidV1Filter` — agent spoke with `says` v1 part)
* Tracks processed message ids in `ctx.state`
* Spawns ephemeral `guardianAngel` runner via `ctx.instances.create`
* Emits `guardian.run` event to trigger runner
### Runner actor [#runner-actor]
* One-shot instance per guardian pass
* Filters on `guardian.run`
* Calls `ctx.llm.infer` with `inject_message` tool
* Posts steering text to monologue with `GUARDIAN_INJECT_META_KEY` in body meta
* Self-destructs after completion
## inject\_message tool [#inject_message-tool]
Built into the runner definition:
```typescript
defineTool({
name: "inject_message",
description: "Inject a steering message into the agent monologue",
parameters: z.object({ text: z.string() }),
execute: async (input, toolCtx) => {
// Posts to monologue with meta.guardianInject
},
});
```
## Join with guardian [#join-with-guardian]
```typescript
await joinConversation(ctx, agent, forumId, {
model: "gpt-4o",
guardianAngel: true,
// or:
guardianAngel: { model: "claude-sonnet-4", systemPrompt: "..." },
});
```
Creates `guardianSubscription` instance scoped to the monologue.
## Default system prompt [#default-system-prompt]
```typescript
export const DEFAULT_GUARDIAN_SYSTEM_PROMPT = `You are a guardian angel overseeing a primary agent's monologue.
Use inject_message only when the agent needs correction or guidance.`;
```
Override via `defineGuardianAngel({ systemPrompt: "..." })`.
## Test provider [#test-provider]
`createGuardianSteerTestProvider` + model `guardian-steer` for tests without API keys.
## Design rationale [#design-rationale]
| Benefit | How |
| ---------------------- | --------------------------------------------------- |
| Separation of concerns | Processor stays simple; guardian is opt-in per join |
| Ephemeral runners | No long-lived guardian state polluting processor |
| Filter flexibility | Any monologue event pattern can trigger oversight |
| No mirror relations | Steering uses normal monologue messages |
## Related [#related]
# Agents (/docs/agents)
Yoke agents are **not** a special runtime mode. An agent is a conversation-store **actor** plus a mesh of **actor instances** (ingestion, processor, egress) that react to `conversation.message.posted` events.
## Quick start [#quick-start]
```typescript
import { createAgent, joinConversation } from "@yoke/sdk";
// 1. Create agent persona + egress (once per agent)
const agent = await createAgent(ctx, { name: "support-bot" });
// 2. Join a forum — creates monologue + ingestion + processor per stint
const join = await joinConversation(ctx, agent, room.id, {
model: "gpt-4o",
});
```
Register SDK actor definitions in your internal `yoke.ts` package (`defineIngestionActor`, `defineProcessorActor`, `defineEgressActor`, …). Your product server embeds the kernel and loads that package — [embed](/docs/getting-started/embed). See [Bootstrap](/docs/agents/bootstrap) for `createAgent` and `joinConversation`.
## Related [#related]
* [Conversations](/docs/conversations) — data model, message types, delivery, HTTP API
* [Mesh pattern](/docs/agents/mesh-pattern) — ingestion → processor → egress
# Inference (/docs/agents/inference)
Yoke's **inference layer** provides a canonical LLM interface for processor actors. All providers accept and return `InferenceMessage[]` using the same `body.parts` schema as the conversation store.
## InferenceMessage [#inferencemessage]
```typescript
type InferenceRole = "user" | "agent";
type InferenceMessage = {
type: string;
role: InferenceRole;
body: MessageBody; // { parts[], meta? }
};
```
Roles are **derived** at inference time (not stored on messages):
| Source | Role |
| ----------------------------------------------- | ------------------------------------------- |
| External shadow actors (ingested forum authors) | `user` |
| `tool_result` parts | `user` |
| `tool_call` / `reasoning` / `refusal` | `agent` |
| Otherwise | Compare author participant → agent actor id |
## defineInferenceStack [#defineinferencestack]
Register models, providers, and routing on the runtime builder:
```typescript
import {
defineInferenceStack,
createOpenAiProvider,
createAnthropicProvider,
createFixtureProvider,
defineRuntime,
} from "@yoke/sdk";
const inferenceStack = defineInferenceStack({
models: {
"gpt-4o": { id: "gpt-4o", contextWindow: 128_000 },
"claude-sonnet-4": { id: "claude-sonnet-4", contextWindow: 200_000 },
"fixture-echo": { id: "fixture-echo" },
},
providers: {
fixture: createFixtureProvider(),
openai: createOpenAiProvider({
apiKey: Deno.env.get("OPENAI_API_KEY") ?? "",
}),
anthropic: createAnthropicProvider({
apiKey: Deno.env.get("ANTHROPIC_API_KEY") ?? "",
}),
},
routing: {
"gpt-4o": ["openai", "fixture"],
"claude-sonnet-4": ["anthropic"],
"fixture-echo": ["fixture"],
},
});
export default defineRuntime()
.inferenceStack(inferenceStack)
.actors(/* ... */);
```
Without `.inferenceStack()`, `ctx.llm.infer()` throws at runtime.
## ctx.llm.infer [#ctxllminfer]
Inside processor `process()`:
```typescript
const outputs = await ctx.llm.infer(inputMessages, {
model: ctx.config.model ?? "fixture-echo",
temperature: 0.7,
maxTokens: 1024,
tools: resolvedTools,
});
```
| Option | Purpose |
| ------------- | ---------------------------------------- |
| `model` | Logical model name from stack |
| `temperature` | Sampling temperature |
| `maxTokens` | Output token limit |
| `tools` | Resolved tools/toolchains for this round |
Returns `InferenceMessage[]` to post into the monologue.
## Router behavior [#router-behavior]
The inference router handles:
* **Provider selection** — first available in routing array for model
* **Round-robin** — load balance across provider instances
* **Retries** — transient failures
* **Failover** — next provider in routing list (e.g. `["openai", "fixture"]`)
## Providers [#providers]
| Factory | Env | Use |
| --------------------------------- | ------------------- | -------------------------- |
| `createOpenAiProvider` | `OPENAI_API_KEY` | GPT models |
| `createAnthropicProvider` | `ANTHROPIC_API_KEY` | Claude models |
| `createFixtureProvider` | none | Tests, CI |
| `createLocalToolLoopProvider` | none | Local tool demo loop |
| `createSaysV1TestProvider` | none | Deterministic test replies |
| `createGuardianSteerTestProvider` | none | Guardian angel E2E |
## Routing example [#routing-example]
```typescript
routing: {
"local-tool-loop": ["local"],
"fixture-echo": ["fixture"],
"gpt-4o": ["openai", "fixture"],
"claude-sonnet-4": ["anthropic"],
"guardian-steer": ["guardian-steer"],
},
```
Join conversation with `model: "local-tool-loop"` for offline demos.
## Processor integration [#processor-integration]
The processor reads monologue transcript → converts to `InferenceMessage[]` → optionally compacts → calls `ctx.llm.infer` → posts output parts to monologue.
One infer step per `conversation.message.posted` event — not a blocking while-loop. See [Processor](/docs/agents/processor).
## Related [#related]
# Agent mesh pattern (/docs/agents/mesh-pattern)
The **agent mesh** is the standard composition for LLM agents on Yoke: three core actor types (ingestion, processor, egress) plus optional follow-up and guardian actors. Each reacts to `conversation.message.posted` on the global stream.
## End-to-end flow [#end-to-end-flow]
```mermaid
flowchart LR
subgraph Forum["Forum F"]
H[Human message]
A[Agent text reply]
end
subgraph Ingestion["Ingestion (per join)"]
I[Copy to monologue]
end
subgraph Monologue["Monologue M"]
U[User lines via external actors]
T[Tool calls / results]
AT[Agent text]
end
subgraph Processor["Processor (per join)"]
P[LLM + tools]
end
subgraph Egress["Egress (per agent)"]
E[Forward agent text to forum]
end
H -->|posted| I
I --> U
U -->|posted| P
P --> T
P --> AT
AT -->|posted| E
E --> A
```
## Ingestion [#ingestion]
**One instance per forum join.** Silently copies forum messages into the agent's active monologue.
| Property | Value |
| ---------- | ----------------------------------------------------------------------------- |
| **Filter** | `conversation.message.posted` where `conversationId === sourceConversationId` |
| **Skips** | `messageType === "reply"`, system messages (no author) |
| **Queue** | FIFO, concurrency 1 |
On each forum message:
1. Resolves the active monologue from forum participant `metadata.monologueId` (supports compaction swaps)
2. Finds or creates an **external shadow actor** for the forum author
3. Posts a copy to the monologue with routing metadata:
```json
{
"parts": [{ "type": "text", "text": "hello", "format": "plain" }],
"meta": {
"sourceConversationId": "forum-id",
"sourceMessageId": "original-message-id",
"sourceActorId": "forum-author-actor-id"
}
}
```
External actors let the processor treat ingested forum lines as **user** turns at inference time without the agent actor authoring them.
## Processor [#processor]
**One instance per forum join.** Runs LLM inference and tool execution on monologue traffic. See [Processor](/docs/agents/processor) for the event-driven loop.
| Property | Value |
| ---------- | --------------------------------------------------------------------- |
| **Filter** | `conversation.message.posted` on monologue, excluding forum traffic |
| **Queue** | FIFO, concurrency 1 |
| **Output** | Posts `type: "message"` to monologue (text, tool\_call, tool\_result) |
The processor does **not** decide user visibility. It persists whatever the LLM and tools produce.
## Egress [#egress]
**One instance per agent** (not per join). Reads monologue message events and forwards agent text to the forum. See [Egress](/docs/agents/egress) for the visibility rules.
| Property | Value |
| ------------ | ---------------------------------------------------------------------------------------- |
| **Filter** | `agent.forum.joined`, `agent.monologue.compacted`, and all `conversation.message.posted` |
| **Forwards** | Agent-authored monologue messages with non-empty text parts |
| **Skips** | tool\_call-only, tool\_result, ingested user copies |
| **Routing** | `ctx.state.joins[forumId] → { monologueId, monologueParticipantId, forumParticipantId }` |
| **Queue** | FIFO, concurrency 1 |
On `agent.forum.joined`:
```typescript
joins[forumId] = { forumParticipantId, monologueId, monologueParticipantId };
```
When an agent posts text to the monologue, egress reads `body.meta.sourceConversationId`, looks up the join, and posts **text parts only** to the forum as the agent's forum participant.
## Optional actors [#optional-actors]
### Follow-up [#follow-up]
Schedules a follow-up message when the agent was last to speak on the forum. Cancels when a human speaks. See [Follow-up](/docs/agents/follow-up).
### Guardian angel [#guardian-angel]
Subscription watches monologue activity; spawns ephemeral runners that can inject steering messages. The primary processor has no knowledge of guardians. See [Guardian angel](/docs/agents/guardian).
## Comparison with legacy three-actor docs [#comparison-with-legacy-three-actor-docs]
Older documentation described **mirror / processor / publisher** actors with `message_relations.mirror`. The current SDK uses:
| Legacy | Current SDK |
| --------------- | ---------------------------------------------------------------------- |
| Mirror actor | **Ingestion** — copies via external shadow actors, no mirror relations |
| Processor actor | **Processor** — same role, event-driven single steps |
| Publisher actor | **Egress** — one per agent, multi-forum routing via joins map |
## Metadata pairing [#metadata-pairing]
Forum stint ↔ monologue pairing is stored in participant metadata, not a separate table:
```
participants.metadata.monologueId → monologue conversation id
```
After [compaction](/docs/agents/compaction), this key is updated and `agent.monologue.compacted` notifies egress.
## Building your mesh [#building-your-mesh]
Register `defineIngestionActor`, `defineProcessorActor`, and `defineEgressActor` in your `yoke.ts`, then use `createAgent` and `joinConversation` from [Bootstrap](/docs/agents/bootstrap). Add [Compaction](/docs/agents/compaction) and [Guardian](/docs/agents/guardian) as needed per join.
# Agent overview (/docs/agents/overview)
An **agent** in Yoke is not a built-in runtime loop. It is a **composition** of ordinary actors that subscribe to stream events and call LLMs when needed.
## Core concepts [#core-concepts]
| Concept | What it is |
| --------------------- | --------------------------------------------------------------------------------------------------- |
| **Chat room (forum)** | A `conversation` you create via the Chat SDK — human-to-human or mixed |
| **Agent** | An `actor` row (persona) + one **egress** instance (routes replies to forums) |
| **Join** | Agent joins a forum → per-stint **monologue** + forum `participant` + **ingestion** + **processor** |
| **Monologue** | Private conversation for one forum stint — `system` + agent + external shadow actors |
| **Registry** | Optional coordinator `ctx.state` (SQLite) storing `JoinHandle` per forum id |
## Forum vs monologue [#forum-vs-monologue]
Public chat and private agent reasoning share the **same conversation store** — different participant sets, not different tables.
```
FORUM F (public) MONOLOGUE M (agent stint)
human system
agent_alice (stint) agent_alice (same actor)
agent_bob
visible to all participants visible only to monologue participants
```
When an agent joins a forum, bootstrap creates a fresh monologue for that stint and stores `metadata.monologueId` on the forum participant row. See [Forum and monologue](/docs/conversations/forum-monologue).
## Where monologue IDs live [#where-monologue-ids-live]
| Layer | Storage | Scope |
| --------------------- | ----------------------------------------------------- | ----------------------------------------------- |
| **Canonical** | Postgres `participants.metadata.monologueId` | One monologue per forum stint |
| **Optional registry** | Coordinator actor `ctx.state` (SQLite `__yoke_state`) | `JoinHandle` per forum id |
| **Runtime routing** | Egress actor `ctx.state.joins` | `forumId → { monologueId, forumParticipantId }` |
`createAgent` does **not** create a monologue. Each `joinConversation` creates a fresh monologue and writes `monologueId` onto that forum participant stint.
## Agent = multiple actor instances [#agent--multiple-actor-instances]
There is no monolithic agent process. A running agent is several **actor instances** sharing one **agent actor** identity:
```mermaid
flowchart TB
subgraph Forum["Forum F"]
H[Human posts message]
end
subgraph Mesh["Agent mesh (per join)"]
I[Ingestion]
P[Processor]
end
subgraph Monologue["Monologue M"]
M1[Mirrored forum lines]
M2[Tool calls / results]
M3["type: reply"]
end
E[Egress — one per agent]
H -->|conversation.message.posted| I
I -->|copy with meta| M1
M1 -->|conversation.message.posted| P
P -->|infer + tools| M2
P -->|final output| M3
M3 -->|conversation.message.posted| E
E -->|type: reply| Forum
```
| Instance | Scope | Job |
| ------------------------- | ----------------------------- | ----------------------------------------------------------------- |
| **Egress** | One per agent | Copy `type: "reply"` monologue messages back to the correct forum |
| **Ingestion** | One per forum join | Copy forum messages into the active monologue |
| **Processor** | One per forum join | Run LLM inference and tool execution on monologue traffic |
| **Follow-up** | One per forum join (optional) | Schedule a nudge when the agent was last to speak |
| **Guardian subscription** | One per forum join (optional) | Watch monologue, spawn ephemeral guardian runners |
## Event map [#event-map]
Merge `conversationEvents` and `agentEvents` from `@yoke/sdk`:
| Event | When |
| ----------------------------- | ------------------------------------------------------- |
| `conversation.message.posted` | Any message written to a conversation |
| `agent.forum.joined` | After `joinConversation` — egress updates its joins map |
| `agent.monologue.compacted` | After compaction swap — egress repoints routing |
| `assistant.follow_up_due` | Follow-up timer fires |
| `guardian.run` | Ephemeral guardian runner starts |
## Design principles [#design-principles]
* **No mirror relations** — ingested copies carry `body.meta.sourceConversationId` for egress routing instead of `message_relations.mirror`.
* **No agents table** — identity lives in `actors`; routing lives in participant metadata and egress state.
* **Agents are actors** — LLM loops are event-driven actor instances, not a special runtime mode.
* **Opaque body** — the runtime does not interpret message content; agents use `body.parts` conventions.
## Next steps [#next-steps]
* [Bootstrap](/docs/agents/bootstrap) — `createAgent` and `joinConversation`
* [Mesh pattern](/docs/agents/mesh-pattern) — detailed ingestion/processor/egress flow
* [Processor](/docs/agents/processor) — the event-driven inference loop
# Processor actor (/docs/agents/processor)
The **processor** is the LLM brain of the agent mesh. Unlike a traditional blocking agent loop, it runs **one short step per event** — each persisted message re-queues the processor via `conversation.message.posted`.
The processor does **not** decide what the user sees. It persists monologue messages and moves on. [Egress](/docs/agents/egress) reads those posts and forwards agent text to the forum.
## Design [#design]
```typescript
/**
* Event-driven LLM processor: one short step per `conversation.message.posted`.
*
* - user text → optional compaction → single infer → persist outputs (type: message)
* - agent tool_call → execute tools → persist tool_result
* - user tool_result → single infer → persist outputs
*
* Each persisted message re-queues the processor via the same event type.
* User visibility is handled by egress, not here.
*/
```
This means the processor never blocks on multi-round tool loops. Tool execution and follow-up inference happen across **separate event deliveries**.
## Trigger filter [#trigger-filter]
```typescript
filterFromConfig: () => (event) => {
if (event.type !== "conversation.message.posted") return false;
// Ignore forum traffic (ingestion handles copy-in) and legacy type: reply.
return event.payload.conversationId !== config.sourceConversationId
&& event.payload.messageType !== "reply";
}
```
The processor handles monologue messages only. It does not filter on whether output is user-visible — that is egress's job.
## State [#state]
```typescript
const processorStateSchema = z.object({
monologueId: z.string().optional(),
agentParticipantId: z.string().optional(),
processed: z.array(z.string()).default([]), // message ids already handled
});
```
Config is seeded on first run from instance config (`monologueId`, `agentParticipantId`). After compaction, state is repointed to the new monologue.
## Decision tree [#decision-tree]
```mermaid
flowchart TD
E[conversation.message.posted] --> C{conversationId === monologueId?}
C -->|no| X[ignore]
C -->|yes| D{already processed?}
D -->|yes| X
D -->|no| T{trigger part types}
T -->|tool_call| TC[Execute tools in parallel]
TC --> TR[Post tool_result as system author]
T -->|tool_result| TI[inferOnce → post outputs]
T -->|user text| COMP{compaction changed list?}
COMP -->|yes| SWAP[Swap monologue + hydrate]
SWAP --> INF[inferOnce → post outputs]
COMP -->|no| INF
TR --> E
INF --> E
INF --> EG[Egress forwards agent text to forum]
```
### User text (role `user`) [#user-text-role-user]
1. Read full monologue transcript → `InferenceMessage[]`
2. Optionally run [compaction](/docs/agents/compaction) — if the list changes, swap to a fresh monologue
3. Call `ctx.llm.infer` once via `inferOnce`
4. Persist outputs as `type: "message"` (text, tool\_call, or both)
### Agent tool\_call (role `agent`) [#agent-tool_call-role-agent]
1. Resolve tool registry (static or lazy)
2. Validate inputs and execute all tools **in parallel**
3. Post `tool_result` parts to monologue as **system** author
4. Mark message processed — the `tool_result` post triggers the next processor step
### User tool\_result (role `user`) [#user-tool_result-role-user]
1. Call `inferOnce` on the full transcript including the new tool results
2. Persist outputs
3. Mark processed
## Configuration [#configuration]
```typescript
const processorConfigSchema = z.object({
sourceConversationId: z.string(), // forum id
forumParticipantId: z.string(),
agentActorId: z.string(),
monologueId: z.string(), // initial — copied to state
agentParticipantId: z.string(),
model: z.string().default("fixture-echo"),
});
```
Created by `joinConversation` with the chosen model id.
## Registration [#registration]
```typescript
defineProcessorActor({
configSchema: processorConfigSchema,
stateSchema: processorStateSchema,
compaction: agentStack.compaction, // optional
tools: agentStack.tools, // optional
});
```
## Contrast with runInferenceLoop [#contrast-with-runinferenceloop]
| | Processor | `runInferenceLoop` |
| ------------------- | ------------------- | -------------------------------------------- |
| **Trigger** | Stream events | Direct function call |
| **Tool rounds** | One round per event | Up to 8 rounds in one call |
| **Use case** | Primary agent mesh | Ephemeral actors (guardian runner) |
| **User visibility** | Does not decide | Does not decide (egress handles mesh agents) |
The processor uses `inferOnce` + event re-queuing. Guardian runners use `runInferenceLoop` for a complete pass in one delivery.
## Queue semantics [#queue-semantics]
```typescript
queue: { mode: "fifo", concurrency: 1 }
```
FIFO preserves monologue ordering. Concurrency 1 prevents parallel inference on the same monologue stint.
## Related [#related]
* [Egress](/docs/agents/egress) — user-visible copy-out from monologue to forum
* [Inference](/docs/agents/inference) — `ctx.llm.infer` and providers
* [Tools](/docs/agents/tools) — tool registration and execution
* [Compaction](/docs/agents/compaction) — transcript compaction before infer
# Subagents (/docs/agents/subagents)
Subagents let a **parent agent** delegate work to another registered agent via LLM tools. Each spawn gets its own conversation and standard mesh (ingestion, processor, egress). Messages flow through **bridge actors** wired by config — no seeded state.
## Mental model [#mental-model]
| Tool | When | What it does |
| --------------------------- | ----------------------------- | ----------------------------------------------------------------------------------------------- |
| **Spawn** (`spawn_`) | First contact with a subagent | Creates conversation, agent, join, bridge instances. Returns **`subagentConversationId`**. |
| **`send_message_to_agent`** | After spawn | Emits `agent.subagent.message.send` → send bridge posts to subagent conversation as system. |
`send_message_to_agent` **never** creates conversations.
## Bootstrap [#bootstrap]
Spawnable agents are **frozen at runtime build**. For each subagent type `K`:
* Register `subagentSendBridge__K` and `subagentReplyBridge__K` (option 2 — per agent at bootstrap).
* Register spawn toolchain from `createToolchainFromAgent("K", ...)`.
* Parent includes `createSubagentMessagingToolchain()` when it can delegate.
```typescript
const researcher = createToolchainFromAgent("researcher", {
agents: expandedAgents,
});
export default defineRuntime()
.events({ ...conversationEvents, ...agentEvents, ...subagentEvents })
.actors(() => ({
...researcher.bridgeActors,
}))
.agents(() => ({
butler: defineAgent({
tools: [
createSubagentMessagingToolchain(),
researcher.spawnToolchain,
],
}),
researcher: defineAgent({ model: "north-default" }),
}));
```
## Flow [#flow]
```mermaid
flowchart LR
P[Parent monologue]
SC[Subagent conversation]
SM[Subagent monologue]
P -->|spawn tool| SC
P -->|send_message_to_agent| E[agent.subagent.message.send]
E --> SB[Send bridge]
SB -->|system post| SC
SC -->|ingestion| SM
SM -->|processor| SM
SC -->|non-system message| RB[Reply bridge]
RB -->|system post| P
```
## Events [#events]
### agent.subagent.message.send [#agentsubagentmessagesend]
```typescript
{
subagentConversationId: string;
message: string;
}
```
Send bridge filters on `config.subagentConversationId === payload.subagentConversationId`.
## Bridge config [#bridge-config]
**Send bridge** (one instance per spawn):
```typescript
{ subagentConversationId: string; systemParticipantId: string }
```
**Reply bridge** (one instance per spawn):
```typescript
{
subagentConversationId: string;
parentMonologueId: string;
parentSystemParticipantId: string;
}
```
Runtime durable queuing applies when filters match — no custom queue state.
## Related [#related]
# Tools (/docs/agents/tools)
**Tools** extend the processor actor with pluggable capabilities the LLM can invoke. They are **agent capabilities inside your product**, not APIs you expose to end users.
A coding harness is the usual example: your server owns the workspace. `defineTool` for `read` / `write` / `shell` lives in the internal package so the agent can work. Users never call those tools — they talk to your session API, which drives Yoke. See [Embed Yoke in your product](/docs/getting-started/embed).
## Types [#types]
```typescript
type ToolOutput = InferenceMessage | InferenceMessage[];
type Tool = {
name: string;
description: string;
parameters?: z.ZodType;
execute: (input: z.infer, ctx: ToolRunContext) => Promise;
};
type Toolchain = {
id: string;
description?: string;
tools: Tool[];
};
```
Tool inputs are validated with Zod before `execute` runs. Outputs must be **inference messages** — not arbitrary JSON.
### Output helpers [#output-helpers]
```typescript
import { toolTextMessage, toolInferenceMessage, normalizeToolOutput } from "@yoke/sdk";
// Single text message
await toolTextMessage("Results: ...");
// From message parts
await toolInferenceMessage([{ type: "text", text: "...", format: "plain" }]);
// Normalize to array
normalizeToolOutput(output); // InferenceMessage[]
```
`tool_result` parts store the inference message array as structured content for the next infer round.
## Registration [#registration]
### Static tools [#static-tools]
```typescript
import {
defineAgentStack,
defineTool,
defineToolchain,
toolTextMessage,
z,
} from "@yoke/sdk";
const echoToolchain = defineToolchain("utilities", [
defineTool({
name: "echo",
description: "Echo text back to the model",
parameters: z.object({ text: z.string() }),
execute: async (input) =>
toolTextMessage(JSON.stringify({ echoed: input.text })),
}),
]);
const agentStack = defineAgentStack({
tools: [echoToolchain],
});
defineProcessorActor({
configSchema: processorConfigSchema,
stateSchema: processorStateSchema,
tools: agentStack.tools,
});
```
### Lazy resolver [#lazy-resolver]
```typescript
defineAgentStack({
tools: async (ctx) => ({
tools: [],
toolchains: ctx.config.model.startsWith("gpt")
? [searchToolchain]
: [],
}),
});
```
Called when building the provider payload each infer round.
## Processor loop [#processor-loop]
1. Resolve tools (static or lazy) → convert Zod to JSON Schema → send to provider
2. If response has `tool_call` parts:
* Post tool calls to monologue (agent author)
* Validate inputs, execute all tools **in parallel**
* Post `tool_result` parts to monologue (system author)
* Run inference again
3. Repeat until no tool calls (max 8 rounds)
4. Post final `type: "reply"` to monologue for egress
## Example echo tool [#example-echo-tool]
```typescript
const echoToolchain = defineToolchain("utilities", [
defineTool({
name: "echo",
description: "Echo text back to the model",
parameters: z.object({ text: z.string() }),
execute: async (input) =>
toolTextMessage(JSON.stringify({ echoed: input.text })),
}),
]);
```
Used with `createLocalToolLoopProvider` and model `local-tool-loop` for E2E demos without API keys.
## Providers [#providers]
| Provider | Tools support |
| ----------------------------- | ------------------------- |
| `createOpenAiProvider` | JSON Schema tools |
| `createAnthropicProvider` | JSON Schema tools |
| `createFixtureProvider` | `echo` tool for tests |
| `createLocalToolLoopProvider` | Local tool loop for demos |
## ToolRunContext [#toolruncontext]
`execute` receives a context with conversation and config access for tools that need to read state or post side-effect messages. Keep tool handlers focused — heavy work should append events for actor processing.
## Related [#related]
# Actors (/docs/concepts/actors)
An **actor** is a registered definition with a `process()` handler, durable state, and a persisted event queue. Actors are Yoke's reactive processors — they run because something happened on the stream.
## Definition vs instance [#definition-vs-instance]
| | Definition | Instance |
| -------- | ------------------------------------------- | -------------------------------- |
| Created | Build time (`defineActor`, `defineRuntime`) | Runtime (`ctx.instances.create`) |
| Contains | `process`, `queue`, schemas, factories | `config`, `filter` |
| Storage | TypeScript module | Postgres `actor_instances` |
One definition can have unlimited instances, each with different config and filters:
```typescript
// Two typescript coders and one python coder — same definition, three instances
await ctx.instances.create("coder", {
config: { model: "gpt-4o", lang: "typescript" },
filter: { type: "work.assigned", where: { "payload.lang": "typescript" } },
});
```
## Three layers of state [#three-layers-of-state]
Each actor instance has three durable layers:
| Layer | Managed by | Contents |
| ------------ | --------------- | ------------------------------------------------- |
| **Queue** | Runtime | Undrained events matched by filter |
| **SQLite** | Actor (via SDK) | Private durable state, actor-owned schema |
| **Postgres** | Runtime | Instance record: config, filter, status, metadata |
All three survive restart.
### Queue [#queue]
When fan-out matches an event to an instance filter, the event is inserted into `actor_queues`. The scheduler claims events per [queue policy](/docs/sdk/queues) and invokes `process()`.
Events are dequeued **only after `process()` returns successfully**. If `process()` throws, the event stays on the queue for retry.
### SQLite [#sqlite]
Each instance gets a private SQLite database — one file per instance, stored via `pg_sqlite` (BYTEA in Postgres) or S3:
| API | Purpose |
| ------------ | ----------------------------------------------------------------- |
| `ctx.state` | Typed document over `__yoke_state` table (`get`, `set`, `update`) |
| `ctx.sqlite` | Raw SQL (`exec`) for custom tables and indexes |
Both access the **same database**. Never write `__yoke_state` directly — use `ctx.state`.
`ctx.state.update()` is atomic: SQLite transaction with optimistic version check via the `state_cas` op.
### Postgres record [#postgres-record]
The `actor_instances` row holds config JSON, serialized filter, queue policy, worker assignment, and lifecycle status. This is what survives across boots to rebuild filters and provision SQLite.
## The process loop [#the-process-loop]
```
1. Event matched by filter → persisted to instance queue
2. Scheduler claims event(s) per queue policy + concurrency
3. await process(ctx, event) // async, impure, non-deterministic
4. process may call ctx.state, ctx.append, ctx.portals.*, ctx.instances.*
5. on success → runtime dequeues event
6. on throw → event stays queued for retry
```
There is **no built-in agent loop**. An LLM actor implements its own loop inside `process`. A routing actor might be a one-liner.
There is no early-ack API. Long-running work blocks the queue slot until `process()` completes. Structure long work as multiple events or use schedules.
## Filters — how actors subscribe [#filters--how-actors-subscribe]
Actors do not declare what they listen to in the definition. **Instances** carry filters:
**Declarative** (fast path, indexed):
```typescript
filter: { type: "work.assigned", where: { "payload.lang": "typescript" } }
filter: { type: "work.*" }
filter: { type: "slack.message.received" }
```
**Function** (slow path, evaluated per event):
```typescript
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
```
Function filters require `filterFromConfig` to survive restart. See [Filters](/docs/sdk/filters).
## Queue policy [#queue-policy]
Each definition sets how its queue is drained:
```typescript
queue: { mode: "fifo", concurrency: 1 } // default
queue: { mode: "lifo", concurrency: 4 }
queue: { mode: "random", concurrency: 2 }
queue: {
mode: "custom",
concurrency: 1,
select: (events) => events.find((e) => e.type === "urgent.alert") ?? events[0],
}
```
`concurrency` controls parallel `process()` invocations per `groupId`. `Infinity` is valid.
Claim order within a group is always **`priority ASC, created_at ASC`** regardless of mode. Modes affect *which* event is selected when using `custom`, or reordering in `lifo`/`random` paths — not the underlying priority sort for default fifo/lifo claims.
## Lifecycle hooks [#lifecycle-hooks]
| Hook | When |
| ---------------- | ---------------------------------- |
| `onCreate(ctx)` | Instance provisioned, SQLite ready |
| `onDestroy(ctx)` | Before teardown, rows deleted |
## Who creates instances [#who-creates-instances]
Anyone with a context:
* Other actors via `ctx.instances.create()`
* Portals in route handlers or `start()` loops
* Portal `start()` hooks at runtime boot
* HTTP control plane API
* CLI
All paths write to Postgres, provision SQLite, register filters, append `instance.created`, and start the drain loop.
## Actor vs portal [#actor-vs-portal]
| | Actor | Portal |
| --------- | ---------------------- | ---------------------------- |
| Trigger | Stream events (filter) | External I/O, timers |
| Queue | Yes | No |
| State | Per-instance SQLite | Optional config only |
| Handler | `process(ctx, event)` | `start(ctx)`, route handlers |
| Instances | Many per definition | One singleton per definition |
## Example coder actor [#example-coder-actor]
```typescript
coder: actor({
configSchema: z.object({ model: z.string(), lang: z.string() }),
stateSchema: z.object({ jobsCompleted: z.number() }),
queue: { mode: "fifo", concurrency: 1 },
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
async process(ctx, event) {
await ctx.append({
type: "code.produced",
payload: { code: `// generated with ${ctx.config.model}` },
});
await ctx.portals.slack.sendMessage("#general", "Code ready");
await ctx.state.update((s) => ({
...s,
jobsCompleted: s.jobsCompleted + 1,
}));
},
}),
```
## Related [#related]
# Durability (/docs/concepts/durability)
Yoke provides **durability by default** — no opt-in checkpoints, no explicit flush calls. Understanding what persists and when helps you reason about failure modes and design reliable systems.
## Three durable layers per actor [#three-durable-layers-per-actor]
| Layer | Storage | Survives restart |
| --------------------- | ------------------------------------- | ---------------- |
| **Queue** | Postgres `actor_queues` | Yes |
| **SQLite state** | Postgres `actor_sqlite` (BYTEA) or S3 | Yes |
| **Instance metadata** | Postgres `actor_instances` | Yes |
Plus the **global stream** in `stream_events` — append-only audit log for all events.
## When writes persist [#when-writes-persist]
| Operation | Persisted when |
| --------------------------------- | ------------------------------------------------------- |
| `ctx.append()` | Immediately — row in `stream_events` before return |
| `ctx.state.get()` | Read — no write |
| `ctx.state.set()` | Immediately — SQLite write + backend flush |
| `ctx.state.update()` | Immediately — transaction + `state_cas` optimistic lock |
| `ctx.sqlite.exec()` | Immediately — direct DB write |
| `ctx.conversations.postMessage()` | Atomically — Postgres INSERT + stream append |
| Queue dequeue | **After** `process()` returns successfully |
| Schedule upsert | Immediately — Postgres `schedules` row |
`ctx.state` is not an in-memory object. It is an async handle backed by SQLite — every `get()`, `set()`, and `update()` round-trips to the storage backend.
## Crash-safe processing [#crash-safe-processing]
The dequeue semantics are the durability contract for event handling:
```
1. Event enqueued to actor_queues (status: pending)
2. Scheduler claims event (status: processing)
3. Deno runs process(ctx, event)
4. SUCCESS → row deleted from queue
5. THROW → status reverted to pending, retry later
```
There is **no early-ack**. If `process()` runs for five minutes, the event occupies a concurrency slot for five minutes. Structure long work as:
* Multiple events (chained via correlation)
* [Schedules](/docs/sdk/schedules) for delayed continuation
* Child instances for parallel work
## Atomic state updates [#atomic-state-updates]
`ctx.state.update()` uses optimistic concurrency:
```typescript
await ctx.state.update((s) => ({
...s,
jobsCompleted: s.jobsCompleted + 1,
}));
```
Under the hood: read version → apply updater in Deno → `state_cas` op with expected version → SQLite transaction. Concurrent updaters serialize; no lost writes.
Direct `ctx.state.set()` overwrites without version check — use `update()` for read-modify-write.
## Storage backends [#storage-backends]
Per-instance SQLite persistence is controlled by `YOKE_STORAGE`:
| Value | Behavior |
| ----------- | ---------------------------------------------------------------------------------- |
| `pg_sqlite` | Serialized SQLite databases stored as BYTEA in Postgres `actor_sqlite` table |
| `s3` | SQLite files in S3 (`YOKE_S3_BUCKET`); local cache dir for rusqlite execution only |
`pg_sqlite` stores BYTEA blobs in Postgres — it is **not** the PostgreSQL `pg_sqlite` extension. SQLite runs locally in the worker process; the serialized file is persisted to Postgres.
The SDK surface (`ctx.state`, `ctx.sqlite`) is identical in both modes.
### S3 sync modes [#s3-sync-modes]
| `YOKE_S3_SYNC` | Behavior |
| ---------------- | ---------------------------------------------- |
| `lazy` (default) | Upload after writes; `flush_all()` on shutdown |
| `strict` | `fsync` local `state.db` before each S3 upload |
With `strict`, a successful `state_set` / `sqlite_exec` / `state_cas` implies the local file is durable before upload begins. A crash between local write and upload may lose data in `lazy` mode.
## Stream durability [#stream-durability]
Stream appends are transactional in Postgres. Once `ctx.append()` resolves, the event has a `seq` and will fan-out (possibly after a short async delay).
Fan-out delivery is idempotent — `fanout_delivered` prevents duplicate enqueues for the same `(tenant_id, stream_seq, instance_id)`.
## Conversation durability [#conversation-durability]
Messages are canonical in Postgres. `conversation.message.posted` stream events carry ids only — handlers read full content from the conversation store when needed.
Kernel ops perform `INSERT` + `stream_append` atomically — user code cannot post to a conversation without persisting it.
## What is not durable [#what-is-not-durable]
| Ephemeral | Why |
| ------------------------------------- | ---------------------------------- |
| In-flight `process()` local variables | Lost on crash — recover from queue |
| Deno module scope without `ctx.state` | Worker restart clears memory |
| Un-awaited promises in `process()` | May not complete before crash |
Always persist important data via `ctx.state`, `ctx.sqlite`, or `ctx.append()` before returning from `process()`.
## Failure recovery [#failure-recovery]
On restart:
1. Load instance records from Postgres
2. Rebuild fan-out index from persisted filters + `filterFromConfig`
3. Provision SQLite from `actor_sqlite` or S3
4. Resume drain loops — pending queue events reprocessed
5. Portal `start()` loops restart
No stream replay. Actors pick up where their queues left off.
## Related [#related]
# Events & Stream (/docs/concepts/events-stream)
The **global stream** is Yoke's communication backbone. Every action that matters in the system becomes a durable, sequenced event in Postgres. Actors never communicate directly — they publish events and subscribe via filters.
## Event shape [#event-shape]
Every persisted event has this structure:
```typescript
type StreamEvent = {
seq: number; // global monotonic position (per tenant)
id: string; // unique event id (UUID)
at: number; // unix timestamp ms
type: string; // open namespace, e.g. "work.assigned"
payload: unknown; // JSON, validated by Zod at append
source?: {
kind: "actor" | "portal";
instanceId: string;
};
correlation?: string; // optional grouping for related work
priority?: number; // queue ordering hint (lower = sooner)
groupId?: string; // concurrency group for queue processing
};
```
| Field | Purpose |
| ---------------------- | ------------------------------------------------------------------------------------ |
| `seq` | Global ordering within a tenant. Assigned atomically on append. |
| `type` | Open namespace string. Convention, not enum — add new types freely. |
| `payload` | Arbitrary JSON. Validated against `defineEvents()` schemas when registered. |
| `source` | Set automatically when appending from an actor or portal context. |
| `correlation` | Links related events into a work chain for tracing and grouping. |
| `priority` / `groupId` | Affect [queue claim order](/docs/runtime/queues) — `priority ASC`, `created_at ASC`. |
## Append-only API [#append-only-api]
User code can **only append** to the stream. There is no tail, subscribe, or read API exposed to TypeScript:
```typescript
await ctx.append({
type: "work.assigned",
payload: { role: "coder", lang: "typescript" },
});
```
The append path:
1. Validate payload against Zod schema (if registered)
2. Insert row into `stream_events` with next `seq`
3. Return immediately to caller
4. **Async fan-out** evaluates actor filters and enqueues matches
The stream **never blocks** on fan-out. A slow or saturated actor affects only its own queue.
## Typed events with defineEvents [#typed-events-with-defineevents]
Register schemas once for compile-time safety and runtime validation:
```typescript
const events = defineEvents({
"slack.message.received": z.object({
channel: z.string(),
text: z.string(),
}),
"work.assigned": z.object({
role: z.string(),
lang: z.string().optional(),
}),
});
export default defineRuntime().events(events) /* ... */;
```
Inside `process()`, known event types narrow automatically:
```typescript
async process(ctx, event) {
if (event.type === "work.assigned") {
event.payload.lang; // typed as string | undefined
}
}
```
Unregistered event types still append and process — payload is `unknown`.
## Correlation [#correlation]
Group related events into one piece of work:
```typescript
ctx.correlation.current; // id of event being processed, if any
ctx.correlation.inherit(); // reuse current or create new
ctx.correlation.create(); // explicit new chain
// Default: inherit correlation while processing
await ctx.append({ type: "step.started", payload: {} });
// Start a new chain
await ctx.append({ type: "step.started", payload: {} }, { correlation: "new" });
// Explicit id
await ctx.append({ type: "step.started", payload: {} }, { correlation: "order-42" });
```
Correlation is stored on the event row and passed through fan-out. It does not affect delivery — only observability and downstream logic that chooses to read it.
## Priority and groupId [#priority-and-groupid]
When fan-out enqueues an event to an actor queue, `priority` and `groupId` control drain behavior:
```typescript
await ctx.append(
{ type: "urgent.alert", payload: { text: "down" } },
{ priority: 0, groupId: "alerts" },
);
```
Queue claim order is **`priority ASC, created_at ASC`** — lower priority numbers are claimed first; ties break on enqueue time. This is independent of queue policy mode (`fifo`, `lifo`, `random`, `custom`).
`groupId` scopes concurrency: the scheduler drains one concurrency slot per group. Use `"inherit"` in conversation `postMessage` to propagate the triggering event's group.
## Stream lifecycle events [#stream-lifecycle-events]
The runtime appends system events:
| Type | When |
| -------------------- | ------------------------------------ |
| `runtime.started` | Boot completes successfully |
| `instance.created` | Actor or portal instance provisioned |
| `instance.destroyed` | Instance torn down |
Conversation and agent packages add more — see [Event catalog](/docs/reference/event-catalog).
## Fan-out delivery [#fan-out-delivery]
After append commits, the fan-out engine:
1. Looks up declarative filters in the fast-path index (by type prefix)
2. Evaluates function filters on the slow path (via Deno `eval_filter`)
3. Inserts matching events into `actor_queues` per instance
4. Records delivery in `fanout_delivered` for idempotency
Fan-out is **per-tenant** and **idempotent** — duplicate delivery for the same `(tenant_id, stream_seq, instance_id)` is prevented.
See [Fan-out](/docs/runtime/fanout) for implementation details.
## Testing with CLI append [#testing-with-cli-append]
Append events without writing code:
```bash
yoke append \
--database "$DATABASE_URL" \
--type slack.message.received \
--payload '{"channel":"#general","text":"hello"}'
```
Multi-tenant deployments scope by `--tenant-id`.
## What the stream is not [#what-the-stream-is-not]
| Misconception | Reality |
| -------------------------- | ----------------------------------------------------------- |
| Event sourcing replay log | History and audit only; actors recover from queues + SQLite |
| Message queue with ACK | Dequeue happens after `process()` succeeds, not on receive |
| Pub/sub with subscriptions | Filters are per-instance, evaluated on every append |
| Kafka topic | Single global log per tenant, not partitioned topics |
## Related [#related]
# Concepts (/docs/concepts)
Yoke is an **always-on, event-driven multi-actor runtime** you embed in your own product. The kernel is Rust; the internal actor package is TypeScript running in an embedded Deno worker. End users talk to your server, not to Yoke.
This section explains the mental model behind the runtime — how pieces fit together before you dive into SDK APIs or the [embed API](/docs/getting-started/embed).
## Three primitives [#three-primitives]
Everything in Yoke reduces to three building blocks:
```
EXTERNAL WORLD
│
┌───────────┴───────────┐
│ PORTALS │
│ routes · start() │
└───────────┬───────────┘
append │
▼
┌───────────────────────┐
│ GLOBAL STREAM │
│ (Postgres, sequenced) │
└───────────┬───────────┘
async fan-out
┌───────────────┼───────────────┐
▼ ▼ ▼
┌──────────┐ ┌──────────┐ ┌──────────┐
│ Actor │ │ Actor │ │ Actor │
│ instance │ │ instance │ │ instance │
└──────────┘ └──────────┘ └──────────┘
```
1. **Global stream** — sequenced, append-only, durable event log
2. **Actors** — reactive; consume events from per-instance queues
3. **Portals** — proactive; connect to the external world
## Core principles [#core-principles]
| Principle | Meaning |
| ---------------------------------------- | ------------------------------------------------------------------------------------------ |
| **Stream is the spine** | All communication goes through the global event stream. No direct actor-to-actor channels. |
| **No targets** | Publishers emit typed events. Consumers self-select via filters. |
| **No replay** | The stream is history and audit, not a program to re-execute. |
| **No purity enforcement** | Actors are async, impure, and non-deterministic. |
| **No stream read API** | User code cannot read the stream directly — security boundary. |
| **Definitions fixed, instances dynamic** | Actor/portal types are registered at build time. Instances are created at runtime. |
Yoke is **not** an LLM framework. An actor is any stateful async event processor — code, LLM calls, human-in-the-loop, or a combination. Multi-agent coordination emerges from actors publishing events and listening via filters.
## Technology stack [#technology-stack]
| Layer | Technology | Role |
| ---------------- | --------------------------------- | ---------------------------------------- |
| Product host | Your Rust server (`yoke-runtime`) | Public API, embed the kernel |
| Internal package | TypeScript (`@yoke/sdk`) | Actor definitions the worker loads |
| Worker | Deno (embedded via libdeno) | Transpile, npm, IPC bridge |
| Kernel | Rust runtime | Stream, fan-out, queues, scheduler |
| Storage | Postgres 16+ | Stream, instances, queues, conversations |
| Actor state | SQLite per instance | `pg_sqlite` (BYTEA in Postgres) or S3 |
## Recommended reading order [#recommended-reading-order]
[Embed Yoke in your product](/docs/getting-started/embed) — host vs internal package.
[Mental model](/docs/concepts/mental-model) — how to think about Yoke systems.
[Events & stream](/docs/concepts/events-stream) — the communication backbone.
[Actors](/docs/concepts/actors) and [Portals](/docs/concepts/portals) — the two processor types.
[Instances](/docs/concepts/instances) — definitions become running objects.
[Durability](/docs/concepts/durability) — what persists and when.
# Instances (/docs/concepts/instances)
**Definitions** are fixed at build time. **Instances** are dynamic runtime objects created from those definitions with per-instance config and filters.
Understanding this distinction is essential for building scalable Yoke systems.
## Definition vs instance [#definition-vs-instance]
```
defineRuntime()
.actors(() => ({
coder: actor({ configSchema, stateSchema, process }), // DEFINITION
}));
// At runtime:
await ctx.instances.create("coder", { // INSTANCE
config: { model: "gpt-4o", lang: "typescript" },
filter: { type: "work.assigned", where: { "payload.lang": "typescript" } },
});
```
| | Definition | Instance |
| -------- | --------------------------- | ----------------------------------------------- |
| Count | One per name in `yoke.ts` | Unlimited |
| Storage | TypeScript | Postgres `actor_instances` / `portal_instances` |
| Identity | `"coder"` (definition name) | UUID `instanceId` |
| Filter | N/A (on instance) | Declarative or function |
| SQLite | Schema in `stateSchema` | Provisioned per instance |
Portal definitions always have **one singleton instance** per tenant. Actor definitions can have many.
## Creating instances [#creating-instances]
### From actor or portal code [#from-actor-or-portal-code]
```typescript
const handle = await ctx.instances.create("coder", {
config: { model: "claude-sonnet", lang: "rust" },
filter: { type: "work.assigned", where: { "payload.lang": "rust" } },
});
console.log(handle.id); // instance UUID
```
`config` is validated against the definition's `configSchema`. `filter` determines stream subscription.
### From bootstrap() [#from-bootstrap]
Packages typically provision fixed topology in **`defineRuntime().bootstrap()`** using stable instance ids:
```typescript
.bootstrap(async (ctx) => {
await ctx.instances.create("router", {
id: "router",
config: { channels: ["#general"] },
filter: { type: "slack.message.received" },
});
})
```
Re-running `bootstrap()` is safe — creation is idempotent by `id`.
### From HTTP / CLI [#from-http--cli]
Multi-tenant control plane and executor APIs can create instances externally. The same Postgres rows and `instance.created` events are emitted.
## Instance options [#instance-options]
```typescript
type CreateInstanceOpts = {
id: string; // required — idempotent key for this instance
config: TConfig;
filter?: DeclarativeFilter | FunctionFilter;
typeHint?: string; // narrows slow-path function filter evaluation
};
```
### Declarative filters [#declarative-filters]
Persisted in Postgres, indexed for fast fan-out:
```typescript
filter: { type: "conversation.message.posted" }
filter: { type: "work.*" }
filter: {
type: "work.assigned",
where: { "payload.lang": "typescript" },
}
```
### Function filters [#function-filters]
Require `filterFromConfig` on the definition:
```typescript
await ctx.instances.create("coder", {
config: { lang: "go" },
filterKind: "function",
});
```
On boot/restart, the runtime calls `filterFromConfig(config)` to rebuild the function. Without the factory, persisted function-filter instances exist but receive **no events**.
## Lifecycle [#lifecycle]
```
create → active → destroy
```
| Event | Stream type | When |
| --------- | -------------------- | --------------------------------------------------- |
| Created | `instance.created` | Row inserted, SQLite provisioned, filter registered |
| Destroyed | `instance.destroyed` | `onDestroy` called, queue cleared, rows deleted |
```typescript
await ctx.instances.destroy(instanceId);
```
Destroy is permanent — SQLite blob deleted, queue rows removed, filter deregistered.
## Instance record in Postgres [#instance-record-in-postgres]
`actor_instances` stores:
| Column | Purpose |
| ----------------- | ------------------------------------------------------------ |
| `id` | Instance UUID |
| `definition_name` | e.g. `"coder"` |
| `config` | JSON blob |
| `filter_kind` | `declarative` or `function` |
| `filter` | Serialized declarative filter or config for function rebuild |
| `queue_policy` | Serialized mode + concurrency |
| `worker_id` | Horizontal scaling assignment |
| `tenant_id` | Multi-tenant isolation |
| `status` | `active`, etc. |
On restart, the runtime loads all instance rows, rebuilds filters, provisions SQLite from `actor_sqlite` or S3, and resumes drain loops.
## Worker assignment [#worker-assignment]
In multi-worker deployments, new instances inherit the creating worker's `worker_id`. Orphan instances (`worker_id IS NULL`) are claimed on worker startup.
Only the owning worker drains an instance's queue. See [Scaling](/docs/deployment/scaling).
## Agent instances [#agent-instances]
Agent bootstrap creates multiple instances per join:
```typescript
const agent = await createAgent(ctx, { name: "support-bot" });
// → { actorId, egressInstanceId }
const join = await joinConversation(ctx, agent, room.id, { model: "gpt-4o" });
// → { ingestionInstanceId, processorInstanceId, monologueId, ... }
```
Each join stint gets fresh ingestion and processor instances. One egress instance per agent routes replies across all forum joins.
## sqliteStoreId [#sqlitestoreid]
Advanced: actors can resolve a logical SQLite store segment per invocation via `sqliteStoreId` on the definition. Default is `instanceId`. Used when one instance manages multiple logical state partitions.
## Related [#related]
# Mental Model (/docs/concepts/mental-model)
Yoke is designed around a small set of constraints that keep complex agentic systems composable. Understanding these constraints upfront prevents fighting the runtime later.
The runtime lives **inside your product**. Your server embeds `yoke-runtime`; the stream, actors, and portals below are implementation details of that process. Users never subscribe to the stream or call `@yoke/sdk`. See [Embed Yoke in your product](/docs/getting-started/embed).
## The stream is the spine [#the-stream-is-the-spine]
Every meaningful action in a Yoke system becomes an **event** on a single global, append-only stream. Actors do not call each other directly. Portals do not push into actor mailboxes. Everything flows through typed events:
```
Portal webhook → append(slack.message.received)
Router actor → append(work.assigned)
Coder actor → append(code.produced)
```
The stream is the **single sequencing authority**. Each event gets a monotonic `seq` number. This is audit history, not a replay log — actors recover from their queues and SQLite state, not by re-executing the stream.
User code has **no stream read API**. Only the Rust kernel tails the stream internally for fan-out. This is a deliberate security boundary.
## No targets — publish and subscribe [#no-targets--publish-and-subscribe]
Traditional message systems route messages to named recipients. Yoke deliberately avoids this:
```typescript
// Correct — publish a typed event
await ctx.append({ type: "work.assigned", payload: { role: "coder", lang: "typescript" } });
// Wrong — does not exist
await ctx.append({ type: "...", target: "coder" });
```
**Coupling is by convention**, not routing tables. A `coder` actor subscribes to `work.assigned` events where `payload.lang` matches its config. A `router` actor publishes those events. Neither knows the other exists.
This enables:
* **Dynamic topology** — spin up N coder instances without reconfiguring publishers
* **Emergent orchestration** — supervisor patterns, pipelines, and meshes arise from filter overlap
* **Loose coupling** — add a new consumer by creating an instance with a filter
## Two kinds of processors [#two-kinds-of-processors]
Yoke has exactly two processor archetypes. They are not interchangeable.
### Actors — reactive [#actors--reactive]
Actors **react** to stream events matched by their filter. Each instance has:
* A **persisted queue** of matched events
* **Durable SQLite state** (`ctx.state`, `ctx.sqlite`)
* A **`process(ctx, event)`** handler invoked by the scheduler
Actors are for work that happens **because something occurred** — a message arrived, a payment cleared, a timer fired.
### Portals — proactive [#portals--proactive]
Portals **connect** to the outside world. Each portal definition has one singleton instance that:
* Runs a long-lived **`start(ctx)`** loop (WebSocket, polling, etc.)
* Registers **HTTP routes** on the runtime server
* Optionally exposes a **typed interface** actors call via `ctx.portals..*`
Portals have **no queues** and **no filters**. They append events (ingress) and serve egress calls.
```
EXTERNAL PORTAL STREAM ACTOR
│ │ │ │
│── webhook ─────►│ │ │
│ │── append ────────►│ │
│ │ │── fan-out ────►│
│ │ │ │── process
│ │ │◄── append ─────│
│◄── portal_call ─│◄──────────────────│ │
```
## Definitions vs instances [#definitions-vs-instances]
| | Definition | Instance |
| ---------- | -------------------------------- | -------------------------------- |
| **When** | Build time (`defineRuntime`) | Runtime (`ctx.instances.create`) |
| **What** | `process`, queue policy, schemas | `config`, `filter` |
| **Count** | Fixed set per package | Unlimited, dynamic |
| **Stored** | TypeScript module | Postgres |
You cannot add new actor or portal **types** at runtime. You **can** create unlimited **instances** of registered definitions.
```typescript
// Build time — register the coder definition
defineRuntime().actors(() => ({
coder: actor({ configSchema, stateSchema, process }),
}));
// Runtime — create instances with different config/filters
await ctx.instances.create("coder", {
config: { model: "gpt-4o", lang: "typescript" },
filter: { type: "work.assigned", where: { "payload.lang": "typescript" } },
});
await ctx.instances.create("coder", {
config: { model: "claude-sonnet", lang: "python" },
filter: { type: "work.assigned", where: { "payload.lang": "python" } },
});
```
## Emergent multi-agent systems [#emergent-multi-agent-systems]
Yoke does not ship orchestration primitives — no supervisor actor, no graph DSL, no built-in agent loop. Complex systems emerge from:
1. **Event type conventions** — `work.assigned`, `code.produced`, `conversation.message.posted`
2. **Filter overlap** — multiple actors subscribe to overlapping event types
3. **Instance creation** — actors spawn child workers via `ctx.instances.create()`
4. **Portal interfaces** — typed egress to external systems
The [agent mesh pattern](/docs/agents/mesh-pattern) (ingestion → processor → egress) is one emergent composition built from standard actors and the conversation store — not a kernel primitive.
## What Yoke is not [#what-yoke-is-not]
| Not Yoke | Instead |
| ----------------------------- | ---------------------------------------------------------- |
| Your product / coding harness | Embed `yoke-runtime` in **your** server; users talk to you |
| Public SDK for end users | `@yoke/sdk` is the internal actor package API |
| LLM framework | Use `ctx.llm.infer()` inside a processor actor |
| Message broker with topics | Global stream + per-instance filters |
| Workflow engine with replay | Queue + SQLite recovery |
| Chat server | Conversation store + stream delivery |
| Serverless functions | Always-on drain loops with durable state |
## System diagram [#system-diagram]
```mermaid
flowchart TB
subgraph external [External World]
HTTP[HTTP / Webhooks]
WS[WebSockets]
end
subgraph portals [Portals]
P[Portal start + routes]
end
subgraph stream [Global Stream]
S[(Postgres stream_events)]
end
subgraph actors [Actors]
A1[Instance A]
A2[Instance B]
Q1[Queue]
Q2[Queue]
DB1[(SQLite)]
DB2[(SQLite)]
end
HTTP --> P
WS --> P
P -->|append| S
S -->|fan-out| Q1
S -->|fan-out| Q2
Q1 --> A1
Q2 --> A2
A1 --> DB1
A2 --> DB2
A1 -->|append| S
A2 -->|append| S
A1 -->|portal_call| P
```
## Next steps [#next-steps]
# Portals (/docs/concepts/portals)
**Portals** connect the Yoke runtime to **integrations it needs** (Slack, GitHub, third-party APIs). They are not actors — they have no queues, no filters, and no reactive `process()` handler.
Your product's public HTTP belongs on **your server** (the embed host), not on a portal. Portal routes should stay private. See [Embed Yoke in your product](/docs/getting-started/embed).
Every portal definition has exactly **one singleton instance** per tenant at runtime boot.
## What portals do [#what-portals-do]
* Register **HTTP/WebSocket routes** on the runtime server
* Run **cron-scheduled route invocations** via `crons` (wake-on-tick, not long-running loops)
* **Append events** to the global stream (ingress)
* Optionally publish a **typed interface** callable from actors (egress)
## What portals do not do [#what-portals-do-not-do]
* Have queues or process stream events reactively
* Filter or subscribe to the stream
* Scale to multiple instances per definition
"Chat channel" or "Slack integration" are **conventions** built with portals — not kernel types. Portal patterns emerge from code, not from a type hierarchy in core.
## Portal shape [#portal-shape]
```typescript
definePortal({
configSchema: z.object({ botToken: z.string() }),
interface: {
async sendMessage(channel: string, text: string) {
await fetch("https://slack.com/api/chat.postMessage", {
method: "POST",
headers: { Authorization: `Bearer ${ctx.config.botToken}` },
body: JSON.stringify({ channel, text }),
});
},
},
routes: [
{
method: "POST",
path: "/webhooks/slack",
handler: async (req, ctx) => {
const body = JSON.parse(new TextDecoder().decode(req.body ?? new Uint8Array()));
await ctx.append({
type: "slack.message.received",
payload: body,
});
return { status: 200, body: "ok" };
},
},
],
});
```
## Ingress — append to stream [#ingress--append-to-stream]
Portals turn external signals into stream events. The webhook handler above is the most common pattern:
```
HTTP POST /webhooks/slack
→ portal route handler
→ ctx.append({ type: "slack.message.received", ... })
→ Postgres stream_events
→ async fan-out to actor queues
```
`crons` on portal definitions are another ingress path — scheduled route invocations that append heartbeat or polled data events.
## Egress — typed interface [#egress--typed-interface]
Actors call portal methods through `ctx.portals..*`:
```typescript
async process(ctx, event) {
await ctx.portals.slack.sendMessage("#general", "Task complete");
}
```
Under the hood: SDK → `portal_call` op → Rust → Deno `portal_invoke` → your interface method.
The interface is typed through `defineFor(events, portals)` — `ctx.portals.slack` knows `sendMessage`'s signature.
## HTTP routes [#http-routes]
Portal routes mount on the runtime HTTP server alongside system endpoints:
| Endpoint | Owner |
| -------------- | -------------------- |
| `GET /health` | Runtime |
| `GET /metrics` | Runtime (Prometheus) |
| Portal routes | User-defined |
Route handlers receive `(req, ctx)` where `ctx` is `PortalCtx` — append, instances, conversations, schedules, but no `state`/`sqlite` (portals are stateless by convention; use actor state for durable portal-adjacent data).
### Path parameters [#path-parameters]
```typescript
{
method: "POST",
path: "/agents/:agentId/join",
handler: async (req, ctx) => {
const agentId = req.path.split("/")[2];
// ...
},
}
```
Register agent bootstrap at `/agents` and `/agents/:agentId/join` on an executor portal — not `/rooms`.
## Portal context [#portal-context]
`PortalCtx` provides:
| API | Available |
| ------------------- | ---------------------- |
| `ctx.config` | Portal instance config |
| `ctx.append()` | Stream ingress |
| `ctx.instances` | Create actor instances |
| `ctx.conversations` | Conversation store |
| `ctx.schedules` | Keyed delayed events |
| `ctx.correlation` | Event correlation |
Portals do **not** have `ctx.state` or `ctx.sqlite` — they are singleton connectors, not stateful processors.
## Boot provisioning [#boot-provisioning]
Use **`defineRuntime().bootstrap(fn)`** to provision fixed actor topology once at boot (idempotent by instance `id`):
```typescript
export default defineRuntime()
.events(events)
.bootstrap(async (ctx) => {
await ctx.instances.create("router", {
id: "router",
config: { channels: ["#general"] },
filter: { type: "slack.message.received" },
});
})
.portals(/* ... */)
.actors(/* ... */)
.build();
```
The runtime ensures portal singleton rows exist in Postgres, runs `bootstrap()` if defined, then registers portal routes and crons.
Portals do **not** support long-running `start()` hooks. Use `bootstrap()` for instance provisioning and portal `crons` or HTTP routes for recurring ingress.
## Multi-tenant ingress [#multi-tenant-ingress]
In hosted deployments, portal HTTP may arrive at a tenant executor after the control plane scheduler wakes a cold runtime. The route dispatch path is the same — `PortalHttp` activity signals keep the tenant hot.
Control plane HTTP (conversations, blobs, schedules) runs on port **8090** by default — separate from the executor portal port.
## Design patterns [#design-patterns]
Register `POST` routes that parse request bodies and `ctx.append()` typed events. Common for Shopify, Stripe, and other webhook providers.
Portal `crons` that invoke a route on an interval. Append heartbeat or fetched data events from the handler.
Expose REST endpoints that call `createAgent` / `joinConversation` — see [Agent bootstrap](/docs/agents/bootstrap).
## Related [#related]
# Branches (/docs/conversations/branches)
**Branches** enable soft forks within a conversation. Forking at message N creates a `branches` row; subsequent messages on that line set `branch_id`. Main timeline messages keep `branch_id = null`.
## Visual model [#visual-model]
```
main: … → msg#10 → msg#11 → msg#12
branch B (fork at #10): msg#10' → msg#11' → msg#12'
```
Both timelines share the same `conversation_id`. The fork point is recorded on the branch row.
## Schema [#schema]
```
branches
├── id (PK)
├── tenant_id
├── conversation_id
├── forked_from_message_id → messages.id
└── created_at
```
Messages on a branch:
```
messages.branch_id → branches.id
```
Main timeline: `branch_id IS NULL`.
## Creating a branch [#creating-a-branch]
SDK:
```typescript
const branch = await ctx.conversations.createBranch(
conversationId,
forkedFromMessageId,
);
```
HTTP:
```
POST /v1/tenants/:tenant_id/conversations/:conversation_id/branches
{ "forkedFromMessageId": "..." }
```
Emits `conversation.branch.created` on the stream.
## Posting on a branch [#posting-on-a-branch]
```typescript
await ctx.conversations.postMessage(conversationId, body, {
branchId: branch.id,
type: "message",
authorParticipantId: participantId,
});
```
## Querying [#querying]
```sql
-- Main timeline only
SELECT * FROM messages
WHERE conversation_id = $1 AND branch_id IS NULL
ORDER BY created_at;
-- Specific branch
SELECT * FROM messages
WHERE conversation_id = $1 AND branch_id = $2
ORDER BY created_at;
```
HTTP pagination accepts `branchId` query param — omit for main timeline.
```
GET .../messages?branchId=&limit=50
```
## Use cases [#use-cases]
| Scenario | Branch pattern |
| -------------------------- | ------------------------------------------------------------------------------ |
| "Try again" from message N | Fork, continue alternate agent response |
| A/B comparison | Parallel branch timelines |
| Edit history | Prefer `type: edit` + relations for replacements; branches for divergent lines |
## Agent processors [#agent-processors]
Processors typically read the main timeline (`branch_id IS NULL`) unless configured for branch-aware UIs. Branch traffic still delivers via `conversation.message.posted` with `branchId` in payload.
## Related [#related]
# Data Model (/docs/conversations/data-model)
The conversation store lives in Postgres alongside the runtime stream. All tables are scoped by `tenant_id` for multi-tenant isolation.
## Entity relationship [#entity-relationship]
```
actors (identity)
│
├── participants (stints in conversations)
│ │
│ └── conversations
│ │
│ ├── messages
│ ├── branches
│ └── message_relations
│
blobs (media metadata + S3 storage_key)
```
## actors [#actors]
Chat identity — human, system, agent persona:
```
actors
├── id (PK)
├── tenant_id
├── created_at
└── metadata (jsonb)
```
An `actors` row may link to a yoke `actor_instances` id via `metadata` when backed by a running instance. Humans have actor rows without instances.
## conversations [#conversations]
```
conversations
├── id (PK)
├── tenant_id
├── created_at
└── metadata (jsonb) // title, monologue pairing hints, etc.
```
No `kind` column — forum vs monologue is determined by participant sets and product conventions.
## participants [#participants]
One **stint** of an actor in one conversation:
```
participants
├── id (PK)
├── tenant_id
├── conversation_id → conversations.id
├── actor_id → actors.id
├── joined_at
├── left_at (nullable — null = active)
└── metadata (jsonb) // monologueId for agent forum stints
```
Rejoining a forum creates a **new participant row** (new stint), not reuse of the old one.
### Actor vs participant [#actor-vs-participant]
```
alice (actor) ──► p1 forum F joined t0 left t1
──► p3 forum F joined t2 left null ← current stint
──► p4 monologue M joined t0 left null
```
Visibility is **membership**. Non-participants cannot see conversation messages.
## messages [#messages]
```
messages
├── id (PK)
├── tenant_id
├── conversation_id
├── type string (message, reply, reaction, system, edit, …)
├── author_participant_id → participants.id (NULL = system)
├── body jsonb — { parts[], meta? }
├── branch_id → branches.id (nullable, null = main timeline)
└── created_at
```
No `parent_id`, `role`, or `thread_root_id` columns — structure lives in `type`, `body.parts`, and `message_relations`.
## message\_relations [#message_relations]
```
message_relations
├── id
├── tenant_id
├── message_id
├── relation_type reply | edit | reaction | mirror
├── target_message_id
└── created_at
```
| Action | Insert pattern |
| -------- | ------------------------------------------------ |
| Reply | `type: message` + `relation: reply → target` |
| Edit | `type: edit` + `relation: edit → original` |
| Reaction | `type: reaction` + `relation: reaction → target` |
| Mirror | `type: message` + `relation: mirror → source` |
## branches [#branches]
Soft forks within a conversation:
```
branches
├── id (PK)
├── tenant_id
├── conversation_id
├── forked_from_message_id → messages.id
└── created_at
```
Main timeline: `branch_id IS NULL`. Branch timeline: messages set `branch_id`.
## blobs [#blobs]
Media metadata (bytes in S3 when `YOKE_STORAGE=s3`):
```
blobs
├── id, tenant_id
├── mime_type, size_bytes, sha256, storage_key
├── filename?, width?, height?, duration_ms?
└── metadata, created_at
```
## Query cheatsheet [#query-cheatsheet]
```sql
-- Forum transcript (main timeline)
SELECT * FROM messages
WHERE tenant_id = $1 AND conversation_id = $2 AND branch_id IS NULL
ORDER BY created_at;
-- Active participants
SELECT * FROM participants
WHERE tenant_id = $1 AND conversation_id = $2 AND left_at IS NULL;
```
## Kernel ops [#kernel-ops]
| Op | Description |
| -------------------------------- | -------------------------------------- |
| `conversation_create_actor` | Create chat identity |
| `conversation_create` | Create conversation |
| `conversation_add_participant` | Add stint |
| `conversation_leave_participant` | End stint |
| `conversation_post_message` | Insert + `conversation.message.posted` |
| `conversation_create_branch` | Soft fork |
| `conversation_list_messages` | Transcript read |
| `conversation_list_participants` | Participant list |
All message posts are atomic: Postgres INSERT + stream append.
## Related [#related]
# Delivery (/docs/conversations/delivery)
The conversation store does **not** maintain an inbox or delivery queue. When a message is posted, delivery happens immediately through the global event stream and async fan-out to actor queues.
## Delivery path [#delivery-path]
```
conversation_post_message (kernel op)
│
├─ INSERT messages (canonical Postgres)
│
└─ stream_append: conversation.message.posted
│
▼
async fan-out → actor_queues
│
▼
actor process() — agent-specific logic
```
Payloads on `conversation.message.posted` carry **ids only**:
```typescript
{
conversationId: string;
messageId: string;
messageType: string;
authorParticipantId?: string;
branchId?: string;
}
```
Handlers read full message content from Postgres via `ctx.conversations.listMessages` or kernel list ops when needed.
## No inbox table [#no-inbox-table]
| Traditional chat | Yoke |
| ------------------------------------------- | ---------------------------------------------- |
| INSERT message + INSERT inbox per recipient | INSERT message + stream append |
| Poll/sync inbox | Actor filters on `conversation.message.posted` |
| Delivery receipts table | Stream is audit; dequeue after process success |
Agents that need mirrored copies implement that in actor logic (ingestion actor), not in the conversation store.
## Fan-out to agents [#fan-out-to-agents]
Agent instances subscribe with declarative or function filters:
```typescript
// Ingestion — forum messages
filter: { type: "conversation.message.posted", where: { "payload.conversationId": forumId } }
// Processor — monologue messages
filter: { type: "conversation.message.posted", where: { "payload.conversationId": monologueId } }
// Egress — agent replies
filter: { type: "conversation.message.posted", where: { "payload.conversationId": monologueId, "payload.messageType": "reply" } }
```
## Atomic post [#atomic-post]
User code cannot post to a conversation without persisting it. Kernel ops perform INSERT + stream append in one transaction — no orphaned stream events or ghost messages.
## Priority and grouping [#priority-and-grouping]
Conversation posts can set queue hints:
```typescript
await ctx.conversations.postMessage(conversationId, body, {
priority: 0,
groupId: "inherit", // propagate from triggering event
});
```
Queue claim order: `priority ASC`, `created_at ASC`.
## Offline agents [#offline-agents]
If no actor instance is listening, the message is still in Postgres. When an agent boots and subscribes, it does **not** automatically receive historical messages via fan-out — fan-out is append-triggered only.
Patterns for catch-up:
* Read transcript on `onCreate` or first `process`
* Backfill via `listMessages` when joining a room
* Ingestion actor handles live traffic only; processor reads full transcript each turn
## Human clients [#human-clients]
Headless apps poll or subscribe via HTTP/WebSocket layers **outside** Yoke core. The control plane exposes transcript pagination:
```
GET /v1/tenants/:tenant_id/conversations/:id/messages?latest=true&limit=50
```
See [HTTP API](/docs/conversations/http-api).
## Related [#related]
# Forum & Monologue (/docs/conversations/forum-monologue)
**Forum** and **monologue** are product roles, not separate database tables. Both use the same `conversations`, `participants`, and `messages` schema — they differ in who participates and how agents route between them.
## Forum (public chat) [#forum-public-chat]
A forum is a conversation with multiple participants — humans and agent personas:
```
FORUM F
human_alice
human_bob
agent_support (stint p3)
messages visible to all active participants
```
Create via Chat SDK:
```typescript
const room = await chat.createRoom({ title: "support" });
await chat.addMember(room.id, humanActorId);
```
## Monologue (agent internal) [#monologue-agent-internal]
When an agent joins a forum, bootstrap creates a **separate conversation** — the monologue — with typically `system` + the agent actor:
```
MONOLOGUE M (agent support's stint)
system
agent_support (stint p4)
messages visible only to monologue participants
(ingestion copies, tool calls, reasoning, replies)
```
Humans in the forum do **not** see monologue traffic unless egress publishes a `type: "reply"` back to the forum.
## Pairing [#pairing]
```
FORUM F MONOLOGUE M
│ ▲
│ ingestion (per join) │
└────────────────────────────────────┘
│
processor + egress
```
| Layer | Storage | Scope |
| --------------------- | -------------------------------------------------- | ----------------------------------------------- |
| **Canonical** | `participants.metadata.monologueId` on forum stint | One monologue per forum join stint |
| **Optional registry** | Coordinator actor `ctx.state` | `JoinHandle` per forum id |
| **Runtime routing** | Egress `ctx.state.joins` | `forumId → { monologueId, forumParticipantId }` |
`createAgent` does **not** create a monologue. Each `joinConversation` creates a fresh monologue and writes `monologueId` onto the forum participant stint.
## Per-join instances [#per-join-instances]
Each `joinConversation` call spins up:
| Instance | Role |
| -------------------------- | ---------------------------------------------------------------------- |
| **ingestion** | Copy forum messages into monologue (external shadow actors per author) |
| **processor** | LLM + tools on monologue traffic |
| **egress** (one per agent) | Route `type: "reply"` from any monologue stint to correct forum |
## Ingestion without mirror relations [#ingestion-without-mirror-relations]
Ingested copies carry routing metadata in `body.meta`:
```json
{
"meta": {
"sourceConversationId": "forum-id",
"sourceMessageId": "original-message-id",
"sourceActorId": "forum-author-actor-id"
}
}
```
No `message_relations` mirror table — egress uses `sourceConversationId` to pick the forum.
## Compaction swap [#compaction-swap]
When [compaction](/docs/agents/compaction) changes the transcript, the forum participant's `metadata.monologueId` is updated to point at a new monologue. Ingestion resolves the active monologue from participant metadata on each event.
Event `agent.monologue.compacted` notifies egress to update its joins map.
## 1:1 is just a small forum [#11-is-just-a-small-forum]
A direct message between system and agent is a forum with two participants — same bootstrap path as multi-party support rooms.
## Related [#related]
# HTTP API (/docs/conversations/http-api)
The **control plane HTTP API** provides headless chat without embedded agents. It runs on port **8090** by default (`YOKE_CONTROL_PLANE_PORT`).
Agent bootstrap routes (`POST /agents`, `POST /agents/:agentId/join`) live on the **executor portal** HTTP server, not the control plane. See [Agent bootstrap](/docs/agents/bootstrap).
## Base URL [#base-url]
```
http://localhost:8090/v1/tenants/:tenant_id
```
Multi-tenant: replace `:tenant_id` with your tenant slug (e.g. `default`).
## Actors [#actors]
```
POST /actors
GET /actors/:actor_id
```
Create chat identity (human, system, agent persona).
## Conversations [#conversations]
```
POST /conversations
GET /conversations/:conversation_id
```
## Participants [#participants]
```
POST /conversations/:conversation_id/participants
POST /conversations/:conversation_id/participants/:participant_id/leave
GET /conversations/:conversation_id/participants
```
## Messages [#messages]
```
POST /conversations/:conversation_id/messages
GET /conversations/:conversation_id/messages
```
### Transcript pagination [#transcript-pagination]
| Param | Description |
| ------------- | --------------------------------------------- |
| `latest=true` | Most recent N messages (default for chat UIs) |
| `limit` | Page size (1–100, default 50) |
| `before` | Message id cursor — older messages |
| `after` | Message id cursor — newer messages |
| `branchId` | Branch timeline (omit for main) |
Response includes:
```json
{
"messages": [...],
"pagination": {
"hasMoreBefore": true,
"hasMoreAfter": false,
"oldestCursor": "...",
"newestCursor": "..."
}
}
```
## Branches [#branches]
```
POST /conversations/:conversation_id/branches
GET /conversations/:conversation_id/branches
```
## Blobs [#blobs]
```
POST /blobs
GET /blobs/:blob_id
GET /blobs/:blob_id/content
```
Upload raw bytes:
```
POST /v1/tenants/:tenant_id/blobs
Content-Type: image/png
Body:
```
Optional query: `filename`, `width`, `height`, `durationMs`.
Requires `YOKE_STORAGE=s3` and `YOKE_S3_BUCKET`.
## Schedules [#schedules]
```
POST /schedules
GET /schedules
DELETE /schedules/:key
```
Keyed delayed events — see [Schedules](/docs/sdk/schedules).
## Tenant ingress (agents) [#tenant-ingress-agents]
Control plane can proxy agent bootstrap to executor:
```
POST /v1/tenants/:tenant_id/ingress/...
```
SDK helper:
```typescript
import { createAgentBootstrapHttpClient } from "@yoke/sdk";
const agents = createAgentBootstrapHttpClient({
baseUrl: `${controlPlaneUrl}/v1/tenants/${tenantId}/ingress`,
});
```
## SDK HTTP clients [#sdk-http-clients]
```typescript
import { createChatHttpClient, createAgentBootstrapHttpClient } from "@yoke/sdk";
const chat = createChatHttpClient({ baseUrl: controlPlaneUrl, tenantId });
const room = await chat.createRoom({ title: "support" });
```
## Health [#health]
Executor portal (separate port):
```
GET /health → ok
GET /metrics → Prometheus text format
```
## Related [#related]
# Conversations (/docs/conversations)
The **conversation store** provides durable, queryable message history for forums (multi-party chats) and monologues (agent-internal threads). All communication still flows through the **global stream** for delivery; Postgres holds the canonical transcript.
## Principles [#principles]
| Principle | Meaning |
| ------------------------------- | --------------------------------------------------------------------- |
| **Everything is conversations** | No `kind` column, no internal table, no visibility flags |
| **Forum vs monologue** | Product roles — same tables, different participant sets |
| **Actor ≠ participant** | `actors` is identity; `participants` is one stint in one conversation |
| **No inbox table** | Delivery is immediate via stream append + fan-out |
| **Agent = actor instances** | An "agent" is composed actor instances, not one monolithic loop |
| **Opaque body** | `messages.body` is jsonb; runtime does not interpret content |
## Quick start (SDK) [#quick-start-sdk]
```typescript
import { createChatClient, conversationEvents } from "@yoke/sdk";
const chat = createChatClient(ctx.conversations);
const room = await chat.createRoom({ title: "support" });
const human = await chat.createActor({ name: "alice" });
const humanP = await chat.addMember(room.id, human.id);
await chat.send(room.id, "Hello!", { authorParticipantId: humanP.id });
```
Merge `conversationEvents` into your `defineEvents()` for typed stream payloads.
## Stream events [#stream-events]
| Event | When |
| --------------------------------- | -------------------- |
| `conversation.created` | New conversation row |
| `conversation.participant.joined` | Participant added |
| `conversation.participant.left` | `left_at` set |
| `conversation.message.posted` | Message inserted |
| `conversation.branch.created` | Soft fork created |
Payloads carry ids only — handlers read full content from Postgres when needed.
## Agent integration [#agent-integration]
Agents use the conversation store alongside the actor mesh:
1. Create forum (chat room) via Chat SDK
2. `createAgent` → actor + egress instance
3. `joinConversation` → monologue stint + ingestion + processor per join
See [Agents](/docs/agents) for the full bootstrap flow.
## Headless HTTP [#headless-http]
For apps without embedded agents, use the control plane REST API on port **8090**:
```typescript
import { createChatHttpClient } from "@yoke/sdk";
const chat = createChatHttpClient({
baseUrl: "http://localhost:8090",
tenantId: "default",
});
```
See [HTTP API](/docs/conversations/http-api).
## Related [#related]
# Message Types (/docs/conversations/message-types)
Conversation messages store structure in **columns and relations**, not envelope metadata. The runtime normalizes `body` on write — only `parts` and whitelisted `meta` keys are kept.
## messages row [#messages-row]
```
messages
├── type string column (message, reply, reaction, system, edit, …)
├── author_participant_id
├── body jsonb
├── branch_id soft fork only
└── created_at
```
Author is always `author_participant_id`. Message kind is `type`.
## body shape [#body-shape]
```json
{
"parts": [
{ "type": "text", "text": "hello", "format": "plain" }
],
"meta": {
"clientMsgId": "optional-dedup-id",
"sourceConversationId": "forum-id-for-egress-routing",
"sourceMessageId": "original-forum-message-id",
"sourceActorId": "forum-author-actor-id"
}
}
```
Whitelisted `meta` keys: `clientMsgId`, `sourceConversationId`, `sourceMessageId`, `sourceActorId`, plus agent-specific keys like `followUp` and `guardianInject`.
## Part types (v1) [#part-types-v1]
| `type` | Fields |
| ------------- | ------------------------------------------------ |
| `text` | `text`, `format` (`plain` \| `mrkdwn` \| `html`) |
| `image` | `blobId`, `alt?` |
| `audio` | `blobId`, `transcript?` |
| `video` | `blobId`, `caption?`, `thumbBlobId?` |
| `document` | `blobId`, `filename?`, `caption?` |
| `reasoning` | `text` |
| `refusal` | `text` |
| `tool_call` | `id`, `name`, `input` |
| `tool_result` | `toolCallId`, `content`, `isError?` |
Media parts reference `blobs.id` — no inline base64 in jsonb.
## Message type conventions [#message-type-conventions]
| `type` | Use |
| ---------- | ------------------------------------------------------------------------- |
| `message` | Normal line — agent LLM output, tool traffic, forum traffic |
| `reply` | Thread/reply semantics — **not** the user-visible signal for agent output |
| `reaction` | Emoji reaction message |
| `system` | Join/leave notifications |
| `edit` | Replacement content for earlier line |
Tool traffic in monologues uses `type: message` with `tool_call` / `tool_result` parts.
Agent LLM outputs are persisted as `type: message` on the monologue. User visibility is determined by [egress](/docs/agents/egress): if the message is agent-authored and contains text parts, egress forwards that text to the forum.
## message\_relations [#message_relations]
Post via `conversation_post_message` with optional `relations`:
```typescript
await ctx.conversations.postMessage(conversationId, {
parts: [{ type: "text", text: "copied", format: "plain" }],
}, {
type: "message",
relations: [{ relationType: "mirror", targetMessageId: sourceId }],
});
```
Threads are chains of `reply` relations — no parent column on messages.
## blobs [#blobs]
Register metadata, upload bytes out of band (S3):
```
POST /v1/tenants/:tenant_id/blobs
Content-Type: image/png
Body:
```
Requires `YOKE_STORAGE=s3` and `YOKE_S3_BUCKET`.
## Three-actor pipeline (message flow) [#three-actor-pipeline-message-flow]
```
Forum: human posts (type: message)
Ingestion: monologue copy (type: message, meta.source*)
Processor: tool_call + tool_result + agent text (type: message)
Egress: agent text → forum (type: message) — decides user visibility
Bridge: agent text on forum → chat platform (Slack, North, …)
```
Egress listens to all monologue `conversation.message.posted` events. It forwards agent-authored messages that contain **text parts** — not tool\_call-only or tool\_result traffic. See [Egress](/docs/agents/egress).
The processor and inference loop do not decide what reaches the user. They persist monologue messages; egress reads the stream and copies text to the forum.
## Inference alignment [#inference-alignment]
`InferenceMessage` uses the same `body.parts` schema. Processor converts transcript rows to inference messages before `ctx.llm.infer`. See [Inference](/docs/agents/inference).
## Related [#related]
# CLI (/docs/deployment/cli)
The `yoke` CLI is the **authoring and ops** entry point — iterate on an internal package, compile a CLI crate or an embeddable library. It is not the product your users run. Production is [compile `--crate` into your workspace](/docs/getting-started/embed).
## Command overview [#command-overview]
| Command | Purpose |
| ---------------------- | ------------------------------------------------------------------ |
| `yoke dev` | Author hot path — run package from source |
| `yoke start` | Alias for `yoke dev` |
| `yoke compile` | CLI binary crate (`run` / `control-plane` / `worker`, TS baked in) |
| `yoke compile --crate` | Library crate for product workspaces (no binary) |
| `yoke run` | Colocated production — CP + worker, one process |
| `yoke control-plane` | Multi-tenant scheduler + HTTP API |
| `yoke worker start` | Standalone tenant executor |
| `yoke append` | Inject a test event |
| `yoke migrate` | Run database migrations |
| `yoke tenant-create` | Register a tenant row (package ref) |
| `yoke executor start` | Start a tenant-scoped executor (Docker sandbox entrypoint) |
See [Compile & deploy](/docs/deployment/compile) for the full compile/run workflow.
## yoke dev / yoke start [#yoke-dev--yoke-start]
Boot a user package from source (single tenant, no control plane):
```bash
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
yoke dev ./your-package --database "$DATABASE_URL"
```
| Flag | Env | Default | Purpose |
| ------------- | ---------------- | ------------ | -------------------- |
| `package` | | required | Path to user package |
| `--database` | `DATABASE_URL` | | Postgres URL |
| `--storage` | `YOKE_STORAGE` | `pg_sqlite` | `pg_sqlite` or `s3` |
| `--tenant-id` | `YOKE_TENANT_ID` | `default` | Tenant scope |
| `--port` | `YOKE_PORT` | `8080` | Portal HTTP port |
| `--data-dir` | `YOKE_DATA_DIR` | `.yoke-data` | Local runtime data |
## yoke compile [#yoke-compile]
```bash
yoke compile ./your-package -o dist/my-app
yoke compile ./packages/runtime -o ./crates/yoke --crate
```
| Flag | Purpose |
| ---------------- | ---------------------------------------------------- |
| `-o`, `--output` | Output directory for the generated crate |
| `--no-bundle` | Skip `deno bundle` |
| `--crate` | Library crate (no binary); omit for CLI binary crate |
## yoke run [#yoke-run]
Colocated single-tenant — control plane + in-process worker:
```bash
yoke run \
--package ./your-package \
--database "$DATABASE_URL" \
--cp-port 8090
```
| Flag | Env | Default | Purpose |
| ------------- | ------------------------- | ------------ | --------------------------- |
| `--package` | | required | Compiled dir or source tree |
| `--database` | `DATABASE_URL` | | Postgres URL |
| `--cp-port` | `YOKE_CONTROL_PLANE_PORT` | `8090` | Control plane HTTP |
| `--tenant-id` | `YOKE_TENANT_ID` | `default` | Tenant scope |
| `--storage` | `YOKE_STORAGE` | `pg_sqlite` | Storage backend |
| `--data-dir` | `YOKE_DATA_DIR` | `.yoke-data` | Runtime data |
## yoke append [#yoke-append]
Inject test events without writing code:
```bash
yoke append \
--database "$DATABASE_URL" \
--type slack.message.received \
--payload '{"channel":"#general","text":"hello"}'
```
## Control plane [#control-plane]
Start multi-tenant control plane (scheduler + HTTP API):
```bash
yoke control-plane --database "$DATABASE_URL" --port 8090
```
Exposes conversation HTTP API, schedules, tenant management, and tenant ingress proxy. See [HTTP API](/docs/reference/http-api).
## yoke tenant-create [#yoke-tenant-create]
Register a tenant in Postgres before using control plane APIs:
```bash
yoke tenant-create \
--database "$DATABASE_URL" \
--id acme \
--package-ref ./your-package
```
Equivalent to `POST /v1/tenants`. The scheduler uses `package_ref` to resolve the user package when waking a cold executor.
## Worker [#worker]
Standalone tenant executor:
```bash
yoke worker start \
--database "$DATABASE_URL" \
--tenant-id acme \
--package ./your-package
```
## yoke executor start [#yoke-executor-start]
Tenant-scoped executor entrypoint (used inside Docker sandboxes; sets `YOKE_EXECUTOR_MODE`):
```bash
yoke executor start \
--database "$DATABASE_URL" \
--tenant-id acme \
--package-ref ./your-package \
--data-dir /data \
--port 8080
```
| Flag | Env | Default | Purpose |
| --------------- | ------------------ | ----------- | -------------------------------- |
| `--tenant-id` | `YOKE_TENANT_ID` | required | Tenant scope |
| `--package-ref` | `YOKE_PACKAGE_REF` | required | Local path or remote package ref |
| `--data-dir` | `YOKE_DATA_DIR` | `/data` | Runtime data directory |
| `--storage` | `YOKE_STORAGE` | `pg_sqlite` | Storage backend |
| `--port` | `YOKE_PORT` | `8080` | Portal HTTP port |
If `package_ref` is a local path that exists, it is used directly. Otherwise the runtime resolves the ref into `--data-dir/package`.
## Rust crate embedding [#rust-crate-embedding]
`yoke compile --crate` writes a **library** with the package baked in. Your crates depend on that — they do not pass a `yoke.ts` path.
```rust
let rt = my_pkg::start_colocated(database, data_dir, "default", 8090).await?;
let worker = my_pkg::start_worker("default", database, data_dir, 8080).await?;
```
For a standalone executor CLI, `yoke compile` (no `--crate`) emits a binary crate instead.
Full walkthrough: [Embed Yoke in your product](/docs/getting-started/embed).
## Development workflow [#development-workflow]
```bash
# Terminal 1: Postgres
docker compose up -d
# Terminal 2: Dev runtime
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
yoke dev ./your-package --database "$DATABASE_URL"
# Terminal 3: Test event
yoke append \
--database "$DATABASE_URL" \
--type slack.message.received \
--payload '{"channel":"#general","text":"build feature X"}'
```
## Related [#related]
# Compile & deploy (/docs/deployment/compile)
Yoke separates **authoring** from **embedding**:
| Mode | Command | Use when |
| --------------------- | ------------------------------------- | ------------------------------------------------------------------------ |
| **Dev** | `yoke dev` / `yoke start` | Iterating on the internal package from source |
| **Compile (CLI)** | `yoke compile -o dist/cli` | Binary crate: `run` / `control-plane` / `worker` with TS baked into Rust |
| **Compile (library)** | `yoke compile --crate -o crates/yoke` | Library crate for your CLI/server workspace — **no binary** |
| **Run** | `yoke run` | Dev colocated host from a source tree without compiling |
Both compile modes embed the TypeScript package with `include_dir`. The generated crate does **not** take a path to `yoke.ts`. It still depends on `yoke-runtime` / `yoke-core`.
## yoke compile (CLI binary) [#yoke-compile-cli-binary]
Emit a **binary crate** you `cargo build` and run as a Yoke CLI (worker, control-plane, colocated `run`):
```bash
yoke compile ./your-package -o dist/my-app
cd dist/my-app && cargo build --release
./target/release/my_app run
./target/release/my_app control-plane --database "$DATABASE_URL"
./target/release/my_app worker --database "$DATABASE_URL" --tenant-id default
```
| Flag | Purpose |
| ---------------- | ---------------------------------------- |
| `-o`, `--output` | Output directory for the generated crate |
| `--no-bundle` | Copy source only; skip `deno bundle` |
When Deno is available, compile attempts `deno bundle` and embeds `bundle.js`. Otherwise it embeds the copied `yoke.ts` tree (`format: source`). Either way the files live **inside the Rust crate** (`package/` + `include_dir!`).
### Output layout (CLI) [#output-layout-cli]
```
dist/my-app/
Cargo.toml # [[bin]] only — no [lib]
src/main.rs # run / control-plane / worker
package/ # baked in via include_dir
README.md
```
## yoke compile --crate (library) [#yoke-compile---crate-library]
Emit a **library crate** for a product workspace. **No binary.** Other crates (`server`, `cli`, …) depend on it. TypeScript is baked in; those crates never receive a `yoke.ts` path.
```bash
yoke compile ./packages/runtime -o ./crates/yoke --crate
```
```
crates/yoke/
Cargo.toml # [lib] only — no [[bin]]
src/lib.rs # start_colocated, start_worker, materialize_package
package/ # include_dir
README.md
```
```toml
# crates/server/Cargo.toml
my_product_yoke = { path = "../yoke" }
```
```rust
let rt = my_product_yoke::start_colocated(database, data_dir, "default", 8090).await?;
```
When `format` is `bundle`, the host does not need the TypeScript source tree. With `format: source`, Deno/npm resolution still runs inside the embedded libdeno worker after the crate materializes the baked-in files to a temp dir.
## yoke dev [#yoke-dev]
Hot path for package authors — loads source directly, same as `yoke start`:
```bash
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
yoke dev ./your-package --database "$DATABASE_URL"
```
Boots a single tenant executor with portal HTTP on port **8080**. No control plane scheduler — use this while editing `yoke.ts` and actor definitions.
## yoke run [#yoke-run]
Colocated single-tenant — control plane and worker in **one process**, from a **source tree** (not a compiled crate):
```bash
export DATABASE_URL=postgres://...
yoke run \
--package ./your-package \
--database "$DATABASE_URL" \
--cp-port 8090
```
Sets automatically:
| Variable | Value |
| -------------------- | ------------ |
| `YOKE_DEPLOYMENT` | `colocated` |
| `YOKE_SANDBOX` | `in_process` |
| `YOKE_SCALE_TO_ZERO` | `0` |
For a shippable CLI, use `yoke compile` (no `--crate`) and run the generated binary. For a product workspace, use `yoke compile --crate` and call `start_colocated` from your crates.
## Deployment models [#deployment-models]
```bash
yoke compile ./packages/runtime -o ./crates/yoke --crate
```
Depend on `crates/yoke` from `crates/server` and `crates/cli`. Intended production path — [embed](/docs/getting-started/embed).
```bash
yoke compile ./your-package -o dist/my-app
cd dist/my-app && cargo build --release
./target/release/my_app run
```
Single executable with the package baked in. No TypeScript tree on the server when format is `bundle`.
```bash
yoke run --package ./your-package --database "$DATABASE_URL"
```
Authoring host when the `yoke` CLI is on the machine.
## Rust embedding [#rust-embedding]
The intended production path is `yoke compile --crate` plus a dependency from **your** crates. Full walkthrough: [Embed Yoke in your product](/docs/getting-started/embed).
```rust
// your server crate — depends on the generated library, not on a yoke.ts path
let rt = my_package::start_colocated(database, data_dir, "default", 8090).await?;
let yoke = /* Worker::runtime() after start_worker, or conversation_ops via control plane */;
```
The generated library re-exports `YokeRuntime`, `Worker`, `EventInput`, and `start_colocated` / `start_worker`. Product crates should not pass `package_path` to a source checkout.
See [CLI](/docs/deployment/cli) for authoring subcommands and [Environment](/docs/deployment/environment) for variables.
## Related [#related]
# Docker (/docs/deployment/docker)
Docker simplifies local Postgres and provides the foundation for production executor images.
## Local Postgres (compose) [#local-postgres-compose]
```bash
docker compose up -d
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
```
A typical `docker-compose.yml` provides Postgres 16 with `yoke` user/database.
## Run runtime locally (no Docker for binary) [#run-runtime-locally-no-docker-for-binary]
```bash
yoke dev ./your-package \
--database "$DATABASE_URL" \
--storage pg_sqlite
```
## Control plane [#control-plane]
```bash
yoke control-plane \
--database "$DATABASE_URL" \
--port 8090
```
Control plane defaults to port **8090** (`YOKE_CONTROL_PLANE_PORT`). Executor portal HTTP defaults to **8080**. Do not confuse them — the control plane serves conversation and tenant APIs; the executor serves portal routes and `/health`.
## Multi-tenant Docker sandbox [#multi-tenant-docker-sandbox]
The Docker sandbox provider runs the **yoke-executor** OCI image per hot tenant:
```
TenantSchedulerService
→ DockerSandbox.provision()
→ container with YOKE_TENANT_ID, DATABASE_URL, YOKE_PACKAGE_REF
→ executor boots user package
```
Stop when idle: no pending events and no active dispatch.
## Executor image contents [#executor-image-contents]
Standard image includes:
* Rust `yoke-executor` binary
* libdeno embedded runtime
* Embedded Deno worker project
* No user package baked in — mounted or fetched via `YOKE_PACKAGE_REF`
## Example compose layout (conceptual) [#example-compose-layout-conceptual]
```yaml
services:
postgres:
image: postgres:16
environment:
POSTGRES_USER: yoke
POSTGRES_PASSWORD: yoke
POSTGRES_DB: yoke
ports:
- "5432:5432"
control-plane:
build: .
command: yoke control-plane --database postgres://yoke:yoke@postgres:5432/yoke --port 8090
ports:
- "8090:8090"
depends_on:
- postgres
# Executor containers spawned dynamically by scheduler — not static in compose
```
## Environment in containers [#environment-in-containers]
```dockerfile
ENV DATABASE_URL=postgres://yoke:yoke@postgres:5432/yoke
ENV YOKE_STORAGE=pg_sqlite
# or for hosted:
ENV YOKE_STORAGE=s3
ENV YOKE_S3_BUCKET=my-bucket
ENV YOKE_TENANT_ID=default
```
## Volume mounts [#volume-mounts]
| Mount | Purpose |
| ------------------------- | ----------------------------------------- |
| `./your-package:/package` | Dev package hot-reload (requires restart) |
| S3 / object storage | Production package refs |
## Health checks [#health-checks]
Configure Docker/orchestrator health checks against executor portal:
```
GET http://localhost:8080/health
```
## Related [#related]
# Environment Variables (/docs/deployment/environment)
## Required [#required]
| Variable | Purpose |
| -------------- | ------------------------------------------------- |
| `DATABASE_URL` | Postgres connection string |
| `YOKE_STORAGE` | `pg_sqlite` or `s3` — per-instance SQLite backend |
## Storage [#storage]
| Variable | When | Purpose |
| ---------------- | --------- | ------------------------------------------------ |
| `YOKE_STORAGE` | Always | `pg_sqlite` (BYTEA in Postgres) or `s3` |
| `YOKE_S3_BUCKET` | `s3` mode | S3 bucket for actor SQLite + blobs |
| `YOKE_S3_PREFIX` | Optional | Key prefix (default empty) |
| `YOKE_S3_SYNC` | Optional | `lazy` (default) or `strict` fsync before upload |
## Worker / Deno [#worker--deno]
| Variable | Default | Purpose |
| --------------------- | ----------------- | --------------------------------------------------------------- |
| `YOKE_WORKER` | (runtime default) | Worker entry script — rarely overridden when authoring packages |
| `YOKE_WORKER_PROJECT` | (runtime default) | Deno project dir for worker `deno.json` |
## Tenant / package [#tenant--package]
| Variable | Default | Purpose |
| ------------------ | --------- | ----------------------------------- |
| `YOKE_TENANT_ID` | `default` | Tenant scope for all ops |
| `YOKE_PACKAGE_REF` | | S3/git ref to user package (hosted) |
## Horizontal scaling [#horizontal-scaling]
| Variable | Default | Purpose |
| ---------------- | ------------------- | ------------------------------------------ |
| `YOKE_WORKER_ID` | `{hostname}-{uuid}` | Stable worker identity for queue ownership |
## Deployment mode [#deployment-mode]
| Variable | Default | Purpose |
| --------------------------- | ------------- | -------------------------------------------------------------- |
| `YOKE_DEPLOYMENT` | `distributed` | `colocated` or `distributed` — set automatically by `yoke run` |
| `YOKE_SANDBOX` | `docker` | `in_process` (colocated) or Docker sandbox provider |
| `YOKE_SCALE_TO_ZERO` | `1` | `0` disables scale-to-zero (colocated single-tenant) |
| `YOKE_EMBEDDED_PACKAGE_DIR` | temp dir | Override where deploy binaries extract embedded packages |
## Multi-tenant hosted [#multi-tenant-hosted]
| Variable | Purpose |
| ------------------------- | ------------------------------------- |
| `YOKE_LEASE_HOLDER` | Executor lease identity for heartbeat |
| `YOKE_CONTROL_PLANE_PORT` | Control plane HTTP (default **8090**) |
## LLM providers (user package) [#llm-providers-user-package]
Set in Deno env, read by user package at runtime:
| Variable | Provider |
| ------------------- | ------------------------- |
| `OPENAI_API_KEY` | `createOpenAiProvider` |
| `ANTHROPIC_API_KEY` | `createAnthropicProvider` |
## Example — local dev [#example--local-dev]
```bash
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
```
## Example — S3 production [#example--s3-production]
```bash
export DATABASE_URL=postgres://...
export YOKE_STORAGE=s3
export YOKE_S3_BUCKET=my-yoke-state
export YOKE_S3_SYNC=strict
```
## Example — multi-tenant executor [#example--multi-tenant-executor]
```bash
export DATABASE_URL=postgres://...
export YOKE_TENANT_ID=customer-abc
export YOKE_STORAGE=s3
export YOKE_S3_BUCKET=my-yoke-state
export YOKE_S3_PREFIX=customer-abc
export YOKE_PACKAGE_REF=s3://packages/customer-abc/latest
export YOKE_LEASE_HOLDER=executor-pod-xyz
```
## Related [#related]
# Deployment (/docs/deployment)
This section covers running Yoke — as a library inside **your** server (the intended production path), and with the authoring CLI while you write the internal package.
## Deployment models [#deployment-models]
**Intended production path.** Your binary depends on `yoke-runtime` and calls `YokeRuntime::start` or `start_colocated`. Public HTTP is yours. Yoke ports stay private. See [Embed](/docs/getting-started/embed).
`yoke dev` for authoring the internal package; `yoke compile` + deploy binary or `yoke run` when you want the generated host instead of your own. One tenant, colocated CP + worker, Postgres + optional S3.
Always-on Postgres with `tenant_id` partitioning. Compute scales to zero per tenant via scheduler + sandbox provider. Control plane on port **8090**. Still typically wrapped by your control plane, not exposed as "the Yoke product."
## Minimum production requirements [#minimum-production-requirements]
| Component | Requirement |
| -------------- | --------------------------------------------------------------- |
| Postgres | 16+, durable storage, backups |
| `YOKE_STORAGE` | `pg_sqlite` or `s3` (required) |
| Deno | 2.x (embedded at runtime; needed for dev) |
| Rust binary | `yoke` CLI, compiled deploy crate, or `yoke-executor` OCI image |
## Quick production checklist [#quick-production-checklist]
Set
`DATABASE_URL`
and run migrations (automatic on start).
Choose
`YOKE_STORAGE=pg_sqlite`
or
`s3`
with
`YOKE_S3_BUCKET`
.
Compile a library into your workspace (
`yoke compile --crate`
) or a CLI crate (
`yoke compile`
).
Configure portal HTTP ingress (load balancer, TLS).
Enable
`GET /metrics`
scraping for observability.
# Multi-tenant (/docs/deployment/multi-tenant)
Yoke supports **multi-tenant hosted deployments** where many customers share one Postgres cluster (`tenant_id` on every row) while **compute scales to zero** per idle tenant.
"Serverless" here means compute starts and stops on demand — Postgres stays always on.
## Architecture [#architecture]
```
INGRESS (per tenant)
append API · portal HTTP · webhooks
│
▼
Always-on Postgres
stream · queues · leases
│
▼
TenantSchedulerService
(serverful reconcile loop)
│
▼
SandboxProvider (Docker, Modal, Fly, …)
│
▼
yoke-executor microVM per hot tenant
Rust kernel + libdeno + user package
│
▼
S3 (actor SQLite per tenant prefix)
```
## Separation of concerns [#separation-of-concerns]
| Component | Scales to zero? | Role |
| ------------------ | --------------- | ---------------------------------------- |
| **Postgres** | No | Durable stream, queues, registry, leases |
| **Control plane** | No | HTTP API port **8090**, tenant ingress |
| **Scheduler** | No (serverful) | Start/stop decisions on activity |
| **Sandbox pool** | Yes | 0..N executor containers/VMs |
| **Tenant runtime** | Yes | Kernel + Deno for one tenant |
## Tenant lifecycle [#tenant-lifecycle]
```
COLD ──(activity)──► STARTING ──(ready)──► HOT ──(idle)──► STOPPING ──► COLD
```
* **COLD**: No executor. State in Postgres + S3 only.
* **STARTING**: Scheduler acquired lease; sandbox provisioning.
* **HOT**: Fan-out, drain, portal HTTP for tenant.
* **STOPPING**: Flush S3, graceful stop, release lease.
**Stop rule:** keep tenant container running while `pending_events > 0` OR `active_dispatch > 0`.
## Leases [#leases]
`tenant_runtimes` table prevents duplicate executors:
* Scheduler CAS on `lease_holder` + `lease_expires_at`
* Executor heartbeats every 15s via `YOKE_LEASE_HOLDER`
* Expired leases → reconciler stops orphan VMs
## Activity signals [#activity-signals]
| Kind | Source |
| -------------- | ---------------------------------- |
| `StreamAppend` | After durable stream write |
| `PortalHttp` | Before portal route dispatch |
| `QueueDepth` | Reconciler sees pending rows |
| `Cron` | External tick for portal `start()` |
| `Reconcile` | Periodic idle sweep |
## Sandbox provider trait [#sandbox-provider-trait]
| Method | Purpose |
| ----------------- | ---------------------------- |
| `provision(spec)` | Create executor VM/container |
| `destroy(handle)` | Tear down |
| `status(handle)` | Running / failed / gone |
`DockerSandbox` included for local development.
## Executor environment [#executor-environment]
```bash
YOKE_TENANT_ID=customer-abc
DATABASE_URL=postgres://... # shared; queries scoped by tenant_id
YOKE_PACKAGE_REF=s3://...
YOKE_STORAGE=s3
YOKE_S3_PREFIX=customer-abc
```
## Control plane API [#control-plane-api]
```bash
yoke control-plane --database "$DATABASE_URL" --port 8090
```
Headless chat + schedules + tenant ingress proxy. See [HTTP API](/docs/conversations/http-api).
## Related [#related]
# Observability (/docs/deployment/observability)
Yoke exposes **Prometheus text format** metrics and a simple health endpoint on the portal HTTP server.
## Endpoints [#endpoints]
| Endpoint | Response |
| -------------- | --------------------------------- |
| `GET /health` | `ok` — runtime booted and serving |
| `GET /metrics` | Prometheus exposition format |
Both are served on the executor portal port (default `8080`), alongside user-defined portal routes.
Control plane (port **8090**) may expose separate health depending on deployment — check your `control-plane` configuration.
## Metrics [#metrics]
| Metric | Type | Labels | Description |
| --------------------------------- | ------- | ------------- | ----------------------------- |
| `yoke_stream_appends_total` | counter | | Stream append operations |
| `yoke_fanout_enqueues_total` | counter | | Events enqueued via fan-out |
| `yoke_actor_process_total` | counter | | `process()` invocations |
| `yoke_actor_process_errors_total` | counter | | Failed `process()` calls |
| `yoke_queue_dead_letters_total` | counter | | Dead-lettered events |
| `yoke_queue_depth` | gauge | `instance_id` | Per-instance queue depth |
| `yoke_runtime_workers` | gauge | | Workers with recent heartbeat |
## Prometheus scrape config [#prometheus-scrape-config]
```yaml
scrape_configs:
- job_name: yoke
static_configs:
- targets: ["localhost:8080"]
metrics_path: /metrics
```
## Alerting suggestions [#alerting-suggestions]
| Signal | Threshold idea | Meaning |
| -------------------------------------- | --------------------------- | ------------------------------------------------------- |
| `yoke_queue_depth` | Sustained high per instance | Backlogged actor — scale concurrency or split instances |
| `yoke_actor_process_errors_total` rate | Spike | Poison messages or code bugs |
| `yoke_queue_dead_letters_total` | > 0 | Events gave up retrying |
| `yoke_runtime_workers` | \< expected | Worker crash or network partition |
| `/health` down | | Boot failure or process crash |
## Logging [#logging]
Runtime logs to stderr:
* Boot sequence steps and failures
* Deno worker load errors
* `process()` uncaught exceptions (before requeue)
* Portal `start()` loop errors
* Scheduler claim/dispatch (debug level)
Structured logging integration is deployment-specific — wrap the binary with your log collector.
## Tracing [#tracing]
Correlation ids on events (`ctx.correlation`) enable cross-event tracing in external systems. Propagate `correlation` through `ctx.append()` chains and log it in `process()` handlers.
Stream `seq` provides global ordering for post-hoc debugging.
## Multi-tenant monitoring [#multi-tenant-monitoring]
Per-tenant metrics require label filtering at scrape time or separate executor endpoints per tenant. Control plane scheduler metrics (pending tenants, active sandboxes) are deployment-specific.
## Related [#related]
# Scaling (/docs/deployment/scaling)
Multiple Yoke runtime **workers** can share one Postgres database. Each worker drains queues only for actor instances it owns — partitioning compute without partitioning the append stream.
## Worker model [#worker-model]
```
GLOBAL STREAM (shared)
│
async fan-out (all workers)
│
┌─────────────────┼─────────────────┐
▼ ▼ ▼
Worker A Worker B Worker C
instances 1,2 instances 3,4 instances 5,6
drain loops drain loops drain loops
```
| Property | Behavior |
| ------------- | ----------------------------------------------------------- |
| Stream append | Shared — any worker or external client can append |
| Fan-out | All workers share fan-out index; enqueue is global |
| Queue drain | **Partitioned** — only owning worker dispatches `process()` |
## Configuration [#configuration]
```bash
export YOKE_WORKER_ID=worker-us-east-1a
```
Default: `{hostname}-{uuid}` — unique per process.
## Instance assignment [#instance-assignment]
| Event | `worker_id` |
| ------------------------------------ | ----------------------------- |
| `ctx.instances.create()` on worker A | Set to A's `YOKE_WORKER_ID` |
| Boot with `worker_id IS NULL` rows | Claimed by starting worker |
| Portal/HTTP create | Creating executor's worker id |
## Worker registry [#worker-registry]
Workers register in `runtime_workers` and heartbeat every **15 seconds**.
Stale workers (no heartbeat for **60s**) are excluded from `yoke_runtime_workers` gauge. Instance ownership is **not** automatically migrated yet — orphan instances (`worker_id` pointing to dead worker) idle until manually reassigned or a worker claims NULL rows.
## Scaling guidelines [#scaling-guidelines]
| Concern | Guidance |
| ------------------- | -------------------------------------------- |
| CPU-bound actors | Add workers; create instances across workers |
| Single hot instance | Increase `queue.concurrency`, not workers |
| Stream throughput | Postgres write scaling; fan-out is async |
| Stateful actors | SQLite stays with instance on owning worker |
## Multi-tenant vs horizontal [#multi-tenant-vs-horizontal]
| Pattern | Scope |
| -------------------------- | ------------------------------------------------ |
| **Horizontal workers** | Multiple processes, same tenant, shared Postgres |
| **Multi-tenant sandboxes** | One executor per hot tenant, scheduler-managed |
They compose: each tenant executor can run multiple workers or a single worker, depending on your deployment.
## Worker partitioning [#worker-partitioning]
Workers partition **actor instances**, not the append stream. All workers for a tenant share the same global stream; each worker owns a subset of actor instances and drains their queues.
## Metrics [#metrics]
| Metric | Purpose |
| ---------------------- | --------------------------------------- |
| `yoke_runtime_workers` | Active workers with recent heartbeat |
| `yoke_queue_depth` | Per-instance backlog — detect hot spots |
## Related [#related]
# Single-tenant (/docs/deployment/single-tenant)
**Single-tenant** deployment runs one logical tenant (`YOKE_TENANT_ID=default`) against one Postgres database. The intended production shape is **your product binary embedding `yoke-runtime`**, not exposing the `yoke` CLI. See [Embed Yoke in your product](/docs/getting-started/embed).
The CLI `yoke run` / `yoke compile` (CLI crate) path below is the same kernel without a custom host — useful until your server exists, or as an executor sidecar. Product workspaces use `yoke compile --crate` and depend on the library.
## Recommended production path [#recommended-production-path]
**Embed the compiled library in your product** (intended):
```bash
yoke compile ./packages/runtime -o ./crates/yoke --crate
```
```rust
let rt = strum_yoke::start_colocated(database, data_dir, "default", 8090).await?;
// Your axum/actix server is the public API.
```
See [Embed Yoke in your product](/docs/getting-started/embed).
**Generated CLI** (same kernel, no custom `main` yet):
```bash
yoke compile ./my-agents -o dist/my-agents
cd dist/my-agents && cargo build --release
export DATABASE_URL=postgres://...
./target/release/my_agents run
```
See [Compile & deploy](/docs/deployment/compile) for the generated CLI crate and `yoke run` from a source tree.
## Architecture (colocated) [#architecture-colocated]
```
your-product binary
embeds start_colocated / YokeRuntime
→ your public HTTP (users land here)
→ Postgres (stream, queues, instances, conversations)
→ Control plane API :8090 (loopback / private)
→ In-process tenant worker (embedded Deno + internal package)
→ Portal HTTP (executor port — keep private)
→ pg_sqlite BYTEA or S3 actor state
```
Scale-to-zero is **off** in colocated mode — the worker stays hot. `yoke run --package ...` is this architecture without your own `main`.
## Dev setup [#dev-setup]
For local authoring, use `yoke dev` instead of compile:
```bash
docker compose up -d
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
yoke dev ./your-package --database "$DATABASE_URL"
```
## Storage choice [#storage-choice]
Recommended for single-tenant. Actor SQLite serialized as BYTEA in `actor_sqlite` — one Postgres handles everything. No S3 dependency.
Use when SQLite files grow large or you want compute/storage separation. Set `YOKE_S3_BUCKET` and prefer `YOKE_S3_SYNC=strict` for durability.
## Ingress [#ingress]
Expose portal HTTP behind TLS reverse proxy:
* Webhook routes from `definePortal().routes`
* `GET /health` for load balancer checks
* `GET /metrics` for Prometheus
In colocated mode, portal routes are served by the in-process executor. Control plane conversation API is on port **8090**.
## User package deployment [#user-package-deployment]
The TypeScript package is loaded by **your host**. How you ship the host:
| Method | Description |
| ----------------------- | ----------------------------------------------------------------------------------------------------- |
| **Your product crates** | `yoke compile --crate` → depend on the library from CLI/server — [embed](/docs/getting-started/embed) |
| **Generated CLI** | `yoke compile -o dist/app` → `cargo build --release` → ship that binary |
| **Dev source tree** | `yoke dev` / `yoke run --package ./packages/runtime` while authoring |
Package updates require runtime restart to reload definitions (definitions are fixed at boot). Don't expose portal or control-plane ports as the product API.
## Instance data [#instance-data]
Actor instances and queues persist in Postgres across restarts. No additional backup beyond Postgres (+ S3 if used) required for recovery.
## When to move to multi-tenant [#when-to-move-to-multi-tenant]
Consider [multi-tenant](/docs/deployment/multi-tenant) when:
* Hosting many isolated customers on shared infra
* Scale-to-zero per customer compute
* Centralized control plane API on port 8090 with Docker sandbox workers
Single-tenant colocated mode remains valid for dedicated customer VMs or internal tools.
## Related [#related]
# Embed Yoke in your product (/docs/getting-started/embed)
The intended production path is:
1. Write the internal actor package in TypeScript (`packages/runtime` or similar).
2. `yoke compile --crate` emits a **Rust library** with that package baked in (`include_dir`).
3. Your product workspace has **more than one crate** (CLI, server, …). Each depends on that library. Users never see Yoke.
`yoke dev ./packages/runtime` is only for iterating on actors. Do not ship it.
`yoke dev` / `yoke start` exist so you can iterate on the internal actor package. Production is `yoke compile --crate` plus your own crates depending on the generated library.
## Intended shape [#intended-shape]
```
End users
│ your brand, your UX
▼
Your product (CLI crate, server crate, web app)
│
▼
Your crates ← auth, workspaces, billing, public HTTP
│ depend on
▼
generated crate (yoke compile --crate)
├── baked-in package (yoke.ts) ← actors, agents, tools — private
├── yoke-runtime ← kernel
├── Postgres
└── Deno worker
```
A coding harness is the usual example. You would **not** expose filesystem utilities as a Yoke feature to end users. Your server already owns the workspace. Agents inside the compiled package get tools that operate on that workspace. Users open a session in **your** product; your crates call `start_colocated` / `append` on the generated library.
## Product layout (Cargo + JS workspaces) [#product-layout-cargo--js-workspaces]
A product that embeds Yoke is its **own** workspace: a Cargo workspace for Rust and a JS workspace (Turborepo or similar) for TypeScript.
```
your-product/
Cargo.toml # [workspace] members = ["crates/*"]
package.json # npm/pnpm workspaces + turbo
turbo.json
crates/
yoke/ # GENERATED: yoke compile --crate -o crates/yoke
server/ # depends on the generated library
cli/ # depends on the generated library (and/or server)
packages/
runtime/ # internal actor package (Deno, yoke.ts)
deno.json
yoke.ts
web/ # optional UI — turbo pipeline, not Yoke
```
Compile the actor package into a crate other members can depend on:
```bash
yoke compile ./packages/runtime -o ./crates/yoke --crate
```
The generated `Cargo.toml` is a **library** (`[lib]`) only — **no binary**. TypeScript is baked in with `include_dir`; dependents never take a path to `yoke.ts`.
```toml
# crates/server/Cargo.toml
[dependencies]
# crate name comes from packages/runtime/deno.json "name"
my_product_yoke = { path = "../yoke" }
```
```rust
use my_product_yoke::{start_colocated, EventInput};
let rt = start_colocated(database, data_dir, "default", 8090).await?;
// handlers on YOUR server call into the library — they do not proxy Yoke ports
```
CLI and server can both depend on `crates/yoke`. That is why `--crate` emits a library rather than a deploy binary. For a standalone executor CLI (`run` / `control-plane` / `worker`), use `yoke compile` without `--crate`.
## Two surfaces [#two-surfaces]
| Surface | Crate / package | Who calls it | Role |
| ------------------- | ----------------------------- | ---------------------------------- | ------------------------------------------------------------------------------------ |
| **Generated crate** | `yoke compile --crate` output | Your CLI, server, and other crates | `start_colocated`, `start_worker`, `materialize_package`, `append` via `YokeRuntime` |
| **Author SDK** | `@yoke/sdk` | `packages/runtime` only | `defineRuntime`, actors, portals, tools |
`@yoke/sdk` is **not** your public SDK. It is the language the internal package uses. Your public SDK is whatever you design for your product.
The kernel crates (`yoke-runtime`, `yoke-core`) are dependencies of the **generated** crate, not something every product crate has to take a direct dep on. Product crates depend on the compiled package crate.
## Generated library API [#generated-library-api]
The crate written to `crates/yoke` exposes:
| Item | Use from CLI / server crates |
| ------------------------------------------------------ | ------------------------------------------------------------------------------ |
| `start_colocated(database, data_dir, tenant, cp_port)` | Boot CP + worker with the baked-in package |
| `start_worker(tenant, database, data_dir, port)` | Worker only |
| `materialize_package()` | Path to the extracted package tree (if you need `YokeRuntime::start` yourself) |
| `YokeRuntime` / `Worker` / `EventInput` | Re-exported host types |
After `start_colocated` / `start_worker`, drive work from **your** handlers (`append`, `conversation_ops`, `create_actor` on `Worker::runtime()`). Keep portal HTTP and `:8090` on loopback.
`--crate` does **not** emit a binary. `yoke compile` (no `--crate`) writes a separate CLI crate with `run` / `control-plane` / `worker` for a standalone executor.
## Authoring the TypeScript package [#authoring-the-typescript-package]
`packages/runtime/yoke.ts` follows [Your first package](/docs/getting-started/first-package). Coding tools (`defineTool` for `read`, `write`, `shell`) live here because the **agent** needs them — not as user-facing APIs.
Iterate without compiling the crate:
```bash
yoke dev ./packages/runtime --database "$DATABASE_URL"
```
A turbo task on that package can wrap `yoke compile --crate -o ../../crates/yoke`.
Product crates should depend on the compiled library from `yoke compile --crate`, not on a loose `yoke.ts` folder at runtime.
## What this is not [#what-this-is-not]
| Don't | Do instead |
| --------------------------------------------------------------------------- | ------------------------------------------------------------------------------------- |
| Ship `yoke start ./package` as the product | `yoke compile --crate` and depend on the library from your crates |
| One product `main.rs` that is the only crate | Cargo workspace: generated yoke crate + CLI + server (at least) |
| Give end users `@yoke/sdk` or Yoke HTTP | Give them your CLI / web / API |
| Put product UX on portal routes and call that the app | Keep portal HTTP private; your crates own public HTTP |
| Add product features (billing, workspaces, FS sandboxes) to the Yoke kernel | Keep them in your crates; promote a primitive to Yoke only when many products need it |
## Next steps [#next-steps]
# Your First Package (/docs/getting-started/first-package)
A **user package** is a folder of TypeScript that exports a single `defineRuntime()` definition. The Yoke kernel loads it in an embedded Deno worker at boot, registers actor and portal definitions, and runs instances that react to events on the global stream.
This package is **internal to your product**. `yoke compile --crate` bakes it into a Rust library your server depends on — it does not take a filesystem path to `yoke.ts`. End users never run `yoke.ts` or `@yoke/sdk` — see [Embed Yoke in your product](/docs/getting-started/embed).
This page walks through the anatomy of a minimal package — a Slack portal, a `router` actor, and a `coder` actor.
## Package layout [#package-layout]
Inside your product the TypeScript the kernel loads lives in the JS workspace. You compile it into a crate the rest of the Cargo workspace depends on — see [Embed Yoke in your product](/docs/getting-started/embed).
```
your-product/
├── Cargo.toml # Rust workspace — crates/cli, crates/server, crates/yoke
├── package.json # JS workspace (Turborepo)
├── turbo.json
├── crates/
│ ├── yoke/ # generated by yoke compile --crate
│ ├── server/
│ └── cli/
└── packages/
└── runtime/ # internal — yoke.ts the compiler bakes in
├── deno.json
├── yoke.ts
├── actors/
└── portals/
```
While learning, you can author just the inner package folder (`./your-package`) without the full product workspace. Add the Cargo workspace when you [embed](/docs/getting-started/embed).
The worker always loads `yoke.ts` at the **package** root (`package_path`). The default export must be a `defineRuntime()` builder (with or without `.build()`).
## deno.json [#denojson]
Your package is a Deno project. Import `@yoke/sdk` from JSR and declare any npm dependencies:
```json
{
"imports": {
"@yoke/sdk": "jsr:@yoke/sdk@^0.1.0"
}
}
```
The embedded worker resolves `deno.json`, import maps, npm packages, and `node:` builtins when the runtime starts.
## defineEvents — typed event schemas [#defineevents--typed-event-schemas]
Events are the only communication channel between actors. Declare payload schemas once with Zod:
```typescript
import { defineEvents, z } from "@yoke/sdk";
const events = defineEvents({
"slack.message.received": z.object({
channel: z.string(),
text: z.string(),
user: z.string().optional(),
}),
"work.assigned": z.object({
text: z.unknown().optional(),
role: z.string(),
lang: z.string().optional(),
}),
"code.produced": z.object({
code: z.string(),
}),
});
```
`defineEvents()` returns a typed event map used by `ctx.append()` and `process()` for compile-time payload narrowing. The worker validates payloads at append time against these schemas.
## defineRuntime — the builder [#defineruntime--the-builder]
`defineRuntime()` is a fluent builder that registers everything the kernel needs:
```typescript
export default defineRuntime()
.events(events)
.bootstrap(async (ctx) => {
await ctx.instances.create("router", {
id: "router",
config: { channels: ["#general"] },
filter: { type: "slack.message.received" },
});
await ctx.instances.create("coder", {
id: "coder",
config: { model: "claude-sonnet", lang: "typescript" },
});
})
.portals(({ events }) => { /* ... */ })
.actors(({ events, portals }) => { /* ... */ })
.build();
```
| Method | Purpose |
| ------------------------ | -------------------------------------------------------------- |
| `.events(map)` | Register Zod schemas per event type |
| `.bootstrap(fn)` | One-shot boot hook — provision instances with stable ids |
| `.portals(fn)` | Register portal definitions (singleton ingress/egress) |
| `.actors(fn)` | Register actor definitions (reactive processors) |
| `.agents(fn)` | Register high-level agent definitions (expands to actor stack) |
| `.chatPlatforms(fn)` | Register chat platforms (expands to portal + bridge actor) |
| `.inferenceStack(stack)` | Optional LLM routing for agent processors |
| `.build()` | Produce the `RuntimeDefinition` the worker loads |
## defineFor — scoped definitions [#definefor--scoped-definitions]
Inside `.portals()` and `.actors()`, use `defineFor(events)` or `defineFor(events, portals)` so event types and portal interfaces flow into actor `process()` handlers without repeating the `events` field:
```typescript
.portals(({ events }) => {
const { portal } = defineFor(events);
return { slack: portal({ /* ... */ }) };
})
.actors(({ events, portals }) => {
const { actor } = defineFor(events, portals);
return {
router: actor({ /* ... */ }),
coder: actor({ /* ctx.portals.slack is typed */ }),
};
})
```
## definePortal — connect to the outside world [#defineportal--connect-to-the-outside-world]
Portals are **not** actors. They connect the runtime to external systems — HTTP webhooks, WebSockets, polling loops — and optionally expose a typed interface for actors to call:
```typescript
type SlackApi = {
sendMessage(channel: string, text: string): Promise;
};
slack: portal({
configSchema: z.object({ botToken: z.string() }),
interface: {
async sendMessage(channel, text) {
console.log(`slack: ${channel} <- ${text}`);
},
} satisfies SlackApi,
routes: [
{
method: "POST",
path: "/webhooks/slack",
handler: async (req, ctx) => {
const body = JSON.parse(new TextDecoder().decode(req.body ?? new Uint8Array()));
await ctx.append({
type: "slack.message.received",
payload: {
channel: body.channel ?? "#general",
text: body.text ?? "",
},
});
return { status: 200, body: "ok" };
},
},
],
}),
```
Portals have **no queues** and **no filters**. They append events to the stream (ingress) and expose methods actors call via `ctx.portals..*` (egress).
## defineActor — reactive event processors [#defineactor--reactive-event-processors]
Actors are stateful, queued handlers. Each definition requires `configSchema`, `stateSchema`, and `process`:
```typescript
router: actor({
configSchema: z.object({ channels: z.array(z.string()) }),
stateSchema: z.object({ lastSeq: z.number() }),
queue: { mode: "fifo", concurrency: 1 },
async process(ctx, event) {
if (event.type === "slack.message.received") {
await ctx.append({
type: "work.assigned",
payload: {
text: event.payload.text,
role: "coder",
lang: "typescript",
},
});
}
await ctx.state.update((s) => ({ ...s, lastSeq: event.seq }));
},
}),
```
### Function filters [#function-filters]
When an instance needs logic beyond declarative filters, provide `filterFromConfig`:
```typescript
coder: actor({
configSchema: z.object({ model: z.string(), lang: z.string() }),
stateSchema: z.object({ jobsCompleted: z.number() }),
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
async process(ctx, event) {
await ctx.append({ type: "code.produced", payload: { code: "// ..." } });
await ctx.portals.slack.sendMessage("#general", "Code ready");
await ctx.state.update((s) => ({
...s,
jobsCompleted: s.jobsCompleted + 1,
}));
},
}),
```
Function filters are **not serializable**. On restart, the runtime rebuilds them from persisted instance config via `filterFromConfig`. Without it, function-filter instances survive in Postgres but receive no events until recreated.
## Context inside process() [#context-inside-process]
Inside `process()`, `ctx` provides the full runtime surface:
| API | Purpose |
| ------------------------ | -------------------------------------------------- |
| `ctx.config` | Instance config (validated against `configSchema`) |
| `ctx.state` | Typed durable document (`get`, `set`, `update`) |
| `ctx.sqlite` | Raw SQL on the same per-instance database |
| `ctx.append()` | Publish events to the global stream |
| `ctx.portals.*` | Call portal interface methods |
| `ctx.instances.create()` | Spawn new actor instances |
| `ctx.correlation` | Group related events into work chains |
| `ctx.schedules` | Keyed delayed events |
| `ctx.conversations` | Conversation store CRUD (when enabled) |
See [Context API](/docs/sdk/context-api) for the complete reference.
## Loading the package [#loading-the-package]
**Production:** `yoke compile --crate -o crates/yoke` bakes this folder into a **library crate**. Your CLI and server depend on that crate — they never take a filesystem path to `yoke.ts`. See [embed](/docs/getting-started/embed).
**Authoring:** the CLI loads the same folder so you can iterate without compiling:
```bash
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
yoke dev ./your-package \
--database "$DATABASE_URL" \
--storage pg_sqlite
```
The CLI boots Postgres migrations, loads your package in Deno, ensures portal singletons, runs `bootstrap()` if defined, and begins actor drain loops. It is a development host, not the product.
Create `./your-package` with `deno.json` and `yoke.ts` (see [Quick start](/docs/getting-started/quick-start) for a full inline example).
Edit `yoke.ts` — add events, actors, and portals.
Add `.bootstrap()` to create instances programmatically with stable ids.
Run `yoke dev ./your-package` and append test events with `yoke append`.
## Next steps [#next-steps]
# Getting Started (/docs/getting-started)
Yoke is an **embeddable runtime** for always-on agentic systems. You build a product — a coding harness, an assistant, a pipeline — and your **server** runs Yoke. End users never see Yoke.
This section covers that host model, then the local tools you use while writing the internal actor package.
## What you are building [#what-you-are-building]
Two layers, one product:
```
Your product server (Rust) ← public HTTP, CLI, workspaces, auth
embeds yoke-runtime
→ Postgres (stream, instances, queues, conversations)
→ Deno worker
→ your internal package (defineRuntime / process)
```
The **internal package** is a folder of TypeScript that exports `defineRuntime()`. The kernel loads it in an embedded Deno worker. That package is implementation detail. The `yoke` CLI (`yoke dev ./your-package`) is how you iterate on it — not how customers run your product.
Yoke is **not** an LLM framework and **not** a coding harness. An actor is any stateful async event processor. Your product decides what users see; Yoke runs the event-driven mesh underneath.
## Recommended path [#recommended-path]
Read [Embed Yoke in your product](/docs/getting-started/embed) — host vs internal package, embed API vs `@yoke/sdk`.
Complete [Installation](/docs/getting-started/installation) — Postgres, Rust toolchain, Deno 2, and `YOKE_STORAGE`.
Follow [Quick start](/docs/getting-started/quick-start) to build a minimal package and see stream fan-out locally.
Read [Your first package](/docs/getting-started/first-package) to structure the TypeScript the kernel loads.
Learn the [mental model](/docs/concepts/mental-model), then put `YokeRuntime::start` in your server.
## Minimal package shape [#minimal-package-shape]
A typical learning package is a Slack portal, a `router` actor, and a `coder` actor — enough to see ingress, fan-out, and egress without building a full product host. Real products wrap a package like this (or a full agent mesh) inside their own server. See [Quick start](/docs/getting-started/quick-start) for inline code you can copy into `./your-package`.
# Installation (/docs/getting-started/installation)
Yoke is a Rust crate you embed in **your** server. It boots an in-process Deno worker for your internal TypeScript package and uses Postgres as the durable runtime store. This page covers what the host machine needs.
The `yoke` CLI is the same kernel, packaged as an authoring tool so you can load a package without writing a host first. See [Embed Yoke in your product](/docs/getting-started/embed).
## Prerequisites [#prerequisites]
| Requirement | Version | Notes |
| ------------ | ------- | ------------------------------------------------------------------------------- |
| **Rust** | stable | Build tooling for compile output and embed workflows |
| **Deno** | 2.x | Used by the embedded worker for TypeScript transpilation and npm/jsr resolution |
| **Postgres** | 16+ | Compulsory runtime storage — stream, instances, queues |
| **yoke CLI** | latest | Authoring commands (`yoke dev`, `yoke compile`, `yoke append`) |
Postgres is required. There is no single-node mode without a database. Actor SQLite state is stored either in Postgres (`pg_sqlite`) or S3 (`s3`), not as standalone local files you manage directly.
## Start Postgres with Docker Compose [#start-postgres-with-docker-compose]
Use Docker Compose (or any Postgres 16+ instance) for local development:
```bash
docker compose up -d
```
Default connection string:
```bash
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
```
For local single-tenant development you only need Postgres. Multi-tenant deployments add a control plane service on port **8090**.
## Install the yoke CLI [#install-the-yoke-cli]
Install the `yoke` CLI for your platform, then verify:
```bash
yoke --help
```
During development you can also build the CLI from source if you are working on the runtime itself.
## Required environment variables [#required-environment-variables]
| Variable | Required | Default | Purpose |
| ---------------- | -------- | ------------ | -------------------------------------------------- |
| `DATABASE_URL` | Yes | — | Postgres connection string |
| `YOKE_STORAGE` | Yes | `pg_sqlite` | Actor SQLite backend: `pg_sqlite` or `s3` |
| `YOKE_PORT` | No | `8080` | HTTP port for portal routes, `/health`, `/metrics` |
| `YOKE_DATA_DIR` | No | `.yoke-data` | Local cache directory (S3 mode, worker metadata) |
| `YOKE_TENANT_ID` | No | `default` | Tenant id for row isolation |
Minimal local setup:
```bash
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
```
When `YOKE_STORAGE=s3`, actor SQLite databases are persisted to S3:
| Variable | Purpose |
| ---------------- | -------------------------------------------------------------------------- |
| `YOKE_S3_BUCKET` | S3 bucket for per-instance `state.db` objects |
| `YOKE_S3_PREFIX` | Key prefix (default empty) |
| `YOKE_S3_SYNC` | `lazy` (default) or `strict` — see [Durability](/docs/concepts/durability) |
Objects are stored at `{prefix}/instances/{instance_id}/state.db`. A local cache directory is used only for rusqlite execution.
| Variable | Purpose |
| ---------------- | ------------------------------------------------------------------------------- |
| `YOKE_WORKER_ID` | Stable worker identity for horizontal scaling (defaults to `{hostname}-{uuid}`) |
Multiple workers can share one Postgres database; each drains queues only for instances it owns.
Advanced deployments may override the embedded worker entry script and Deno project directory via `YOKE_WORKER` and `YOKE_WORKER_PROJECT`. Defaults are set by the runtime and rarely need changing when authoring packages.
## CLI flags [#cli-flags]
The `yoke dev` command accepts the same settings as environment variables:
```bash
yoke dev ./your-package \
--database "$DATABASE_URL" \
--storage pg_sqlite \
--port 8080 \
--data-dir .yoke-data \
--tenant-id default
```
Run migrations explicitly if needed:
```bash
yoke migrate --database "$DATABASE_URL"
```
`yoke dev` runs migrations automatically on boot.
## User package dependencies [#user-package-dependencies]
Your package needs a `deno.json` with an import map to `@yoke/sdk`:
```json
{
"name": "your-package",
"imports": {
"@yoke/sdk": "jsr:@yoke/sdk@^0.1.0",
"zod": "npm:zod@^4.5.4"
}
}
```
Import `@yoke/sdk` from JSR in published packages. During local development you may use a path import if you are co-developing the SDK.
## Verify the install [#verify-the-install]
Confirm Postgres is reachable:
```bash
psql "$DATABASE_URL" -c 'SELECT 1'
```
Build and start the runtime (see [Quick start](/docs/getting-started/quick-start)):
```bash
yoke dev ./your-package \
--database "$DATABASE_URL" \
--storage pg_sqlite
```
Hit the health endpoint on the portal HTTP port:
```bash
curl http://localhost:8080/health
```
`pg_sqlite` stores serialized SQLite files in the Postgres `actor_sqlite` table as **BYTEA** blobs — it is not a Postgres extension. Both stream metadata and actor state share the same database connection in this mode.
# Quick Start (/docs/getting-started/quick-start)
This walkthrough uses a **minimal inline package** — a Slack portal, a `router` actor, and a `coder` actor — so you can see stream fan-out without writing a host yet. Production still means [embedding the kernel in your server](/docs/getting-started/embed); `yoke dev` is not the product interface.
By the end you will have a live runtime, an appended `slack.message.received` event, and actors that react through the global stream.
## What your package does [#what-your-package-does]
The package wires three pieces together:
1. **Slack portal** — HTTP webhook ingress and a typed `sendMessage` interface
2. **Router actor** — listens for `slack.message.received`, emits `work.assigned`
3. **Coder actor** — listens for `work.assigned` (function filter on `lang`), emits `code.produced`, calls `ctx.portals.slack.sendMessage`
## Step 1 — Prerequisites [#step-1--prerequisites]
Start Postgres and set environment variables (see [Installation](/docs/getting-started/installation)):
```bash
docker compose up -d
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
```
Install the `yoke` CLI (see [Installation](/docs/getting-started/installation)).
## Step 2 — Create your package [#step-2--create-your-package]
Create a folder `./your-package` with `deno.json` and `yoke.ts`:
```json
// your-package/deno.json
{
"imports": {
"@yoke/sdk": "jsr:@yoke/sdk@^0.1.0"
}
}
```
```typescript
// your-package/yoke.ts
import { defineEvents, defineFor, defineRuntime, z } from "@yoke/sdk";
const events = defineEvents({
"slack.message.received": z.object({
channel: z.string(),
text: z.string(),
}),
"work.assigned": z.object({
text: z.unknown().optional(),
role: z.string(),
lang: z.string().optional(),
}),
"code.produced": z.object({ code: z.string() }),
});
export default defineRuntime()
.events(events)
.bootstrap(async (ctx) => {
await ctx.instances.create("router", {
id: "router",
config: { channels: ["#general"] },
filter: { type: "slack.message.received" },
});
await ctx.instances.create("coder", {
id: "coder",
config: { model: "claude-sonnet", lang: "typescript" },
});
})
.portals(({ events }) => {
const { portal } = defineFor(events);
return {
slack: portal({
configSchema: z.object({ botToken: z.string().optional() }),
interface: {
async sendMessage(channel, text) {
console.log(`slack: ${channel} <- ${text}`);
},
},
routes: [
{
method: "POST",
path: "/webhooks/slack",
handler: async (req, ctx) => {
const body = JSON.parse(
new TextDecoder().decode(req.body ?? new Uint8Array()),
);
await ctx.append({
type: "slack.message.received",
payload: {
channel: body.channel ?? "#general",
text: body.text ?? "",
},
});
return { status: 200, body: "ok" };
},
},
],
}),
};
})
.actors(({ events, portals }) => {
const { actor } = defineFor(events, portals);
return {
router: actor({
configSchema: z.object({ channels: z.array(z.string()) }),
stateSchema: z.object({ lastSeq: z.number() }),
queue: { mode: "fifo", concurrency: 1 },
async process(ctx, event) {
if (event.type === "slack.message.received") {
await ctx.append({
type: "work.assigned",
payload: {
text: event.payload.text,
role: "coder",
lang: "typescript",
},
});
}
await ctx.state.update((s) => ({ ...s, lastSeq: event.seq }));
},
}),
coder: actor({
configSchema: z.object({ model: z.string(), lang: z.string() }),
stateSchema: z.object({ jobsCompleted: z.number() }),
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
queue: { mode: "fifo", concurrency: 1 },
async process(ctx, event) {
await ctx.append({
type: "code.produced",
payload: { code: "// generated" },
});
await ctx.portals.slack.sendMessage("#general", "Code ready");
await ctx.state.update((s) => ({
...s,
jobsCompleted: s.jobsCompleted + 1,
}));
},
}),
};
})
.build();
```
See [Your first package](/docs/getting-started/first-package) for a deeper walkthrough of each piece.
## Step 3 — Start the runtime [#step-3--start-the-runtime]
```bash
yoke dev ./your-package \
--database "$DATABASE_URL" \
--storage pg_sqlite
```
(`yoke start` is an alias for `yoke dev`. For production compile/deploy, see [Compile & deploy](/docs/deployment/compile).)
On boot the runtime:
1. Runs Postgres migrations
2. Spawns the embedded Deno worker and loads `yoke.ts`
3. Ensures portal singletons and runs `bootstrap()` (creates router/coder instances)
4. Starts fan-out and actor drain loops
5. Registers portal HTTP routes (including `POST /webhooks/slack`)
Leave this process running in a terminal. On success you should see the Deno worker load and the runtime reach a ready state (health check returns `ok`).
## Step 4 — Append a test event [#step-4--append-a-test-event]
In a second terminal, use the CLI `append` subcommand:
```bash
yoke append \
--database "$DATABASE_URL" \
--storage pg_sqlite \
--type slack.message.received \
--payload '{"channel":"#general","text":"hello"}'
```
The command prints the assigned stream position:
```
appended seq=1 id=
```
Alternatively, POST to the portal webhook while the runtime is running:
```bash
curl -X POST http://localhost:8080/webhooks/slack \
-H 'Content-Type: application/json' \
-d '{"channel":"#general","text":"hello from webhook"}'
```
## Step 5 — Watch actors process [#step-5--watch-actors-process]
In the runtime terminal you should see processing unfold:
1. **Router** receives `slack.message.received` (declarative filter from instance creation)
2. Router appends `work.assigned` with `role: "coder"` and `lang: "typescript"`
3. **Coder** matches via its function filter (`filterFromConfig`)
4. Coder appends `code.produced` and calls `ctx.portals.slack.sendMessage("#general", "Code ready")`
Expected console output from the coder path:
```
slack: #general <- Code ready
```
## Event flow diagram [#event-flow-diagram]
```mermaid
sequenceDiagram
participant CLI as CLI / Webhook
participant Stream as Global Stream
participant Router as router instance
participant Coder as coder instance
participant Slack as slack portal
CLI->>Stream: append slack.message.received
Stream->>Router: fan-out to queue
Router->>Stream: append work.assigned
Stream->>Coder: fan-out to queue
Coder->>Stream: append code.produced
Coder->>Slack: sendMessage("#general", "Code ready")
```
## Inspect runtime state [#inspect-runtime-state]
| Endpoint | Purpose |
| -------------- | ----------------------------------------------------------------------- |
| `GET /health` | Liveness check — returns `ok` |
| `GET /metrics` | Prometheus metrics (`yoke_stream_appends_total`, `yoke_queue_depth`, …) |
Check queue depth per instance:
```bash
curl -s http://localhost:8080/metrics | grep yoke_queue_depth
```
## Stop the runtime [#stop-the-runtime]
Press `Ctrl+C` in the runtime terminal. The runtime shuts down the Deno worker and flushes storage backends.
The `append` CLI command starts a minimal runtime, appends one event, and stops. For sustained processing while authoring, use `yoke dev` with your package directory. To ship, [embed the kernel](/docs/getting-started/embed) in your server.
## Next steps [#next-steps]
* Read [Embed Yoke in your product](/docs/getting-started/embed) to put this kernel inside your own server
* Read [Your first package](/docs/getting-started/first-package) to understand the `yoke.ts` source in more detail
* Learn how instances are created in [Instances](/docs/concepts/instances)
* Learn the mental model in [Concepts](/docs/concepts/mental-model)
# Event Catalog (/docs/reference/event-catalog)
Event types use an **open namespace** — register schemas with `defineEvents()` for validation and typing. This catalog lists events emitted by the runtime and SDK packages.
## Runtime system events [#runtime-system-events]
| Type | When | Payload |
| -------------------- | -------------------- | ----------------- |
| `runtime.started` | Boot completes | (minimal) |
| `instance.created` | Instance provisioned | instance metadata |
| `instance.destroyed` | Instance torn down | instance id |
## Conversation events [#conversation-events]
From `conversationEvents` in `@yoke/sdk`:
| Type | When |
| ---------------------------------- | ---------------------------- |
| `conversation.actor.created` | Actor row created |
| `conversation.created` | Conversation row created |
| `conversation.participant.joined` | Participant added |
| `conversation.participant.left` | `left_at` set |
| `conversation.participant.updated` | Participant metadata updated |
| `conversation.message.posted` | Message inserted |
| `conversation.branch.created` | Soft fork created |
### conversation.message.posted payload [#conversationmessageposted-payload]
```typescript
{
conversationId: string;
messageId: string;
messageType: string;
authorParticipantId?: string | null;
branchId?: string | null;
relations?: Array<{
id: string;
messageId: string;
relationType: string;
targetMessageId: string;
createdAt: number;
}>;
}
```
Ids only — read full content from Postgres.
## Chat platform events [#chat-platform-events]
From `chatPlatformEvents` in `@yoke/sdk` (used by `defineChatPlatform` / bridge actors):
| Type | When |
| -------------------------------- | ----------------------------------------------------- |
| `chat.platform.message.received` | Inbound message parsed from external platform webhook |
| `chat.platform.message.sent` | Outbound message delivered to external platform |
| `chat.platform.channel.linked` | External channel linked to a Yoke conversation |
### chat.platform.message.received [#chatplatformmessagereceived]
```typescript
{
platform: string;
chatPlatformName: string;
externalMessageId: string;
externalChannelId: string;
externalThreadId?: string;
externalUserId: string;
externalUserDisplayName?: string;
body: MessageBody;
timestamp: number;
}
```
## Agent events [#agent-events]
From `agentEvents` in `@yoke/sdk`:
| Type | When |
| --------------------------- | --------------------------------------------------- |
| `agent.forum.joined` | After `joinConversation` — egress updates joins map |
| `agent.monologue.compacted` | After compaction monologue swap |
| `assistant.follow_up_due` | Follow-up schedule fired |
| `guardian.run` | Guardian runner spawned |
### agent.forum.joined [#agentforumjoined]
```typescript
{
agentActorId: string;
monologueId: string;
monologueParticipantId: string;
forumId: string;
forumParticipantId: string;
}
```
### agent.monologue.compacted [#agentmonologuecompacted]
```typescript
{
agentActorId: string;
forumId: string;
forumParticipantId: string;
previousMonologueId: string;
monologueId: string;
monologueParticipantId: string;
}
```
## Tutorial package events [#tutorial-package-events]
From the [Quick start](/docs/getting-started/quick-start) minimal package:
| Type | Publisher |
| ------------------------ | ------------ |
| `slack.message.received` | slack portal |
| `work.assigned` | router actor |
| `code.produced` | coder actor |
## Multi-portal workflow events [#multi-portal-workflow-events]
Common patterns when multiple portals feed a coordinator actor:
| Type | Stage |
| ------------------------ | -------------- |
| `ingress.shopify.order` | Shopify portal |
| `ingress.stripe.payment` | Stripe portal |
| `ingress.forwarded` | deduper actor |
| `case.opened` | ingress-router |
| `case.payment_received` | case actors |
| `case.ready_to_fulfill` | coordinator |
| `case.fulfilled` | fulfillment |
| `case.resolved` | coordinator |
## Registering in your package [#registering-in-your-package]
```typescript
import { defineEvents, conversationEvents, agentEvents } from "@yoke/sdk";
const events = defineEvents({
...conversationEvents,
...agentEvents,
"myapp.custom": z.object({ id: z.string() }),
});
export default defineRuntime().events(events) /* ... */;
```
## Filter examples [#filter-examples]
```typescript
// All conversation messages in a room
{ type: "conversation.message.posted", where: { "payload.conversationId": forumId } }
// Agent replies on monologue
{ type: "conversation.message.posted", where: {
"payload.conversationId": monologueId,
"payload.messageType": "reply",
}}
// Work assignment
{ type: "work.assigned", where: { "payload.lang": "typescript" } }
```
## Related [#related]
# HTTP API (/docs/reference/http-api)
Yoke exposes HTTP on two logical servers in multi-tenant deployments:
| Server | Default port | Purpose |
| ------------------- | ------------ | ------------------------------------------------ |
| **Executor portal** | 8080 | User portal routes, `/health`, `/metrics` |
| **Control plane** | **8090** | Conversations, actors, schedules, tenant ingress |
## Executor — system routes [#executor--system-routes]
```
GET /health → ok
GET /metrics → Prometheus text format
```
## Executor — portal routes [#executor--portal-routes]
User-defined in `definePortal().routes`:
```typescript
{ method: "POST", path: "/webhooks/slack", handler: async (req, ctx) => { ... } }
```
Path parameters parsed from `req.path` (e.g. `/agents/:agentId/join`).
### Agent bootstrap routes [#agent-bootstrap-routes]
Register these on an executor portal when exposing agent creation from HTTP:
```
POST /agents
POST /agents/:agentId/join
```
**Not** `/rooms` — forums are conversations created via Chat SDK or control plane. See [Agent bootstrap](/docs/agents/bootstrap).
## Control plane — base path [#control-plane--base-path]
```
/v1/tenants/:tenant_id
```
Start with:
```bash
yoke control-plane --port 8090
```
### Tenants [#tenants]
Register and inspect tenants; append events to wake cold executors.
```
POST /v1/tenants
GET /v1/tenants/:tenant_id
POST /v1/tenants/:tenant_id/events
```
**Create tenant**
```bash
curl -X POST http://localhost:8090/v1/tenants \
-H 'content-type: application/json' \
-d '{"id":"acme","packageRef":"./your-package"}'
```
Response:
```json
{
"id": "acme",
"packageRef": "./your-package",
"runtimeStatus": "cold"
}
```
**Get tenant**
```bash
curl http://localhost:8090/v1/tenants/acme
```
Returns `packageRef` and `runtimeStatus` (`cold`, `starting`, `hot`, …).
**Append event**
```bash
curl -X POST http://localhost:8090/v1/tenants/acme/events \
-H 'content-type: application/json' \
-d '{"type":"slack.message.received","payload":{"channel":"#general","text":"hello"}}'
```
Response:
```json
{ "seq": 1, "id": "" }
```
Appends to the tenant stream and signals the scheduler (may wake a cold executor).
### Actors [#actors]
```
POST /actors
GET /actors/:actor_id
```
### Conversations [#conversations]
```
POST /conversations
GET /conversations/:conversation_id
```
### Participants [#participants]
```
POST /conversations/:conversation_id/participants
POST /conversations/:conversation_id/participants/:participant_id/leave
GET /conversations/:conversation_id/participants
```
### Messages [#messages]
```
POST /conversations/:conversation_id/messages
GET /conversations/:conversation_id/messages
```
Query params: `latest`, `limit`, `before`, `after`, `branchId`.
### Branches [#branches]
```
POST /conversations/:conversation_id/branches
GET /conversations/:conversation_id/branches
```
### Blobs [#blobs]
```
POST /blobs
GET /blobs/:blob_id
GET /blobs/:blob_id/content
```
Requires `YOKE_STORAGE=s3` and `YOKE_S3_BUCKET`.
### Schedules [#schedules]
```
POST /schedules
GET /schedules
DELETE /schedules/:key
```
### Tenant ingress (agent bootstrap proxy) [#tenant-ingress-agent-bootstrap-proxy]
```
POST /v1/tenants/:tenant_id/ingress/*
```
Proxies to tenant executor for `createAgent` / `joinConversation` when executor is hot.
## SDK HTTP clients [#sdk-http-clients]
```typescript
import { createChatHttpClient, createAgentBootstrapHttpClient } from "@yoke/sdk";
const chat = createChatHttpClient({ baseUrl: "http://localhost:8090", tenantId: "default" });
const agents = createAgentBootstrapHttpClient({
baseUrl: "http://localhost:8090/v1/tenants/default/ingress",
});
```
## Response formats [#response-formats]
* JSON bodies for CRUD endpoints
* Raw bytes for blob content (`GET .../blobs/:id/content`)
* Prometheus text for `/metrics`
## Authentication [#authentication]
Authentication is deployment-specific — not enforced by core runtime. Place API gateways, mTLS, or JWT validation in front of control plane ingress in production.
## Related [#related]
# Reference (/docs/reference)
Low-level reference material for authors integrating Yoke into their products.
## Quick links [#quick-links]
| Topic | Location |
| --------------- | ---------------------------------------------------- |
| TypeScript SDK | [SDK docs](/docs/sdk) |
| Rust runtime | [Runtime docs](/docs/runtime) |
| Author tutorial | [Quick start](/docs/getting-started/quick-start) |
| Embed guide | [Embed in your product](/docs/getting-started/embed) |
## Accuracy notes [#accuracy-notes]
When integrating, prefer this reference for:
* `state_cas` op (not `state_update`)
* Queue claim: `priority ASC, created_at ASC`
* No `mode: "priority"` queue policy
* `pg_sqlite` = BYTEA blobs, not Postgres extension
* Agent routes: `/agents` not `/rooms`
* Control plane port: **8090**
# Migrations (/docs/reference/migrations)
Yoke runs SQL migrations automatically on `yoke dev`, `yoke start`, and `control-plane` boot.
## Migration files [#migration-files]
| File | Purpose |
| ------------------------------------- | ----------------------------------------------------------------------------------- |
| `001_initial.sql` | Core tables: `stream_events`, `actor_instances`, `portal_instances`, `actor_queues` |
| `002_actor_sqlite.sql` | `actor_sqlite` BYTEA storage for per-instance SQLite |
| `003_queue_retries_dead_letters.sql` | Retry counts, `dead_letters` table |
| `004_worker_scaling.sql` | `runtime_workers` registry, heartbeats |
| `005_multi_tenant.sql` | `tenants`, `tenant_runtimes`, lease columns |
| `006_tenant_scoped.sql` | `tenant_id` on stream, instances, queues |
| `007_notify_rls_scheduler_events.sql` | Scheduler notifications |
| `008_fanout_idempotency.sql` | `fanout_delivered`, `runtime_metadata` cursor |
| `009_conversations.sql` | Conversation store: actors, conversations, participants, messages |
| `010_message_schema.sql` | `message_relations`, `blobs`, `branches` |
| `011_scheduled_events.sql` | Keyed schedules table |
| `012_actor_sqlite_stores.sql` | `store_id` column for sqliteStoreId segments |
| `013_queue_priority_groups.sql` | `priority`, `group_id` on `actor_queues` |
## Core runtime tables [#core-runtime-tables]
```
stream_events — global append-only log
tenant_stream_counters — per-tenant seq allocator
actor_instances — instance registry
portal_instances — portal singletons
actor_queues — per-instance work queues
dead_letters — exhausted retries
actor_sqlite — serialized SQLite (BYTEA)
fanout_delivered — fan-out idempotency
runtime_workers — horizontal scaling
runtime_metadata — fan-out cursor
scheduled_events — keyed delayed appends
```
## Conversation tables [#conversation-tables]
```
actors
conversations
participants
messages
message_relations
blobs
branches
```
All scoped by `tenant_id`.
## Multi-tenant tables [#multi-tenant-tables]
```
tenants
tenant_runtimes — lease_holder, lease_expires_at, last_activity
```
## Applying migrations [#applying-migrations]
Automatic on boot:
```bash
yoke dev ./your-package --database "$DATABASE_URL" --storage pg_sqlite
```
Manual migration command:
```bash
yoke migrate --database "$DATABASE_URL"
```
Inspect applied schema:
```bash
psql "$DATABASE_URL" -c "\dt"
```
## Schema versioning [#schema-versioning]
Migrations are sequential and idempotent where possible. The runtime tracks applied migrations in Postgres — do not modify applied migration files in production; add new numbered files.
## Backup guidance [#backup-guidance]
| Deployment | Backup |
| ------------ | ------------------------------------------------------------- |
| `pg_sqlite` | Postgres dump includes actor SQLite BYTEA |
| `s3` | Postgres dump + S3 bucket replication |
| Multi-tenant | Single Postgres cluster backup; tenant isolation is row-level |
## Related [#related]
# Ops Bridge (/docs/reference/ops-bridge)
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) [#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) [#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 [#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 [#state_cas-detail]
```typescript
// 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 [#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 [#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) [#system-endpoints-rust-http-not-ops]
| Endpoint | Purpose |
| -------------- | --------------- |
| `GET /health` | `ok` |
| `GET /metrics` | Prometheus text |
## Type flow [#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
```
## Related [#related]
# Runtime Architecture (/docs/runtime/architecture)
`YokeRuntime` is the top-level assembly of every kernel subsystem. It implements the `Runtime` trait and wires Postgres, fan-out, scheduling, storage, and the embedded Deno host into one long-lived process.
## Component map [#component-map]
`YokeRuntime` holds:
| Field | Responsibility |
| ------------------ | --------------------------------------------------------------------------- |
| `services` | `StreamWithFanout`, instance registry, `Storage` — shared by the op handler |
| `deno` | `EmbeddedDenoHost` — package load, `dispatch_process`, filter eval |
| `fanout` | `FanoutEngine` — index rebuild, async enqueue |
| `scheduler` | `QueueDrainScheduler` — per-instance queue drain loop |
| `storage` | `Storage` enum (`PgSqlite` \| `S3`) — actor SQLite + `ctx.state` |
| `_worker_registry` | Registers in `runtime_workers`, claims unassigned instances |
| `_stream_catchup` | Background tail for cold-start / multi-executor fan-out |
| `_timer_dispatch` | Timer dispatch — fires due rows from `scheduled_events` |
| `_lease_heartbeat` | Optional tenant lease extension (`YOKE_EXECUTOR_MODE`) |
`RuntimeServices` groups the three handles the JSON-RPC op handler needs: the stream (with fan-out wrapper), the instance registry, and storage.
## Data flow diagram [#data-flow-diagram]
```mermaid
flowchart TB
subgraph external [External]
PortalHTTP[Portal HTTP routes]
UserTS[User TypeScript package]
end
subgraph deno [Embedded Deno]
WorkerRPC[WorkerRpc JSON-RPC]
ProcessFn["actor.process(ctx, event)"]
PortalIface[Portal interface methods]
end
subgraph rust [YokeRuntime Rust]
OpHandler[RuntimeOpHandler]
StreamWF[StreamWithFanout]
PgStream[PgStream]
Fanout[FanoutEngine]
Queue[PgActorQueue]
Scheduler[QueueDrainScheduler]
Storage[Storage pg_sqlite or S3]
end
subgraph pg [Postgres]
StreamEvents[(stream_events)]
ActorQueues[(actor_queues)]
Instances[(actor_instances)]
ActorSqlite[(actor_sqlite)]
end
PortalHTTP --> OpHandler
UserTS --> WorkerRPC
WorkerRPC <-->|unix socket| OpHandler
OpHandler --> StreamWF
StreamWF --> PgStream
PgStream --> StreamEvents
StreamWF -->|async| Fanout
Fanout --> Queue
Queue --> ActorQueues
Scheduler --> Queue
Scheduler -->|dispatch_process| WorkerRPC
WorkerRPC --> ProcessFn
ProcessFn -->|stream_append op| OpHandler
ProcessFn -->|portal_call op| PortalIface
OpHandler --> Storage
Storage --> ActorSqlite
Fanout --> Instances
Scheduler --> Instances
```
## Append path (hot path) [#append-path-hot-path]
1. Deno SDK calls `stream_append` op → `RuntimeOpHandler`.
2. `StreamWithFanout::append_and_fanout` writes via `PgStream::append` (assigns `seq`, inserts row).
3. Append returns `AppendResult { seq, id }` to the caller immediately.
4. Fan-out dispatches asynchronously (or deferred during nested actor ops — see [Fan-out](/docs/runtime/fanout)).
5. `record_dispatched` updates the per-tenant fan-out cursor in `runtime_metadata`.
Pseudocode:
```
append_and_fanout(input):
result = stream.append(input)
record metrics
if should_defer_fanout():
spawn fanout.dispatch(event) // avoid re-entering Deno
else
fanout.dispatch(event)
return result
```
## Process path (actor drain) [#process-path-actor-drain]
1. `QueueDrainScheduler` tick loop lists actor instances and active `group_id` values.
2. Per `(instance_id, group_id)` it acquires semaphore permits up to `queue_policy.concurrency`.
3. `claim_next` marks a row `processing` in `actor_queues`.
4. `EmbeddedDenoHost::dispatch_process` RPC runs `process()` in Deno.
5. On success → `queue.complete` deletes the row. On failure → `mark_failed` retries or dead-letters.
See [Scheduler](/docs/runtime/scheduler) and [Queues](/docs/runtime/queues).
## Op handler bridge [#op-handler-bridge]
`RuntimeOpHandler` implements `OpHandler` and is installed on the Deno worker RPC channel. Every SDK kernel call (`ctx.append`, `ctx.state`, `ctx.instances.create`, conversations, schedules) maps to a method string handled in the runtime.
The handler holds references to stream, instances, fan-out, storage, and conversations so Deno never touches Postgres directly.
## Background tasks [#background-tasks]
| Task | Trigger | Purpose |
| ---------------- | -------------------------- | ----------------------------------------------- |
| Stream catch-up | `LISTEN` + 30s poll | Fan out events appended while executor was cold |
| Timer dispatch | `LISTEN` + next-fire sleep | Append due `scheduled_events` to stream |
| Worker heartbeat | 15s interval | Update `runtime_workers.last_heartbeat` |
| Executor lease | 15s interval (optional) | Extend `tenant_runtimes` lease TTL |
## Horizontal scaling [#horizontal-scaling]
Multiple runtime workers share one Postgres database. Each registers in `runtime_workers` and claims `actor_instances` rows where `worker_id IS NULL`. New instances are assigned to the creating worker's ID. The scheduler only drains queues for instances owned by the local worker.
| Variable | Purpose |
| ---------------- | -------------------------------------------------------- |
| `YOKE_WORKER_ID` | Stable worker identity (defaults to `{hostname}-{uuid}`) |
Workers with no heartbeat for 60 seconds are excluded from the `yoke_runtime_workers` gauge. Instance ownership is not automatically migrated yet — plan worker drains before shutdown.
## Control plane vs executor [#control-plane-vs-executor]
The control-plane HTTP API uses `NoOpFanout` so appends write to `stream_events` only. Tenant executors fan-out via `stream_catchup` (boot and `LISTEN yoke_stream_append` while hot). Duplicate delivery is prevented by `fanout_delivered` (see [Fan-out](/docs/runtime/fanout)).
## Subsystem modules [#subsystem-modules]
The Rust runtime is organized around these responsibilities:
| Area | Role |
| ----------------------------- | ------------------------------------------------- |
| Runtime assembly + op handler | Boot, JSON-RPC bridge, kernel orchestration |
| Stream | Per-tenant sequenced append to `stream_events` |
| Fan-out | Two-tier filters → `actor_queues` |
| Queues | Persisted per-instance work queues |
| Scheduler | Drain loops, concurrency, Deno `dispatch_process` |
| Timer dispatch | `scheduled_events` → stream |
| Tenant scheduler | Runtime lifecycle for hosted deployments |
| Storage | `pg_sqlite` vs S3 backend selection |
| SQLite layer | Per-instance file + `__yoke_state` CAS |
| Embedded Deno | libdeno worker host |
| Worker RPC | JSON-RPC framing |
| Portal HTTP | Routes, `/health`, `/metrics` |
| Migrations | SQL schema applied at boot |
# Boot Sequence (/docs/runtime/boot-sequence)
When your server calls `YokeRuntime::start` (or you run `yoke start ./my-package` while authoring), the runtime executes a deterministic boot sequence. Understanding this order helps debug startup failures and reason about what exists before the first event is appended.
## Sequence [#sequence]
```
1. Connect Postgres, run migrations
2. Boot embedded Deno worker (libdeno)
3. Load user package (resolve npm deps, transpile TS)
4. Execute yoke.ts → collect defineRuntime() definitions
5. Register definitions in kernel
6. Load instance records from Postgres
7. Rebuild fan-out index (declarative + filterFromConfig)
8. Provision SQLite per instance (pg_sqlite BYTEA or S3)
9. Ensure portal singletons exist in Postgres
10. Run one-shot runtime bootstrap() if defined
11. Register portal crons and HTTP routes
12. Start actor drain loops (scheduler)
13. Append runtime.started to stream
14. Run forever (wake on events, HTTP, schedules — then idle)
```
## Step-by-step detail [#step-by-step-detail]
### 1. Postgres and migrations [#1-postgres-and-migrations]
Connect using `DATABASE_URL`. Run SQL migrations automatically. Creates tables: `stream_events`, `actor_instances`, `actor_queues`, `actor_sqlite`, conversations schema, multi-tenant tables, etc.
### 2–4. Deno worker and package [#24-deno-worker-and-package]
Spawn embedded worker. Resolve user package `deno.json`, npm lockfile, import maps. Execute `yoke.ts` default export.
Failure here: missing deps, TypeScript errors, invalid `defineRuntime()` shape.
### 5. Register definitions [#5-register-definitions]
Extract portal and actor definitions — config schemas, state schemas, queue policies, `filterFromConfig`, `queueSelectFromConfig`, routes, interfaces, inference stack.
Definitions are a **fixed set** — no dynamic registration after this point.
### 6. Load persisted instances [#6-load-persisted-instances]
Read `actor_instances` and `portal_instances` for the tenant. Instances created in previous runs are restored.
### 7. Fan-out index [#7-fan-out-index]
Rebuild fast-path index from declarative filters. Register function-filter instances for slow-path `eval_filter`. Load fan-out cursor from `runtime_metadata`.
### 8. SQLite provisioning [#8-sqlite-provisioning]
For each actor instance: load BYTEA from `actor_sqlite` or download from S3 → open rusqlite. Call `onCreate` only for **new** instances.
### 9. Portal provisioning [#9-portal-provisioning]
For each portal definition: ensure a singleton portal instance row exists in Postgres (create with `{}` config if missing). No long-running portal loops.
### 10. Runtime bootstrap [#10-runtime-bootstrap]
If the package defines `defineRuntime().bootstrap(fn)`, run it once (must complete — no infinite loops). Typical use: idempotent `ctx.instances.create({ id, ... })` for fixed topology.
### 11. Portal crons [#11-portal-crons]
Register declarative `crons` on portal definitions as scheduled route invocations (`yoke.portal.route` internally).
### 12. HTTP routes [#12-http-routes]
Mount portal routes on the executor HTTP server.
System routes available immediately: `GET /health`, `GET /metrics`.
### 13. Scheduler [#13-scheduler]
Start drain loops for all active actor instances owned by this worker. Claim orphaned instances (`worker_id IS NULL`) in multi-worker mode.
### 14. runtime.started [#14-runtimestarted]
Append system event to stream — signals boot complete. Fan-out delivers to any instance filtering on this type (uncommon but useful for coordinators).
## CLI invocation [#cli-invocation]
```bash
export DATABASE_URL=postgres://yoke:yoke@localhost:5432/yoke
export YOKE_STORAGE=pg_sqlite
yoke dev ./your-package \
--database "$DATABASE_URL" \
--storage pg_sqlite
```
| Flag | Purpose |
| ------------- | -------------------------------------- |
| `--package` | Path to user package |
| `--database` | Postgres URL |
| `--storage` | `pg_sqlite` or `s3` |
| `--tenant-id` | Multi-tenant scope (default `default`) |
| `--port` | Portal HTTP port |
Control plane (separate binary invocation): port **8090** default via `YOKE_CONTROL_PLANE_PORT`.
## Boot failure modes [#boot-failure-modes]
| Symptom | Likely cause |
| ----------------------- | ---------------------------------------------- |
| Migration error | Postgres version, permissions, stale schema |
| Package load error | deno.json, missing @yoke/sdk, TS compile error |
| Config validation error | instance config doesn't match Zod schema |
| Portal start throw | Uncaught exception in `start()` — check logs |
| Health check fails | Boot did not reach step 12 |
## Multi-tenant boot [#multi-tenant-boot]
Tenant executors boot the same sequence scoped by `YOKE_TENANT_ID`. Cold start triggered by `RuntimeLifecycleService` when ingress activity arrives. Package ref from `YOKE_PACKAGE_REF` (S3/git).
Per-phase boot timings are recorded when `YOKE_BOOT_METRICS` is enabled (default on). Filter phases with `YOKE_BOOT_METRIC_PHASES` (`all` or comma-separated names like `deno_spawn,package_load`).
## After boot [#after-boot]
* Append events via `ctx.append`, portal routes, CLI `yoke append`, or control plane API
* Fan-out delivers to queues asynchronously
* Scheduler drains continuously until shutdown
Graceful shutdown: flush S3 (`lazy` mode), stop scheduler, tear down Deno worker, release leases.
## Related [#related]
# Deno Worker (/docs/runtime/deno-worker)
User TypeScript runs in an **embedded in-process Deno runtime** via [libdeno](https://docs.rs/libdeno/latest/libdeno/). The Rust kernel communicates with the worker over a unix socket using newline-delimited JSON-RPC — the same protocol as the earlier subprocess model, but without an external `deno` binary.
## Architecture [#architecture]
```
YokeRuntime (Rust)
↕ unix socket, JSON-RPC
Embedded Deno worker
→ loads user package yoke.ts
→ registers definitions
→ handles dispatch_process, eval_filter, portal_invoke
```
| Env | Purpose |
| --------------------- | --------------------------------------------------------- |
| `YOKE_WORKER` | Worker entry script (runtime default; rarely overridden) |
| `YOKE_WORKER_PROJECT` | Deno project dir for worker `deno.json` (runtime default) |
## Package loading [#package-loading]
On boot:
1. Resolve user package path from CLI `--package` or `YOKE_PACKAGE_REF`
2. Load `deno.json`, import maps, npm dependencies
3. Transpile and execute `yoke.ts`
4. Collect `defineRuntime()` definitions
5. Register actor/portal specs in kernel via `get_actor_specs`
The worker supports npm, jsr, `node:` builtins, and TypeScript transpilation — full Deno module resolution.
## RPC protocol [#rpc-protocol]
### TypeScript → Rust (host calls) [#typescript--rust-host-calls]
| Op | Purpose |
| ------------------------------------------------------- | --------------------------------------- |
| `stream_append` | Publish event (Zod-validated in worker) |
| `instances_create` | Create actor instance |
| `instances_destroy` | Tear down instance |
| `sqlite_exec` | Actor SQLite query |
| `state_get` / `state_set` / `state_cas` | Typed state with optimistic locking |
| `portal_call` | Invoke portal interface method |
| `schedule_upsert` / `schedule_cancel` / `schedule_list` | Keyed delayed events |
| `conversation_*` | Conversation store CRUD |
### Rust → TypeScript (worker calls) [#rust--typescript-worker-calls]
| RPC | Purpose |
| ------------------- | -------------------------------------------------- |
| `eval_filter` | Function filter check for fan-out slow path |
| `eval_queue_select` | Custom queue selector over pending events |
| `dispatch_process` | Run actor `process()`; dequeue on success |
| `get_actor_specs` | Queue policy, default state, filter hints for boot |
| `portal_invoke` | Execute portal interface method from actor |
## dispatch\_process flow [#dispatch_process-flow]
```
Rust scheduler claims queue event
→ RPC dispatch_process { instanceId, event }
→ Worker builds ActorCtx for instance
→ await definition.process(ctx, event)
→ ctx.append() → stream_append op → Postgres → fan-out
→ ctx.portals.* → portal_call → portal_invoke
→ return success | throw
→ Rust: complete or requeue
```
## SDK as thin wrapper [#sdk-as-thin-wrapper]
User code never calls ops directly. `@yoke/sdk` wraps each op:
```typescript
// Simplified — @yoke/sdk wraps host ops
async append(input) {
return callHost("stream_append", { ...input, correlation: resolved });
}
```
Rust serializes payloads; it does not interpret business logic.
## Validation in worker [#validation-in-worker]
Event payloads are validated against `defineEvents()` Zod schemas in the worker **before** `stream_append` reaches Postgres. Invalid payloads throw in user code with schema errors.
Instance config is validated against `configSchema` on `instances_create`.
## Portal interface dispatch [#portal-interface-dispatch]
```
Actor: ctx.portals.slack.sendMessage("#general", "hi")
→ portal_call op (Rust)
→ portal_invoke RPC (Deno)
→ portal definition interface.sendMessage(...)
```
Portal `start()` loops and route handlers run in the same worker isolate.
## Permissions [#permissions]
Currently `allow_all_permissions` for trusted host code.
## Worker entry [#worker-entry]
The embedded worker boots the RPC server, registers op handlers, and exposes the host bridge to `@yoke/sdk` context factories. libdeno embedding replaces an external Deno subprocess — no external Deno process is required at runtime (Deno 2.x is still needed for development tooling).
## Related [#related]
# Fan-out (/docs/runtime/fanout)
**Fan-out** is the mechanism that delivers stream events to actor instance queues. It runs **asynchronously** after every append — the stream never blocks on delivery.
## Append → fan-out flow [#append--fan-out-flow]
```
1. ctx.append() or kernel append
2. INSERT stream_events (assign seq)
3. Return to caller immediately
4. Async fan-out task:
a. Claim fan-out idempotency slot (released on failure)
b. Load fan-out index for event type
c. Evaluate declarative filters (fast path)
d. Evaluate function filters (slow path via Deno)
e. INSERT actor_queues rows for matches
f. Advance per-tenant fan-out cursor on success
```
A slow or saturated actor affects **only its own queue**. The append path and other actors continue normally.
## Two-tier filter evaluation [#two-tier-filter-evaluation]
| Tier | Filter type | Path | Persisted |
| ---- | ----------- | ------------------------ | ----------------- |
| Fast | Declarative | Indexed by type prefix | Yes (in Postgres) |
| Slow | Function | Evaluated on every event | Config only |
### Fast path — declarative filters [#fast-path--declarative-filters]
Declarative filters are stored in Postgres and indexed by event type prefix:
```typescript
{ type: "work.assigned", where: { "payload.lang": "typescript" } }
{ type: "work.*" }
{ type: "slack.message.received" }
```
The fan-out index maps type prefixes to instance ids. On append, the kernel looks up candidates by `event.type` without scanning all instances.
### Slow path — function filters [#slow-path--function-filters]
Function filters cannot be serialized. They are rebuilt at boot from `filterFromConfig`:
```typescript
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
```
For each append, instances with function filters are evaluated via Deno RPC `eval_filter`. Optional `typeHint` on the instance narrows which events reach the slow path (optimization, not a guarantee).
Without `filterFromConfig`, function-filter instances survive restart in Postgres but receive **no events** until recreated.
## Idempotent delivery [#idempotent-delivery]
`fanout_delivered` records `(tenant_id, stream_seq, instance_id)` tuples. Re-running fan-out for the same event (crash recovery, catch-up) does not double-enqueue.
Per-tenant fan-out cursor in `runtime_metadata` tracks catch-up position for workers joining late.
## Enqueue semantics [#enqueue-semantics]
Matched events insert into `actor_queues`:
| Column | Purpose |
| ------------- | ------------------------------------- |
| `instance_id` | Target actor instance |
| `event_id` | Source stream event id |
| `event` | Full serialized event JSON |
| `status` | `pending` or `processing` |
| `priority` | From event (default 0) |
| `group_id` | Concurrency group (default `default`) |
| `created_at` | Enqueue timestamp |
| `tenant_id` | Multi-tenant isolation |
Claim order depends on the instance queue policy (`fifo`, `lifo`, or `random`). See [Queues](/docs/runtime/queues).
## Control plane fan-out [#control-plane-fan-out]
The multi-tenant control plane HTTP API **appends only** (`ControlPlaneStreamService` + `NoOpFanout`). Fan-out to `actor_queues` runs on the tenant executor via stream catch-up (`LISTEN yoke_stream_append` while hot, full catch-up on boot). The scheduler is signaled to wake executors after control-plane appends.
## Fan-out vs scheduler [#fan-out-vs-scheduler]
| Component | Role |
| ------------- | ------------------------------------------ |
| **Fan-out** | Write path — stream event → queue row |
| **Scheduler** | Read path — queue row → `dispatch_process` |
Fan-out is append-triggered. The scheduler polls or is notified when queue depth changes.
## Catch-up on boot [#catch-up-on-boot]
Boot sequence step 7 rebuilds the fan-out index:
1. Load all `actor_instances` with declarative filters
2. Rebuild type-prefix index
3. Register function-filter instances for slow-path evaluation
4. Resume fan-out cursor from `runtime_metadata`
Events appended while the runtime was down are in `stream_events` — catch-up fan-out processes any seq after the stored cursor.
## Metrics [#metrics]
| Metric | Description |
| ---------------------------- | --------------------------- |
| `yoke_fanout_enqueues_total` | Events enqueued via fan-out |
| `yoke_stream_appends_total` | Stream append operations |
See [Observability](/docs/deployment/observability).
## Multi-tenant fan-out [#multi-tenant-fan-out]
All fan-out operations are scoped by `tenant_id`. Stream seq is per-tenant (via `tenant_stream_counters`). Cross-tenant delivery is impossible at the storage layer.
## Related [#related]
# Rust Runtime (/docs/runtime)
The **Rust runtime** is Yoke's kernel. It owns durable storage, the global event stream, fan-out to actor queues, actor scheduling, portal HTTP, and the embedded Deno worker that runs user TypeScript.
Application builders interact with Yoke through the TypeScript SDK. The runtime is what makes append-only events, filter-based delivery, and per-instance SQLite actually durable across restarts.
## Responsibilities [#responsibilities]
| Subsystem | Role |
| ----------------- | ------------------------------------------------- |
| **Orchestration** | `YokeRuntime::start`, op handler, boot |
| **Stream** | Per-tenant sequenced append to `stream_events` |
| **Fan-out** | Two-tier filters → `actor_queues` |
| **Queues** | Persisted per-instance work queues |
| **Scheduler** | Drain loops, concurrency, Deno `dispatch_process` |
| **Storage** | Per-instance SQLite (`pg_sqlite` or S3) |
| **Deno** | In-process TypeScript via libdeno + JSON-RPC |
| **HTTP** | Portal routes, `/health`, `/metrics` |
## Stack position [#stack-position]
```
User package (yoke.ts)
↓ load / dispatch
Embedded Deno worker
↕ newline-delimited JSON-RPC (unix socket)
YokeRuntime (Rust)
→ Postgres — stream, instances, queues, conversations
→ SQLite — per-actor state (pg_sqlite BYTEA or S3 objects)
→ HTTP — portal ingress/egress
```
## Core invariants [#core-invariants]
1. **Stream is the spine** — all communication is append-only events; no direct actor channels.
2. **Append never blocks on fan-out** — fan-out runs asynchronously after the row is committed.
3. **No stream read API for user code** — only the kernel tails the stream internally.
4. **Crash-safe dequeue** — events leave the queue only after `process()` returns successfully in Deno.
5. **Tenant isolation** — `tenant_id` scopes stream seq, instances, queues, and fan-out idempotency.
## Postgres tables (runtime) [#postgres-tables-runtime]
| Table | Purpose |
| ------------------------ | ------------------------------------------------------------ |
| `stream_events` | Global append-only event log (`tenant_id`, `seq`) |
| `tenant_stream_counters` | Monotonic seq allocator per tenant |
| `actor_instances` | Actor instance registry (config, filter, queue policy) |
| `portal_instances` | One portal instance per definition per tenant |
| `actor_queues` | Per-instance pending/processing events |
| `dead_letters` | Saturated or exhausted-retry events |
| `actor_sqlite` | Serialized SQLite files (`store_id`, `instance_id`, `db`) |
| `fanout_delivered` | Idempotency ledger for fan-out per `(tenant_id, stream_seq)` |
| `runtime_workers` | Horizontal scaling worker registry + heartbeats |
| `runtime_metadata` | Per-tenant fan-out catch-up cursor |
SQL migrations run automatically at boot.
## Environment variables (summary) [#environment-variables-summary]
| Variable | Required | Default | Purpose |
| ------------------------------------ | -------- | ------------------- | -------------------------------------------- |
| `DATABASE_URL` | Yes | — | Postgres connection string |
| `YOKE_STORAGE` | Yes | `pg_sqlite` (CLI) | `pg_sqlite` or `s3` |
| `YOKE_TENANT_ID` | No | `default` | Tenant scope for this process |
| `YOKE_DATA_DIR` | No | `.yoke-data` | Local SQLite cache directory |
| `YOKE_PORT` | No | `8080` | Portal HTTP port |
| `YOKE_WORKER` | No | (runtime default) | Deno worker entry script |
| `YOKE_WORKER_PROJECT` | No | (runtime default) | Deno project dir (`deno.json`) |
| `YOKE_WORKER_ID` | No | `{hostname}-{uuid}` | Stable worker identity |
| `YOKE_MAX_QUEUE_DEPTH` | No | `1000` | Per-instance queue cap |
| `YOKE_MAX_RETRIES` | No | `5` | `process()` failures before dead letter |
| `YOKE_S3_BUCKET` | S3 only | — | S3 bucket for actor SQLite |
| `YOKE_S3_PREFIX` | No | `""` | Key prefix (scoped under `tenant_id`) |
| `YOKE_S3_SYNC` | No | `lazy` | `lazy` or `strict` local fsync before upload |
| `YOKE_EXECUTOR_MODE` | No | — | `1` / `true` enables lease heartbeat |
| `YOKE_LEASE_HOLDER` | No | — | Lease identity for hosted executors |
| `YOKE_LEASE_HEARTBEAT_INTERVAL_SECS` | No | `15` | Executor lease tick interval |
See [Storage](/docs/runtime/storage) and [Deno worker](/docs/runtime/deno-worker) for backend-specific details.
## Pages in this section [#pages-in-this-section]
## Observability [#observability]
`GET /metrics` on the portal HTTP port exposes Prometheus text format (alongside `GET /health`).
| Metric | Type | Description |
| --------------------------------- | ------- | ---------------------------------------------- |
| `yoke_stream_appends_total` | counter | Stream append operations |
| `yoke_fanout_enqueues_total` | counter | Events enqueued via fan-out |
| `yoke_actor_process_total` | counter | `process()` invocations |
| `yoke_actor_process_errors_total` | counter | Failed `process()` calls |
| `yoke_queue_dead_letters_total` | counter | Dead-lettered events |
| `yoke_queue_depth` | gauge | Per-instance queue depth (`instance_id` label) |
| `yoke_runtime_workers` | gauge | Workers with recent heartbeat |
## Related docs [#related-docs]
* [Mental model](/docs/concepts/mental-model) — stream, actors, portals from the application side
* [Mental model](/docs/concepts/mental-model) — design principles from the author perspective
# Queues (/docs/runtime/queues)
Each actor instance has a **persisted queue** in Postgres (`actor_queues`). Fan-out enqueues matched stream events; the scheduler claims and dispatches them to Deno `process()`.
## Queue row schema [#queue-row-schema]
```sql
actor_queues (
instance_id,
event_id,
event, -- full StreamEvent JSON
status, -- pending | processing
created_at,
retry_count,
tenant_id,
priority,
group_id
)
```
## Claim order [#claim-order]
When the scheduler claims the next event, Postgres selects by queue policy mode:
| Mode | SQL order |
| ---------------- | -------------------------------------------------------------- |
| `fifo` (default) | `ORDER BY priority ASC, created_at ASC` |
| `lifo` | `ORDER BY priority ASC, created_at DESC` |
| `random` | `ORDER BY priority ASC, random()` |
| `custom` | Deno `select()` picks an id; falls back to fifo if unclaimable |
```sql
LIMIT 1
FOR UPDATE SKIP LOCKED
```
| Sort key | Meaning |
| -------------- | ----------------------------------------- |
| `priority ASC` | Lower numbers claimed first (0 before 10) |
| `created_at` | Direction depends on fifo/lifo mode |
**There is no `mode: "priority"` queue policy** — priority is a per-event field set at append or fan-out time.
## Stale processing reclaim [#stale-processing-reclaim]
If a worker crashes mid-`process()`, rows stay in `processing` until reclaimed. The scheduler periodically resets rows where `claimed_at` is older than `YOKE_PROCESSING_TIMEOUT_SECS` (default 300) back to `pending`.
## Queue policy modes [#queue-policy-modes]
Queue policy is set on the actor **definition** and serialized on the instance record:
```typescript
type QueuePolicy =
| { mode: "fifo"; concurrency: number }
| { mode: "lifo"; concurrency: number }
| { mode: "random"; concurrency: number }
| { mode: "custom"; select: (queue) => event | events[]; concurrency: number };
```
| Mode | Behavior |
| -------- | ------------------------------------------------------------------------------------------------------------- |
| `fifo` | Claim lowest priority + earliest `created_at` (default) |
| `lifo` | Claim lowest priority + latest `created_at` |
| `random` | Random selection among pending events at each priority level |
| `custom` | Deno evaluates `select()` via `eval_queue_select` RPC; falls back to fifo when the chosen id is not claimable |
`concurrency` controls parallel `process()` invocations **per `group_id`**. `Infinity` is valid.
### Custom selectors after restart [#custom-selectors-after-restart]
Custom `select` functions are not serializable. Provide `queueSelectFromConfig` on the definition:
```typescript
queueSelectFromConfig: (config) => (events) =>
events.find((e) => e.type === config.preferredType) ?? events[0],
```
## Processing lifecycle [#processing-lifecycle]
```
pending → processing → (deleted on success)
→ pending (on throw, retry)
→ dead_letters (on saturation / max retries)
```
1. Scheduler claims event (`status = processing`)
2. `dispatch_process` RPC → Deno `process(ctx, event)`
3. **Success** → row deleted
4. **Throw** → `status` reverted to `pending`, `retry_count` incremented
There is no early-ack. The event stays `processing` for the entire `process()` duration.
## Concurrency groups [#concurrency-groups]
`groupId` on events scopes concurrency slots:
```typescript
await ctx.append(
{ type: "work.item", payload: { id: "a" } },
{ groupId: "batch-1" },
);
await ctx.append(
{ type: "work.item", payload: { id: "b" } },
{ groupId: "batch-2" },
);
```
With `concurrency: 1`, both groups process in parallel. Within a group, only one `process()` runs at a time (per concurrency setting).
The scheduler calls `list_active_groups` to find groups with pending work, then claims per group independently.
## Dead letters [#dead-letters]
When queues saturate or retries exhaust, events move to `dead_letters`:
| Column | Purpose |
| ------------- | ----------------- |
| `instance_id` | Source instance |
| `event_id` | Original event id |
| `event` | Full event JSON |
| `reason` | Why dead-lettered |
| `created_at` | When moved |
Metric: `yoke_queue_dead_letters_total`.
## Backpressure [#backpressure]
Fan-out always enqueues — it does not block on queue depth. Very deep queues increase latency for that instance but do not affect the stream or other instances.
Design patterns for high-volume instances:
* Increase `concurrency`
* Split work across instances with narrower filters
* Use `priority` to prefer urgent events
* Dead-letter policy for poison messages
## Queue depth metrics [#queue-depth-metrics]
`yoke_queue_depth` gauge with `instance_id` label — per-instance pending + processing count.
## Implementation [#implementation]
The Postgres queue backend (`PgActorQueue`) implements these key methods:
| Method | Purpose |
| -------------------------- | ------------------------------------------- |
| `enqueue` | Insert row from fan-out |
| `claim_next` | Transactional claim with SKIP LOCKED |
| `list_pending` | List for custom selector evaluation |
| `complete` | Delete row after successful process |
| `reclaim_stale_processing` | Reset stuck `processing` rows after timeout |
| `requeue` | Revert to pending on failure |
| `dead_letter` | Move to dead\_letters table |
## Related [#related]
# Scheduler (/docs/runtime/scheduler)
Yoke has **three schedulers** with distinct jobs. Naming them separately avoids conflating tenant cold-start with per-actor queue draining.
| Layer | Type / module | Responsibility |
| --------------------- | ------------------------- | --------------------------------------------------------- |
| **Runtime lifecycle** | `RuntimeLifecycleService` | Start/stop tenant executor sandboxes from ingress signals |
| **Queue drain** | `QueueDrainScheduler` | Claim `actor_queues` rows and invoke `process()` |
| **Timer dispatch** | `timer_dispatch` | Fire due `scheduled_events` rows into the stream |
Legacy names `TenantSchedulerService` and `ActorScheduler` remain as type aliases.
## Queue drain [#queue-drain]
The **queue drain scheduler** (`QueueDrainScheduler`) drives actor processing. It watches per-instance queues, claims events according to queue policy, and invokes `process()` in the embedded Deno worker.
## Responsibilities [#responsibilities]
| Task | Description |
| -------------------- | --------------------------------------------------------- |
| **Drain loops** | Per-instance (or per-group) loops that claim and dispatch |
| **Concurrency** | Respect `queue.concurrency` per `group_id` |
| **Worker ownership** | Only drain instances assigned to this worker |
| **RPC dispatch** | Call `dispatch_process` → Deno `process(ctx, event)` |
| **Dequeue** | Delete queue row on success; requeue on throw |
## Dispatch flow [#dispatch-flow]
```
Scheduler loop
→ list_active_groups(instance_id)
→ for each group under concurrency limit:
→ claim_next(instance_id, group_id, policy)
→ worker_rpc.dispatch_process(instance_id, event)
→ Deno: await definition.process(ctx, event)
→ success: complete(event_id)
→ throw: requeue(event_id)
```
Custom queue mode inserts `eval_queue_select` between `list_pending` and `claim_next`:
1. List pending events for group
2. Deno runs `queueSelectFromConfig(config)` or inline `select`
3. Claim chosen `event_id` specifically
## Concurrency model [#concurrency-model]
```typescript
queue: { mode: "fifo", concurrency: 3 }
```
Up to three concurrent `process()` calls per instance, scoped by `group_id`. Each invocation gets its own `ActorCtx` with the same `instanceId` but independent call stack.
Shared mutable state must go through `ctx.state.update()` — not module-level variables. Concurrent `process()` calls can interleave.
## Worker assignment [#worker-assignment]
Horizontal scaling partitions instances across workers:
| Mechanism | Behavior |
| ---------------- | --------------------------------------------------------- |
| `YOKE_WORKER_ID` | Stable worker identity (default `{hostname}-{uuid}`) |
| Create path | New instances get creating worker's `worker_id` |
| Boot claim | Worker claims `actor_instances` where `worker_id IS NULL` |
| Drain scope | Scheduler only processes locally-owned instances |
Workers register in `runtime_workers` and heartbeat every 15 seconds.
Stale workers (no heartbeat for 60s) are excluded from `yoke_runtime_workers` gauge. Instance ownership is not automatically migrated yet — orphan instances are claimed on next worker boot.
## Interaction with fan-out [#interaction-with-fan-out]
Fan-out enqueues asynchronously. The scheduler is notified (or polls) when:
* New rows appear in `actor_queues`
* A `process()` completes and frees a concurrency slot
* Boot finishes and instances become active
There is no global "tick" — drain is event-driven per instance.
## Portal and multi-tenant scheduling [#portal-and-multi-tenant-scheduling]
**Portals** are not scheduled — `start()` loops run independently.
**Multi-tenant** hosted deployments run `RuntimeLifecycleService` (alias `TenantSchedulerService`) on the control plane. It decides when to start/stop tenant executor sandboxes from stream appends, portal HTTP, queue depth, cron due rows, and periodic reconcile sweeps. See [Multi-tenant](/docs/deployment/multi-tenant).
| Env var | Default | Purpose |
| ----------------------------------- | ------- | ------------------------------------------------- |
| `YOKE_HOT_RECONCILE_INTERVAL_SECS` | 5 | Extend lease / stop idle hot tenants |
| `YOKE_COLD_RECONCILE_INTERVAL_SECS` | 30 | Cold sweep — stuck starting, pending queues |
| `YOKE_SCHEDULE_WAKE_POLL_SECS` | 10 | Wake cold tenants with overdue `scheduled_events` |
| `YOKE_NOTIFY_POLL_BACKUP_SECS` | 15 | Postgres poll when NOTIFY was missed |
Stop rule for tenant containers: keep running while `pending_events > 0` OR `active_dispatch > 0`.
## Timer dispatch [#timer-dispatch]
The **timer dispatch** loop (spawned from `YokeRuntime`) fires due rows from `scheduled_events` by appending to the stream. It listens on `LISTEN yoke_schedule_change` and sleeps until the next `fire_at_ms`.
## Metrics [#metrics]
| Metric | Type | Description |
| ---------------------------------- | --------- | -------------------------------------- |
| `yoke_actor_process_total` | counter | `process()` invocations |
| `yoke_actor_process_errors_total` | counter | Failed `process()` calls |
| `yoke_queue_depth` | gauge | Per-instance queue depth |
| `yoke_runtime_workers` | gauge | Workers with recent heartbeat |
| `yoke_executor_boot_phase_seconds` | histogram | Per-phase executor cold-start duration |
## Failure behavior [#failure-behavior]
| Failure | Effect |
| -------------------- | ----------------------------------------------------- |
| `process()` throws | Event requeued, error metric incremented |
| Deno worker crash | In-flight events revert to pending on restart |
| Worker dies | Owned instances idle until another worker claims them |
| Postgres unavailable | Scheduler blocks; stream appends fail |
## Related [#related]
# Storage (/docs/runtime/storage)
Actor instances store private durable state in **per-instance SQLite databases**. The Rust kernel abstracts two storage backends behind a `StorageBackend` trait — selected by `YOKE_STORAGE` at runtime start.
## YOKE\_STORAGE [#yoke_storage]
| Value | Behavior |
| ----------- | ------------------------------------------------------------------------ |
| `pg_sqlite` | Serialized SQLite files stored as BYTEA in Postgres `actor_sqlite` table |
| `s3` | SQLite files in S3; local cache directory for rusqlite execution only |
`YOKE_STORAGE` is **required**. There is no local-file-only production mode.
**`pg_sqlite` is not the PostgreSQL `pg_sqlite` extension.** Yoke serializes entire SQLite database files into a `BYTEA` column (`actor_sqlite.db`). SQLite executes locally in the worker process; persistence is the serialized blob in Postgres.
## Postgres actor\_sqlite table [#postgres-actor_sqlite-table]
```sql
actor_sqlite (
tenant_id,
instance_id,
store_id, -- default instanceId; supports sqliteStoreId segments
db BYTEA, -- serialized SQLite file
version, -- optimistic lock for state_cas
updated_at
)
```
On instance start: load BYTEA → open rusqlite in memory/temp → serve `sqlite_exec` and `state_*` ops.
On write: execute SQL → serialize → upsert BYTEA.
## S3 backend [#s3-backend]
Objects stored at:
```
{YOKE_S3_PREFIX}/instances/{instance_id}/state.db
```
| Env | Purpose |
| ---------------- | ---------------------------- |
| `YOKE_S3_BUCKET` | Required for S3 mode |
| `YOKE_S3_PREFIX` | Key prefix (optional) |
| `YOKE_S3_SYNC` | `lazy` (default) or `strict` |
### Sync modes [#sync-modes]
| `YOKE_S3_SYNC` | Behavior |
| -------------- | ---------------------------------------------------- |
| `lazy` | Upload after writes; `flush_all()` on shutdown |
| `strict` | `fsync` local `state.db` before each S3 `put_object` |
**Guarantee:** With `strict`, successful `state_set` / `sqlite_exec` / `state_cas` implies local file durability before upload begins. S3 durability follows AWS semantics. Crashes between local write and upload may lose data in `lazy` mode.
## SDK surface (identical in both modes) [#sdk-surface-identical-in-both-modes]
| API | Backing |
| -------------------------- | ------------------------------------------------------------------ |
| `ctx.state.get/set/update` | `__yoke_state` table via `state_get`, `state_set`, `state_cas` ops |
| `ctx.sqlite.exec` | Arbitrary SQL via `sqlite_exec` op |
Never write `__yoke_state` directly — use `ctx.state`.
### state\_cas [#state_cas]
Atomic compare-and-swap for optimistic locking:
```
state_get → { data, version }
state_cas(store_id, expected_version, new_data) → { data, version } | conflict
```
The worker RPC exposes `state_cas` (not `state_update`). `ctx.state.update()` wraps this.
## sqliteStoreId [#sqlitestoreid]
Actors can partition SQLite into logical stores per invocation:
```typescript
defineActor({
sqliteStoreId: (ctx) => `${ctx.instanceId}:${ctx.config.shardKey}`,
// ...
});
```
Each `store_id` gets its own row in `actor_sqlite` or S3 object.
## Blob storage (conversations) [#blob-storage-conversations]
Conversation media blobs use S3 when `YOKE_STORAGE=s3`:
```
POST /v1/tenants/:tenant_id/blobs
```
Requires `YOKE_S3_BUCKET`. Metadata in Postgres `blobs` table; bytes in S3.
## Choosing a backend [#choosing-a-backend]
**Best for:** single-tenant dev, smaller deployments, strongest single-system durability. One Postgres connection handles stream, queues, and actor state.
**Best for:** multi-tenant scale-to-zero, large SQLite files, geographic distribution. Postgres stays always-on; compute scales independently.
## Implementation layers [#implementation-layers]
| Layer | Role |
| ---------------- | ------------------------------------------------------ |
| Storage dispatch | Routes ops to the active backend (`pg_sqlite` or `s3`) |
| SQLite engine | rusqlite open, `state_cas`, exec |
| Postgres backend | BYTEA read/write for `pg_sqlite` mode |
| S3 backend | S3 get/put, local cache |
## Related [#related]
# Global Stream (/docs/runtime/stream)
The global stream is Yoke's single sequencing authority. Every action that matters — actor work, portal ingress, instance lifecycle, conversation messages — becomes a durable row in `stream_events`.
User code can **append only**. There is no tail, subscribe, or read API exposed to TypeScript. The Rust kernel tails internally for fan-out catch-up and audit.
## Event shape [#event-shape]
Events stored in Postgres match the SDK `StreamEvent` type:
| Field | Column | Description |
| ------------- | ------------- | ----------------------------------------------------- |
| `seq` | `seq` | Monotonic position per tenant |
| `id` | `id` | Unique event UUID |
| `at` | `at` | Unix timestamp (ms) |
| `type` | `type` | Open namespace string (e.g. `work.assigned`) |
| `payload` | `payload` | JSONB body |
| `source` | `source` | Optional `{ kind, instanceId }` |
| `correlation` | `correlation` | Optional work-chain id |
| `priority` | `priority` | Queue ordering hint (lower = sooner, default `0`) |
| `groupId` | `group_id` | Concurrency scope for queue drain (default `default`) |
| — | `tenant_id` | Tenant partition key |
`priority` and `group_id` propagate from stream events into `actor_queues` on fan-out.
## PgStream [#pgstream]
`PgStream` implements the `Stream` trait. It holds a Postgres connection pool and a `TenantId`.
### Per-tenant sequencing [#per-tenant-sequencing]
Sequence numbers are **not** global across tenants. Each tenant has a row in `tenant_stream_counters`. `next_seq` atomically increments and returns the assigned sequence:
```sql
INSERT INTO tenant_stream_counters (tenant_id, next_seq)
VALUES ($tenant, 2)
ON CONFLICT (tenant_id) DO UPDATE
SET next_seq = tenant_stream_counters.next_seq + 1
RETURNING next_seq - 1 AS seq
```
The `INSERT ... ON CONFLICT DO UPDATE` pattern ensures the first event for a tenant gets `seq = 1`. The primary key on `stream_events` is `(tenant_id, seq)`.
### Append implementation [#append-implementation]
On append, the runtime:
1. Generates a unique event id and current timestamp.
2. Calls `next_seq()` to allocate the tenant-scoped sequence number.
3. Inserts the full row into `stream_events` (type, payload, source, correlation, priority, group\_id, tenant\_id).
4. Returns `AppendResult { seq, id }` to the caller.
Before each query, the runtime sets the Postgres session variable used by row-level security policies for tenant isolation.
## Append path end-to-end [#append-path-end-to-end]
```mermaid
sequenceDiagram
participant SDK as Deno SDK
participant Op as RuntimeOpHandler
participant SWF as StreamWithFanout
participant PG as PgStream
participant DB as Postgres
participant FO as FanoutEngine
SDK->>Op: stream_append
Op->>SWF: append_and_fanout
SWF->>PG: append
PG->>DB: tenant_stream_counters bump
PG->>DB: INSERT stream_events
PG-->>SWF: AppendResult
SWF-->>Op: AppendResult
Op->>DB: record_dispatched (metadata)
Op-->>SDK: { seq, id }
SWF->>FO: dispatch (async)
```
Steps in order:
1. **Allocate seq** — `tenant_stream_counters` UPSERT.
2. **Insert row** — full event JSON into `stream_events`.
3. **Return to caller** — `AppendResult` before fan-out completes.
4. **Fan-out** — `FanoutEngine::dispatch` evaluates filters and enqueues (see [Fan-out](/docs/runtime/fanout)).
5. **Record cursor** — stores `fanout_last_seq:{tenant_id}` in `runtime_metadata`.
Append commits and returns even when target actor queues are saturated. Backpressure is per-instance queue depth, not on the append path.
## StreamWithFanout [#streamwithfanout]
Callers use `StreamWithFanout` rather than raw `PgStream`. It wraps append with metrics and fan-out dispatch:
* Increments `yoke_stream_appends_total`.
* Builds a `StreamEvent` from the append result + input fields.
* Dispatches to `Fanout` trait impl (`FanoutEngine` in production, `NoOpFanout` on control plane).
When the op handler is processing a nested Deno call (e.g. `instances_create` during portal bootstrap), fan-out is deferred and spawned on a separate task to avoid re-entering Deno synchronously.
## Internal read API [#internal-read-api]
`list_events_after(after_seq)` is **kernel-only**. Stream catch-up uses it to replay events the executor missed while cold:
```
SELECT ... FROM stream_events
WHERE tenant_id = $tenant AND seq > $after_seq
ORDER BY seq ASC
```
This is not exposed to user TypeScript — it is the security boundary between author code and durable history.
## Schema reference [#schema-reference]
Initial table:
```sql
CREATE TABLE stream_events (
seq BIGSERIAL PRIMARY KEY, -- later: (tenant_id, seq) composite PK
id TEXT NOT NULL UNIQUE,
at BIGINT NOT NULL,
type TEXT NOT NULL,
payload JSONB NOT NULL DEFAULT '{}',
source JSONB,
correlation TEXT
);
```
Indexes:
* `idx_stream_events_type` — type lookups
* `idx_stream_events_tenant_seq` — per-tenant tail
* `uq_stream_events_tenant_seq` — unique `(tenant_id, seq)`
## Environment variables [#environment-variables]
| Variable | Required | Purpose |
| ---------------- | -------- | -------------------------------------------------------- |
| `DATABASE_URL` | Yes | Postgres connection for `PgStream` pool |
| `YOKE_TENANT_ID` | No | Tenant scope (set automatically at `YokeRuntime::start`) |
## SDK surface [#sdk-surface]
Application code appends via `ctx.append()`:
```typescript
await ctx.append({ type: "work.assigned", payload: { role: "coder" } });
```
Optional fields `priority` and `groupId` on the event input flow through to queue ordering and scheduler concurrency scopes.
There are **no targets** — routing is entirely filter-based on the consumer side.
# Actors (/docs/sdk/actors)
An **actor** is a registered definition with a `process()` handler, durable state, and a persisted event queue. Definitions are fixed at build time; **instances** are created at runtime with per-instance config and filters.
## defineActor [#defineactor]
```typescript
import { defineActor, defineEvents, z } from "@yoke/sdk";
const events = defineEvents({
"work.assigned": z.object({ lang: z.string(), role: z.string() }),
});
const coder = defineActor({
events,
configSchema: z.object({
model: z.string(),
lang: z.string(),
}),
stateSchema: z.object({
jobsCompleted: z.number(),
}),
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
async process(ctx, event) {
await ctx.state.update((s) => ({
...s,
jobsCompleted: s.jobsCompleted + 1,
}));
},
});
```
Inside `defineRuntime().actors()`, prefer `defineFor(events, portals).actor({ ... })` — it omits the `events` field.
### ActorDefinition fields [#actordefinition-fields]
| Field | Required | Description |
| ----------------------- | -------- | ----------------------------------------------------------------------------------- |
| `configSchema` | yes | Zod schema for instance config (validated on create) |
| `stateSchema` | yes | Zod schema for initial/default durable state |
| `process` | yes | `(ctx, event) => Promise` — main handler |
| `queue` | no | Drain policy; default `{ mode: "fifo", concurrency: 1 }` |
| `filterFromConfig` | no | Factory for function filters (required for function-filter survival across restart) |
| `queueSelectFromConfig` | no | Factory for custom queue selectors after restart |
| `sqliteStoreId` | no | Resolve SQLite store segment per invocation |
| `onCreate` | no | Called once when instance is provisioned |
| `onDestroy` | no | Called before instance teardown |
## process [#process]
```typescript
async process(ctx, event) {
// ctx — full ActorCtx (see Context API)
// event — StreamEventInput with typed payload for known types
if (event.type === "work.assigned") {
event.payload.lang; // typed when in EventMap
}
}
```
Processing semantics:
1. Event matches instance filter → enqueued (Postgres `actor_queues`).
2. Scheduler claims event(s) per queue policy and concurrency group.
3. Worker calls `process(ctx, event)` in Deno.
4. **Success** → event dequeued.
5. **Throw** → event stays on queue for retry.
There is no early-ack API. Long work should be split across events or schedules.
### matchEvent [#matchevent]
Narrow event types in `process()`:
```typescript
import { matchEvent } from "@yoke/sdk";
async process(ctx, event) {
if (matchEvent(event, "work.assigned")) {
event.payload.lang; // narrowed
}
}
```
## onCreate / onDestroy [#oncreate--ondestroy]
Lifecycle hooks receive the same `ActorCtx` as `process()` (without a triggering event — default priority/group, default correlation).
```typescript
actor({
configSchema: z.object({ webhookUrl: z.string() }),
stateSchema: z.object({ ready: z.boolean() }),
async onCreate(ctx) {
await ctx.sqlite.exec(
"CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, status TEXT)",
);
await ctx.state.set({ ready: true });
},
async onDestroy(ctx) {
await ctx.portals.notify.send(ctx.config.webhookUrl, "instance stopped");
},
async process(ctx, event) { /* ... */ },
})
```
| Hook | When | Typical use |
| ----------- | --------------------------------------- | --------------------------------------------------- |
| `onCreate` | After instance row + SQLite provisioned | Schema migration, initial state, side subscriptions |
| `onDestroy` | Before rows deleted | Cleanup, final notifications |
## queue [#queue]
Default queue policy:
```typescript
import { DEFAULT_QUEUE_POLICY } from "@yoke/sdk";
// { mode: "fifo", concurrency: 1 }
```
Override on the definition:
```typescript
queue: { mode: "fifo", concurrency: 4 }
queue: { mode: "lifo", concurrency: 1 }
queue: { mode: "random", concurrency: 2 }
queue: {
mode: "custom",
concurrency: 1,
select: (pending) => pending.find((e) => e.type === "urgent") ?? pending[0],
}
```
See [Queues](/docs/sdk/queues) for modes, priority groups, and custom selectors.
## filterFromConfig [#filterfromconfig]
Function filters are **not serializable**. Persisted instances store config only; on restart the runtime rebuilds the filter by calling `filterFromConfig(config)`:
```typescript
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
```
| Scenario | Behavior |
| ---------------------------------------- | --------------------------------------------------------------------------------------- |
| Definition **has** `filterFromConfig` | Function filters work after restart via `eval_filter` RPC |
| Definition **lacks** `filterFromConfig` | Function-filter instances persist in Postgres but **receive no events** until recreated |
| Instance created with declarative filter | Fast-path fan-out; no `filterFromConfig` needed |
At instance creation you can pass either filter style:
```typescript
await ctx.instances.create("coder", {
config: { model: "claude-sonnet", lang: "typescript" },
filter: { type: "work.assigned", where: { "payload.lang": "typescript" } },
});
await ctx.instances.create("coder", {
config: { model: "claude-sonnet", lang: "typescript" },
filter: (event) =>
event.type === "work.assigned" && event.payload.lang === "typescript",
});
```
Optional `typeHint` on `CreateInstanceOpts` narrows slow-path evaluation for function filters (optimization, not a guarantee).
## queueSelectFromConfig [#queueselectfromconfig]
Custom `queue.select` functions are also not serializable. Provide a factory mirroring `filterFromConfig`:
```typescript
queueSelectFromConfig: (config) => (pending) => {
const preferred = pending.filter((e) => e.type === config.preferredType);
return preferred[0] ?? pending[0];
},
```
At boot, `get_actor_specs` reports `hasCustomQueueSelect: true` and the runtime uses `eval_queue_select` RPC to run the selector over pending events.
If you inline `queue: { mode: "custom", select, concurrency }` without `queueSelectFromConfig`, custom selection works until restart, then fails unless the factory is added.
## sqliteStoreId [#sqlitestoreid]
By default each actor instance has **one SQLite database** keyed by `instanceId`. Use `sqliteStoreId` to shard storage within an instance:
```typescript
import { sqliteStoreSegments, resolveSqliteStoreId } from "@yoke/sdk";
actor({
configSchema: z.object({ tenantId: z.string() }),
stateSchema: z.object({ count: z.number() }),
sqliteStoreId: (config) => async (ctx, event) => {
if (event?.type === "session.started") {
return sqliteStoreSegments.perStint(event.payload.sessionId);
}
return sqliteStoreSegments.perInstance();
},
async process(ctx, event) {
ctx.sqliteStoreId; // e.g. "inst-abc:stint:sess-123"
await ctx.state.get(); // scoped to resolved store
},
})
```
### resolveSqliteStoreId [#resolvesqlitestoreid]
```typescript
resolveSqliteStoreId("inst-abc", ""); // "inst-abc"
resolveSqliteStoreId("inst-abc", "default"); // "inst-abc"
resolveSqliteStoreId("inst-abc", "stint:x"); // "inst-abc:stint:x"
```
### sqliteStoreSegments helpers [#sqlitestoresegments-helpers]
| Helper | Segment | Use case |
| ---------------------- | --------------------- | --------------------------------------- |
| `perInstance()` | `""` | Default — one DB per instance |
| `perStint(id)` | `stint:${id}` | Agent monologue / forum stint isolation |
| `perInvocation(event)` | `inv:${event.id}` | One store per queued event |
| `perEventType(event)` | `event:${event.type}` | Partition by event type |
## Definition vs instance [#definition-vs-instance]
| | Definition | Instance |
| -------- | ------------------------------------ | --------------------------------------- |
| When | `defineRuntime().actors()` | `ctx.instances.create("name", opts)` |
| Stored | TypeScript module | Postgres (`instances`, queues, filters) |
| Contains | `process`, schemas, queue, factories | `config`, `filter`, runtime status |
| Count | Fixed set of names | Unlimited dynamic instances |
## ActorRegistration [#actorregistration]
```typescript
type ActorRegistration = {
readonly name: Name;
readonly definition: ActorDefinition<...>;
};
```
Inference helpers: `InferActorConfig`, `InferActorState`, `InferActorConfigFromRegistry`, `InferActorStateFromRegistry`.
## Full example [#full-example]
```typescript
export default defineRuntime()
.events(events)
.portals(({ events }) => {
const { portal } = defineFor(events);
return {
slack: portal({
configSchema: z.object({ botToken: z.string() }),
interface: {
async sendMessage(channel: string, text: string) {
console.log(channel, text);
},
},
}),
};
})
.actors(({ events, portals }) => {
const { actor } = defineFor(events, portals);
return {
coder: actor({
configSchema: z.object({ model: z.string(), lang: z.string() }),
stateSchema: z.object({ jobsCompleted: z.number() }),
queue: { mode: "fifo", concurrency: 1 },
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
async onCreate(ctx) {
await ctx.state.set({ jobsCompleted: 0 });
},
async process(ctx, event) {
await ctx.append({
type: "code.produced",
payload: { code: "// ..." },
});
await ctx.portals.slack.sendMessage("#general", "Code ready");
await ctx.state.update((s) => ({
...s,
jobsCompleted: s.jobsCompleted + 1,
}));
},
}),
};
})
.build();
```
# Context API (/docs/sdk/context-api)
`ctx` is the runtime handle passed to actor `process()`, portal `start()`, route handlers, and lifecycle hooks. The SDK wraps yoke ops into a typed surface — user code never calls Rust directly.
## ActorCtx [#actorctx]
Full surface for actor handlers:
```typescript
type ActorCtx = {
instanceId: string;
config: TConfig;
sqliteStoreId: string;
state: DurableState;
sqlite: SQLiteDatabase;
correlation: CorrelationContext;
append(input, options?): Promise;
instances: InstanceManager;
portals: PortalClients;
conversations: ConversationClient;
schedules: ScheduleManager;
llm: LlmClient; // requires inferenceStack on runtime
};
```
### instanceId and config [#instanceid-and-config]
```typescript
async process(ctx, event) {
console.log(ctx.instanceId); // UUID
console.log(ctx.config.model); // validated against configSchema
}
```
### state and sqlite [#state-and-sqlite]
See [State & SQLite](/docs/sdk/state-sqlite).
```typescript
await ctx.state.update((s) => ({ ...s, count: s.count + 1 }));
await ctx.sqlite.exec("SELECT 1");
```
### append [#append]
```typescript
await ctx.append({
type: "work.assigned",
payload: { role: "coder", lang: "typescript" },
});
await ctx.append(
{ type: "urgent.alert", payload: {} },
{ correlation: "new", priority: 0, groupId: "alerts" },
);
```
Typed for events in `defineEvents()`. Source is set automatically (`kind: "actor"`, `instanceId`).
### correlation [#correlation]
```typescript
const id = ctx.correlation.inherit();
await ctx.append({ type: "step.done", payload: {} }); // inherits id
```
| Method | Behavior |
| ----------- | --------------------------------------- |
| `current` | Correlation id of event being processed |
| `inherit()` | Reuse current or create new |
| `create()` | Always new id |
### instances [#instances]
```typescript
const handle = await ctx.instances.create("helper", {
config: { shard: "us-east" },
filter: { type: "work.assigned" },
});
await ctx.instances.destroy(handle.id);
```
Typed when using `defineFor` — definition names autocomplete.
### portals [#portals]
```typescript
await ctx.portals.slack.sendMessage("#general", "Done!");
```
Typed from portal `interface` definitions. Routes through `portal_call` → `portal_invoke`.
### conversations [#conversations]
Conversation store CRUD — available when tenant has conversation migrations:
```typescript
const actor = await ctx.conversations.createActor({ name: "alice" });
const room = await ctx.conversations.createConversation({ title: "support" });
const p = await ctx.conversations.addParticipant(room.id, actor.id);
await ctx.conversations.postMessage(room.id, {
parts: [{ type: "text", text: "Hello", format: "plain" }],
}, { authorParticipantId: p.id, type: "message" });
```
Prefer `createChatClient(ctx.conversations)` for ergonomic wrappers.
### schedules [#schedules]
```typescript
await ctx.schedules.schedule({
key: "retry:42",
delay: { seconds: 30 },
event: { type: "retry.tick", payload: { id: "42" } },
});
```
See [Schedules](/docs/sdk/schedules).
### llm [#llm]
Requires `.inferenceStack()` on the runtime builder:
```typescript
const outputs = await ctx.llm.infer(messages, {
model: "gpt-4o",
temperature: 0.7,
maxTokens: 1024,
tools: resolvedTools,
});
```
See [Inference](/docs/agents/inference).
## PortalCtx [#portalctx]
Portals receive a subset — no `state`, `sqlite`, or `llm`:
```typescript
type PortalCtx = {
instanceId: string;
config: TConfig;
correlation: CorrelationContext;
append(input, options?): Promise;
instances: InstanceManager;
conversations: ConversationClient;
schedules: ScheduleManager;
};
```
Route handlers and `start()` use `PortalCtx`:
```typescript
routes: [{
method: "POST",
path: "/webhooks/shopify",
handler: async (req, ctx) => {
await ctx.append({ type: "ingress.shopify.order", payload: parsed });
return { status: 200, body: "ok" };
},
}],
```
## Lifecycle hooks [#lifecycle-hooks]
```typescript
defineActor({
async onCreate(ctx) {
await ctx.sqlite.exec("CREATE TABLE IF NOT EXISTS ...");
},
async onDestroy(ctx) {
// cleanup before teardown
},
async process(ctx, event) { /* ... */ },
});
```
`onCreate` runs once after SQLite provisioning. `onDestroy` runs before instance rows are deleted.
## Type flow [#type-flow]
`defineFor(events, portals)` threads types through:
```
defineRuntime().events(events).portals(...).actors(...)
→ ActorCtx.config: CoderConfig
→ ActorCtx.state: CoderState
→ ActorCtx.portals.slack: SlackApi
→ process event payload: narrowed by EventMap
```
## Related [#related]
# defineRuntime (/docs/sdk/define-runtime)
`defineRuntime()` returns a fluent **runtime builder** that registers events, portals, actors, and an optional inference stack. Call `.build()` to produce a `RuntimeDefinition` the worker loads at boot.
## Basic shape [#basic-shape]
```typescript
import {
defineEvents,
defineFor,
defineRuntime,
z,
} from "@yoke/sdk";
const events = defineEvents({
"slack.message.received": z.object({
channel: z.string(),
text: z.string(),
}),
"work.assigned": z.object({
role: z.string(),
lang: z.string().optional(),
}),
});
export default defineRuntime()
.events(events)
.portals(({ events }) => {
const { portal } = defineFor(events);
return {
slack: portal({
interface: {
async sendMessage(channel: string, text: string) {
// ...
},
},
}),
};
})
.actors(({ events, portals }) => {
const { actor } = defineFor(events, portals);
return {
router: actor({
configSchema: z.object({ channels: z.array(z.string()) }),
stateSchema: z.object({ lastSeq: z.number() }),
async process(ctx, event) {
// ctx.portals.slack is fully typed
},
}),
};
})
.build();
```
## RuntimeBuilder methods [#runtimebuilder-methods]
### `.events(map)` [#eventsmap]
Register event type → Zod payload schemas. Merges with any previously registered events.
```typescript
const events = defineEvents({
"order.created": z.object({ orderId: z.string() }),
});
defineRuntime()
.events(events)
.events({
"order.shipped": z.object({ orderId: z.string(), tracking: z.string() }),
});
```
Returns a builder with an expanded `EventMap` type used for typed `ctx.append()` and `process()` event narrowing.
### `.inferenceStack(stack)` [#inferencestackstack]
Attach an LLM routing configuration used by `ctx.llm.infer()` and agent processors. Without this, agent inference throws at runtime.
```typescript
import { defineInferenceStack } from "@yoke/sdk";
const stack = defineInferenceStack({
models: {
"claude-sonnet": { id: "claude-sonnet-4-20250514", contextWindow: 200_000 },
},
providers: {
anthropic: anthropicProvider({ apiKey: process.env.ANTHROPIC_API_KEY! }),
},
routing: {
"claude-sonnet": ["anthropic"],
},
defaultModel: "claude-sonnet",
});
export default defineRuntime()
.events(events)
.inferenceStack(stack)
// ...
.build();
```
`InferenceStackConfig` fields:
| Field | Type | Purpose |
| -------------- | ----------------------------- | ------------------------------------------------------ |
| `models` | `Record` | Logical model ids → provider model metadata |
| `providers` | `Record` | Named provider implementations |
| `routing` | `Record` | Model id → ordered provider ids (round-robin failover) |
| `defaultModel` | `string?` | Default when callers omit `runConfig.model` |
### `.agents(fn)` [#agentsfn]
Register high-level agent definitions. The SDK expands each into the standard actor stack (ingestion, egress, processor, optional follow-up and guardian actors).
```typescript
.agents(() => ({
support: defineAgent({
name: "support-bot",
model: "local-tool-loop",
tools: [echoToolchain],
followUp: false,
guardianAngel: false,
}),
}))
```
The callback receives `{ events, inferenceStack }`. Returns a map of agent keys → `defineAgent()` inputs. Agent refs (`agents.support`) are available to `.chatPlatforms()`.
### `.chatPlatforms(fn)` [#chatplatformsfn]
Register chat platform integrations. Each expands to a portal (webhook ingress + outbound API) and a bridge actor that wires the platform adapter to an agent via `createAgent` / `joinConversation`.
```typescript
.chatPlatforms(({ agents }) => ({
slack: defineChatPlatform({
adapter: createStubChatAdapter("slack"),
agent: agents.support,
webhookPath: "/webhooks/slack",
}),
}))
```
Requires `.agents()` (or manual agent refs) when binding `agent`. See [Agents](/docs/agents) for chat platform integration patterns.
### `.bootstrap(fn)` [#bootstrapfn]
One-shot boot hook — provision instances with stable ids. Must complete quickly; no long-running loops.
```typescript
.bootstrap(async (ctx) => {
await ctx.instances.create("router", {
id: "router",
config: { channels: ["#general"] },
filter: { type: "slack.message.received" },
});
})
```
`BootstrapCtx` exposes `ctx.instances` and `ctx.append()`. Idempotent by instance `id`. Runs after portal singleton rows exist, before HTTP routes mount.
### `.portals(fn)` [#portalsfn]
Register portal definitions. The callback receives `{ events }` — the accumulated event map.
```typescript
.portals(({ events }) => {
const { portal } = defineFor(events);
return {
slack: portal({ /* PortalDefinition */ }),
webhook: portal({ /* ... */ }),
};
})
```
Updates builder types:
* `InferPortalRegistry` — registered portal names
* `InferPortalApis` — portal interface types available in `.actors()` as `portals`
### `.actors(fn)` [#actorsfn]
Register actor definitions. The callback receives `{ events, portals }`.
```typescript
.actors(({ events, portals }) => {
const { actor } = defineFor(events, portals);
return {
coder: actor({ /* ActorDefinition */ }),
router: actor({ /* ... */ }),
};
})
```
Updates `InferActorRegistry` with typed config/state per actor name.
### `.portal(name, definition)` / `.actor(name, definition)` [#portalname-definition--actorname-definition]
Incremental registration without callbacks. Equivalent to merging one definition at a time:
```typescript
defineRuntime()
.events(events)
.portal("slack", slackDefinition)
.actor("coder", coderDefinition)
.build();
```
Prefer `.portals()` / `.actors()` for type inference across the registry.
### `.build()` [#build]
Returns `RuntimeDefinition`:
```typescript
type RuntimeDefinition = {
actors: ActorRegistry;
portals: PortalRegistry;
events: EventMap;
inferenceStack?: InferenceStackConfig;
agents?: Record;
chatPlatforms?: Record;
bootstrap?: (ctx: BootstrapCtx) => Promise;
};
```
The worker reads this object (or calls `.build()` on a builder export) to register specs, event schemas, and portal routes.
## defineEvents [#defineevents]
`defineEvents()` preserves literal event type keys for TypeScript inference:
```typescript
const events = defineEvents({
"work.assigned": z.object({ lang: z.string() }),
});
// events is typed as { "work.assigned": ZodObject<...> }
```
Event types not in the map still append at runtime (`EventInput` fallback), but known types get payload checking in the editor and Zod validation in the worker.
## defineFor [#definefor]
Scoped factories that **omit the redundant `events` field** inside runtime callbacks:
```typescript
const { actor, portal } = defineFor(events, portals);
// ^ optional — pass portals in .actors() for typed ctx.portals.*
```
| Factory | Wraps | Injects |
| --------------- | -------------- | --------------------- |
| `actor(input)` | `defineActor` | `events` from closure |
| `portal(input)` | `definePortal` | `events` from closure |
Standalone definitions outside the builder still use `defineActor({ events, ... })` / `definePortal({ events, ... })` directly.
## Type parameters [#type-parameters]
The builder carries generic state through the chain:
```typescript
RuntimeBuilder<
TActors, // registered actor registrations
TPortals, // registered portal registrations
TEventMap, // merged event schemas
TPortalApis, // portal interface map for actors
TInferenceStack
>
```
Utility types:
| Type | Purpose |
| ------------------------------ | ----------------------------------------------------- |
| `Registry` | `{ actors, portals }` pair |
| `RuntimeHandle` | Live runtime with `.portals` client map and `.stop()` |
| `EmptyRegistry` | Initial builder state |
| `InferPortalClients` | Portal API types from registrations |
## User package contract [#user-package-contract]
1. **File**: `yoke.ts` at package root.
2. **Export**: default export must be a `defineRuntime()` builder or its `.build()` result.
3. **Validation**: actor `configSchema` / `stateSchema` and portal `configSchema` are validated when instances are created.
4. **Boot**: Deno worker imports the module, extracts actor specs (`get_actor_specs`), portal routes (`list_portal_routes`), and event schemas for Zod parsing.
```typescript
// Valid — builder (worker calls .build())
export default defineRuntime().events(events).actors(...);
// Valid — definition object
export default defineRuntime().events(events).actors(...).build();
```
## inferenceStack in actor code [#inferencestack-in-actor-code]
Once configured, actors and portals call:
```typescript
async process(ctx, event) {
const messages = await ctx.llm.infer(
[{ type: "message", role: "user", body: { text: "Hello" } }],
{ model: "claude-sonnet", temperature: 0.2 },
);
}
```
See [Context API — LLM client](/docs/sdk/context-api#llm-client) for `InferRunConfig` fields.
# Events (/docs/sdk/events)
All communication in Yoke flows through a **global append-only event stream**. The SDK types distinguish persisted events, append inputs, and the narrowed shapes received in `process()`.
## StreamEvent [#streamevent]
A **persisted** event after append (assigned by the runtime):
```typescript
type StreamEvent = {
seq: number; // monotonic stream sequence
id: string; // unique event id
at: number; // epoch ms timestamp
type: string; // event type key
payload: T;
source?: EventSource;
correlation?: string;
priority?: number;
groupId?: string;
};
type EventSource = {
kind: "actor" | "portal";
instanceId: string;
};
```
| Field | Description |
| ------------- | ----------------------------------------------------------- |
| `seq` | Global ordering in the tenant stream |
| `id` | Stable id used for queue deduplication and custom selectors |
| `at` | Wall-clock append time (ms) |
| `source` | Which instance appended the event |
| `correlation` | Logical work-chain id |
| `priority` | MX-style priority (lower runs sooner within a group) |
| `groupId` | Concurrency partition within an actor instance queue |
## EventInput [#eventinput]
Minimal shape for **new** events (runtime assigns `seq`, `id`, `at`):
```typescript
type EventInput = {
type: string;
payload: T;
correlation?: string;
priority?: number;
groupId?: string;
};
```
## AppendInput [#appendinput]
Typed append union — known event types from your `EventMap` plus an open fallback:
```typescript
type KnownAppendInput = {
[K in keyof TMap & string]: {
type: K;
payload: z.core.output;
correlation?: string;
priority?: number;
groupId?: string;
};
}[keyof TMap & string];
type AppendInput =
| KnownAppendInput
| EventInput;
```
Example with registered types:
```typescript
const events = defineEvents({
"work.assigned": z.object({ role: z.string(), lang: z.string() }),
});
// Typed payload — editor knows fields
await ctx.append({
type: "work.assigned",
payload: { role: "coder", lang: "typescript" },
});
// Unregistered types still allowed
await ctx.append({
type: "custom.debug",
payload: { note: "anything" },
});
```
## StreamEventInput [#streameventinput]
Shape received in `process(ctx, event)`:
```typescript
type KnownStreamEvent = {
[K in keyof TMap & string]: {
seq: number;
id: string;
at: number;
type: K;
payload: z.core.output;
source?: EventSource;
correlation?: string;
priority?: number;
groupId?: string;
};
}[keyof TMap & string];
type StreamEventInput =
keyof TMap & string extends never
? StreamEvent
: KnownStreamEvent;
```
When an `EventMap` is wired through the actor definition, **only known types appear in the union** — full discriminated narrowing on `event.type`.
## AppendResult [#appendresult]
```typescript
type AppendResult = {
seq: number;
id: string;
};
```
Returned from `ctx.append()`.
## Zod validation [#zod-validation]
Event payloads are validated in the Deno worker at:
1. **`ctx.append()`** — before `stream_append` RPC
2. **`process()`** — before handler runs (re-validates dequeued payload)
Validation logic:
```typescript
function validateEvent(input: AppendInput): AppendInput {
const schema = loaded.eventSchemas[input.type];
if (!schema) return input; // unknown type — pass through
const payload = schema.parse(input.payload); // throws on invalid
return { ...input, payload };
}
```
Register schemas with `defineEvents()` / `.events()`:
```typescript
const events = defineEvents({
"order.created": z.object({
orderId: z.string().uuid(),
total: z.number().positive(),
}),
});
export default defineRuntime().events(events).build();
```
| Event type | Schema registered | Validation |
| ---------- | ----------------- | ---------------------------------------------- |
| Known | yes | Strict Zod parse; throws `ZodError` on failure |
| Unknown | no | Payload accepted as-is |
Actor `configSchema` and `stateSchema` are separate — validated on instance create and state operations, not on the stream.
## ctx.append options [#ctxappend-options]
```typescript
await ctx.append(
{ type: "work.assigned", payload: { role: "coder", lang: "ts" } },
{
correlation: "inherit" | "new" | "my-chain-id",
priority: 0 | "inherit",
groupId: "default" | "inherit" | "monologue-abc",
},
);
```
### Correlation [#correlation]
Group related events into one logical piece of work:
| Value | Behavior |
| -------------------------------------- | -------------------------------------------------- |
| `"inherit"` (default while processing) | Reuse `ctx.correlation.current`, or create if none |
| `"new"` | Start a fresh correlation chain |
| explicit string | Use that id |
```typescript
await ctx.append({ type: "step.started", payload: {} }); // inherits
await ctx.append({ type: "step.done", payload: {} }, { correlation: "new" });
```
### priority [#priority]
MX-style ordering **within a concurrency group** on an actor queue:
* **Lower number runs sooner** (default `0`)
* `"inherit"` copies `ctx.priority` from the event being processed
* Omitted on append → resolved via `resolveQueueDispatch` (inherit while processing, else default `0`)
```typescript
await ctx.append(
{ type: "guardian.run", payload: {} },
{ priority: -10 }, // urgent — runs before priority 0 in same group
);
```
### groupId [#groupid]
Partitions concurrency **within one actor instance**:
* Events with different `groupId` values drain **in parallel** up to `queue.concurrency` **per group**
* Default group: `"default"` (`DEFAULT_GROUP_ID`)
* `"inherit"` copies `ctx.groupId` from the current event
```typescript
await ctx.append(
{ type: "agent.think", payload: { monologueId: "m-1" } },
{ groupId: "m-1" }, // isolated drain lane for this monologue
);
```
Scheduler claims pending events with:
```sql
ORDER BY priority ASC, created_at ASC
```
within each `(instance_id, group_id)` partition.
## resolveQueueDispatch [#resolvequeuedispatch]
Utility used internally by append and conversation postMessage:
```typescript
import { resolveQueueDispatch, DEFAULT_PRIORITY, DEFAULT_GROUP_ID } from "@yoke/sdk";
const { priority, groupId } = resolveQueueDispatch(
{ priority: ctx.priority, groupId: ctx.groupId },
{ priority: "inherit", groupId: "my-group" },
);
```
## matchEvent [#matchevent]
Type guard for known event types:
```typescript
import { matchEvent } from "@yoke/sdk";
async process(ctx, event) {
if (matchEvent(event, "work.assigned")) {
event.payload.lang; // narrowed
return;
}
// other known types or exhaustiveness checking
}
```
## Event flow [#event-flow]
```
ctx.append(input)
→ validateEvent (Zod if registered)
→ stream_append RPC
→ Postgres stream_events
→ fan-out to matching instance queues
→ scheduler → process(ctx, event)
```
See [Filters](/docs/sdk/filters) for subscription matching and [Queues](/docs/sdk/queues) for drain ordering.
# Filters (/docs/sdk/filters)
Actor instances **subscribe** to stream events via a **filter**. When an appended event matches, the runtime enqueues it on that instance's persisted queue. Filters are provided at **instance creation** (`CreateInstanceOpts.filter`) or derived from actor defaults — not on the definition alone.
## InstanceFilter [#instancefilter]
```typescript
type InstanceFilter =
| DeclarativeFilter
| FunctionFilter;
```
## Declarative filters [#declarative-filters]
Indexed, persisted, **fast-path** fan-out:
```typescript
type DeclarativeFilter = {
/** Exact type or prefix glob */
type: keyof TEventMap & string | (string & {});
/** Dot-path equality constraints on the event object */
where?: Record;
};
```
Examples:
```typescript
// Exact type
{ type: "slack.message.received" }
// Prefix glob
{ type: "work.*" }
// Type + payload constraint
{
type: "work.assigned",
where: { "payload.lang": "typescript" },
}
// Nested paths
{
type: "github.event",
where: { "payload.repository.name": "yoke" },
}
```
`where` keys use **dot paths** into the serialized event (e.g. `"payload.lang"`, `"source.instanceId"`).
### isDeclarativeFilter [#isdeclarativefilter]
```typescript
import { isDeclarativeFilter } from "@yoke/sdk";
if (isDeclarativeFilter(filter)) {
filter.type;
filter.where;
} else {
filter(event); // function
}
```
## Function filters [#function-filters]
In-memory predicate, **slow-path** fan-out — evaluated via Deno `eval_filter` RPC:
```typescript
type FunctionFilter = (
event: StreamEventInput,
) => boolean;
// Example
(event) =>
event.type === "work.assigned" && event.payload.lang === "typescript";
```
Function filters are **not serializable**. Only config is persisted in Postgres.
## Two-tier evaluation [#two-tier-evaluation]
| Tier | Filter type | Evaluation | Persisted |
| -------- | ----------- | -------------------------------------------------------- | ---------------------- |
| **Fast** | Declarative | `FanoutIndex` lookup by type / glob + `where` match | Full filter JSON in DB |
| **Slow** | Function | Every candidate event (or type-narrowed with `typeHint`) | Config only |
```
append → FanoutIndex.candidate_ids(type)
→ exact type bucket
→ glob patterns (e.g. work.*)
→ ALL function-filter instances (slow path)
→ for each candidate: match filter → enqueue
```
## FanoutIndex [#fanoutindex]
Rust-side index rebuilt at boot and on instance changes (`FanoutEngine.rebuild_index`):
```rust
pub struct FanoutIndex {
exact: HashMap>, // type → instance ids
globs: Vec<(String, String)>, // "work.*" → instance id
function_instances: Vec, // all function-filter instances
}
```
`candidate_ids(event_type)` returns:
1. Instances with exact declarative `type` match
2. Instances with glob patterns matching the event type
3. **All** function-filter instance ids (always candidates — predicate runs in Deno)
Declarative `where` clauses are evaluated **after** index lookup when enqueueing.
### Why function filters hit every event type [#why-function-filters-hit-every-event-type]
The index cannot precompute arbitrary predicates. Function-filter instances are included in every fan-out candidate set; `eval_filter` short-circuits false matches.
**Optimization**: pass `typeHint` on instance create to narrow slow-path evaluation (not a hard guarantee, but reduces work in practice).
## filterFromConfig requirement [#filterfromconfig-requirement]
Definitions that expect **function filters** after restart **must** provide `filterFromConfig`:
```typescript
actor({
configSchema: z.object({ lang: z.string() }),
stateSchema: z.object({ count: z.number() }),
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" && event.payload.lang === config.lang,
async process(ctx, event) { /* ... */ },
})
```
| Definition has `filterFromConfig` | Restart behavior |
| --------------------------------- | ---------------------------------------------------------------------------------------------------------- |
| **Yes** | Runtime calls `filterFromConfig(persistedConfig)` → `eval_filter` |
| **No** | Function-filter instances exist in DB but **receive no events** until recreated with a matching definition |
Declarative filters do **not** need `filterFromConfig` — the full filter is stored in Postgres.
### Boot spec [#boot-spec]
`get_actor_specs` reports `hasFunctionFilter: !!def.filterFromConfig` per actor name.
## Creating instances with filters [#creating-instances-with-filters]
```typescript
// Declarative — preferred for hot paths
await ctx.instances.create("coder", {
config: { lang: "typescript", model: "claude-sonnet" },
filter: {
type: "work.assigned",
where: { "payload.lang": "typescript" },
},
});
// Function — flexible, slow path
await ctx.instances.create("coder", {
config: { lang: "typescript", model: "claude-sonnet" },
filter: (event) =>
event.type === "work.assigned" &&
event.payload.lang === "typescript",
typeHint: "work.assigned", // optional optimization
});
```
If `filter` is omitted, the instance receives **no events** until updated (no default subscription).
## Declarative vs function — when to use [#declarative-vs-function--when-to-use]
| Use declarative when | Use function when |
| ------------------------------------------------ | ---------------------------------------------- |
| Matching on type + static payload fields | Complex logic, cross-field conditions |
| High event volume | Low volume or narrow typeHint |
| Filters must survive restart without code deploy | Prototyping (with `filterFromConfig` for prod) |
## filterFromConfig vs instance filter [#filterfromconfig-vs-instance-filter]
| | `filterFromConfig` | Instance `filter` |
| ----------- | ------------------------------------- | ------------------------------------------- |
| Location | Actor **definition** | Instance **create** opts |
| Purpose | Rebuild function filter after restart | Initial subscription |
| Declarative | Not used | Stored in Postgres |
| Function | Factory for eval\_filter | Initial predicate (same factory on restart) |
Typically the instance filter mirrors what `filterFromConfig(config)` would produce for that config.
## Complete example [#complete-example]
```typescript
const events = defineEvents({
"work.assigned": z.object({ lang: z.string(), role: z.string() }),
"work.*": z.object({}).passthrough(), // if using glob types in map
});
export default defineRuntime()
.events(events)
.actors(({ events }) => {
const { actor } = defineFor(events);
return {
tsCoder: actor({
configSchema: z.object({ model: z.string() }),
stateSchema: z.object({ jobs: z.number() }),
filterFromConfig: (config) => (event) =>
event.type === "work.assigned" &&
event.payload.lang === "typescript",
async process(ctx, event) {
await ctx.state.update((s) => ({ ...s, jobs: s.jobs + 1 }));
},
}),
workRouter: actor({
configSchema: z.object({}),
stateSchema: z.object({}),
async process(ctx, event) {
if (event.type === "work.assigned") {
await ctx.instances.create("tsCoder", {
config: { model: "default" },
filter: {
type: "work.assigned",
where: { "payload.lang": event.payload.lang },
},
});
}
},
}),
};
})
.build();
```
See [Actors — filterFromConfig](/docs/sdk/actors#filterfromconfig) and [Runtime fan-out](/docs/runtime/fanout) for kernel details.
# TypeScript SDK (/docs/sdk)
The `@yoke/sdk` package is the **author API for the internal TypeScript package** the kernel loads. It is not your product's public SDK. Your users never import it.
Your product server embeds `yoke-runtime` (Rust). This package is what that runtime executes inside Deno — `defineRuntime()`, actors, portals, tools. See [Embed Yoke in your product](/docs/getting-started/embed).
Every runtime loads a single default export from `yoke.ts` built with `defineRuntime()`.
```typescript
// yoke.ts — default export required
import { defineEvents, defineFor, defineRuntime, z } from "@yoke/sdk";
const events = defineEvents({
"work.assigned": z.object({ role: z.string(), lang: z.string() }),
});
export default defineRuntime()
.events(events)
.portals(({ events }) => { /* ... */ })
.actors(({ events, portals }) => { /* ... */ })
.build();
```
## Package contract [#package-contract]
| Requirement | Detail |
| ------------------ | ----------------------------------------------------------------------------------------------------- |
| **Entry file** | `yoke.ts` at package root |
| **Default export** | Must be a `defineRuntime()` builder (call `.build()` or export the builder — the worker accepts both) |
| **Events** | Declare once with `defineEvents()`; Zod schemas validate payloads at append and in `process()` |
| **Zod re-export** | Import `z` from `@yoke/sdk` (Zod 4) |
The Deno worker loads your package, registers definitions in the Rust kernel, and bridges `ctx.*` calls over JSON-RPC. See [Runtime](/docs/runtime) for the IPC surface.
## Core concepts [#core-concepts]
| Concept | SDK entry | Role |
| ------------- | -------------------------------------- | ----------------------------------------------------------------------- |
| **Events** | `defineEvents`, `ctx.append` | Global append-only stream; all communication flows through typed events |
| **Actors** | `defineActor`, `defineFor().actor` | Stateful, queued handlers; one or many instances per definition |
| **Portals** | `definePortal`, `defineFor().portal` | Singleton ingress/egress (HTTP routes + typed interface) |
| **Filters** | `InstanceFilter`, `filterFromConfig` | Instance subscription to stream events |
| **Queues** | `QueuePolicy`, `queueSelectFromConfig` | Per-instance drain order and concurrency |
| **State** | `ctx.state`, `ctx.sqlite` | Durable per-instance SQLite (typed document + raw SQL) |
| **Schedules** | `ctx.schedules`, `ctx.schedule` | Keyed delayed events owned by an instance |
| **Context** | `ActorCtx`, `PortalCtx` | Full runtime surface inside `process`, `start`, routes, lifecycle hooks |
## Runtime builder flow [#runtime-builder-flow]
```
defineRuntime()
.events(eventMap) // register Zod payload schemas
.inferenceStack(stack) // optional LLM routing
.agents(() => …) // optional high-level agent definitions
.chatPlatforms(({ agents }) => …) // optional chat platform integrations
.bootstrap(async (ctx) => …) // optional one-shot instance provisioning
.portals(({ events }) => …) // portal definitions; APIs flow to actors
.actors(({ events, portals }) => …)
.build() // RuntimeDefinition → loaded by DenoHost
```
Prefer the **callback style** (`.portals()`, `.actors()`) so events and portal APIs are threaded through types automatically. Legacy `.portal(name, def)` / `.actor(name, def)` methods remain available.
## Reference pages [#reference-pages]
| Page | Topics |
| ----------------------------------------- | ------------------------------------------------------------------------------------------------------ |
| [defineRuntime](/docs/sdk/define-runtime) | Builder methods, `defineFor`, `defineEvents`, `agents`, `chatPlatforms`, `bootstrap`, `inferenceStack` |
| [Actors](/docs/sdk/actors) | `defineActor`, `process`, lifecycle, queue, filters, `sqliteStoreId` |
| [Portals](/docs/sdk/portals) | `definePortal`, `routes`, `start`, `interface` |
| [Events](/docs/sdk/events) | `StreamEvent`, `AppendInput`, Zod validation, priority, `groupId` |
| [Filters](/docs/sdk/filters) | Declarative vs function filters, `FanoutIndex`, `filterFromConfig` |
| [Queues](/docs/sdk/queues) | FIFO/LIFO/random/custom, concurrency, priority groups, inherit |
| [State & SQLite](/docs/sdk/state-sqlite) | `ctx.state`, CAS updates, `__yoke_state`, multi-store segments |
| [Schedules](/docs/sdk/schedules) | `ScheduleManager`, keyed delayed events, HTTP client |
| [Context API](/docs/sdk/context-api) | Full `ActorCtx` / `PortalCtx` surface, conversations, instances, LLM |
## Type helpers [#type-helpers]
Common inference utilities exported from `@yoke/sdk`:
```typescript
import type {
InferSchema,
InferConfig,
InferState,
InferActorConfig,
InferActorState,
InferPortalApi,
InferPortalClients,
} from "@yoke/sdk";
```
## Related documentation [#related-documentation]
* [Embed Yoke in your product](/docs/getting-started/embed) — Rust host API vs this package
* [Runtime architecture](/docs/runtime/architecture) — kernel components and data flow
* [Ops bridge](/docs/reference/ops-bridge) — Rust ↔ Deno JSON-RPC ops
* [Mental model](/docs/concepts/mental-model) — stream, fan-out, instances
* [Quick start](/docs/getting-started/quick-start) — minimal inline package tutorial
# Portals (/docs/sdk/portals)
A **portal** is a singleton definition (one instance per portal name at runtime boot) that bridges **integrations the runtime needs** (Slack, GitHub webhooks, third-party APIs) and the event stream.
Your product's public HTTP is **not** a portal — it is your host server, which embeds Yoke and calls `append` / `conversation_ops`. Keep portal routes private. See [Embed Yoke in your product](/docs/getting-started/embed).
Portals expose:
* **`interface`** — typed methods actors call via `ctx.portals..()`
* **`routes`** — HTTP ingress handlers that append events or run side effects
* **`crons`** — scheduled invocations of a portal route with a JSON payload (wake-on-tick, not long-running loops)
Portals do **not** support long-running `start()` hooks. The compute model wakes runtimes for HTTP ingress, stream events, and scheduled ticks — then goes cold.
## definePortal [#defineportal]
```typescript
import { definePortal, defineEvents, z } from "@yoke/sdk";
const events = defineEvents({
"slack.message.received": z.object({
channel: z.string(),
text: z.string(),
user: z.string().optional(),
}),
});
const slackPortal = definePortal({
events,
configSchema: z.object({ botToken: z.string() }),
interface: {
async sendMessage(channel: string, text: string) {
// call Slack API using bot token
},
},
routes: [
{
method: "POST",
path: "/webhooks/slack",
handler: async (req, ctx) => {
const body = req.body ? JSON.parse(new TextDecoder().decode(req.body)) : {};
await ctx.append({
type: "slack.message.received",
payload: {
channel: body.channel ?? "#general",
text: body.text ?? "",
user: body.user,
},
});
return { status: 200, body: "ok" };
},
},
],
});
```
Inside `defineRuntime().portals()`, use `defineFor(events).portal({ ... })`.
### PortalDefinition fields [#portaldefinition-fields]
| Field | Required | Description |
| -------------- | -------- | -------------------------------------------------- |
| `configSchema` | no | Zod schema for portal instance config |
| `interface` | no | API object — methods invoked via `portal_call` RPC |
| `routes` | no | HTTP route table |
| `crons` | no | Recurring route invocations with payload |
Generic portal typing:
```typescript
type SlackApi = {
sendMessage(channel: string, text: string): Promise;
};
portal({
configSchema: slackConfigSchema,
interface: { /* ... */ } satisfies SlackApi,
});
```
## routes [#routes]
Each route maps to runtime HTTP dispatch (`portal_route_dispatch` RPC):
```typescript
type PortalRoute = {
method: "GET" | "POST" | "PUT" | "PATCH" | "DELETE";
path: string;
handler: PortalRouteHandler;
};
```
### PortalRequest [#portalrequest]
```typescript
type PortalRequest = {
method: string;
path: string;
headers: Record;
body: Uint8Array | null;
};
```
### PortalResponse [#portalresponse]
```typescript
type PortalResponse = {
status: number;
headers?: Record;
body?: Uint8Array | string;
};
```
### Handler signature [#handler-signature]
```typescript
type PortalRouteHandler = (
req: PortalRequest,
ctx: PortalCtx,
) => Promise | PortalResponse;
```
Example webhook:
```typescript
routes: [
{
method: "POST",
path: "/v1/github",
handler: async (req, ctx) => {
if (req.headers["x-hub-signature"] !== expected) {
return { status: 401, body: "invalid signature" };
}
const payload = JSON.parse(new TextDecoder().decode(req.body!));
await ctx.append({
type: "github.push",
payload: { repo: payload.repository.full_name },
});
return { status: 202, body: JSON.stringify({ accepted: true }) };
},
},
],
```
At boot the worker returns all routes via `list_portal_routes`. The Rust runtime mounts them under the portal's configured base path.
## crons [#crons]
Declare periodic work as route invocations — the runtime wakes, dispatches the handler, and returns to idle:
```typescript
crons: [
{
key: "poll-inbox",
cadence: { kind: "every", everyMs: 60_000 },
route: { method: "POST", path: "/internal/poll" },
payload: { since: 0 },
},
],
```
Each tick POSTs `payload` to the route handler (same as an HTTP request body). Use `defineRuntime().bootstrap()` for one-shot instance provisioning at cold start.
## interface [#interface]
The **typed egress surface** for actors. Each method becomes a JSON-RPC `portal_call`:
```typescript
// In actor process():
await ctx.portals.slack.sendMessage("#deploys", "Shipped v2.1");
```
Implementation runs in the Deno worker against the portal singleton's `interface` object:
```typescript
interface: {
async sendMessage(channel: string, text: string) {
const res = await fetch("https://slack.com/api/chat.postMessage", {
method: "POST",
headers: { Authorization: `Bearer ${botToken}` },
body: JSON.stringify({ channel, text }),
});
if (!res.ok) throw new Error(await res.text());
},
async uploadFile(channel: string, bytes: Uint8Array, filename: string) {
// ...
},
},
```
Portal methods can close over module-level state or read config injected at boot. They do **not** receive `PortalCtx` automatically — use closures or module scope for credentials.
### Cross-portal calls [#cross-portal-calls]
Portals can call other portals through `ctx.portals` inside `start` and route handlers (proxy to other singleton interfaces).
## Portal vs actor [#portal-vs-actor]
| | Portal | Actor |
| --------- | ----------------------------- | ---------------------------- |
| Instances | One singleton per definition | Many dynamic instances |
| State | No `ctx.state` / `ctx.sqlite` | Full durable state |
| Ingress | HTTP `routes` | Event filters only |
| Egress | `interface` methods | `ctx.portals.*` |
| Queue | None | Persisted per-instance queue |
## PortalRegistration [#portalregistration]
```typescript
type PortalRegistration = {
readonly name: Name;
readonly definition: PortalDefinition<...>;
};
```
`InferPortalApi` extracts the interface type from a registration.
## Runtime registration [#runtime-registration]
```typescript
export default defineRuntime()
.events(events)
.portals(({ events }) => {
const { portal } = defineFor(events);
return {
slack: portal({ /* ... */ }),
controlPlane: portal({
routes: [{ method: "GET", path: "/health", handler: () => ({ status: 200, body: "ok" }) }],
}),
};
})
.actors(({ events, portals }) => {
const { actor } = defineFor(events, portals);
return {
notifier: actor({
configSchema: z.object({ channel: z.string() }),
stateSchema: z.object({}),
async process(ctx, event) {
await ctx.portals.slack.sendMessage(ctx.config.channel, "ping");
},
}),
};
})
.build();
```
The `.portals()` callback types `portals.slack` inside `.actors()` from the registered interface.
# Queues (/docs/sdk/queues)
Each actor definition can specify a **queue policy** that controls how the runtime drains matched events from the per-instance queue. Queue policy is set at definition time and serialized on each instance record.
## QueuePolicy type [#queuepolicy-type]
```typescript
type QueuePolicy =
| { mode: "fifo"; concurrency: number }
| { mode: "lifo"; concurrency: number }
| { mode: "random"; concurrency: number }
| { mode: "custom"; select: (queue) => event | event[]; concurrency: number };
```
Default: `{ mode: "fifo", concurrency: 1 }` (`DEFAULT_QUEUE_POLICY`).
There is **no `mode: "priority"`** queue policy. Per-event priority is set at append time via `{ priority: number }` and affects claim order before the mode-specific tie-breaker (`created_at` or `random()`).
## Modes [#modes]
### fifo [#fifo]
First-in-first-out among pending events, respecting priority sort:
```typescript
queue: { mode: "fifo", concurrency: 1 },
```
Best for: ordered processing, chat ingestion, sequential workflows.
### lifo [#lifo]
Most recently enqueued events processed first (after priority sort):
```typescript
queue: { mode: "lifo", concurrency: 2 },
```
Best for: "latest state wins" patterns, stack-like processing.
### random [#random]
Random selection among pending events:
```typescript
queue: { mode: "random", concurrency: 4 },
```
Best for: load spreading when order does not matter.
### custom [#custom]
Application-defined selection over the pending queue:
```typescript
queue: {
mode: "custom",
concurrency: 1,
select: (events) => {
const urgent = events.find((e) => e.type === "alert.critical");
return urgent ?? events[0];
},
},
```
The runtime lists pending events, sends them to Deno via `eval_queue_select`, and claims the returned `event_id`.
## Concurrency [#concurrency]
`concurrency` controls parallel `process()` invocations **per `group_id`**:
```typescript
queue: { mode: "fifo", concurrency: Infinity }
```
| Value | Effect |
| ---------- | -------------------------------------------- |
| `1` | One event at a time per group (default) |
| `N` | Up to N parallel `process()` calls per group |
| `Infinity` | Unbounded parallelism per group |
Use `groupId` on appended events to scope concurrency:
```typescript
await ctx.append(
{ type: "work.batch", payload: { id: "a" } },
{ groupId: "worker-a" },
);
```
## Priority at append time [#priority-at-append-time]
```typescript
await ctx.append(
{ type: "alert.warning", payload: {} },
{ priority: 10 },
);
await ctx.append(
{ type: "alert.critical", payload: {} },
{ priority: 0 },
);
```
Lower `priority` values are claimed first. Tie-break: `created_at ASC`.
## queueSelectFromConfig [#queueselectfromconfig]
Custom `select` functions are **not serializable**. Survive restart with a factory:
```typescript
const worker = defineActor({
configSchema: z.object({ preferredType: z.string() }),
stateSchema: z.object({}),
queue: { mode: "custom", concurrency: 1, select: () => { throw new Error("rebuilt"); } },
queueSelectFromConfig: (config) => (events) =>
events.find((e) => e.type === config.preferredType) ?? events[0],
// ...
});
```
On boot, the runtime calls `queueSelectFromConfig(config)` to rebuild the selector.
Without the factory, custom-queue instances fall back to broken inline `select` after restart.
## Definition example [#definition-example]
From a minimal router/coder package:
```typescript
router: actor({
queue: { mode: "fifo", concurrency: 1 },
// ...
}),
coder: actor({
queue: { mode: "fifo", concurrency: 1 },
// ...
}),
```
Agent SDK actors default to `{ mode: "fifo", concurrency: 1 }` — preserving monologue message order.
## Interaction with dequeue semantics [#interaction-with-dequeue-semantics]
Regardless of mode:
* Event dequeued **only after** `process()` succeeds
* Throw → event stays on queue
* No early-ack API
## Related [#related]
# Schedules (/docs/sdk/schedules)
**Schedules** let actor instances fire delayed stream events. Each schedule is keyed — upserting the same key replaces the previous fire time. When due, the runtime appends the configured event to the global stream (same durability as `ctx.append`).
## API [#api]
Schedules are available on `ActorCtx` and `PortalCtx`:
```typescript
// Relative delay
await ctx.schedules.schedule({
key: "reminder:order-42",
delay: { minutes: 30 },
event: {
type: "order.reminder",
payload: { orderId: "42" },
},
});
// Absolute time
await ctx.schedules.schedule({
key: "reminder:order-42",
at: Date.now() + 3_600_000,
event: { type: "order.reminder", payload: { orderId: "42" } },
});
// Cancel
await ctx.schedules.cancel("reminder:order-42");
// List
const pending = await ctx.schedules.list({ keyPrefix: "reminder:" });
```
### ScheduleInput [#scheduleinput]
```typescript
type ScheduleInput = {
key: string; // stable id for replace/cancel
event: AppendInput; // typed event to append when due
at?: number; // absolute fire time (epoch ms)
delay?: Duration; // relative — mutually exclusive with at
correlation?: AppendOptions["correlation"];
};
type Duration =
| number
| { ms?: number; seconds?: number; minutes?: number; hours?: number };
```
## Ownership [#ownership]
Schedules are **owned by the creating instance**. The kernel stores `instance_id` with each schedule row. When an instance is destroyed, its schedules are cancelled.
Use descriptive key prefixes to avoid collisions within an instance:
```typescript
function scheduleKey(forumId: string) {
return `follow-up:${forumId}`;
}
```
## Fire semantics [#fire-semantics]
When `fire_at` elapses:
1. Runtime appends the configured event to `stream_events`
2. Async fan-out delivers to matching instances
3. Schedule row is deleted (one-shot) or rescheduled (if upserted again)
Correlation from the schedule input is applied to the appended event.
## HTTP API (control plane) [#http-api-control-plane]
Headless apps can manage schedules without an actor context:
```
POST /v1/tenants/:tenant_id/schedules
GET /v1/tenants/:tenant_id/schedules
DELETE /v1/tenants/:tenant_id/schedules/:key
```
See [HTTP API reference](/docs/reference/http-api).
## Follow-up actor pattern [#follow-up-actor-pattern]
The SDK `defineFollowUpActor` uses schedules for agent nudges:
```typescript
// On agent message in forum → schedule follow-up
await ctx.schedules.schedule({
key: scheduleKey(forumId),
delay: { ms: ctx.config.delayMs },
event: {
type: "assistant.follow_up_due",
payload: {
forumId,
forumParticipantId: ctx.config.forumParticipantId,
},
},
});
// On human message → cancel
await ctx.schedules.cancel(scheduleKey(forumId));
```
When `assistant.follow_up_due` fires, the follow-up actor posts a forum message with `meta.followUp: true` so it does not re-arm itself.
## Ops bridge [#ops-bridge]
| Op | Direction | Purpose |
| ----------------- | --------- | --------------------------- |
| `schedule_upsert` | TS → Rust | Create/replace schedule |
| `schedule_cancel` | TS → Rust | Cancel by instance + key |
| `schedule_list` | TS → Rust | List schedules for instance |
## Helpers [#helpers]
```typescript
import { resolveDuration, resolveFireAt } from "@yoke/sdk";
resolveDuration({ minutes: 5, seconds: 30 }); // 330_000 ms
resolveFireAt({ delay: { hours: 1 } }); // Date.now() + 3_600_000
```
## Design notes [#design-notes]
| Use schedules for | Use append + actor filter for |
| ----------------------------------- | ------------------------------------------ |
| Delayed one-shot actions | Immediate reactions |
| Debounced reminders | Continuous polling |
| Replacing pending timers (same key) | Recurring cron (use portal `start()` loop) |
Schedules persist in Postgres — they survive runtime restart and fire when due even if the instance was idle.
## Related [#related]
# State & SQLite (/docs/sdk/state-sqlite)
Each actor instance has a private **SQLite database**. The SDK exposes two APIs on the same underlying store: a typed document abstraction (`ctx.state`) and raw SQL (`ctx.sqlite`).
## Shared database [#shared-database]
```
┌─────────────────────────────────┐
│ Per-instance SQLite database │
│ │
│ __yoke_state ← ctx.state │
│ custom tables ← ctx.sqlite │
└─────────────────────────────────┘
│
▼
pg_sqlite BYTEA in Postgres
or S3 object (YOKE_STORAGE)
```
**Never write `__yoke_state` directly.** Use `ctx.state` for the typed document. Direct SQL against other tables is fine via `ctx.sqlite`.
## ctx.state — typed document [#ctxstate--typed-document]
Backed by reserved `__yoke_state` table with JSON blob + version column.
```typescript
const coder = defineActor({
stateSchema: z.object({
jobsCompleted: z.number(),
lastModel: z.string().optional(),
}),
// ...
});
async process(ctx, event) {
const s = await ctx.state.get();
await ctx.state.set({ ...s, jobsCompleted: s.jobsCompleted + 1 });
// Atomic read-modify-write (preferred)
await ctx.state.update((s) => ({
...s,
jobsCompleted: s.jobsCompleted + 1,
lastModel: ctx.config.model,
}));
}
```
### DurableState API [#durablestate-api]
| Method | Behavior |
| ------------ | ------------------------------------------------------- |
| `get()` | Read current document (validated against `stateSchema`) |
| `set(value)` | Overwrite entire document |
| `update(fn)` | Atomic read-modify-write with optimistic locking |
`update()` calls the `state_cas` op under the hood:
```
1. state_get → { data, version }
2. Apply updater function in Deno
3. state_cas(expected_version, new_data)
4. On conflict → retry or throw
```
Concurrent updaters serialize — no lost writes.
## ctx.sqlite — raw SQL [#ctxsqlite--raw-sql]
For indexes, custom tables, and queries:
```typescript
await ctx.sqlite.exec(`
CREATE TABLE IF NOT EXISTS jobs (
id TEXT PRIMARY KEY,
status TEXT NOT NULL,
created_at INTEGER NOT NULL
)
`);
await ctx.sqlite.exec(
"INSERT INTO jobs (id, status, created_at) VALUES (?, ?, ?)",
[jobId, "pending", Date.now()],
);
const rows = await ctx.sqlite.exec(
"SELECT id, status FROM jobs WHERE status = ?",
["pending"],
);
```
Every `exec()` persists immediately to the storage backend.
## sqliteStoreId [#sqlitestoreid]
Advanced partitioning via `sqliteStoreId` on the actor definition:
```typescript
defineActor({
sqliteStoreId: (ctx) => `${ctx.instanceId}:${ctx.config.tenantShard}`,
// ...
});
```
`ctx.sqliteStoreId` in `process()` reflects the resolved store key. Default is `instanceId`.
## Initial state [#initial-state]
`stateSchema` defines the default document shape. On instance creation, the runtime initializes `__yoke_state` from the schema default (typically via Zod `.default()` or empty object parsing).
## Storage backends [#storage-backends]
| `YOKE_STORAGE` | Persistence |
| -------------- | -------------------------------------------- |
| `pg_sqlite` | BYTEA in `actor_sqlite` table |
| `s3` | Object at `{prefix}/instances/{id}/state.db` |
SDK API is identical. See [Storage](/docs/runtime/storage).
## Portal state [#portal-state]
Portals do **not** have `ctx.state` or `ctx.sqlite`. Store portal-adjacent durable data in an actor instance or Postgres via conversations API.
## Patterns [#patterns]
Use `ctx.state.update()` for simple counters and flags — atomic, typed, minimal boilerplate.
Use `ctx.sqlite.exec()` for relational data — job tables, idempotency keys, queryable history.
Agent coordinators store `JoinHandle` maps in `ctx.state` — optional registry alongside Postgres participant metadata.
## Related [#related]