> ## Documentation Index
> Fetch the complete documentation index at: https://iii.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Observers

> When one function runs, run another: worker.observe(a, b).

## Observe a function

`observe(a, b)` means: when `a` runs, `b` runs.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe("orders::create", "alerts::new-order");
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe("orders::create", "alerts::new-order")
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe("orders::create", "alerts::new-order");
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.ObserveFunction("orders::create", "alerts::new-order")
> ```

An observer connects two functions in one direction. After each successful call of the observed
function (`a`), iii runs the target (`b`) with the call's input, its result and its caller.

The examples on this page react to `orders::create`, a function that another worker registers.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { registerWorker, type Observed } from "iii-sdk";
> import { Logger } from "@iii-dev/helpers/observability";
>
> const url = process.env.III_URL;
> if (!url) throw new Error("III_URL must be set");
> const worker = registerWorker(url);
> const logger = new Logger();
>
> type Order = { id: string; email: string; total: number };
>
> // The target: a regular function. It receives the Observed payload of each call.
> worker.registerFunction("alerts::new-order", async (event: Observed<Order>) => {
>   logger.info(`order ${event.input?.id} totals ${event.input?.total}`);
>   return null;
> });
>
> // When orders::create runs, run alerts::new-order.
> const newOrders = worker.observe("orders::create", "alerts::new-order");
> await newOrders.ready(); // resolves once the observer is active
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
> from iii import InitOptions, register_worker
> from iii_helpers.observability import Logger
>
> worker = register_worker(os.environ.get("III_URL"), InitOptions(worker_name="alerts"))
> logger = Logger()
>
> # The target: a regular function. It receives the Observed payload of each call.
> def new_order(event: dict) -> None:
>     order = event.get("input", {})
>     logger.info(f"order {order.get('id')} totals {order.get('total')}")
>
> worker.register_function("alerts::new-order", new_order)
>
> # When orders::create runs, run alerts::new-order.
> new_orders = worker.observe("orders::create", "alerts::new-order")
> new_orders.ready()  # blocks until active; use `await new_orders.ready_async()` in asyncio code
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_helpers::observability::Logger;
> use iii_sdk::{Error, InitOptions, Observed, RegisterFunction, register_worker};
> use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
> use serde::{Deserialize, Serialize};
>
> #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
> struct Order {
>     id: String,
>     email: String,
>     total: f64,
> }
>
> let url = std::env::var("III_URL").expect("III_URL must be set");
> let worker = register_worker(&url, InitOptions::default());
>
> // The target: a regular function. It receives the Observed payload of each call.
> worker.register_function(
>     "alerts::new-order",
>     RegisterFunction::new(|event: Observed<Order>| {
>         if let Some(order) = &event.input {
>             Logger::new().info(&format!("order {} totals {}", order.id, order.total), None);
>         }
>         Ok::<_, Error>(())
>     }),
> );
>
> // When orders::create runs, run alerts::new-order.
> let new_orders = worker.observe("orders::create", "alerts::new-order");
> new_orders.ready().await?; // resolves once the observer is active
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import "github.com/iii-hq/iii/sdk/packages/go/iii/observability"
>
> type Order struct {
>     ID    string  `json:"id"`
>     Email string  `json:"email"`
>     Total float64 `json:"total"`
> }
>
> var logger observability.Logger
>
> worker := iii.RegisterWorker(os.Getenv("III_URL"))
>
> // The target: a regular function. It receives the Observed payload of each call.
> worker.RegisterFunction("alerts::new-order", func(ctx context.Context, data json.RawMessage) (any, error) {
>     var event iii.Observed
>     if err := json.Unmarshal(data, &event); err != nil {
>         return nil, err
>     }
>     logger.Info(ctx, "new order", event.Input)
>     return nil, nil
> })
>
> // When orders::create runs, run alerts::new-order.
> newOrders, err := worker.ObserveFunction("orders::create", "alerts::new-order")
> if err != nil {
>     log.Fatal(err) // the request could not be built (a missing id)
> }
> if err := newOrders.Ready(ctx); err != nil {
>     log.Fatal(err) // an *iii.ObserveRejectionError carries the engine's code
> }
> ```

### What can be observed

The observed function can be any function on the engine: one that your team wrote, or one a worker
hosts to expose something that happens, such as an HTTP request (`http::on-request`), a queue
message (`queue::on-message`) or a row change (`database::on-row-changed`). The target is any
registered function. Any worker can observe any function, private ones included, unless the function
is registered with `observable: false` or RBAC restricts it (see [Authorization](#authorization)).

By default, an observer is accepted as soon as it is added, and its target runs on every call its
`capture` fires on and its filters match. The worker that hosts the function can override each part
of that with a hook: `onObserverAdded` sets something up for the observer, or rejects it;
`onObservationMatch` decides which calls reach which observer; and `onObservationDeliver` delivers
the calls itself and decides what the caller gets back (see
[Hooks](https://iii.dev/docs/next/creating-workers/observable-functions.md#hooks)). If unsure, refer to that worker's
documentation.

### Defaults

* The target runs after each successful call of the observed function.
* The target receives the [`Observed` payload](#the-payload) of the call, with its `input`,
  its `output`, and its `caller`.
* The caller of the observed function gets its result without waiting for the target. Each call runs
  the target at most once, and a failure of the target does not change the call.
* The observer stays `pending` until both functions are registered, in any order. The
  [handle](#observer-handle) that `observe` returns reports this in its `state` property, and its
  `ready()` resolves once the observer is `active`. Calls that happen before then can be missed.
* You call `observe` once at startup, like `registerFunction`. It returns immediately, even before
  the connection is up: the SDK sends the observer when it connects and replays it with the same id
  after every reconnect.

> [!NOTE]
>
> This page is about running your function when another one runs. To decide what happens when the
> functions your worker hosts are observed, see [Observable
> functions](https://iii.dev/docs/next/creating-workers/observable-functions.md). Coming from 0.25 triggers? `registerTrigger`
> becomes `observe`: see [Upgrading from 0.25.x to 0.26.x](https://iii.dev/docs/next/upgrading/from-0-25-x.md).

## Args and filters

An observer can carry `args` and `filters`. `args` go to the worker that hosts the function, which
reads them to set something up for the observer (a route, a database trigger, a queue consumer) and
to filter which calls the observer receives. This is custom logic: each function defines the `args`
its use case needs. `filters` are standard: the iii engine applies them to each call, the same way
for every function.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe(
>   {
>     function_id: "database::on-row-changed",
>     args: { db: "primary", table: "orders", ops: ["insert", "update"] }, // for the database worker
>     filters: { "caller.parent.worker_name": { $ne: "order-projector" } }, // for iii
>   },
>   "orders::project",
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe(
>     {
>         "function_id": "database::on-row-changed",
>         "args": {"db": "primary", "table": "orders", "ops": ["insert", "update"]},  # for the database worker
>         "filters": {"caller.parent.worker_name": {"$ne": "order-projector"}},  # for iii
>     },
>     "orders::project",
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::Source;
> use serde_json::json;
>
> worker.observe(
>     Source::function("database::on-row-changed")
>         // for the database worker
>         .args(json!({ "db": "primary", "table": "orders", "ops": ["insert", "update"] }))
>         // for iii
>         .filters(json!({ "caller.parent.worker_name": { "$ne": "order-projector" } })),
>     "orders::project",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.Observe(iii.Source{
>     FunctionID: "database::on-row-changed",
>     // for the database worker
>     Args: iii.Args{"db": "primary", "table": "orders", "ops": []string{"insert", "update"}},
>     // for iii
>     Filters: iii.Filters{"caller.parent.worker_name": iii.Op{"$ne": "order-projector"}},
> }, "orders::project")
> ```

### Args

The engine checks `args` against the `args_format` of the function, a JSON Schema the function's
worker declares when it registers the function. The observer fails with `INVALID_ARGS` when the
`args` do not match, or when the function has no `args_format` and the `args` are not empty. The
worker can also refuse an observer whose `args` it cannot serve, with a code of its own (for example
`UNKNOWN_DB` or `HTTP_ROUTE_CONFLICT`). Each function documents its `args`: see its worker's page on
[workers.iii.dev](https://workers.iii.dev).

### Filters

A filter is a JSON object. Each key is a path in the call that starts with `input`, `output`,
`error`, `caller`, or `metadata`. Each value is an operator object, for example `{ "$eq": "EUR" }`.
A literal value such as `"EUR"` is not allowed.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.registerFunction("alerts::big-order", async (event: Observed<Order>) => {
>   logger.info(
>     `order ${event.input?.id} totals ${event.input?.total}, placed by ${event.caller?.worker_name}`,
>   );
>   return null;
> });
>
> // After each successful call of orders::create with total >= 500, run alerts::big-order.
> await worker
>   .observe(
>     { function_id: "orders::create", filters: { "input.total": { $gte: 500 } } },
>     "alerts::big-order",
>   )
>   .ready();
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> def big_order(event: dict) -> None:
>     order = event.get("input", {})
>     caller = event.get("caller", {})
>     logger.info(f"order {order.get('id')} totals {order.get('total')}, placed by {caller.get('worker_name')}")
>
> worker.register_function("alerts::big-order", big_order)
>
> # After each successful call of orders::create with total >= 500, run alerts::big-order.
> worker.observe(
>     {"function_id": "orders::create", "filters": {"input.total": {"$gte": 500}}},
>     "alerts::big-order",
> ).ready()
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::Source;
> use serde_json::json;
>
> worker.register_function(
>     "alerts::big-order",
>     RegisterFunction::new(|event: Observed<Order>| {
>         if let Some(order) = &event.input {
>             Logger::new().info(&format!("order {} totals {}", order.id, order.total), None);
>         }
>         Ok::<_, Error>(())
>     }),
> );
>
> // After each successful call of orders::create with total >= 500, run alerts::big-order.
> worker
>     .observe(
>         Source::function("orders::create").filters(json!({ "input.total": { "$gte": 500 } })),
>         "alerts::big-order",
>     )
>     .ready()
>     .await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.RegisterFunction("alerts::big-order", func(ctx context.Context, data json.RawMessage) (any, error) {
>     var event iii.Observed
>     if err := json.Unmarshal(data, &event); err != nil {
>         return nil, err
>     }
>     logger.Info(ctx, "big order", event.Input)
>     return nil, nil
> })
>
> // After each successful call of orders::create with total >= 500, run alerts::big-order.
> bigOrders, err := worker.Observe(iii.Source{
>     FunctionID: "orders::create",
>     Filters:    iii.Filters{"input.total": iii.Op{"$gte": 500}},
> }, "alerts::big-order")
> if err != nil {
>     log.Fatal(err)
> }
> if err := bigOrders.Ready(ctx); err != nil {
>     log.Fatal(err)
> }
> ```

#### Operators

The operators of one operator object must all match:
`{ "input.total": { "$gte": 100, "$lt": 500 } }` is a range. See the
[filter language reference](https://iii.dev/docs/next/reference/filters.md) for the full rules and the limits.

##### Comparison

`$eq`, `$ne`, `$in`, `$nin`, `$gt`, `$gte`, `$lt`, `$lte`. Equality is JSON deep equality, with no
coercion between types. Ordering applies within one type only: numbers compare numerically, strings
by code point.

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{
  "input.currency": { "$eq": "EUR" },
  "caller.worker_name": { "$ne": "importer" },
  "input.country": { "$in": ["DE", "FR"] },
  "input.status": { "$nin": ["draft", "cancelled"] },
  "input.total": { "$gt": 0, "$lte": 10000 },
  "input.quantity": { "$gte": 2 },
  "input.created_at": { "$lt": "2025-01-01" }
}
```

##### Existence and `null`

`$exists` takes a boolean and matches on the presence of the path. `null` is a value:
`{ "$exists": true }` matches `null`, and `{ "$eq": null }` matches only `null`, and will not match
a missing path. Only `$ne`, `$nin`, and `$exists: false` match a missing path.

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{
  "error": { "$exists": true },
  "input.coupon": { "$exists": false },
  "input.parent_id": { "$eq": null }
}
```

##### Strings

`$prefix` matches a string that starts with the operand.

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{ "input.path": { "$prefix": "/api/" } }
```

##### Arrays

A path does not read inside an array. When `input.items` is an array of objects, the path
`input.items.sku` is a missing path: it does not read the `sku` of each item. To match on the
elements, use `$any`. It matches when at least one element matches its operand. For scalar elements,
the operand is an operator object. For object elements, the operand is a filter whose paths start at
`@`, the current element.

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{
  "input.tags": { "$any": { "$eq": "vip" } },
  "input.items": { "$any": { "@.sku": { "$eq": "A-1" }, "@.qty": { "$gte": 10 } } }
}
```

##### Logical

`$and` and `$or` take a non-empty array of filters. `$not` takes one filter.

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{
  "$and": [{ "input.total": { "$gte": 500 } }, { "input.currency": { "$eq": "EUR" } }],
  "$or": [{ "input.priority": { "$eq": "high" } }, { "input.total": { "$gte": 10000 } }],
  "$not": { "input.path": { "$prefix": "/health" } }
}
```

> [!NOTE]
>
> `$and`, `$or` and `$not` are clauses of a filter. They cannot go inside an operator object on a
> path.

#### Validation

When the observer is created, the SDK and the engine check the filter. If it is not valid, the
observer fails with `INVALID_FILTER`. The engine also checks the `input.*` and `output.*` paths
against the `request_format` and `response_format` of the function. If a closed schema does not have
a path, the observer fails with `INVALID_FILTER_PATH`. If the function is not registered yet, the
engine does this check when the function registers.

## What the target receives

`capture` sets what the target receives, and with it after which calls the target runs.

### The payload

The target receives the `Observed` payload of the call. It always has the ids, `status`, and the
timings. It has `input`, `output`, `error`, `caller`, and `metadata` when [`capture`](#capture)
includes them and the call has them: `output` only after a successful call, `error` only after a
failed one. A field that is not captured, or that the call does not have, is absent, never `null`.
This differs from 0.25, which sent `"output": null` to an observer that captured `output` when a
successful call reported no output.

| Field | Meaning |
| - | - |
| `observer_id`, `activation_id` | The observer, and this activation of it. |
| `function_id`, `namespace` | The observed function. |
| `invocation_id` | The observed call. |
| `status` | `success` or `error`. |
| `input`, `output`, `error` | The observed payload, its result, or its `{ code, message, stacktrace? }` error. |
| `caller` | Who made the observed call. See [Caller](#caller-and-causality). |
| `metadata` | The metadata of the observed call. |
| `started_at_ms`, `duration_ms` | When the observed invocation started, and how long it ran. |

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.registerFunction("emails::send-receipt", async ({ input }: Observed<Order>) => {
>   await sendReceipt(input.email, input);
>   return null;
> });
>
> // After each successful call of orders::create, run emails::send-receipt with the order.
> worker.observe({ function_id: "orders::create", capture: ["input"] }, "emails::send-receipt");
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> def send_receipt(observed: dict) -> None:
>     order = observed["input"]
>     mailer.send_receipt(order["email"], order)
>
> worker.register_function("emails::send-receipt", send_receipt)
>
> # After each successful call of orders::create, run emails::send-receipt with the order.
> worker.observe({"function_id": "orders::create", "capture": ["input"]}, "emails::send-receipt")
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Capture, Observed, Source};
>
> worker.register_function(
>     "emails::send-receipt",
>     RegisterFunction::new(|observed: Observed<Order>| {
>         let order = observed.input.expect("captured");
>         send_receipt(&order.email, &order);
>         Ok::<_, Error>(())
>     }),
> );
>
> // After each successful call of orders::create, run emails::send-receipt with the order.
> worker.observe(
>     Source::function("orders::create").capture([Capture::Input]),
>     "emails::send-receipt",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.RegisterFunction("emails::send-receipt", func(ctx context.Context, data json.RawMessage) (any, error) {
>     var observed iii.Observed
>     if err := json.Unmarshal(data, &observed); err != nil {
>         return nil, err
>     }
>     var order Order
>     if err := json.Unmarshal(observed.Input, &order); err != nil {
>         return nil, err
>     }
>     return nil, sendReceipt(ctx, order.Email, order)
> })
>
> // After each successful call of orders::create, run emails::send-receipt with the order.
> worker.Observe(iii.Source{
>     FunctionID: "orders::create",
>     Capture:    []iii.Capture{iii.CaptureInput},
> }, "emails::send-receipt")
> ```

### `capture`

`capture` lists the fields the payload carries: any of `input`, `output`, `error`, `caller`, and
`metadata`. The default is `["input", "output", "caller"]`.

`metadata` is opt-in: the call's metadata reaches the target only when `capture` names it. Earlier
0.26 release candidates of this change added `metadata` to the default; 0.26 does not. To route
calls by their metadata (for example a `for_observer` key), use the hosting worker's
[`onObservationMatch` or `onObservationDeliver`](../creating-workers/observable-functions#hooks)
hook, which receives the call's metadata whatever the observers capture. A [filter](#filters) can
also read `metadata.*` without `metadata` in `capture`.

The fields also decide after which calls the target runs:

* `input` and `output` run it after a successful call.
* `error` runs it after a failed call.
* `caller` and `metadata` add to the payload only. A `capture` must name at least one of `input`,
  `output`, and `error`; one with only `caller` or `metadata` fails with `INVALID_ARGS`. The
  exception is `capture: []`, which runs the target after successful calls with no payload fields.
* `[]` runs the target after a successful call, with the ids, `status`, and the timings only: no
  payload moves to it. An observer that only counts calls uses it.

| `capture` | The target runs after |
| - | - |
| default, `["input"]`, `["output", "caller"]` | a successful call |
| `["error"]`, `["error", "caller"]` | a failed call |
| `["input", "error"]`, `["output", "error"]` | each call, successful or failed |
| `[]` | a successful call (no payload) |

The default does not include `error`, so a default observer runs after successful calls only. A
call the engine cancels (for example, a call it routes whose caller stopped waiting) does not run
the target, for any `capture`. The engine counts it as a dropped call with the reason `CANCELLED`
in the `iii.observer.dropped.total` metric. A call the hosting worker runs in its own process is
not cancelled when its caller stops waiting: its handler runs to the end, and its outcome reaches
the observers like any other.

Once `capture` has decided which outcomes run the target, a [filter](#filters) can read any field
of the call, including the fields that `capture` leaves out of the payload.

To keep observed calls small, capture only what the target reads. See
[Observed call size](#observed-call-size).

To run the target after failed calls with their error, capture `error`:

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // After each failed call: the error, and who made the call.
> worker.observe(
>   { function_id: "orders::create", capture: ["error", "caller"] },
>   "alerts::order-failed",
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> # After each failed call: the error, and who made the call.
> worker.observe(
>     {"function_id": "orders::create", "capture": ["error", "caller"]},
>     "alerts::order-failed",
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Capture, Source};
>
> // After each failed call: the error, and who made the call.
> worker.observe(
>     Source::function("orders::create").capture([Capture::Error, Capture::Caller]),
>     "alerts::order-failed",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // After each failed call: the error, and who made the call.
> worker.Observe(iii.Source{
>     FunctionID: "orders::create",
>     Capture:    []iii.Capture{iii.CaptureError, iii.CaptureCaller},
> }, "alerts::order-failed")
> ```

To get the input of every call, successful or failed, capture `input` and `error`: the target runs
after both, and the payload's `status` says which.

#### Failed calls with their input

To run the target after failed calls only, and still get each call's input, capture `input` and
`error` and keep only the calls that have an `error` with a filter:

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // After each failed call only: its input and its error.
> worker.observe(
>   {
>     function_id: "orders::create",
>     capture: ["input", "error"],
>     filters: { error: { $exists: true } },
>   },
>   "alerts::order-failed",
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> # After each failed call only: its input and its error.
> worker.observe(
>     {
>         "function_id": "orders::create",
>         "capture": ["input", "error"],
>         "filters": {"error": {"$exists": True}},
>     },
>     "alerts::order-failed",
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Capture, Source};
> use serde_json::json;
>
> // After each failed call only: its input and its error.
> worker.observe(
>     Source::function("orders::create")
>         .capture([Capture::Input, Capture::Error])
>         .filters(json!({ "error": { "$exists": true } })),
>     "alerts::order-failed",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // After each failed call only: its input and its error.
> worker.Observe(iii.Source{
>     FunctionID: "orders::create",
>     Capture:    []iii.Capture{iii.CaptureInput, iii.CaptureError},
>     Filters:    iii.Filters{"error": iii.Op{"$exists": true}},
> }, "alerts::order-failed")
> ```

`capture: ["input", "error"]` fires on both outcomes, and the filter drops the successful calls
before they reach the observing worker. The engine evaluates it for the calls it routes and starts
no target for a successful one. For the calls a worker on a 0.26 SDK runs in-process, that SDK
evaluates it and does not report a successful call to the engine at all. After each failed call
the target gets `status: "error"`, the `input`, and the `error`. Capturing `error` alone would
fire on failures only, but its payload would not carry the input.

The observing worker needs a 0.26 SDK. An observer from a worker without the `observe.capture`
feature keeps the 0.25 semantics and fires on successful calls only, so with this filter it never
fires (see [Compatibility](#compatibility)).

### Metadata

The observer's `metadata` option, set in the [policy](#policy-and-metadata), reaches the target
function as its second argument on every activation.

> [!NOTE]
>
> The payload's `metadata` field is the metadata of the observed call, present only when `capture`
> names `metadata`: the default leaves it out. The observer's `metadata` option is the target
> function's second argument.

### The target function's return value

The iii engine does not use the return values of target functions. However, an observable function
can use
[`onObservationDeliver`](../creating-workers/observable-functions#onobservationdelivercall-observers)
to get that return value from the target function and use it, for example to answer an HTTP request
or to acknowledge a queue message. The function's documentation says what its target functions must
return. Likewise, observers of a target function receive the return value as the `output` of its
call. See [Chain reactions](#chain-reactions).

## Policy and metadata

An observer can carry a policy. The policy sets limits for the observer itself. An activation is one
run of the target.

| Option | Default | Meaning |
| - | - | - |
| `id` | a random UUID | The observer id. An `observe` with an id that a live handle of this worker, or another worker's observer, already holds fails with `OBSERVER_ID_CONFLICT`. The id is free again once that handle is disposed. With an id of your own, a restarted worker takes back its `suspended` observer (see [Observer states](#observer-states)). |
| `max_fires` | unlimited | The observer becomes `disposed` after this number of activations (reason `MAX_FIRES`). |
| `expires_in_ms` | never | The observer becomes `disposed` after this time (reason `EXPIRED`). |
| `max_in_flight` | `observers.default_max_in_flight` (64) | The maximum number of activations that run at the same time. More activations wait in a queue of `observers.max_queue_per_observer` (1024). When the queue is full, or when the queues of all observers hold 64 MiB of observed calls, the engine drops new activations (`OVERFLOW`). With `1`, activations run in the order they arrive. |
| `metadata` | none | JSON that each activation gives to the target as its second argument. See [Metadata](#metadata). |

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe(
>   {
>     function_id: "payments::charge",
>     filters: { "input.amount": { $gte: 500 } },
>     capture: ["error", "caller"],
>   },
>   "alerts::payment-failed",
>   { max_in_flight: 4, metadata: { team: "billing" } },
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe(
>     {
>         "function_id": "payments::charge",
>         "filters": {"input.amount": {"$gte": 500}},
>         "capture": ["error", "caller"],
>     },
>     "alerts::payment-failed",
>     {"max_in_flight": 4, "metadata": {"team": "billing"}},
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Capture, ObserverOptions, Source};
> use serde_json::json;
>
> worker.observe_with(
>     Source::function("payments::charge")
>         .filters(json!({ "input.amount": { "$gte": 500 } }))
>         .capture([Capture::Error, Capture::Caller]),
>     "alerts::payment-failed",
>     ObserverOptions::default()
>         .max_in_flight(4)
>         .metadata(json!({ "team": "billing" })),
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.Observe(iii.Source{
>     FunctionID: "payments::charge",
>     Filters:    iii.Filters{"input.amount": iii.Op{"$gte": 500}},
>     Capture:    []iii.Capture{iii.CaptureError, iii.CaptureCaller},
> }, "alerts::payment-failed", iii.ObserverOptions{
>     MaxInFlight: 4,
>     Metadata:    json.RawMessage(`{"team":"billing"}`),
> })
> ```

The engine does not retry a failed activation. A worker that delivers calls to the observers itself
can retry on its own terms, which its documentation states. The engine counts each dropped call in
the `iii.observer.dropped.total` metric, by function and reason.

## Observer handle

`observe` returns a handle with the observer's `id` and `state`, and three methods:

* `ready()` resolves when the observer is `active`. It rejects with an `ObserveRejection` that
  carries the reason's code when the observer turns `error` or `disposed` first.
* `onStateChange(listener)` reports each state change with its reason, and returns a function that
  unsubscribes. Listeners run outside the connection's read path, so a listener can call the worker.
* `dispose()` removes the observer. The SDK does not replay a disposed observer after a reconnect.
  On an observer the engine refused, `dispose()` has nothing to remove and makes no request to the
  engine.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> const newOrders = worker.observe("orders::create", "alerts::new-order");
>
> newOrders.onStateChange((state, reason) => {
>   logger.info(`observer ${newOrders.id}: ${state}`, { reason });
> });
> await newOrders.ready();
>
> // later: remove the observer
> await newOrders.dispose();
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> new_orders = worker.observe("orders::create", "alerts::new-order")
>
> def log_state(state, reason):
>     logger.info(f"observer {new_orders.id}: {state}")
>
> new_orders.on_state_change(log_state)
> new_orders.ready()
>
> # later: remove the observer
> new_orders.dispose()  # or: await new_orders.dispose_async()
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> let new_orders = worker.observe("orders::create", "alerts::new-order");
>
> let _listener = new_orders.on_state_change(|state, reason| {
>     println!("observer state: {state:?} {reason:?}");
> });
> new_orders.ready().await?;
>
> // later: remove the observer
> new_orders.dispose().await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> newOrders, err := worker.ObserveFunction("orders::create", "alerts::new-order")
> if err != nil {
>     log.Fatal(err)
> }
>
> unsubscribe := newOrders.OnStateChange(func(state iii.ObserverState, reason *iii.ObserveReason) {
>     log.Printf("observer %s: %s", newOrders.ID(), state)
> })
> defer unsubscribe()
> if err := newOrders.Ready(ctx); err != nil {
>     log.Fatal(err)
> }
>
> // later: remove the observer
> if err := newOrders.Dispose(ctx); err != nil {
>     log.Fatal(err)
> }
> ```

A target that runs in the observing worker can stop its own observer: it calls `dispose()` on the
handle, or it calls `engine::observers::dispose` with the `observer_id` from the payload. To stop
after a number of calls or after a time, set `max_fires` or `expires_in_ms` in the
[policy](#policy-and-metadata).

### Observer states

| State | Meaning |
| - | - |
| `pending` | Waiting. No worker hosts the function yet (`OWNER_UNAVAILABLE`), the target is not registered yet (`TARGET_UNAVAILABLE`), or the worker that hosts the function has not accepted the observer yet. |
| `active` | The target runs for each matching call. |
| `suspended` | The observing worker disconnected (`OBSERVER_DISCONNECTED`), or the target was removed after the observer was `active` (`TARGET_UNAVAILABLE`). |
| `error` | Rejected. The reason has the code, for example `INVALID_ARGS`, `FORBIDDEN`, or a code from the function's worker. |
| `disposed` | Removed permanently, by `dispose()`, `max_fires`, `expires_in_ms`, or because the observing worker did not reconnect in time. |

* An observer belongs to the connection that created it. When that worker disconnects, its observers
  become `suspended` for `observers.observer_grace_ms` (30 s by default). If the worker reconnects
  in this time, it keeps its observers and their counters. If not, the observers become `disposed`.
* A restarted worker is a new connection. With an `id` of your own (see
  [Policy and metadata](#policy-and-metadata)), the restarted worker's observe of that `id` replaces
  its `suspended` observer, when both connections have the same worker name, namespace and
  principal: both on ports without `rbac`, or both RBAC sessions whose auth function gave them the
  same non-empty `context`. An observer with a random `id` stays `suspended` until the grace period
  ends.
* Set the worker name with `III_WORKER_NAME` or the worker name option of the SDK; a worker that
  [iii compose](https://iii.dev/docs/next/using-iii/compose.md) starts gets `III_WORKER_NAME` from it. The default name, `hostname:pid`,
  changes when the process restarts, so the restarted worker's observe of the `id` fails with
  `OBSERVER_ID_CONFLICT` until the grace period ends.
* When a worker reconnects, its SDK sends each of its observers to the engine again, with the same
  id. The engine keeps a record of a `disposed` observer for `observers.disposed_retention_ms` (1
  hour by default). In this time, the engine does not create the observer again, and the handle
  becomes `disposed`, so `max_fires` and `expires_in_ms` stay in effect after a reconnect. After
  this time, the engine creates the observer again with new counters, as after an engine restart.
* An observer with neither `max_fires` nor `expires_in_ms` is not lost when its worker is away for
  longer than the grace period. When the engine answers that it was disposed because the worker was
  away (`OBSERVER_DISCONNECTED`), the SDK observes it again at once, with the same `id`, source,
  target, metadata and options, and logs a line naming the `id`. The handle goes to `pending` and
  then `active`, and does not become `disposed`. The engine creates a new observer, so its count of
  fires starts again at 0, and the worker that hosts the observed function gets it as a new observer
  (`onObserverAdded` runs). An observer you are disposing is not observed again.
* To know when an observer becomes `disposed`, and why, use `onStateChange` on its handle.
  `engine::observers::list` and `get` do not show disposed observers.
* When the worker that hosts the observed function disconnects, its observers go back to `pending`
  and become `active` again when the function is registered again.

## Engine-hosted event sources

The engine hosts event sources as functions in the `default` namespace, and you observe them with
filters on the event. A worker in any namespace observes `engine::*` and `configuration::*` sources
without naming a namespace. For `observability::*` sources, a worker in another namespace names
`namespace: "default"`.

| Function | Event |
| - | - |
| `engine::on-functions-changed` | `{ event: "functions_changed", functions: [{ function_id, namespace, worker_name, description?, metadata?, args_format?, private?, observable?, observer_count }] }`. `args_format`, `private: true`, and `observable: false` appear when the function sets them. Changes within about 100 ms are coalesced into one event. |
| `engine::on-workers-changed` | `{ event, worker_id }`, `event` one of `worker_connected`, `worker_disconnected`, `worker_metadata_updated` |
| `engine::on-observers-changed` | `{ observer_id, function_id, target, state, reason? }`, one event for each state change of an observer. |
| `configuration::on-updated` | `{ type: "configuration", event_type, id, name, description, schema, metadata? }`. `event_type` is `configuration:registered`, `configuration:updated`, or `configuration:deleted`. The event carries no value: read it with `configuration::get`. |
| `observability::on-log` | One log record: `severity_text`, `severity_number`, `body`, `attributes`, `service_name`, `trace_id`, `span_id`, ... |
| `observability::on-span` | `{ trace_ids: [...] }`: the traces that recorded spans in one window of about 300 ms. |

Each of them is a private function whose handler returns the event. So `input` and `output` are both
the event: `capture: ["input"]` or `capture: ["output"]` gives the target the event in the payload.
Their calls never fail, so an observer that captures `error` but neither `input` nor `output` is
accepted, but its target never runs. Their
`request_format` is closed at the top level, so an `input.*` filter path that is not in the event
fails with `INVALID_FILTER_PATH`. Only the engine calls them. A call from a worker fails with
`FORBIDDEN`. The SDK exports their ids as `EngineSources`, for example
`EngineSources.WORKERS_CHANGED`.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // Reload the billing worker's settings when its configuration entry changes.
> worker.observe(
>   {
>     function_id: "configuration::on-updated",
>     capture: ["input"],
>     filters: {
>       "input.id": { $eq: "billing" },
>       "input.event_type": { $eq: "configuration:updated" },
>     },
>   },
>   "billing::reload",
> );
>
> // Alert on error logs.
> worker.observe(
>   {
>     function_id: "observability::on-log",
>     namespace: "default",
>     capture: ["input"],
>     filters: { "input.severity_number": { $gte: 17 } }, // OpenTelemetry ERROR and above
>   },
>   "alerts::log",
> );
>
> // Refresh a cache when workers change.
> worker.observe({ function_id: "engine::on-workers-changed", capture: ["input"] }, "ops::refresh-workers");
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe(
>     {
>         "function_id": "configuration::on-updated",
>         "capture": ["input"],
>         "filters": {
>             "input.id": {"$eq": "billing"},
>             "input.event_type": {"$eq": "configuration:updated"},
>         },
>     },
>     "billing::reload",
> )
>
> # Alert on error logs.
> worker.observe(
>     {
>         "function_id": "observability::on-log",
>         "namespace": "default",
>         "capture": ["input"],
>         "filters": {"input.severity_number": {"$gte": 17}},  # OpenTelemetry ERROR and above
>     },
>     "alerts::log",
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Capture, Source};
> use serde_json::json;
>
> worker.observe(
>     Source::function("configuration::on-updated")
>         .capture([Capture::Input])
>         .filters(json!({
>             "input.id": { "$eq": "billing" },
>             "input.event_type": { "$eq": "configuration:updated" },
>         })),
>     "billing::reload",
> );
>
> // Alert on error logs.
> worker.observe(
>     Source::function("observability::on-log")
>         .namespace("default")
>         .capture([Capture::Input])
>         .filters(json!({ "input.severity_number": { "$gte": 17 } })),
>     "alerts::log",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.Observe(iii.Source{
>     FunctionID: "configuration::on-updated",
>     Capture:    []iii.Capture{iii.CaptureInput},
>     Filters: iii.Filters{
>         "input.id":         iii.Op{"$eq": "billing"},
>         "input.event_type": iii.Op{"$eq": "configuration:updated"},
>     },
> }, "billing::reload")
>
> // Alert on error logs.
> worker.Observe(iii.Source{
>     FunctionID: "observability::on-log",
>     Namespace:  "default",
>     Capture:    []iii.Capture{iii.CaptureInput},
>     Filters:    iii.Filters{"input.severity_number": iii.Op{"$gte": 17}},
> }, "alerts::log")
> ```

## Chain reactions

A call of a target is a normal call, so another observer can observe the target. In a chain, each
observer observes the target of the observer before it, and `capture: ["output"]` hands it the
previous step's return value as the payload's `output`.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // Inventory worker: after each successful call of orders::create, reserve the items.
> inventoryWorker.registerFunction("inventory::reserve", async ({ input }: Observed<Order>) =>
>   reserveItems(input),
> );
> inventoryWorker.observe({ function_id: "orders::create", capture: ["input"] }, "inventory::reserve");
>
> // Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
> shippingWorker.registerFunction("shipping::schedule", async ({ output }: Observed<unknown, Reservation>) =>
>   scheduleShipment(output),
> );
> shippingWorker.observe(
>   { function_id: "inventory::reserve", capture: ["output"] },
>   "shipping::schedule",
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> # Inventory worker: after each successful call of orders::create, reserve the items.
> inventory_worker.register_function("inventory::reserve", lambda observed: reserve_items(observed["input"]))
> inventory_worker.observe({"function_id": "orders::create", "capture": ["input"]}, "inventory::reserve")
>
> # Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
> shipping_worker.register_function("shipping::schedule", lambda observed: schedule_shipment(observed["output"]))
> shipping_worker.observe({"function_id": "inventory::reserve", "capture": ["output"]}, "shipping::schedule")
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Capture, Source};
>
> // Inventory worker: after each successful call of orders::create, reserve the items.
> inventory_worker.register_function("inventory::reserve", RegisterFunction::new(reserve_items));
> inventory_worker.observe(
>     Source::function("orders::create").capture([Capture::Input]),
>     "inventory::reserve",
> );
>
> // Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
> shipping_worker.register_function("shipping::schedule", RegisterFunction::new(schedule_shipment));
> shipping_worker.observe(
>     Source::function("inventory::reserve").capture([Capture::Output]),
>     "shipping::schedule",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // Inventory worker: after each successful call of orders::create, reserve the items.
> inventoryWorker.RegisterFunction("inventory::reserve", reserveItems)
> inventoryWorker.Observe(iii.Source{
>     FunctionID: "orders::create",
>     Capture:    []iii.Capture{iii.CaptureInput},
> }, "inventory::reserve")
>
> // Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
> shippingWorker.RegisterFunction("shipping::schedule", scheduleShipment)
> shippingWorker.Observe(iii.Source{
>     FunctionID: "inventory::reserve",
>     Capture:    []iii.Capture{iii.CaptureOutput},
> }, "shipping::schedule")
> ```

In Rust and Go, `reserve_items` / `reserveItems` and `schedule_shipment` / `scheduleShipment` take
the `Observed` payload and read its `input` or `output`.

To add a step, observe the last function in the chain. To branch, add two observers to the same
function. Each step adds 1 to the depth of the call: see [Loop guard](#loop-guard).

## Inspect observers

The engine lists observers through three built-in functions:

| Function | Input | Returns |
| - | - | - |
| `engine::observers::list` | `{ function_id?, namespace?, worker? }` | `{ observers: [...] }`: id, observed function, args, normalized filters, target, `capture`, `fires_on` (the outcomes that fire the observer: `success`, `error`, or both), observing worker, state, reason, `fires` (the number of activations), drops and activation failures by reason, policy. |
| `engine::observers::get` | `{ id }` | One observer, or `OBSERVER_NOT_FOUND`. |
| `engine::observers::dispose` | `{ id }` | Disposes an observer. Only the observing worker, or a caller without RBAC restrictions, may do it. |

The output does not include the observer's `metadata` option. With RBAC, the output holds only the
observers of functions the session's grants cover.

```bash theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
iii call engine::observers::list function_id=orders::create
```

To react to observers as they change, observe
[`engine::on-observers-changed`](#engine-hosted-event-sources).

## Errors

When an observer cannot be created, it becomes `error`, and `ready()` fails with one of these codes:

| Code | When |
| - | - |
| `UNSUPPORTED` | The engine is older than 0.25 (the SDK gives this code), or `capture` has a value that the engine does not support. |
| `INVALID_ARGS` | `args` do not match the `args_format` of the function, or the function takes no `args`. Also: an empty or too-long observer id, an empty `function_id`, or `max_fires`, `expires_in_ms`, or `max_in_flight` set to `0`, or a non-empty `capture` that names none of `input`, `output`, and `error` (not checked for an observer that gets the 0.25 semantics: see [Legacy `on` and `deliver`](../reference/engine-protocol#legacy-on-and-deliver)). |
| `INVALID_FILTER` | The filter is not valid, uses a literal value in place of an operator object, or is larger than a limit. |
| `INVALID_FILTER_PATH` | A filter path is not in the function's closed `request_format` or `response_format`. |
| `NOT_OBSERVABLE` | The function is registered with `observable: false`. |
| `FORBIDDEN` | RBAC does not allow the session to call the observed function or the target, the registration hook refused the observer, or the target is a private function. |
| `SELF_TARGET` | The target is the observed function. |
| `LIMIT_EXCEEDED` | The worker has `observers.max_per_connection` observers (1024 by default), or the function has `observers.max_per_function` observers (4096 by default). |
| `OBSERVER_ID_CONFLICT` | An observer with the same id and different options exists, or another worker's observer holds the id. A restarted worker under the same name and namespace replaces its own `suspended` observer with this id. |
| `OWNER_TIMEOUT` | The worker that hosts the function did not accept the observer in 30 seconds. |
| `OBSERVE_REJECTED` | The hosting worker's `onObserverAdded` threw an error other than `ObserveRejection`. |

A worker can also reject an observer with its own code, which its documentation lists (for example
`HTTP_ROUTE_CONFLICT` or `UNKNOWN_DB`).

## Observed call size

The engine keeps one copy of each observed call for all observers of the function. The copy has each
field that any of these observers captures or filters on. If observers capture `input` and `output`,
both count: a function that takes a 40 MiB input and returns a 40 MiB result makes an observed call
of about 80 MiB. The engine does not deliver an observed call in these cases:

* The fields the engine keeps come to more than 64 MiB (approximate in-memory size).
* The activation is larger than the target worker accepts in one message: 64 MiB, or 16 MiB for
  browser workers and some older SDKs. See
  [Payload size](https://iii.dev/docs/next/creating-workers/channels.md#payload-size).
* For calls that run in the function's own worker ([local dispatch](#local-dispatch)), that SDK's
  report of the call is larger than its 8 MiB report buffer.

Each such call counts as a `TOO_LARGE` drop in the `iii.observer.dropped.total` metric. In the first
two cases it also counts in the observer's `dropped` counts
([`engine::observers::list`](#inspect-observers)), and the engine logs a warning naming the
observer, the function, the size, and the limit, at most once a minute per observer.

To keep observed calls small, capture only what the target reads, for example `capture: ["input"]`.
For payloads larger than these limits, pass a
[channel](https://iii.dev/docs/next/creating-workers/channels.md) reference in place of the data.

> [!NOTE]
>
> `OVERFLOW` is a different drop: the queues were full when the call arrived. A call of the same
> size can go through when the queues have space again.

## Caller and causality

Every invocation carries a `caller`: the worker that made the call and, when the call was made from
inside a handler, the function and invocation it was made from.

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{
  "worker_id": "3f1c...",
  "worker_name": "billing",
  "function_id": "billing::close-invoice",
  "invocation_id": "c3f...",
  "parent": { "worker_name": "api", "function_id": "api::checkout", "invocation_id": "77..." }
}
```

The engine sets `worker_id` and `worker_name` from the calling connection. It replaces any values
that the SDK sends. The calling SDK sends `function_id`, `invocation_id`, and `parent` (the caller
of that handler, one level only). Use these 3 fields for routing and filters only. A filter such as
`{ "caller.parent.worker_name": { "$ne": "search-indexer" } }` stops an indexer from reacting to its
own writes.

When the engine activates a target, the target's caller is the engine (`worker_name: "iii-engine"`,
no `worker_id`), with the observed function and invocation as `function_id` and `invocation_id`. The
current invocation's `activation` holds the `observer_id` and `activation_id`.

The SDK propagates the caller automatically from the handler you are in, and exposes the current
invocation (`function_id`, `invocation_id`, `caller`, `activation`, `observe_depth`) as
`worker.currentInvocation()`, carried across `await`.

## Loop guard

An observer's target can call the observed function again, directly or through other observers. This
makes a loop. To stop loops, every invocation carries an `observe_depth`, which increases by 1 on
each activation. Activations deeper than `observers.max_observe_depth` (8 by default) are dropped
and counted as `LOOP_GUARD`, and an observer whose target is the observed function itself is
rejected at creation with `SELF_TARGET`.

> [!WARNING]
>
> Workers on SDKs older than 0.25 do not send the depth. For their calls, the engine uses the
> largest depth of the invocations still in flight on that worker.

## Local dispatch

When a worker calls a function that it registered, the SDK runs the handler in the same process. The
call does not go through the engine. If the function has no observers, the SDK sends nothing to the
engine. If the function has observers, the SDK applies their filters and sends the matching calls to
the engine in batches. Observers get local calls and engine calls in the same way.

> [!NOTE]
>
> The hosting SDK reports the observed values of local calls to the engine. Values JSON cannot hold,
> such as a number no double can represent, arrive as `null`, and a lone surrogate in a string or
> key arrives as U+FFFD. Filters read the values the same way. Reports are at most once: a report
> lost with a dropped connection is not sent again.

A call runs locally when all of these hold:

1. This worker registered the function with a handler. HTTP-invoked functions always go through the
   engine. Local dispatch is on for the function.
2. The engine confirmed local dispatch for the registration. This requires an engine at 0.25 or
   later with `observers.local_dispatch` on. If the worker connects through RBAC, its session must
   have no `middleware_function_id` (unless the worker hosts the function privately), and the
   session must be allowed to call the function.
3. The call resolves to the namespace the function is registered in.
4. The action is `Void` or a normal call. An `Enqueue` call always goes through the engine.
5. The call does not set `dispatch: "remote"`.
6. The call is fewer than 64 local calls deep. Deeper calls go through the engine.

A local call returns the same result, error shape, and timeout as a remote one. The differences are
in [Upgrading from 0.24.x](https://iii.dev/docs/next/upgrading/from-0-24-x.md#local-dispatch). To keep a function on the
engine path:

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // per function
> worker.registerFunction("reports::build", buildReport, { localDispatch: false });
>
> // per call
> await worker.call({ function_id: "reports::build", payload, dispatch: "remote" });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> # per function
> worker.register_function("reports::build", build_report, local_dispatch=False)
>
> # per call
> worker.call({"function_id": "reports::build", "payload": payload, "dispatch": "remote"})
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::protocol::CallRequest;
> use iii_sdk::{Dispatch, RegisterFunction};
>
> // per function
> worker.register_function(
>     "reports::build",
>     RegisterFunction::new_async(build_report).local_dispatch(false),
> );
>
> // per call
> worker
>     .call(CallRequest::new("reports::build", payload).dispatch(Dispatch::Remote))
>     .await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // per function
> off := false
> worker.RegisterFunction("reports::build", buildReport, iii.FunctionOptions{LocalDispatch: &off})
>
> // per call
> worker.Call(ctx, iii.CallRequest{FunctionID: "reports::build", Payload: payload, Dispatch: iii.DispatchRemote})
> ```

To turn it off for the whole engine, set `observers.local_dispatch: false` in the
[engine configuration](https://iii.dev/docs/next/using-iii/configuration.md#observers).

## Authorization

If your worker connects with RBAC, it can observe a function only when its grants
(`expose_functions`, `namespaces`) cover both the observed function and the target. Otherwise
`observe` fails with `FORBIDDEN`. A private function is observable by any session whose grants cover
it. When the observed function or the target registers again later, the engine does the check again.
To change or deny observers in the RBAC configuration, see
[Worker manager](https://iii.dev/docs/next/creating-workers/worker-manager.md#intercept-calls-with-middleware).

## Compatibility

* With an engine older than 0.25, `observe` fails with `UNSUPPORTED`.
* With a 0.25 engine, observers work. A function registered with `args_format`, `private: true`, or
  `observable: false`, or that sets `onObservationMatch` or `onObservationDeliver`, needs a 0.26
  engine: the SDK does not register it on a 0.25 engine, so its observers stay `pending`. The event
  functions of workers such as `http` and `queue` are such functions.
* Functions hosted by workers on older SDKs are observable. The engine evaluates the filters of
  every routed call, whichever SDK hosts the function, and the hosting SDK reports its own local
  calls.
* Workers on SDKs older than 0.25 cannot observe.
* `on` and `deliver` in an observer's source are deprecated. A 0.26 engine translates them to their
  0.25 behavior and logs a warning, and an observer from a worker on a 0.25 SDK or a 0.26 release
  candidate SDK always gets the 0.25 semantics. Replace them with `capture`: see
  [Legacy `on` and `deliver`](../reference/engine-protocol#legacy-on-and-deliver). Support will be
  removed in a future release.
* A 0.26 SDK connected to a 0.25 engine or a 0.26 release candidate engine sends each observer with
  the legacy `on` its `capture` implies, and logs once that this is deprecated. Upgrade the engine.
  See the [compatibility matrix](https://iii.dev/docs/next/upgrading/from-0-25-x.md#compatibility).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.