Inngest

# Inngest

## Overview

The module lives in `features/inngest/`. It defines the shared Inngest clients,
typed event names, and function files that run Amble background work outside the
request lifecycle.

Inngest is the async runtime behind [Actions](/docs/features/actions),
[Sessions](/docs/features/sessions), document processing, outbound
[Webhooks](/docs/features/webhooks), notification digests, feedback reruns,
entity embeddings, and platform maintenance jobs. The functions are served by
five domain-specific Next.js routes under `app/api/inngest/*`, not by a single
aggregate route.

## Key Concepts

**Domain clients** - `features/inngest/client.ts` exports one `Inngest` client
per route:

```ts
export const inngestActions = new Inngest({
  id: "amble-actions",
  ...baseClientOptions,
});
export const inngestSessions = new Inngest({
  id: "amble-sessions",
  ...baseClientOptions,
});
export const inngestDocuments = new Inngest({
  id: "amble-documents",
  ...baseClientOptions,
});
export const inngestComms = new Inngest({
  id: "amble-comms",
  ...baseClientOptions,
});
export const inngestEntities = new Inngest({
  id: "amble-entities",
  ...baseClientOptions,
});
```

Every function must be created on the matching per-domain client. The exported
`inngest` alias points at `inngestActions` only so older event publishers can
keep calling `.send()`.

**Event names** - `EVENT_NAMES` in `features/inngest/client.ts` is the platform
event-name registry. It includes `entity/created`, `entity/updated`,
`document/uploaded`, `webhook/fire`, `actions/tick`, `session/execute`,
`session/completed`, `session/resume`, `integrations/webhook.received`, and
other platform events.

**Routes** - each route exports `maxDuration = 300` and serves one domain
client:

| Route                                | Client             | Function group                                                                                                                                        |
| ------------------------------------ | ------------------ | ----------------------------------------------------------------------------------------------------------------------------------------------------- |
| `app/api/inngest/actions/route.ts`   | `inngestActions`   | action cron, entity-event action dispatch, action ticks, integration webhook receipt, model sync, observability archival, shared-knowledge sync       |
| `app/api/inngest/sessions/route.ts`  | `inngestSessions`  | session executor, session resume, elicitation resume, session completion, budget rollup, session reaper, feedback-view reaper                         |
| `app/api/inngest/documents/route.ts` | `inngestDocuments` | document processing, entity embedding, video render                                                                                                   |
| `app/api/inngest/comms/route.ts`     | `inngestComms`     | email delivery, email processing, notification digest, notification digest cron, webhook delivery                                                     |
| `app/api/inngest/entities/route.ts`  | `inngestEntities`  | entity enrichment, feedback rerun, claim aggregate recompute, edge-score repair, capture routing, automation-ratio snapshots, custom entity functions |

**Cron gate** - scheduled functions call `isCronAllowed()` from `lib/env.ts`.
Production Vercel runs cron. Preview and development Vercel always skip cron.
Local development and CI skip cron unless `ENABLE_LOCAL_CRON=1` is set.

```ts
export function isCronAllowed(): boolean {
  if (process.env.VERCEL_ENV === "production") return true;
  if (process.env.VERCEL_ENV) return false;
  return process.env.ENABLE_LOCAL_CRON === "1";
}
```

**Sentry failure middleware** - `features/inngest/client.ts` defines
`SentryFailureMiddleware`, which reports terminal Inngest function failures in
`onRunError` only when `isFinalAttempt` is true.

## Architecture And Data Flow

### Action Cron And Action Tick

`features/inngest/functions/action-cron.ts` is the once-per-minute scanner for
cron actions:

```ts
export const actionCron = inngestActions.createFunction(
  {
    id: "task-cron",
    name: "Action Cron Scanner",
    concurrency: [{ limit: 1, scope: "fn" }],
    retries: 0,
    triggers: [{ cron: "*/1 * * * *" }],
  },
  async ({ event, step }) => {
    // ...
  },
);
```

After the cron gate passes, it finds active cron `actions` rows whose
`trigger_config.schedule` matches the current minute and emits
`EVENT_NAMES.ACTIONS_TICK` events.

`features/inngest/functions/action-tick.ts` is the unified dispatch seam from
ADR-0016. It runs with a per-action concurrency key and branches in this
priority order:

1. `queue_config` is set: call `runQueueTick(action)` and emit
   `session/execute` for spawned sessions.
2. An action definition key is present: run `executeActionDefinition()` inside
   `withToolContext({ tenantId, workspaceId })`.
3. `metadata.loop` is set: validate the `LoopSpec` and call `runLoopBranch()`.
4. `agent_slug` is set: create a scheduled-agent session and call
   `invokeAgentAutonomousRun()` inside `withToolContext({ tenantId, agentSlug })`.
5. Fallback: call `triggerTask()` and emit `session/execute`.

The source code validates the claimed action row with `actionTickRowSchema` at
the seam before branching. A malformed dispatch-shaping field throws a
`NonRetriableError`.

### Session Execution

`features/inngest/functions/session-executor.ts` consumes `session/execute` and
`session/completed`. It loads parent and child sessions, runs ready agent
sessions through `executeAgentSession()`, marks blocked or human sessions, and
emits completion events so downstream sessions advance.

The same file exports `sessionResume`, which consumes `session/resume` for
approval-tool continuations.

### Documents, Communications, And Entities

`document-processing` loads uploaded files from Supabase Storage, parses them,
chunks pages, can run Gemini vision extraction for image or scanned PDF
candidates, stores document pages, generates embeddings, and triggers entity
extraction when the document is linked to an entity.

`webhook-delivery` consumes `webhook/fire`, selects enabled
`webhook_endpoints` for the tenant and optional workspace, signs a JSON body
with `signPayload()`, posts to the endpoint, and updates failure counters.

`notification-digest-cron` is gated by `isCronAllowed()`, lists tenants, and
emits `notification/digest` events. `notification-digest` applies user
frequency preferences and emits email work.

Entity-domain functions cover feedback reruns, entity enrichment, edge-score
repair, capture URL routing, claim aggregate recompute, and automation-ratio
snapshots.

## Key APIs

```ts
export { inngest, type Events } from "./client";
```

`features/inngest/index.ts` re-exports the compatibility `inngest` alias and
the typed `Events` map.

```ts
export const EVENT_NAMES = {
  ENTITY_CREATED: "entity/created",
  ENTITY_UPDATED: "entity/updated",
  DOCUMENT_UPLOADED: "document/uploaded",
  WEBHOOK_FIRE: "webhook/fire",
  ACTIONS_TICK: "actions/tick",
  SESSION_EXECUTE: "session/execute",
  SESSION_COMPLETED: "session/completed",
  SESSION_RESUME: "session/resume",
  INTEGRATIONS_WEBHOOK_RECEIVED: "integrations/webhook.received",
} as const;
```

Use the real `EVENT_NAMES` export instead of raw strings when adding a producer
or consumer.

```ts
export const actionTick = inngestActions.createFunction(
  {
    id: "action-tick",
    name: "Action Tick (queue / agent / single-fire)",
    concurrency: [{ limit: 1, scope: "fn", key: "event.data.actionId" }],
    retries: 2,
    triggers: [{ event: EVENT_NAMES.ACTIONS_TICK }],
  },
  async ({ event, step, runId, attempt, maxAttempts }) => {
    // ...
  },
);
```

`actionTick` is the extension point for new action dispatch modes. Add a new
branch here only when the branch belongs to the action/session spine rather
than a separate orchestrator.

```ts
export function isCronAllowed(): boolean;
```

Use this gate in every scheduled Inngest function before tenant fan-out or AI
work.

## Agent Instructions

When adding an Inngest function:

1. Add the event name to `EVENT_NAMES` and the `Events` type in
   `features/inngest/client.ts`.
2. Create the function in `features/inngest/functions/` using the correct
   domain client, for example `inngestActions.createFunction(...)` for action
   dispatch or `inngestSessions.createFunction(...)` for session work.
3. Register the function in exactly one domain route under `app/api/inngest/*`.
4. Use `isCronAllowed()` for scheduled functions before scanning tenants,
   actions, or other global rows.
5. For tenant-scoped work, pass tenant context explicitly through
   `withToolContext({ tenantId })`, `withToolContext({ tenantId, workspaceId })`,
   or a service/admin path that filters by `tenant_id`.

Inngest workers must not call `requireAuth()` or `requireAdmin()`. They do not
run inside an HTTP request and have no browser cookies or tenant headers.
Background work should use tenant-scoped service access, `withToolContext()`,
and explicit `tenant_id` filters.

Do not restore a root `app/api/inngest/route.ts` aggregate endpoint. The
current runtime is the five-domain split described in
`features/inngest/README.md`.