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

> Declare how your function is observed: react when observers are added, route events to them by their args, and replace a trigger type with a function.

Every function you register can be observed. By default, each successful call of the function is an
event, and the engine runs the target of each [observer](https://iii.dev/docs/0-25-0/using-iii/observers.md) whose filters
match. The function does not need an `observe` descriptor or any other change. This is **auto
delivery**, the default. See [Observing functions](#observing-functions).

The rest of this page is for the worker that hosts a function (its owner). It shows how the owner
can:

* publish events that do not come from a function call
  ([Publish your own events](#publish-your-own-events));
* check for observers, and run code when an observer is added or removed
  ([Know when a function is observed](#know-when-a-function-is-observed));
* send different events to different observers, based on the `args` of each observer
  ([Advanced function observation with owner delivery](#advanced-function-observation-with-owner-delivery));
* accept the old `registerTrigger` calls of a trigger type the function replaces
  ([Accept old triggers](#accept-old-triggers)).

> [!WARNING]
>
> With owner delivery, the function's code delivers each event. An observer's target runs only when
> the function's code calls `obs.emit` or `obs.call` for that observer. To send every call to every
> observer, use auto delivery.

In 0.25 this replaces `registerTriggerType`: a trigger type becomes a function with an `observe`
descriptor, and a trigger becomes an observer. To move an existing trigger type, see
[Upgrading from 0.24.x](https://iii.dev/docs/0-25-0/upgrading/from-0-24-x.md#move-a-trigger-type-to-an-observable-function).

## Observing functions

In this example, `demo::double` is a regular function, so it uses auto delivery: each successful
call of `demo::double` is an event. It has no `observe` descriptor and no hooks. The observer runs
`demo::on-double` after each event, and `demo::on-double` gets the
[`Observed` envelope](../using-iii/observers#target-arguments) of the call. Here the source, the
target, and the observer are in one worker. They can also be in different workers.

> [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();
>
> // The source: a regular function. Auto delivery is the default.
> worker.registerFunction("demo::double", async (input: { n: number }) => ({ n: input.n * 2 }));
>
> // The target: it gets the Observed envelope of each successful call of demo::double.
> worker.registerFunction("demo::on-double", async (event: Observed<{ n: number }, { n: number }>) => {
>   logger.info("demo::double was called", { input: event.input, output: event.output });
>   return null;
> });
>
> // Run demo::on-double after each successful call of demo::double.
> await worker.observe({ function_id: "demo::double" }, "demo::on-double").ready();
>
> await worker.trigger({ function_id: "demo::double", payload: { n: 21 } });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> from iii_helpers.observability import Logger
>
> logger = Logger()
>
> # The source: a regular function. Auto delivery is the default.
> worker.register_function("demo::double", lambda input: {"n": input["n"] * 2})
>
> # The target: it gets the Observed envelope of each successful call of demo::double.
> def on_double(event: dict) -> None:
>     logger.info("demo::double was called", {"input": event.get("input"), "output": event.get("output")})
>
> worker.register_function("demo::on-double", on_double)
>
> # Run demo::on-double after each successful call of demo::double.
> worker.observe({"function_id": "demo::double"}, "demo::on-double").ready()
>
> worker.trigger({"function_id": "demo::double", "payload": {"n": 21}})
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_helpers::observability::Logger;
> use iii_sdk::protocol::TriggerRequest;
> use iii_sdk::{Error, Observed, RegisterFunction, Source};
> use serde_json::{json, Value};
>
> // The source: a regular function. Auto delivery is the default.
> worker.register_function(
>     "demo::double",
>     RegisterFunction::new(|input: Value| {
>         Ok::<_, Error>(json!({ "n": input["n"].as_i64().unwrap_or(0) * 2 }))
>     }),
> );
>
> // The target: it gets the Observed envelope of each successful call of demo::double.
> worker.register_function(
>     "demo::on-double",
>     RegisterFunction::new(|event: Observed| {
>         Logger::new().info(
>             "demo::double was called",
>             Some(json!({ "input": event.input, "output": event.output })),
>         );
>         Ok::<_, Error>(())
>     }),
> );
>
> // Run demo::on-double after each successful call of demo::double.
> worker
>     .observe(Source::function("demo::double"), "demo::on-double")
>     .ready()
>     .await?;
>
> worker
>     .trigger(TriggerRequest::new("demo::double", json!({ "n": 21 })))
>     .await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> // The source: a regular function. Auto delivery is the default.
> worker.RegisterFunction("demo::double", func(ctx context.Context, data json.RawMessage) (any, error) {
>     var input struct {
>         N int `json:"n"`
>     }
>     if err := json.Unmarshal(data, &input); err != nil {
>         return nil, err
>     }
>     return map[string]int{"n": input.N * 2}, nil
> })
>
> // The target: it gets the Observed envelope of each successful call of demo::double.
> worker.RegisterFunction("demo::on-double", 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("demo::double was called: input %s, output %s", event.Input, event.Output)
>     return nil, nil
> })
>
> // Run demo::on-double after each successful call of demo::double.
> observer, err := worker.Observe(iii.Source{FunctionID: "demo::double"}, "demo::on-double")
> if err != nil {
>     log.Fatal(err)
> }
> if err := observer.Ready(ctx); err != nil {
>     log.Fatal(err)
> }
>
> payload, _ := json.Marshal(map[string]int{"n": 21})
> worker.Trigger(ctx, iii.TriggerRequest{FunctionID: "demo::double", Data: payload})
> ```

After the call, `demo::on-double` logs `demo::double was called` with the input `{ n: 21 }` and the
output `{ n: 42 }`.

## Publish your own events

Some events do not come from a function call, for example a file change or a message from an outside
system. To publish such an event, register a function that stands for it, and call that function
each time the event occurs. The function can return its input unchanged. Each call is an event for
the observers of the function.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> type FileChanged = { path: string; event: "created" | "modified" | "deleted" };
>
> worker.registerFunction("files::on-change", async (change: FileChanged) => change);
>
> // whenever a file changes:
> await worker.trigger({ function_id: "files::on-change", payload: change });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.register_function("files::on-change", lambda change: change)
>
> # whenever a file changes:
> worker.trigger({"function_id": "files::on-change", "payload": {"path": path, "event": "modified"}})
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_sdk::protocol::TriggerRequest;
> use iii_sdk::{Error, RegisterFunction};
> use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
> use serde::{Deserialize, Serialize};
> use serde_json::json;
>
> #[derive(Serialize, Deserialize, JsonSchema)]
> struct FileChanged {
>     path: String,
>     event: String,
> }
>
> worker.register_function(
>     "files::on-change",
>     RegisterFunction::new(|change: FileChanged| Ok::<_, Error>(change)),
> );
>
> // whenever a file changes:
> let change = json!({ "path": path, "event": "modified" });
> worker.trigger(TriggerRequest::new("files::on-change", change)).await?;
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.RegisterFunction("files::on-change", func(ctx context.Context, data json.RawMessage) (any, error) {
>     return data, nil
> })
>
> // whenever a file changes:
> change, _ := json.Marshal(map[string]string{"path": path, "event": "modified"})
> worker.Trigger(ctx, iii.TriggerRequest{FunctionID: "files::on-change", Data: change})
> ```

## Know when a function is observed

`FunctionRef.isObserved()` (`is_observed()` in Python and Rust, `worker.IsObserved(id)` in Go)
returns true when the function has at least 1 active observer. It is fast, so you can call it before
you build an event, and skip the event when no observer gets it.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> const prices = worker.registerFunction("prices::on-change", async (change: PriceChange) => change);
>
> // in the price feed loop
> if (prices.isObserved()) {
>   await worker.trigger({ function_id: "prices::on-change", payload: change });
> }
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> prices = worker.register_function("prices::on-change", lambda change: change)
>
> # in the price feed loop
> if prices.is_observed():
>     worker.trigger({"function_id": "prices::on-change", "payload": change})
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> let prices = worker.register_function(
>     "prices::on-change",
>     RegisterFunction::new(|change: PriceChange| Ok::<_, Error>(change)),
> );
>
> // in the price feed loop
> if prices.is_observed() {
>     worker.trigger(TriggerRequest::new("prices::on-change", json!(change))).await?;
> }
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.RegisterFunction("prices::on-change", func(ctx context.Context, data json.RawMessage) (any, error) {
>     return data, nil
> })
>
> // in the price feed loop
> if worker.IsObserved("prices::on-change") {
>     worker.Trigger(ctx, iii.TriggerRequest{FunctionID: "prices::on-change", Data: change})
> }
> ```

Because the worker calls its own function, the call runs in-process
([local dispatch](https://iii.dev/docs/0-25-0/using-iii/observers.md#local-dispatch)) and reaches the engine only when an
observer's filters match.

To run code when an observer is added or removed, give the function `onObserverAdded` and
`onObserverRemoved` hooks. Auto delivery does not change: each call still goes to each observer.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { Logger } from "@iii-dev/helpers/observability";
>
> const logger = new Logger();
>
> worker.registerFunction("prices::on-change", async (change: PriceChange) => change, {
>   observe: {
>     onObserverAdded(obs) {
>       logger.info(`${obs.target} now observes prices::on-change`);
>     },
>   },
> });
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> from iii import ObserveHooks
> from iii_helpers.observability import Logger
>
> logger = Logger()
>
> worker.register_function(
>     "prices::on-change",
>     lambda change: change,
>     observe=ObserveHooks(
>         on_observer_added=lambda obs: logger.info(f"{obs.target} now observes prices::on-change"),
>     ),
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use iii_helpers::observability::Logger;
> use iii_sdk::{Observe, ObserveHandler, ObserveRejection, Observer};
>
> struct LogObservers;
>
> #[async_trait::async_trait]
> impl ObserveHandler for LogObservers {
>     async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
>         Logger::new().info(&format!("{} now observes prices::on-change", obs.target()), None);
>         Ok(())
>     }
> }
>
> worker.register_function(
>     "prices::on-change",
>     RegisterFunction::new(|change: PriceChange| Ok::<_, Error>(change))
>         .observe(Observe::auto().handler(LogObservers)),
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> worker.RegisterFunction("prices::on-change", func(ctx context.Context, data json.RawMessage) (any, error) {
>     return data, nil
> }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
>     OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
>         log.Printf("%s now observes prices::on-change", obs.Target)
>         return nil
>     },
> }})
> ```

## When to use owner delivery

| You want | Use |
| - | - |
| Observers to react to every call of the function (filters select calls) | Auto delivery: call the function. |
| To start or stop work while anyone observes, but still deliver every call | Auto delivery with hooks. See [Know when a function is observed](#know-when-a-function-is-observed). |
| Each observer to set something up and get only its own events (an interval, a path, a table) | Owner delivery. |
| An answer back from the observer's target (an HTTP response, a queue ack) | Owner delivery with `obs.call`. |

## Advanced function observation with owner delivery

> [!NOTE]
>
> Use owner delivery when different observers of the function must get different events, and
> [filters](https://iii.dev/docs/0-25-0/using-iii/observers.md#filters) cannot select those events. For example, each observer
> of a file watcher gives a path in `args` and gets the changes to that path only. Also use owner
> delivery when the function needs the return value of the observer's target (`obs.call`).

With `delivery: 'owner'`, the function's code decides which observers each event is for, from their
`args`, and hands it over. In this case, an observer receives an event only when the function's code
calls `obs.emit()` or `obs.call()` on it. `obs.emit(event)` sends the event and does not wait for
the target, and `obs.call(payload)` runs the target and returns its result to the function.

| | Auto (the default) | Owner |
| - | - | - |
| An event is | every invocation of the function | what the function's code hands to an observer with `obs.emit` or `obs.call` |
| Observer `args` | not accepted (`INVALID_ARGS`) | validated against the function's `args_format`, then read by the function's code |
| Filters evaluated by | the engine (routed calls) or the hosting SDK (local calls) | the owner's SDK, each time the function's code hands an event to the observer |
| When the target runs | after the observed invocation completes, per `on` | when the function's code decides, usually inside the function's own handler |
| The target's result | discarded | discarded for `obs.emit`; returned to the function's code for `obs.call` |
| Examples | any regular function, the engine-hosted event sources | sources that need setup per observer: a schedule, a route, a table to watch |

The examples below use this pattern:

1. **`onObserverAdded` runs once for each new observer**, before the observer becomes `active`. In
   the examples, it starts what produces events for the observer's `args` (a timer, a watcher). To
   refuse the observer, it throws. After 30 seconds, the observer is rejected with `OWNER_TIMEOUT`.
2. **The producer calls the function** when something happens.
3. **The function's handler** gets the observers the event is for (from `observers()` or from its
   own index) and calls `obs.emit(event)` or `obs.call(payload)` on each.

`onObserverRemoved` runs once for each removed observer. In the examples, it stops the producer when
no observer needs it.

The two ways to hand over an event:

| | `obs.emit(event)` | `obs.call(payload)` |
| - | - | - |
| Use for | events whose result you do not need: a tick, a row change | requests that need an answer: an HTTP response, a queue acknowledgement |
| The target gets | the [`Observed` envelope](../using-iii/observers#target-arguments) with `input` (or the event alone with `deliver: 'input'`) | the `payload` |
| Returns | `true` when the event was handed over, `false` when the observer is not active or the filters rejected it | the target's result |
| Wire cost | one batched report item, then one activation from the engine | one invocation through the engine |

Only the hosting worker can call an owner-delivered function. The engine refuses calls from other
workers, and from the engine, with `FORBIDDEN`.

The SDK applies the observer's filters to `emit` and `call`. If the filters reject the event, `emit`
returns `false`, `call` fails with `FILTERS_NOT_MATCHED`, and the SDK sends nothing. To check before
you send, use `obs.matches({ input })`. Both go through the engine, which applies the observer's
policy (`max_fires`, expiry, `max_in_flight`), adds its metadata, and enforces the loop guard. Both
throw `NOT_OWNER_DELIVERED` on an auto function.

`emit` does not wait. If the function emits faster than the targets can run, the observers' queues
become full. The engine then stops reading the worker's connection until the queues have space, for
a maximum of 250 ms. During this hold:

* The SDK keeps events in its report buffer. When the buffer is full, it drops events as
  `BUFFER_FULL`.
* All other messages from the worker (its calls and its answers) wait behind the reports.

After each hold, the engine reads the connection for at least 1/4 of the hold time before the next
hold. Events that overflow during this time are dropped as `OVERFLOW`. Thus the engine holds the
worker for a maximum of 4/5 of the time.

An emitted event travels in the SDK's report, like a
[local call](https://iii.dev/docs/0-25-0/using-iii/observers.md#local-dispatch): values JSON cannot hold arrive as `null` (or
U+FFFD for a lone surrogate), an event larger than the 8 MiB report buffer is dropped as
`TOO_LARGE`, and a report lost with a dropped connection is not sent again.

The event's `caller` depends on where the function's code calls `emit`. Inside the function's own
handler, the event is that invocation, with its id and its caller. Inside a different handler, that
handler is the caller. Outside all handlers, the caller is only the worker. That is why the examples
below trigger their own function and emit from its handler.

The event's `invocation_id` follows the same rule. Inside the function's own handler, the event has
the id of that invocation, and all events emitted during the same invocation have the same id.
Anywhere else, each event gets a new id. `emit` cannot set the id. To pass an id on, put it in the
event.

### Example: a file watcher that emits

`files::watch` is a complete owner-delivered source. Each observer gives a file path in its args
(`{ path }`) and gets the changes to that file only. The numbered comments show the steps above. The
worker polls each watched file 1 time each second. When many observers watch the same path, the
worker still polls that path only 1 time each second.

This function uses owner delivery and the two hooks because its observers decide which files to
watch. The worker cannot poll every file and then filter the changes. It must learn the path of each
observer before it can produce events for that path, and it must stop when no observer wants that
path, or its pollers use resources for nothing. The hooks are the only places where the function
learns that an observer arrived or left.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { Logger } from "@iii-dev/helpers/observability";
> import { type Stats, unwatchFile, watchFile } from "node:fs";
> import type { Observer } from "iii-sdk";
>
> const logger = new Logger();
>
> type FileChanged = { path: string; size_bytes: number };
>
> const watched = new Map<string, Map<string, Observer>>(); // path → its observers
>
> worker.registerFunction(
>   "files::watch",
>   async (change: FileChanged) => {
>     // 3. Hand the change to the observers of that path.
>     for (const obs of watched.get(change.path)?.values() ?? []) {
>       obs.emit(change);
>     }
>     return null;
>   },
>   {
>     description: "Fires when a watched file changes. Observe it with args { path }.",
>     observe: {
>       delivery: "owner",
>       args_format: {
>         type: "object",
>         additionalProperties: false,
>         required: ["path"],
>         properties: { path: { type: "string" } },
>       },
>       onObserverAdded(obs) {
>         // 1. Keep the observer, and start polling its path if nobody else watches it.
>         const path = obs.args.path as string;
>         let observers = watched.get(path);
>         if (!observers) {
>           observers = new Map();
>           watched.set(path, observers);
>           watchFile(path, { interval: 1000 }, (curr: Stats) => {
>             // 2. The poller calls the function.
>             worker
>               .trigger({ function_id: "files::watch", payload: { path, size_bytes: curr.size } })
>               .catch((err) => logger.error("file change failed", err));
>           });
>         }
>         observers.set(obs.id, obs);
>       },
>       onObserverRemoved(obs) {
>         const path = obs.args.path as string;
>         const observers = watched.get(path);
>         observers?.delete(obs.id);
>         if (observers?.size === 0) {
>           unwatchFile(path); // last observer of this path: stop polling it
>           watched.delete(path);
>         }
>       },
>     },
>   },
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import os
> import threading
>
> from iii import ObserveHooks
>
> watched: dict[str, dict] = {}  # path → {"observers": {id: Observer}, "stop": Event}
> lock = threading.Lock()
>
> def poll(path: str, stop: threading.Event) -> None:
>     last = None
>     while not stop.wait(1.0):
>         try:
>             st = os.stat(path)
>         except FileNotFoundError:
>             continue
>         if last is not None and st.st_mtime_ns != last:
>             # 2. The poller calls the function.
>             worker.trigger({
>                 "function_id": "files::watch",
>                 "payload": {"path": path, "size_bytes": st.st_size},
>             })
>         last = st.st_mtime_ns
>
> class WatchHooks(ObserveHooks):
>     delivery = "owner"
>     args_format = {
>         "type": "object",
>         "additionalProperties": False,
>         "required": ["path"],
>         "properties": {"path": {"type": "string"}},
>     }
>
>     def on_observer_added(self, obs):
>         # 1. Keep the observer, and start polling its path if nobody else watches it.
>         path = obs.args["path"]
>         with lock:
>             entry = watched.get(path)
>             if entry is None:
>                 stop = threading.Event()
>                 threading.Thread(target=poll, args=(path, stop), daemon=True).start()
>                 entry = watched[path] = {"observers": {}, "stop": stop}
>             entry["observers"][obs.id] = obs
>
>     def on_observer_removed(self, obs):
>         path = obs.args["path"]
>         with lock:
>             entry = watched.get(path)
>             if entry is None:
>                 return
>             entry["observers"].pop(obs.id, None)
>             if not entry["observers"]:
>                 entry["stop"].set()  # last observer of this path: stop polling it
>                 del watched[path]
>
> def on_change(change: dict) -> None:
>     # 3. Hand the change to the observers of that path.
>     with lock:
>         entry = watched.get(change["path"])
>         observers = list(entry["observers"].values()) if entry else []
>     for obs in observers:
>         obs.emit(change)
>
> worker.register_function(
>     "files::watch",
>     on_change,
>     description="Fires when a watched file changes. Observe it with args { path }.",
>     observe=WatchHooks(),
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::collections::HashMap;
> use std::sync::{Arc, Mutex};
> use std::time::Duration;
>
> use iii_sdk::protocol::TriggerRequest;
> use iii_sdk::{
>     Error, IIIClient, Observe, ObserveHandler, ObserveRejection, Observer, RegisterFunction,
> };
> use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
> use serde::{Deserialize, Serialize};
> use serde_json::json;
>
> #[derive(Deserialize, JsonSchema)]
> #[serde(deny_unknown_fields)]
> struct WatchArgs {
>     path: String,
> }
>
> #[derive(Serialize, Deserialize, JsonSchema)]
> struct FileChanged {
>     path: String,
>     size_bytes: u64,
> }
>
> /// path → the observers of that path and the task that polls it
> #[derive(Clone)]
> struct Watched {
>     worker: IIIClient,
>     by_path: Arc<Mutex<HashMap<String, (Vec<Observer>, tokio::task::JoinHandle<()>)>>>,
> }
>
> async fn poll(worker: IIIClient, path: String) {
>     let mut last = None;
>     let mut interval = tokio::time::interval(Duration::from_secs(1));
>     loop {
>         interval.tick().await;
>         let Ok(meta) = tokio::fs::metadata(&path).await else { continue };
>         let modified = meta.modified().ok();
>         if last.is_some() && modified != last {
>             // 2. The poller calls the function.
>             let change = json!({ "path": path, "size_bytes": meta.len() });
>             let _ = worker.trigger(TriggerRequest::new("files::watch", change)).await;
>         }
>         last = modified;
>     }
> }
>
> #[async_trait::async_trait]
> impl ObserveHandler for Watched {
>     async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
>         // 1. Keep the observer, and start polling its path if nobody else watches it.
>         let WatchArgs { path } = obs.args()?; // validated by the engine
>         let mut by_path = self.by_path.lock().unwrap();
>         let (observers, _) = by_path.entry(path.clone()).or_insert_with(|| {
>             let task = tokio::spawn(poll(self.worker.clone(), path));
>             (Vec::new(), task)
>         });
>         observers.push(obs.clone());
>         Ok(())
>     }
>
>     async fn on_observer_removed(&self, obs: &Observer) {
>         self.by_path.lock().unwrap().retain(|_, (observers, task)| {
>             observers.retain(|o| o.id() != obs.id());
>             if observers.is_empty() {
>                 task.abort(); // last observer of this path: stop polling it
>             }
>             !observers.is_empty()
>         });
>     }
> }
>
> let watched = Watched { worker: worker.clone(), by_path: Default::default() };
>
> worker.register_function(
>     "files::watch",
>     RegisterFunction::new({
>         let watched = watched.clone();
>         move |change: FileChanged| {
>             // 3. Hand the change to the observers of that path.
>             let observers = watched
>                 .by_path
>                 .lock()
>                 .unwrap()
>                 .get(&change.path)
>                 .map(|(observers, _)| observers.clone())
>                 .unwrap_or_default();
>             for obs in observers {
>                 obs.emit(&change)?;
>             }
>             Ok::<_, Error>(())
>         }
>     })
>     .description("Fires when a watched file changes. Observe it with args { path }.")
>     .observe(Observe::owner().args_format::<WatchArgs>().handler(watched.clone())),
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> type fileChanged struct {
>     Path      string `json:"path"`
>     SizeBytes int64  `json:"size_bytes"`
> }
>
> type watch struct {
>     observers map[string]*iii.Observer
>     stop      chan struct{}
> }
>
> var (
>     watchMu sync.Mutex
>     watched = map[string]*watch{} // path → its observers and their poller
> )
>
> func poll(worker *iii.Client, path string, stop chan struct{}) {
>     ticker := time.NewTicker(time.Second)
>     defer ticker.Stop()
>     var last time.Time
>     for {
>         select {
>         case <-stop:
>             return
>         case <-ticker.C:
>             info, err := os.Stat(path)
>             if err != nil {
>                 continue
>             }
>             if !last.IsZero() && !info.ModTime().Equal(last) {
>                 // 2. The poller calls the function.
>                 data, _ := json.Marshal(fileChanged{Path: path, SizeBytes: info.Size()})
>                 worker.Trigger(context.Background(), iii.TriggerRequest{FunctionID: "files::watch", Data: data})
>             }
>             last = info.ModTime()
>         }
>     }
> }
>
> func watchedPath(obs *iii.Observer) string {
>     var args struct {
>         Path string `json:"path"`
>     }
>     _ = obs.DecodeArgs(&args) // validated by the engine
>     return args.Path
> }
>
> func registerWatch(worker *iii.Client) error {
>     return worker.RegisterFunction("files::watch", func(ctx context.Context, data json.RawMessage) (any, error) {
>         var change fileChanged
>         if err := json.Unmarshal(data, &change); err != nil {
>             return nil, err
>         }
>         // 3. Hand the change to the observers of that path.
>         watchMu.Lock()
>         var observers []*iii.Observer
>         if w := watched[change.Path]; w != nil {
>             for _, obs := range w.observers {
>                 observers = append(observers, obs)
>             }
>         }
>         watchMu.Unlock()
>         for _, obs := range observers {
>             obs.Emit(ctx, change)
>         }
>         return nil, nil
>     }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
>         Delivery:   iii.DeliveryOwner,
>         ArgsFormat: json.RawMessage(`{"type":"object","additionalProperties":false,"required":["path"],"properties":{"path":{"type":"string"}}}`),
>         OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
>             // 1. Keep the observer, and start polling its path if nobody else watches it.
>             path := watchedPath(obs)
>             watchMu.Lock()
>             defer watchMu.Unlock()
>             w := watched[path]
>             if w == nil {
>                 w = &watch{observers: map[string]*iii.Observer{}, stop: make(chan struct{})}
>                 watched[path] = w
>                 go poll(worker, path, w.stop)
>             }
>             w.observers[obs.ID] = obs
>             return nil
>         },
>         OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) {
>             path := watchedPath(obs)
>             watchMu.Lock()
>             defer watchMu.Unlock()
>             if w := watched[path]; w != nil {
>                 delete(w.observers, obs.ID)
>                 if len(w.observers) == 0 {
>                     close(w.stop) // last observer of this path: stop polling it
>                     delete(watched, path)
>                 }
>             }
>         },
>     }})
> }
> ```

A consumer observes it like any other function, with the path as args:

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.observe(
  { function_id: "files::watch", args: { path: "/data/orders.csv" }, deliver: "input" },
  "csv::process",
);
```

Compare it with `files::on-change` at the top of this page. With `files::on-change`, each observer
gets all changes and uses filters to select changes. With `files::watch`, each observer asks for 1
path, and the source watches only the paths that observers ask for.

If an observer's args do not agree with the schema (no `path`, or a field that the schema does not
list), the observer fails with `INVALID_ARGS`. `onObserverAdded` does not run.

### Example: a command router that calls

The file watcher does not wait for a result. This example needs a result, so it uses `obs.call`.
`chat::on-command` answers chat commands. Each observer gives 1 command in its args (`{ command }`).
The router calls the target of that observer and returns the target's result as its own result.

In this example, the function's `onObserverAdded` hook keeps a map from each `command` string to its
observer. If the map already has the observer's `command` string (an exact, case-sensitive match),
the hook rejects the observer with `COMMAND_TAKEN`. This rule and the `COMMAND_TAKEN` code are part
of the example code. The SDK does not compare the args of different observers.

> [tab] Node / TypeScript
>
> ```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> import { ObserveRejection, type Observer } from "iii-sdk";
>
> type CommandRequest = { command: string; text: string; user: string };
> type CommandReply = { reply: string };
>
> const commands = new Map<string, Observer>(); // command → observer
>
> worker.registerFunction(
>   "chat::on-command",
>   async (req: CommandRequest): Promise<CommandReply> => {
>     const obs = commands.get(req.command);
>     if (!obs) return { reply: `unknown command /${req.command}` };
>     return obs.call<CommandReply>(req); // the target's result is this function's result
>   },
>   {
>     description: "One invocation per chat command. Observe it with args { command }.",
>     observe: {
>       delivery: "owner",
>       args_format: {
>         type: "object",
>         additionalProperties: false,
>         required: ["command"],
>         properties: { command: { type: "string" } },
>       },
>       onObserverAdded(obs) {
>         const command = obs.args.command as string;
>         if (commands.has(command)) {
>           throw new ObserveRejection("COMMAND_TAKEN", `/${command} already has a handler`);
>         }
>         commands.set(command, obs);
>       },
>       onObserverRemoved(obs) {
>         const command = obs.args.command as string;
>         if (commands.get(command)?.id === obs.id) commands.delete(command);
>       },
>     },
>   },
> );
> ```

> [tab] Python
>
> ```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> from iii import ObserveHooks, ObserveRejection
>
> commands: dict = {}  # command → Observer
>
> class CommandHooks(ObserveHooks):
>     delivery = "owner"
>     args_format = {
>         "type": "object",
>         "additionalProperties": False,
>         "required": ["command"],
>         "properties": {"command": {"type": "string"}},
>     }
>
>     def on_observer_added(self, obs):
>         command = obs.args["command"]
>         if command in commands:
>             raise ObserveRejection("COMMAND_TAKEN", f"/{command} already has a handler")
>         commands[command] = obs
>
>     def on_observer_removed(self, obs):
>         if commands.get(obs.args["command"]) is obs:
>             del commands[obs.args["command"]]
>
> async def on_command(req: dict) -> dict:
>     obs = commands.get(req["command"])
>     if obs is None:
>         return {"reply": f"unknown command /{req['command']}"}
>     return await obs.call_async(req)  # the target's result is this function's result
>
> worker.register_function(
>     "chat::on-command",
>     on_command,
>     description="One invocation per chat command. Observe it with args { command }.",
>     observe=CommandHooks(),
> )
> ```

> [tab] Rust
>
> ```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> use std::collections::HashMap;
> use std::sync::{Arc, Mutex};
>
> use iii_sdk::{Observe, ObserveHandler, ObserveRejection, Observer, RegisterFunction};
> use schemars::JsonSchema; // schemars 0.8, the version iii-sdk uses
> use serde::{Deserialize, Serialize};
>
> #[derive(Deserialize, JsonSchema)]
> #[serde(deny_unknown_fields)]
> struct CommandArgs {
>     command: String,
> }
>
> #[derive(Serialize, Deserialize, JsonSchema)]
> struct CommandRequest {
>     command: String,
>     text: String,
>     user: String,
> }
>
> #[derive(Serialize, Deserialize, JsonSchema)]
> struct CommandReply {
>     reply: String,
> }
>
> /// command → observer
> #[derive(Clone, Default)]
> struct Commands(Arc<Mutex<HashMap<String, Observer>>>);
>
> #[async_trait::async_trait]
> impl ObserveHandler for Commands {
>     async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
>         let CommandArgs { command } = obs.args()?;
>         let mut commands = self.0.lock().unwrap();
>         if commands.contains_key(&command) {
>             return Err(ObserveRejection::new(
>                 "COMMAND_TAKEN",
>                 format!("/{command} already has a handler"),
>             ));
>         }
>         commands.insert(command, obs.clone());
>         Ok(())
>     }
>
>     async fn on_observer_removed(&self, obs: &Observer) {
>         self.0.lock().unwrap().retain(|_, o| o.id() != obs.id());
>     }
> }
>
> let commands = Commands::default();
>
> worker.register_function(
>     "chat::on-command",
>     RegisterFunction::new_async({
>         let commands = commands.clone();
>         move |req: CommandRequest| {
>             let obs = commands.0.lock().unwrap().get(&req.command).cloned();
>             async move {
>                 match obs {
>                     // the target's result is this function's result
>                     Some(obs) => obs.call::<CommandReply>(&req).await,
>                     None => Ok(CommandReply { reply: format!("unknown command /{}", req.command) }),
>                 }
>             }
>         }
>     })
>     .description("One invocation per chat command. Observe it with args { command }.")
>     .observe(Observe::owner().args_format::<CommandArgs>().handler(commands.clone())),
> );
> ```

> [tab] Go
>
> ```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
> var (
>     cmdMu    sync.Mutex
>     commands = map[string]*iii.Observer{} // command → observer
> )
>
> func commandOf(obs *iii.Observer) string {
>     command, _ := obs.Args["command"].(string) // required by ArgsFormat
>     return command
> }
>
> func registerCommands(worker *iii.Client) error {
>     return worker.RegisterFunction("chat::on-command", func(ctx context.Context, data json.RawMessage) (any, error) {
>         var req struct {
>             Command string `json:"command"`
>         }
>         if err := json.Unmarshal(data, &req); err != nil {
>             return nil, err
>         }
>         cmdMu.Lock()
>         obs := commands[req.Command]
>         cmdMu.Unlock()
>         if obs == nil {
>             return map[string]string{"reply": "unknown command /" + req.Command}, nil
>         }
>         return obs.Call(ctx, data) // the target's result is this function's result
>     }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
>         Delivery:   iii.DeliveryOwner,
>         ArgsFormat: json.RawMessage(`{"type":"object","additionalProperties":false,"required":["command"],"properties":{"command":{"type":"string"}}}`),
>         OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
>             command := commandOf(obs)
>             cmdMu.Lock()
>             defer cmdMu.Unlock()
>             if _, taken := commands[command]; taken {
>                 return iii.ObserveRejection("COMMAND_TAKEN", "/"+command+" already has a handler")
>             }
>             commands[command] = obs
>             return nil
>         },
>         OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) {
>             command := commandOf(obs)
>             cmdMu.Lock()
>             defer cmdMu.Unlock()
>             if commands[command] == obs {
>                 delete(commands, command)
>             }
>         },
>     }})
> }
> ```

A worker that handles `/weather` registers its handler and observes the router with its command:

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerFunction("weather::reply", async (req: { text: string }) => ({
  reply: `Sunny in ${req.text}`,
}));

const weather = worker.observe(
  { function_id: "chat::on-command", args: { command: "weather" } },
  "weather::reply",
);
await weather.ready(); // a second observer of /weather fails with COMMAND_TAKEN
```

Only the router's worker can call `chat::on-command`. When it calls the router, the return value of
`weather::reply` becomes the router's result:

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
await worker.trigger({
  function_id: "chat::on-command",
  payload: { command: "weather", text: "Lisbon", user: "ana" },
}); // { reply: "Sunny in Lisbon" }
```

### Errors of `obs.call`

`obs.call` fails with the target's own error, or with one of these codes:

| Code | When |
| - | - |
| `FILTERS_NOT_MATCHED` | The observer's filters reject the payload. Nothing was sent. |
| `NOT_OWNER_DELIVERED` | The function is auto-delivered. |
| `OBSERVER_NOT_ACTIVE` | The observer is not active yet (for example, inside `onObserverAdded`), or it was removed or rejected before the call. |
| `OBSERVER_NOT_FOUND`, `NOT_FUNCTION_OWNER`, `TARGET_MISMATCH` | The engine does not know the observer, or this worker does not host the observed function. |
| `LOOP_GUARD` | The call is deeper than `observers.max_observe_depth`. |
| `OVERFLOW` | The observer has `max_in_flight` calls running. An owner call does not go into a queue, because the function waits for its result. |
| `TIMEOUT` | The target did not answer within `observers.activation_timeout_ms` (30 s by default). |
| `function_not_found` | The target is not registered. |
| `payload_too_large` | The activation is larger than the target worker accepts in one message. See [Event size](https://iii.dev/docs/0-25-0/using-iii/observers.md#event-size). |

## Reject an observer later

An observer that the function accepted can become incorrect later. For example, a configuration
reload can remove the resource that the observer watches. `obs.reject(code, message)` stops delivery
to the observer immediately. The engine moves the observer to `error` with that code, and the SDK
then runs `onObserverRemoved`. If the connection is lost, the SDK sends the rejection again after it
connects again.

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
for (const obs of feeds.observers()) {
  if (!config.sources.includes(obs.args.source as string)) {
    obs.reject("UNKNOWN_SOURCE", `source ${obs.args.source} was removed`);
  }
}
```

`FunctionRef.observers()` (`observers()` in Python and Rust, `worker.Observers(id)` in Go) returns
the active observers of the function.

## Refuse observers

Some functions must not be observed, for example because their input contains secrets. Register them
with `observe: { allow: false }` (`ObserveHooks(allow=False)` in Python,
`Observe::auto().allow(false)` in Rust, `Allow: &no` in Go). Every `observe` on them fails with
`NOT_OBSERVABLE`.

## Accept old triggers

A function that replaces a trigger type can keep serving workers that still call
`registerTrigger({ type, function_id, config })`. Declare the type in `legacy_trigger_type`. The
excerpt below turns the old `timer` trigger type into `timer::on-tick`, a source shaped like the
[file watcher](#example-a-file-watcher-that-emits) with one timer per interval instead of one poller
per path. Only the parts that differ for old triggers are shown.

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
import { ObserveRejection, type Observer } from "iii-sdk";

type Tick = { every_ms: number; fired_at_ms: number };
type TimerArgs = { every_ms: number };

// Observers made from an old registerTrigger have no args: their config is in obs.legacy.
const argsOf = (obs: Observer): TimerArgs =>
  (obs.legacy ? obs.legacy.config : obs.args) as TimerArgs;

worker.registerFunction(
  "timer::on-tick",
  async (tick: Tick) => {
    // ... hand the tick to the observers of tick.every_ms, as files::watch does per path
    return null;
  },
  {
    observe: {
      delivery: "owner",
      args_format: {
        type: "object",
        additionalProperties: false,
        required: ["every_ms"],
        properties: { every_ms: { type: "integer", minimum: 100 } },
      },
      legacy_trigger_type: "timer",
      onObserverAdded(obs) {
        const { every_ms } = argsOf(obs);
        // The engine does not check a legacy config against args_format: validate it here.
        if (!Number.isInteger(every_ms) || every_ms < 100) {
          throw new ObserveRejection(
            "INVALID_TIMER_CONFIG",
            "every_ms must be an integer of at least 100",
          );
        }
        // ... index the observer by every_ms, as files::watch does by path
      },
      onObserverRemoved(obs) {
        // ... use argsOf(obs) here too, then drop the observer from the index
      },
    },
  },
);
```

When a `registerTrigger` arrives for type `timer` and no worker provides the `timer` trigger type
the old way, the engine changes the trigger into an observer of the function:

* The observer belongs to the worker that called `registerTrigger`; its id is the trigger id.
* It has no args, no filters, and `deliver: 'input'`. Its target gets the emitted event in the same
  form as the old trigger handler got the fired payload.
* The trigger's metadata becomes the observer's metadata sidecar.
* The hook gets `obs.legacy = { trigger_type, config }`. Validate the config as the old
  `TriggerHandler` did, then accept or reject the observer. The old worker gets 1
  `TriggerRegistrationResult` with the result, which includes a rejection.
* `unregisterTrigger` (or `engine::unregister_trigger`) disposes the observer.

The engine looks for the stand-in function where `registerTrigger` looks for a provider: only in the
trigger's `trigger_namespace` when it names one, otherwise in the registering worker's namespace and
then in `default`.

The engine uses `timer::on-tick` for a `registerTrigger` call only when no worker provides the
`timer` trigger type with `registerTriggerType`. Thus, during a migration:

* If a worker still provides `timer` with `registerTriggerType`, new `timer` triggers go to that
  worker. They do not become observers of `timer::on-tick`.
* Triggers that are already observers of `timer::on-tick` stay observers. They do not move to the
  old worker.

To send all `timer` triggers to `timer::on-tick`, stop calling `registerTriggerType("timer")` in the
same release that adds `timer::on-tick`.

## The `observe` descriptor

Pass `observe` when you register the function:

| Field | Default | Meaning |
| - | - | - |
| `delivery` | `auto` | `auto`: each invocation is an event for each observer. `owner`: the function's code gives events to observers. |
| `args_format` | none | Owner delivery only: the JSON Schema of the observers' `args`. The engine rejects other args with `INVALID_ARGS`. |
| `legacy_trigger_type` | none | Accept old `registerTrigger({ type })` calls for this trigger type. See [Accept old triggers](#accept-old-triggers). |
| `allow` | `true` | `false` refuses every observer with `NOT_OBSERVABLE`. |
| `onObserverAdded` | none | Runs once per added observer, before the engine marks it `active`. Throw to reject it. |
| `onObserverRemoved` | none | Runs once per removed observer. It cannot fail; errors are logged. |

The SDK sends only the first 4 fields to the engine. The hooks stay in the worker's process. The
names follow the conventions of each SDK:

| SDK | Where | Hooks |
| - | - | - |
| Node | `registerFunction(id, handler, { observe: { ... } })` | `onObserverAdded(obs)`, `onObserverRemoved(obs)` |
| Browser | same as Node | same as Node |
| Python | `register_function(id, handler, observe=ObserveHooks(...))` or a subclass | `on_observer_added(obs)`, `on_observer_removed(obs)`, sync or async |
| Rust | `RegisterFunction::new(..).observe(Observe::owner().args_format::<T>().handler(h))` | `ObserveHandler::on_observer_added` / `on_observer_removed` |
| Go | `RegisterFunctionOptions{Observe: &iii.ObserveHooks{...}}` | `OnObserverAdded(ctx, obs)`, `OnObserverRemoved(ctx, obs)` |

`args_format` with `delivery: 'auto'` fails at registration with `INVALID_OBSERVE_DESCRIPTOR`. An
auto function sends each invocation to each observer, so it does not use observer args.

### Hook rules

* `onObserverAdded` runs before the SDK acknowledges the observer to the engine. The observer
  becomes `active` only after the hook returns. Thus, the hook's work is complete before the first
  event. Inside the hook, `emit` returns `false` and `call` fails with `OBSERVER_NOT_ACTIVE`.
* Throwing rejects the observer: an `ObserveRejection(code, message)` sets the code the observer's
  `ready()` fails with; any other error is `OBSERVE_REJECTED`.
* The hooks of one function run one at a time, in the engine's order. Hooks of different functions
  may run concurrently.
* A hook still running after 30 seconds counts as failed (`OWNER_TIMEOUT`, the engine enforces the
  same `observers.owner_ack_timeout_ms`), and the next hook may start.
* `onObserverRemoved` runs when an accepted observer leaves the function: it is disposed, or it goes
  to `error`. For example: `dispose()`, `max_fires` or expiry, an observing worker that does not
  connect again within its grace period, a late `obs.reject`, or a reconnect that no longer lists
  the observer.

### The `Observer` object

Hooks get an `Observer`: the observer as the owner sees it. It does not include the observer's
metadata, policy, or observing worker. The object stays the same for the full life of the observer,
so the function's code can keep it, for example in a map.

| Member | Meaning |
| - | - |
| `id` | The observer id. |
| `function_id` | The observed function. |
| `args` | The observer's args, already validated against `args_format` by the engine. |
| `filters` | The observer's filters in [normal form](https://iii.dev/docs/0-25-0/reference/filters.md#normal-form). The SDK applies them for you. |
| `target` | The target function id. |
| `capture`, `on` | What the observer captures, and `on` (auto delivery only). |
| `legacy` | `{ trigger_type, config }` when the engine made the observer from an old `registerTrigger`. |
| `matches(doc)` | Evaluates the observer's filters on `{ input?, output?, error?, caller?, metadata? }`. |
| `emit(event)` | Owner delivery, fire-and-forget. Returns `false` and sends nothing when the filters reject the event or the observer is no longer active. |
| `call(payload)` | Owner delivery, request/response: invokes the target and returns its result. |
| `reject(code, message)` | Late rejection of an active observer. See [Reject an observer later](#reject-an-observer-later). |

Language specifics: Python has `call` (blocking) and `call_async`; Rust has `id()`, `args::<T>()`,
`args_value()`, `legacy()`, `emit(&event) -> Result<bool, Error>` and `call::<R>(payload).await`; Go
has `Args` (a map), `DecodeArgs(&v)`, `Legacy`, `Emit(ctx, event) (bool, error)` and
`Call(ctx, payload) (json.RawMessage, error)`.

## Against an older engine

With an engine older than 0.25, a function with an `observe` descriptor registers as a plain
function. Its hooks do not run, and the SDK logs 1 warning for each function that has hooks or owner
delivery. Local dispatch is off, and all calls go through the engine.


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