electric-sql / electric-sql/electric

Add a typed inbox dispatch helper for entity handlers

Open
#4,519 0 comments 0 reactions 0 assignees View on GitHub
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

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.