> ## 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

> Run a function each time another function has an event: worker.observe(source, target).

`observe` runs a function (the target) each time another function (the source) has an event:

```text theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.observe(source, target)
```

By default, an event is a successful call of the source. Some sources make their own events, and
their behavior may differ. In these cases, consult the documentation of the specific worker.

Any worker can observe any function, unless otherwise restricted by [RBAC](#authorization) or by the
source itself (`allow: false`).

* **`source`** is the function to observe, plus optional [filters](#filters) that select which
  events reach the target.
* **`target`** is the id of the function to run. It can be any function except the source itself.
  The observer stays `pending` until the source and the target are registered.

`worker.observe` replaces `registerTrigger` for sources that are functions. See
[Upgrading from 0.24.x](https://iii.dev/docs/0-25-0/upgrading/from-0-24-x.md) to move existing triggers.

> [!NOTE]
>
> The http, cron, queue, state, stream, pubsub, and database workers still publish trigger types in
> 0.25. Keep binding to them with [`registerTrigger`](./triggers) until they ship source functions;
> the deprecation warning it logs is informational. You can observe any regular function, the
> [engine-hosted event sources](#engine-hosted-event-sources), and providers that already moved to
> [observable functions](https://iii.dev/docs/0-25-0/creating-workers/observable-functions.md).

## Observe a function

The examples on this page build on each other. They all react to `orders::create`, a regular
function registered by another worker.

### React to every call

After each successful call of `orders::create`, run `emails::send-receipt` with the order.
`deliver: "input"` makes the target receive the input of `orders::create` (the order) as its
argument.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { registerWorker } from "iii-sdk";
>
> const url = process.env.III_URL;
> if (!url) throw new Error("III_URL must be set");
> const worker = registerWorker(url);
>
> type Order = { id: string; email: string; total: number };
>
> // The target: a regular function.
> worker.registerFunction("emails::send-receipt", async (order: Order) => {
>   await sendReceipt(order.email, order);
>   return null;
> });
>
> // After each successful call of orders::create, run emails::send-receipt with the order.
> const receipts = worker.observe(
>   { function_id: "orders::create", deliver: "input" },
>   "emails::send-receipt",
> );
> await receipts.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
>
> worker = register_worker(os.environ.get("III_URL"), InitOptions(worker_name="emails"))
>
> # The target: a regular function.
> def send_receipt(order: dict) -> None:
>     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.
> receipts = worker.observe(
>     {"function_id": "orders::create", "deliver": "input"},
>     "emails::send-receipt",
> )
> receipts.ready()  # blocks until active; use `await receipts.ready_async()` in asyncio code
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Deliver, Error, InitOptions, RegisterFunction, Source, 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.
> worker.register_function(
>     "emails::send-receipt",
>     RegisterFunction::new(|order: Order| {
>         send_receipt(&order.email, &order);
>         Ok::<_, Error>(())
>     }),
> );
>
> // After each successful call of orders::create, run emails::send-receipt with the order.
> let receipts = worker.observe(
>     Source::function("orders::create").deliver(Deliver::Input),
>     "emails::send-receipt",
> );
> receipts.ready().await?; // resolves once the observer is active
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> type Order struct {
>     ID    string  `json:"id"`
>     Email string  `json:"email"`
>     Total float64 `json:"total"`
> }
>
> worker := iii.RegisterWorker(os.Getenv("III_URL"))
>
> // The target: a regular function.
> worker.RegisterFunction("emails::send-receipt", func(ctx context.Context, data json.RawMessage) (any, error) {
>     var order Order
>     if err := json.Unmarshal(data, &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.
> receipts, err := worker.Observe(iii.Source{
>     FunctionID: "orders::create",
>     Deliver:    iii.DeliverInput,
> }, "emails::send-receipt")
> if err != nil {
>     log.Fatal(err) // the request could not be built (missing ids, args that are not JSON)
> }
> if err := receipts.Ready(ctx); err != nil {
>     log.Fatal(err) // an *iii.ObserveRejectionError carries the engine's code
> }
> ```

From this point, `emails::send-receipt` runs after each successful call of `orders::create`.

`observe` returns a handle at once, also before the connection is up. The SDK sends the observer
when it connects and replays it with the same id after every reconnect, so you call it once at
startup, like `registerFunction`. The browser SDK (`iii-browser-sdk`) has the same `observe` API as
Node.

> [!NOTE]
>
> The observer can miss events that occur before `ready()` resolves.

### Narrow it with a filter

This observer runs `alerts::big-order` only for orders with a total of 500 or more. A
[filter](#filters) on the source selects which events reach the target. The source does not set
`deliver`, so the target receives the [`Observed` envelope](#target-arguments). With the default
`capture`, the envelope has the order (`input`), the result (`output`), the caller (`caller`), and
the duration of the call (`duration_ms`).

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { Logger } from "@iii-dev/helpers/observability";
> import type { Observed } from "iii-sdk";
>
> const logger = new Logger();
>
> 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"}}
> from iii_helpers.observability import Logger
>
> logger = Logger()
>
> 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_helpers::observability::Logger;
> use iii_sdk::Observed;
> 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
>     }
>     log.Printf("big order: %s", 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)
> }
> ```

Both observers watch the same function, independently. The next examples add
[policy](#policy-and-metadata), [chains](#chain-reactions), and
[engine events](#engine-hosted-event-sources).

### Observe errors or all calls

By default, the target runs only after a successful call of the source. To change this, set `on` in
the source:

| `on` | The target runs after |
| - | - |
| `success` (default) | a successful call |
| `error` | a failed call |
| `complete` | each call, successful or failed |

A cancelled call (for example, its caller stopped waiting) does not run the target, for any value of
`on`. The engine counts it as a dropped event with the reason `CANCELLED` in the
`iii.observer.dropped.total` metric.

The default `capture` is `["input", "output", "caller"]`, without `error`. If `capture` does not
include `error`, the envelope has no `error` field, and the target does not get the error. To give
the target the error, add `error` to `capture`. This observer runs `alerts::order-failed` after each
failed call of `orders::create`. The target receives the [`Observed` envelope](#target-arguments)
with the order (`input`) and the error (`error`):

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

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

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Capture, On, Source};
>
> worker.observe(
>     Source::function("orders::create")
>         .on(On::Error)
>         .capture([Capture::Input, Capture::Error]),
>     "alerts::order-failed",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.Observe(iii.Source{
>     FunctionID: "orders::create",
>     On:         iii.OnError,
>     Capture:    []iii.Capture{iii.CaptureInput, iii.CaptureError},
> }, "alerts::order-failed")
> ```

To run the target after each call, use `on: "complete"`. The envelope's `status` is `success` or
`error`.

## Target arguments

With the default `deliver: "envelope"`, the target's first argument is the `Observed` envelope. The
envelope always has the ids, `status`, and the timings. It has `input`, `output`, `error`, `caller`,
and `metadata` only when `capture` includes them. `output` is present only after a successful call.

| Field | Meaning |
| - | - |
| `observer_id`, `activation_id` | The observer, and this activation of it. |
| `function_id`, `namespace` | The observed function. |
| `invocation_id` | The observed call. For an event that a source makes, see below. |
| `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. |

When a source makes its own events, the `invocation_id` depends on where the source sends the event:

* From the handler of the source function: the id of that call. Events sent during the same call
  have the same `invocation_id`.
* From other code: a new id for each event; for example when a source calls another function that
  then performs the emit. The envelope's `caller` then has the `function_id` and `invocation_id` of
  that function.

Engine-hosted event sources give each event a new id.

With `deliver: "input"` or `deliver: "output"`, the target's first argument is that value alone. If
the value is absent (for example, `output` after a failed call), the argument is `null`.

The observer's `metadata` option is the target's second argument (the `metadata` parameter in
Python, `MetadataFromContext(ctx)` in Go). It is different from the `metadata` field of the
envelope, which is the metadata of the observed call.

iii discards the target's return value, unless the source documents otherwise. Observers of the
target can see it as the `output` of the target's call. See [Chain reactions](#chain-reactions).

## Source options

All fields are optional except `function_id`:

| Field | Default | Meaning |
| - | - | - |
| `function_id` | required | The function to observe. |
| `namespace` | `default` for `engine::*` and `configuration::*` functions; the worker's namespace for all others | The namespace of the function. |
| `filters` | `{}` (every event) | Which events reach the target. See [Filters](#filters). |
| `on` | `success` | Which calls run the target: `success`, `error`, or `complete` (both). See [Observe errors or all calls](#observe-errors-or-all-calls). |
| `capture` | `["input", "output", "caller"]` | Which fields the envelope has: any of `input`, `output`, `error`, `caller`, `metadata`. With `[]`, the envelope has only the ids, `status`, and the timings. |
| `deliver` | `envelope` | What the target's first argument is: the `envelope`, or the observed `input` or `output` alone. |
| `args` | none | Arguments for the source, only for sources that document them. See [Source `args`](#source-args). |

A filter can read any field, including those not defined in `capture`. `capture` sets only which
fields the target function receives.

### Source `args`

Some sources take `args`, for example a path for a file watcher or an interval for a schedule. The
documentation of the source lists its `args`. For example, `files::watch`, the example source in
[Observable functions](https://iii.dev/docs/0-25-0/creating-workers/observable-functions.md#example-a-file-watcher-that-emits),
takes `{ path }`. It is not part of iii:

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // When /data/orders.csv changes, run csv::process.
> worker.observe(
>   { function_id: "files::watch", args: { path: "/data/orders.csv" }, deliver: "input" },
>   "csv::process",
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> # When /data/orders.csv changes, run csv::process.
> worker.observe(
>     {"function_id": "files::watch", "args": {"path": "/data/orders.csv"}, "deliver": "input"},
>     "csv::process",
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // When /data/orders.csv changes, run csv::process.
> worker.observe(
>     Source::function("files::watch")
>         .args(json!({ "path": "/data/orders.csv" }))
>         .deliver(Deliver::Input),
>     "csv::process",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // When /data/orders.csv changes, run csv::process.
> worker.Observe(iii.Source{
>     FunctionID: "files::watch",
>     Args:       iii.Args{"path": "/data/orders.csv"},
>     Deliver:    iii.DeliverInput,
> }, "csv::process")
> ```

`args` tell the source which events to make (the changes to one file). Filters select from those
events, for example only changes that leave the file non-empty:
`{ "input.size_bytes": { "$gt": 0 } }`.

The engine checks `args` against the `args_format` of the source. The observer fails with
`INVALID_ARGS` when the `args` do not match, or when the source has no `args_format` and the `args`
are not empty.

## Filters

A filter is a JSON object. Each key is a path in the event 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.

The operators are `$eq`, `$ne`, `$gt`, `$gte`, `$lt`, `$lte`, `$in`, `$nin`, `$exists`, `$prefix`,
`$and`, `$or`, `$not`, and `$any`.

```json theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
{
  "input.total": { "$gte": 500 },
  "input.currency": { "$in": ["EUR", "USD"] },
  "caller.worker_name": { "$ne": "importer" }
}
```

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.

The engine and each SDK have their own filter implementation. All of them pass the same set of test
cases, so a filter matches the same events in all of them. See the
[filter language reference](https://iii.dev/docs/0-25-0/reference/filters.md) for the operators, the rules for absent and `null`
values, and the limits.

## 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. In this example, the inventory worker
reserves the items of each new order, and the shipping worker schedules a shipment for each
reservation. Each worker registers its own function and observer.

> [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 (order: Order) => reserveItems(order));
> inventoryWorker.observe({ function_id: "orders::create", deliver: "input" }, "inventory::reserve");
>
> // Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
> shippingWorker.registerFunction("shipping::schedule", async (reservation: Reservation) =>
>   scheduleShipment(reservation),
> );
> shippingWorker.observe(
>   { function_id: "inventory::reserve", deliver: "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", reserve_items)
> inventory_worker.observe({"function_id": "orders::create", "deliver": "input"}, "inventory::reserve")
>
> # Shipping worker: after each successful call of inventory::reserve, schedule the shipment.
> shipping_worker.register_function("shipping::schedule", schedule_shipment)
> shipping_worker.observe({"function_id": "inventory::reserve", "deliver": "output"}, "shipping::schedule")
> ```

> [tab] Rust
>
> ```rust 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", RegisterFunction::new(reserve_items));
> inventory_worker.observe(
>     Source::function("orders::create").deliver(Deliver::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").deliver(Deliver::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", Deliver: iii.DeliverInput}, "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", Deliver: iii.DeliverOutput}, "shipping::schedule")
> ```

`inventory::reserve` returns the reservation. With `deliver: "output"`, `shipping::schedule`
receives that return value as its argument. 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 event. When
the depth is more than `observers.max_observe_depth` (8 by default), the engine drops the event with
`LOOP_GUARD`. See [Loop guard](#loop-guard).

## Engine-hosted event sources

The engine hosts event sources as functions in the `default` namespace. Observe them like any
function, with filters on the event. They replace the trigger types of the engine.

| Function | Replaces the trigger type | Event (`input`) |
| - | - | - |
| `engine::on-functions-changed` | `engine::functions-available` | `{ event: "functions_changed", functions: [{ function_id, namespace, worker_name, description?, metadata? }] }`. Changes within about 100 ms are coalesced into one event. |
| `engine::on-workers-changed` | `engine::workers-available` | `{ event, worker_id }`, `event` one of `worker_connected`, `worker_disconnected`, `worker_metadata_updated` |
| `configuration::on-updated` | `configuration` | `{ type: "configuration", event_type, id, name, description, schema, old_value, new_value, metadata? }`. `event_type` is `configuration:registered`, `configuration:updated`, or `configuration:deleted`. |
| `observability::on-log` | `log` | One log record: `severity_text`, `severity_number`, `body`, `attributes`, `service_name`, `trace_id`, `span_id`, ... |
| `observability::on-span` | `trace` | `{ trace_ids: [...] }`: the traces that recorded spans in one window of about 300 ms. |

No function runs for these events. The engine creates each event, puts it in `input`, and handles it
as a successful call with no `output`. This has two results:

* The calls never fail. An observer with `on: "error"` is accepted, but its target never runs.
* There is no `output`. With `deliver: "output"`, the target gets `null`. To give the target the
  event, use `deliver: "input"`.

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 makes their events. A direct call to one of
them fails with `FORBIDDEN`.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // Reload the stream worker's settings when its configuration entry changes.
> worker.observe(
>   {
>     function_id: "configuration::on-updated",
>     deliver: "input",
>     filters: {
>       "input.id": { $eq: "iii-stream" },
>       "input.event_type": { $eq: "configuration:updated" },
>     },
>   },
>   "stream::reload",
> );
>
> // Alert on error logs.
> worker.observe(
>   {
>     function_id: "observability::on-log",
>     deliver: "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", deliver: "input" }, "ops::on-worker");
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe(
>     {
>         "function_id": "configuration::on-updated",
>         "deliver": "input",
>         "filters": {
>             "input.id": {"$eq": "iii-stream"},
>             "input.event_type": {"$eq": "configuration:updated"},
>         },
>     },
>     "stream::reload",
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::{Deliver, Source};
> use serde_json::json;
>
> worker.observe(
>     Source::function("configuration::on-updated")
>         .deliver(Deliver::Input)
>         .filters(json!({
>             "input.id": { "$eq": "iii-stream" },
>             "input.event_type": { "$eq": "configuration:updated" },
>         })),
>     "stream::reload",
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.Observe(iii.Source{
>     FunctionID: "configuration::on-updated",
>     Deliver:    iii.DeliverInput,
>     Filters: iii.Filters{
>         "input.id":         iii.Op{"$eq": "iii-stream"},
>         "input.event_type": iii.Op{"$eq": "configuration:updated"},
>     },
> }, "stream::reload")
> ```

## Policy and metadata

`observe` takes an optional third argument, the 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. The same id with the same options again changes nothing. The same id with different options fails with `OBSERVER_ID_CONFLICT`. |
| `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 events, 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 [Target arguments](#target-arguments). |

This observer runs `alerts::payment-failed` after each failed call of `payments::charge` with an
amount of 500 or more. It runs at most 4 activations at the same time, and it gives each activation
`{ "team": "billing" }` as metadata:

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.observe(
>   {
>     function_id: "payments::charge",
>     on: "error",
>     filters: { "input.amount": { $gte: 500 } },
>     capture: ["input", "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",
>         "on": "error",
>         "filters": {"input.amount": {"$gte": 500}},
>         "capture": ["input", "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, On, Source};
> use serde_json::json;
>
> worker.observe_with(
>     Source::function("payments::charge")
>         .on(On::Error)
>         .filters(json!({ "input.amount": { "$gte": 500 } }))
>         .capture([Capture::Input, 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",
>     On:         iii.OnError,
>     Filters:    iii.Filters{"input.amount": iii.Op{"$gte": 500}},
>     Capture:    []iii.Capture{iii.CaptureInput, iii.CaptureError, iii.CaptureCaller},
> }, "alerts::payment-failed", iii.ObserverOptions{
>     MaxInFlight: 4,
>     Metadata:    json.RawMessage(`{"team":"billing"}`),
> })
> ```

Each event runs the target at most once. The engine does not retry. A failure of the target does not
change the observed call. The engine counts each dropped event in the `iii.observer.dropped.total`
metric, by function and reason.

## Observer states

| State | Meaning |
| - | - |
| `pending` | Waiting. No worker hosts the function yet (`OWNER_UNAVAILABLE`), the target is not registered yet (`TARGET_UNAVAILABLE`), or the source has not accepted the observer yet. |
| `active` | The target runs for each matching event. |
| `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 source. |
| `disposed` | Removed permanently, by `dispose()`, `max_fires`, `expires_in_ms`, or because the observing worker did not reconnect in time. |

* `ready()` resolves when the observer is `active` and rejects with the reason's code when it turns
  `error` or `disposed` first (`ObserveRejection` in Node, Python, and Rust;
  `*iii.ObserveRejectionError` in Go).
* `onStateChange(listener)` (`on_state_change` in Python and Rust, `OnStateChange` in Go) reports
  each state change with its reason. Listeners run outside the connection's read path, so a listener
  can call the worker.
* `dispose()` (`Dispose(ctx)` in Go) removes the observer. A disposed observer is not replayed. 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 envelope. To stop
  after a number of events or after a time, set `max_fires` or `expires_in_ms` in the policy.
* 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`.
* When a worker reconnects, its SDK sends each of its observers to the engine again, with the same
  id. The engine remembers 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.
* 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 observed function's worker disconnects, its observers go back to `pending` and become
  `active` again when the function is registered again.

## 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, delivery, args, normalized filters, target, observing worker, state, reason, fires, 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, a session sees only the
observers of functions that it is allowed to invoke.

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

## Errors

When the engine cannot create an observer, the observer 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 `on`, `capture`, or `deliver` has a value that the engine does not know. |
| `INVALID_ARGS` | `args` do not match the `args_format` of the source, or the source 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`. |
| `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 refuses observers (`allow: false`). |
| `FORBIDDEN` | RBAC does not allow the session to invoke the observed function or the target, the registration hook refused the observer, or the target is an owner-delivered 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. |
| `OWNER_TIMEOUT` | The source's worker did not accept the observer in 30 seconds. |
| `OBSERVE_REJECTED` | The source refused the observer. A source can also reject with its own code, documented by that source. |

## Event 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 returns its 40 MiB input makes an event of about 80 MiB. The engine does
not deliver an event 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/0-25-0/creating-workers/channels.md#payload-size).
* For calls that run in the source's own worker ([local dispatch](#local-dispatch)), the hosting
  SDK's report of the event is larger than its 8 MiB report buffer.

Each such event 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. `OVERFLOW` is
different: the queues were full when the event arrived. An event of the same size can go through
when the queues have space again.

To keep events small, capture only what the target reads, for example `capture: ["input"]`. With
`deliver: "input"` or `deliver: "output"`, set a matching `capture` too: captured fields are kept
even when the target receives one value. For payloads larger than these limits, pass a
[channel](https://iii.dev/docs/0-25-0/creating-workers/channels.md) reference instead of the data.

## 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 SDKs propagate the caller automatically from the handler you are in. Each SDK also exposes the
current invocation (`function_id`, `invocation_id`, `caller`, `activation`, `observe_depth`):

| SDK | Current invocation | Notes |
| - | - | - |
| Node | `worker.currentInvocation()` | Carried across `await` with `AsyncLocalStorage`. |
| Browser | `worker.currentInvocation()` | Known until the handler's first `await`. Read it first and wrap later calls in `worker.withInvocation(inv, fn)`. |
| Python | `worker.current_invocation()` | A `contextvars` context, carried into sync handlers' threads. |
| Rust | `iii_sdk::current_invocation()` | A task-local. Work moved to `tokio::spawn` loses it; wrap the future in `iii_sdk::scope(ctx, fut)`. |
| Go | `iii.InvocationFromContext(ctx)` | Pass the handler's `ctx` to `worker.Trigger` so outgoing calls carry the caller. |

In the browser, a call made after an `await` and outside `withInvocation` gets the depth of the
deepest invocation that runs on the page. When several invocations run at the same time, this depth
can be too high, and the loop guard can stop an observer chain early. While the page is offline, the
functions that it hosts run locally, to a maximum of 256 nested calls. A call past that limit fails
immediately with `LOCAL_DEPTH_EXCEEDED`.

## 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
invocations 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 a 0.25 engine with
   `observers.local_dispatch` on. If the worker connects through RBAC, its session must have no
   `middleware_function_id`, 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 ask for the remote path.
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/0-25-0/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.trigger({ 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.trigger({"function_id": "reports::build", "payload": payload, "dispatch": "remote"})
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::protocol::TriggerRequest;
> use iii_sdk::{Dispatch, RegisterFunction};
>
> // per function
> worker.register_function(
>     "reports::build",
>     RegisterFunction::new_async(build_report).local_dispatch(false),
> );
>
> // per call
> worker
>     .trigger(TriggerRequest::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.RegisterFunctionOptions{LocalDispatch: &off})
>
> // per call
> worker.Trigger(ctx, iii.TriggerRequest{FunctionID: "reports::build", Data: 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/0-25-0/using-iii/configuration.md#observers).

## Create an observable function

To make something observable, register a function for it and call that function whenever the thing
happens. Each call is an event for its observers:

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
// Producer: call this whenever a file changes.
worker.registerFunction("files::on-change", async (change: FileChanged) => change);

// ...in your watcher loop
await worker.trigger({ function_id: "files::on-change", payload: { path, event: "modified" } });

// Consumer, in any worker: when a file changes, process it.
worker.observe({ function_id: "files::on-change" }, "csv::process");
```

A function can also send different events to different observers. For example, a file watcher can
watch only the paths that its observers give in `args`. To write this type of function, see
[Observable functions](https://iii.dev/docs/0-25-0/creating-workers/observable-functions.md). You observe it in the same way,
sometimes with [args](#source-args).

## Authorization

If your worker connects with RBAC, it can observe a function only when it is allowed to call both
the observed function and the target. Otherwise `observe` fails with `FORBIDDEN`. 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/0-25-0/creating-workers/worker-manager.md#intercept-calls-with-middleware).

## Compatibility

* With an engine older than 0.25, `observe` fails with `UNSUPPORTED`.
* Functions hosted by workers on older SDKs are still observable: their invocations are routed
  through the engine, which evaluates the filters.
* Workers on older SDKs cannot observe. They keep using `registerTrigger`.


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