electric-sql / electric-sql/electric
Add a typed inbox dispatch helper for entity handlers
- Dominant language
- TypeScript
- Stars
- 10.4k
- Forks
- 375
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 18
Description
## Summary
Entity handlers currently receive the low-level wake shape:
```ts
handler(ctx, wake)
```
For inbox messages, users typically write logic like:
```ts
if (wake.type === 'inbox' && wake.summary === 'continue-step') {
const payload = wake.payload as { step: number }
// ...
}
```
This works, but it is not very ergonomic:
- inbox message type is exposed via `wake.summary`
- payloads are `unknown`
- users manually cast payloads
- dispatch logic is repetitive
- `inboxSchemas` can describe payloads, but does not currently provide ergonomic typed dispatch inside handlers
We should explore a small helper that lets users define schema-driven typed inbox handlers while keeping the core entity handler model unchanged.
This should likely start as a userland/helper API, not a core runtime semantic change.
## Motivation
Entity handlers often need to do more than dispatch on an event:
- configure `ctx.useAgent(...)`
- configure `ctx.useContext(...)`
- compose tools
- inspect state
- spawn or observe children
- decide whether to call `ctx.agent.run()`
- schedule future self-sends
- call `ctx.sleep()`
- handle non-inbox wakes
So a top-level `handlers: { ... }` API is probably too restrictive.
Instead, we want a composable helper that can be called from inside the existing handler:
```ts
async handler(ctx, wake) {
ctx.useContext(...)
ctx.useAgent(...)
if (await inboxHandlers.handle(ctx, wake)) {
return
}
await ctx.agent.run()
}
```
This preserves the full handler lifecycle while making event routing type-safe.
## Proposed API
### Define inbox schemas
```ts
import { z } from 'zod'
import { defineInboxHandlers } from '@electric-ax/agents-runtime/helpers'
const inboxSchemas = {
continueStep: z.object({
step: z.number(),
}),
cancelJob: z.object({
reason: z.string(),
}),
}
```
### Define typed handlers
```ts
const inboxHandlers = defineInboxHandlers(inboxSchemas, {
async continueStep(ctx, event) {
event.type
// ^? "continueStep"
event.payload.step
// ^? number
await resumeStep(ctx, event.payload.step)
},
async cancelJob(ctx, event) {
event.payload.reason
// ^? string
await cancelJob(ctx, event.payload.reason)
},
})
```
### Use inside the normal entity handler
```ts
export const worker = defineEntity({
inboxSchemas,
async handler(ctx, wake) {
ctx.useContext({
sourceBudget: 20_000,
sources: {
conversation: {
content: () => ctx.timelineMessages(),
cache: 'volatile',
},
},
})
ctx.useAgent({
systemPrompt,
model,
tools,
})
if (await inboxHandlers.handle(ctx, wake)) {
return
}
await ctx.agent.run()
},
})
```
## Example: scheduled self-send
This helper pairs well with self-send scheduling:
```ts
await ctx.send(
ctx.entityUrl,
{ step: 2 },
{
type: 'continueStep',
afterMs: 30_000,
}
)
ctx.sleep()
```
Then later:
```ts
const inboxHandlers = defineInboxHandlers(inboxSchemas, {
async continueStep(ctx, event) {
await runStep(ctx, event.payload.step)
},
})
```
This avoids needing a separate `ctx.sleep({ step: 2 })` API. The runtime already supports the underlying primitive via `ctx.send(..., { afterMs })`.
## Type sketch
```ts
import type { StandardSchemaV1 } from '@standard-schema/spec'
import type {
HandlerContext,
WakeEvent,
} from '@electric-ax/agents-runtime'
type InboxSchemas = Record
type InferInput =
S extends StandardSchemaV1 ? I : never
type InferOutput =
S extends StandardSchemaV1 ? O : never
type InboxEvent<
TInbox extends InboxSchemas,
TType extends keyof TInbox & string,
> = {
kind: 'inbox'
type: TType
payload: InferOutput
rawPayload: unknown
wake: WakeEvent
}
type InboxHandler<
TInbox extends InboxSchemas,
TType extends keyof TInbox & string,
> = (
ctx: HandlerContext,
event: InboxEvent
) => void | Promise
type InboxHandlerMap = {
[K in keyof TInbox & string]?: InboxHandler
}
type InboxHandlerRouter = {
handle(ctx: HandlerContext, wake: WakeEvent): Promise
}
export function defineInboxHandlers(
inbox: TInbox,
handlers: InboxHandlerMap
): InboxHandlerRouter
```
## Runtime behavior
`handle(ctx, wake)` should:
1. Return `false` if `wake.type !== 'inbox'`.
2. Read the inbox message type from `wake.summary`.
3. Return `false` if there is no message type.
4. Return `false` if the message type is not in the schema map.
5. Return `false` if there is no handler for the message type.
6. Validate `wake.payload` against the matching schema.
7. Call the typed handler.
8. Return `true` once the handler runs.
Pseudocode:
```ts
async function handle(ctx, wake) {
if (wake.type !== 'inbox') return false
const messageType = wake.summary
if (!messageType || !(messageType in inbox)) return false
const handler = handlers[messageType]
if (!handler) return false
const schema = inbox[messageType]
const result = await validate(schema, wake.payload)
if (!result.ok) {
throw new Error(
`Invalid inbox payload for "${messageType}": ${formatIssues(result.issues)}`
)
}
await handler(ctx, {
kind: 'inbox',
type: messageType,
payload: result.value,
rawPayload: wake.payload,
wake,
})
return true
}
```
## Open questions
### 1. Where should this live?
Options:
```ts
@electric-ax/agents-runtime/helpers
```
or:
```ts
@electric-ax/agents-runtime/match
```
or initially in docs/examples only.
Recommendation: start as docs/example or a small helper export, not a runtime semantic change.
### 2. Should invalid payloads throw or return an unhandled result?
Possible behaviors:
```ts
throw new Error(...)
```
or:
```ts
return {
handled: false,
reason: 'invalid_payload',
issues,
}
```
Throwing is simple and surfaces as a normal handler failure. Returning structured results gives users more control.
Maybe support both:
```ts
defineInboxHandlers(inbox, handlers, {
onInvalidPayload: 'throw',
})
```
or:
```ts
defineInboxHandlers(inbox, handlers, {
async onInvalidPayload(ctx, event) {
// custom behavior
},
})
```
### 3. Should unknown message types fall through?
Default should probably be fallthrough:
```ts
return false
```
That lets the main handler run the agent loop or fallback behavior.
Optionally support strict mode:
```ts
defineInboxHandlers(inbox, handlers, {
unknown: 'throw',
})
```
### 4. Should handlers be exhaustive?
Default should allow partial handlers:
```ts
defineInboxHandlers(inbox, {
continueStep(ctx, event) {},
})
```
Optional helper for exhaustive handling:
```ts
defineExhaustiveInboxHandlers(inbox, {
continueStep(ctx, event) {},
cancelJob(ctx, event) {},
})
```
Missing keys would be a TypeScript error.
### 5. Should this cover all wake kinds?
Initial scope should be inbox messages only.
Future helpers might support:
- observed child run finished
- cron wakes
- external event source wakes
- shared state/source change wakes
But inbox messages are the clearest first slice because they already have `inboxSchemas` and message type.
### 6. Should `ctx.send` be typed from these schemas?
Eventually, yes.
If the router knows the inbox schema map, it could expose a typed send helper:
```ts
inboxHandlers.send(ctx, ctx.entityUrl, 'continueStep', {
step: 2,
}, {
afterMs: 30_000,
})
```
or:
```ts
await sendInbox(ctx, ctx.entityUrl, inboxSchemas, 'continueStep', {
step: 2,
})
```
But that can be a follow-up.
## Non-goals
- Do not replace `handler(ctx, wake)`.
- Do not require users to define event handlers separately from the normal handler.
- Do not make agent loop execution implicit.
- Do not change runtime wake semantics.
- Do not rename `wake.summary` in this issue, though a future cleanup could add a clearer alias like `wake.messageType` for inbox wakes.
## Why this design
This follows TanStack-style TypeScript ergonomics:
- schema-driven inference
- normal object/function definitions
- callback map keyed by semantic names
- no explicit generics for common usage
- extracted helper preserves inference
- compatibility with lower-level manual handling
The result is a small composable helper that improves developer experience without committing the runtime to a new entity handler model.
Contributor guide
Assessment
This issue has not been assessed yet.