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

# Observable functions

> 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 is observable without any code on your side: other workers call
[`iii.observe`](../using-iii/observers) on it, and each of its invocations is offered to every
observer. This page is for the worker that **hosts** the function (its owner) and wants a say in how
it is observed:

* run code when an observer is added or removed (start a schedule, open a route, watch a table);
* read each observer's `args` and hand each event only to the observers it belongs to (**owner
  delivery**);
* accept the old `registerTrigger` calls of a trigger type the function replaces.

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](/docs/upgrading/from-0-24-x.md#move-a-trigger-type-to-an-observable-function).

## The `observe` descriptor

Pass `observe` when you register the function:

| Field | Default | Meaning |
| - | - | - |
| `delivery` | `auto` | `auto`: every invocation is an event, offered to every observer. `owner`: your code hands 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. |

Only the first four fields travel to the engine; the hooks stay in your process. The names follow
each SDK's conventions:

| 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's events are its invocations, offered to every observer, so there is nothing to route
by.

### Hook rules

* `onObserverAdded` runs before the SDK acknowledges the observer to the engine. Whatever it adds to
  your own index is in place before the observer turns `active`, so the first event after `ready()`
  resolves reaches it.
* 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 on `dispose()`, policy exhaustion (`max_fires`, expiry), when the
  observing worker does not come back within its grace period, and when a reconnect no longer lists
  the observer.

### The `Observer` object

Hooks receive an `Observer`: the observer as its owner sees it. It never exposes the observer's
metadata, policy, or observing worker. The object stays the same for the observer's whole lifetime, so keep
it in your own index.

| Member | Meaning |
| - | - |
| `id` | The observer id. |
| `function_id` | Your function. |
| `args` | The observer's args, already validated against `args_format` by the engine. |
| `filters` | The observer's filters in [normal form](/docs/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)`.

## Know when you are observed

An auto function can declare hooks too, for example to start expensive work only while someone
observes it. `FunctionRef.isObserved()` (`is_observed()` in Python and Rust, `client.IsObserved(id)`
in Go) answers in one load, so a hot path can skip building events nobody receives.

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
const prices = worker.registerFunction("prices::on-change", async (change: PriceChange) => change, {
  observe: {
    onObserverAdded(obs) {
      console.log(`${obs.target} now observes prices::on-change`);
    },
  },
});

// in the price feed loop
if (prices.isObserved()) {
  await worker.trigger({ function_id: "prices::on-change", payload: change });
}
```

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

## Owner delivery

With `delivery: 'owner'`, invocations of your function are no longer offered to every observer. Your
code decides which observers each event is for, from their `args`, and hands it over:

| | `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#what-the-target-receives) with `input` (or the event alone with `deliver: 'input'`) | your `payload` |
| Returns | `true` when the event was handed over, `false` when 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 your worker may invoke an owner-delivered function: the engine refuses a call to it from any
other worker (or from the engine itself) with `FORBIDDEN`, so nobody can pass off an event as yours.

The SDK applies the observer's filters in both: `emit` returns `false`, and `call` fails with
`FILTERS_NOT_MATCHED`, without sending anything. Use `obs.matches({ input })` when you want to know
first. 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` never waits. When you emit faster than the targets take, the observers' queues fill and the
engine stops reading your worker's connection until they have room again, for at most 250 ms.
Meanwhile the SDK keeps what it can in its bounded report buffer and drops the rest, counted as
`BUFFER_FULL`, and everything else your worker sends (its calls, its answers to calls) waits behind
its reports. After each hold the engine reads your worker's connection for at least a quarter as
long before it holds it again, and drops what overflows meanwhile as `OVERFLOW`: your worker is
held at most four fifths of the time.

The event's `caller` follows where you call `emit`: inside your function's own handler, the event is
that invocation (its id and its caller); inside another handler, that handler is the caller; outside
any handler, only your worker. That is why the examples below trigger their own function and emit
from its handler.

### Example: a timer that emits

`timer::on-tick` fires every `every_ms` milliseconds for the observers that asked for that interval.
It runs one timer per distinct interval, however many observers share it.

**Node / TypeScript**

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

type Tick = { every_ms: number; fired_at_ms: number };
type Timer = { observers: Map<string, Observer>; handle: ReturnType<typeof setInterval> };

const timers = new Map<number, Timer>(); // every_ms → observers and their timer

worker.registerFunction(
  "timer::on-tick",
  async (tick: Tick) => {
    // One call per tick: hand it to the observers of that interval.
    for (const obs of timers.get(tick.every_ms)?.observers.values() ?? []) {
      obs.emit(tick); // false when the observer's filters reject the tick
    }
    return null;
  },
  {
    description: "Fires every `every_ms` milliseconds. Observe it with args { every_ms }.",
    request_format: {
      type: "object",
      additionalProperties: false,
      required: ["every_ms", "fired_at_ms"],
      properties: { every_ms: { type: "integer" }, fired_at_ms: { type: "integer" } },
    },
    observe: {
      delivery: "owner",
      args_format: {
        type: "object",
        additionalProperties: false,
        required: ["every_ms"],
        properties: { every_ms: { type: "integer", minimum: 100 } },
      },
      onObserverAdded(obs) {
        const every_ms = obs.args.every_ms as number;
        let timer = timers.get(every_ms);
        if (!timer) {
          // First observer of this interval: start one timer for all of them.
          const handle = setInterval(() => {
            worker
              .trigger({ function_id: "timer::on-tick", payload: { every_ms, fired_at_ms: Date.now() } })
              .catch((err) => console.error("tick failed", err));
          }, every_ms);
          timer = { observers: new Map(), handle };
          timers.set(every_ms, timer);
        }
        timer.observers.set(obs.id, obs);
      },
      onObserverRemoved(obs) {
        const every_ms = obs.args.every_ms as number;
        const timer = timers.get(every_ms);
        if (!timer) return;
        timer.observers.delete(obs.id);
        if (timer.observers.size === 0) {
          // Last observer of this interval: stop its timer.
          clearInterval(timer.handle);
          timers.delete(every_ms);
        }
      },
    },
  },
);
```

**Python**

```python theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
import threading
import time

from iii import ObserveHooks

timers: dict[int, dict] = {}  # every_ms → {"observers": {id: Observer}, "stop": Event}
lock = threading.Lock()

def run_timer(every_ms: int, stop: threading.Event) -> None:
    while not stop.wait(every_ms / 1000):
        worker.trigger({
            "function_id": "timer::on-tick",
            "payload": {"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)},
        })

class TimerHooks(ObserveHooks):
    delivery = "owner"
    args_format = {
        "type": "object",
        "additionalProperties": False,
        "required": ["every_ms"],
        "properties": {"every_ms": {"type": "integer", "minimum": 100}},
    }

    def on_observer_added(self, obs):
        every_ms = obs.args["every_ms"]
        with lock:
            timer = timers.get(every_ms)
            if timer is None:
                # First observer of this interval: start one timer for all of them.
                stop = threading.Event()
                threading.Thread(target=run_timer, args=(every_ms, stop), daemon=True).start()
                timer = timers[every_ms] = {"observers": {}, "stop": stop}
            timer["observers"][obs.id] = obs

    def on_observer_removed(self, obs):
        every_ms = obs.args["every_ms"]
        with lock:
            timer = timers.get(every_ms)
            if timer is None:
                return
            timer["observers"].pop(obs.id, None)
            if not timer["observers"]:
                # Last observer of this interval: stop its timer.
                timer["stop"].set()
                del timers[every_ms]

def on_tick(tick: dict) -> None:
    with lock:
        timer = timers.get(tick["every_ms"])
        observers = list(timer["observers"].values()) if timer else []
    for obs in observers:
        obs.emit(tick)  # False when the observer's filters reject the tick

worker.register_function(
    "timer::on-tick",
    on_tick,
    description="Fires every `every_ms` milliseconds. Observe it with args { every_ms }.",
    observe=TimerHooks(),
)
```

**Rust**

```rust theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use iii_sdk::protocol::TriggerRequest;
use iii_sdk::{
    Error, IIIClient, Observe, ObserveHandler, ObserveRejection, Observer, RegisterFunction,
};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::json;

#[derive(Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
struct TimerArgs {
    every_ms: u64,
}

#[derive(Serialize, Deserialize, JsonSchema)]
struct Tick {
    every_ms: u64,
    fired_at_ms: u64,
}

/// every_ms → the observers of that interval and the task that fires it
#[derive(Clone)]
struct Timers {
    iii: IIIClient,
    by_interval: Arc<Mutex<HashMap<u64, (Vec<Observer>, tokio::task::JoinHandle<()>)>>>,
}

#[async_trait::async_trait]
impl ObserveHandler for Timers {
    async fn on_observer_added(&self, obs: &Observer) -> Result<(), ObserveRejection> {
        let TimerArgs { every_ms } = obs.args()?; // validated by the engine
        let mut timers = self.by_interval.lock().unwrap();
        let (observers, _) = timers.entry(every_ms).or_insert_with(|| {
            // First observer of this interval: start one task for all of them.
            let iii = self.iii.clone();
            let task = tokio::spawn(async move {
                let mut interval = tokio::time::interval(Duration::from_millis(every_ms));
                loop {
                    interval.tick().await;
                    let fired_at_ms = SystemTime::now()
                        .duration_since(UNIX_EPOCH)
                        .map(|d| d.as_millis() as u64)
                        .unwrap_or_default();
                    let tick = json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms });
                    let _ = iii.trigger(TriggerRequest::new("timer::on-tick", tick)).await;
                }
            });
            (Vec::new(), task)
        });
        observers.push(obs.clone());
        Ok(())
    }

    async fn on_observer_removed(&self, obs: &Observer) {
        self.by_interval.lock().unwrap().retain(|_, (observers, task)| {
            observers.retain(|o| o.id() != obs.id());
            if observers.is_empty() {
                task.abort(); // last observer of this interval: stop its task
            }
            !observers.is_empty()
        });
    }
}

let timers = Timers { iii: worker.clone(), by_interval: Default::default() };

worker.register_function(
    "timer::on-tick",
    RegisterFunction::new({
        let timers = timers.clone();
        move |tick: Tick| {
            let observers = timers
                .by_interval
                .lock()
                .unwrap()
                .get(&tick.every_ms)
                .map(|(observers, _)| observers.clone())
                .unwrap_or_default();
            for obs in observers {
                obs.emit(&tick)?; // Ok(false) when the observer's filters reject the tick
            }
            Ok::<_, Error>(())
        }
    })
    .description("Fires every `every_ms` milliseconds. Observe it with args { every_ms }.")
    .observe(Observe::owner().args_format::<TimerArgs>().handler(timers.clone())),
);
```

**Go**

```go theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
type tick struct {
    EveryMs   int64 `json:"every_ms"`
    FiredAtMs int64 `json:"fired_at_ms"`
}

type timer struct {
    observers map[string]*iii.Observer
    stop      chan struct{}
}

var (
    mu     sync.Mutex
    timers = map[int64]*timer{} // every_ms → observers and their timer
)

func runTimer(client *iii.Client, everyMs int64, stop chan struct{}) {
    ticker := time.NewTicker(time.Duration(everyMs) * time.Millisecond)
    defer ticker.Stop()
    for {
        select {
        case <-stop:
            return
        case now := <-ticker.C:
            data, _ := json.Marshal(tick{EveryMs: everyMs, FiredAtMs: now.UnixMilli()})
            client.Trigger(context.Background(), iii.TriggerRequest{FunctionID: "timer::on-tick", Data: data})
        }
    }
}

func everyMs(obs *iii.Observer) int64 {
    var args struct {
        EveryMs int64 `json:"every_ms"`
    }
    _ = obs.DecodeArgs(&args) // validated by the engine
    return args.EveryMs
}

func registerTimer(client *iii.Client) error {
    return client.RegisterFunction("timer::on-tick", func(ctx context.Context, data json.RawMessage) (any, error) {
        var t tick
        if err := json.Unmarshal(data, &t); err != nil {
            return nil, err
        }
        mu.Lock()
        var observers []*iii.Observer
        if tm := timers[t.EveryMs]; tm != nil {
            for _, obs := range tm.observers {
                observers = append(observers, obs)
            }
        }
        mu.Unlock()
        for _, obs := range observers {
            obs.Emit(ctx, t) // false when the observer's filters reject the tick
        }
        return nil, nil
    }, iii.RegisterFunctionOptions{Observe: &iii.ObserveHooks{
        Delivery:   iii.DeliveryOwner,
        ArgsFormat: json.RawMessage(`{"type":"object","additionalProperties":false,"required":["every_ms"],"properties":{"every_ms":{"type":"integer","minimum":100}}}`),
        OnObserverAdded: func(ctx context.Context, obs *iii.Observer) error {
            every := everyMs(obs)
            mu.Lock()
            defer mu.Unlock()
            tm := timers[every]
            if tm == nil {
                // First observer of this interval: start one timer for all of them.
                tm = &timer{observers: map[string]*iii.Observer{}, stop: make(chan struct{})}
                timers[every] = tm
                go runTimer(client, every, tm.stop)
            }
            tm.observers[obs.ID] = obs
            return nil
        },
        OnObserverRemoved: func(ctx context.Context, obs *iii.Observer) {
            every := everyMs(obs)
            mu.Lock()
            defer mu.Unlock()
            if tm := timers[every]; tm != nil {
                delete(tm.observers, obs.ID)
                if len(tm.observers) == 0 {
                    // Last observer of this interval: stop its timer.
                    close(tm.stop)
                    delete(timers, every)
                }
            }
        },
    }})
}
```

Another worker observes it with args and, optionally, filters:

```typescript theme={"theme":{"light":"catppuccin-latte","dark":"dark-plus"}}
worker.registerFunction(
  "reports::refresh",
  async ({ input }: Observed<{ fired_at_ms: number }>) => {
    await refreshReports(input?.fired_at_ms);
    return null;
  },
);

worker.observe({ function_id: "timer::on-tick", args: { every_ms: 60_000 } }, "reports::refresh");
```

An observer with args the schema refuses (`{ every_ms: 10 }`, or no `every_ms`) fails with
`INVALID_ARGS` before `onObserverAdded` runs.

### Example: a command router that calls

`chat::on-command` answers chat commands. Each observer registers one command with args
`{ command }`; the router calls that observer's target and returns its result as its own.

**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);
      },
    },
  },
);
```

**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(),
)
```

**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;
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())),
);
```

**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(client *iii.Client) error {
    return client.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. Its
return value becomes the router's result:

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

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 was removed or rejected meanwhile. |
| `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 fails instead of queueing, because you are waiting 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. |

## Reject an observer later

An observer you accepted can become invalid, for example after a configuration reload removed the
resource it watches. `obs.reject(code, message)` takes it out of delivery at once; the engine moves
it to `error` with your code and then calls `onObserverRemoved`. The rejection survives a reconnect.

```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, `client.Observers(id)` in Go) lists the
active observers as your hooks saw them.

## Refuse observers

Some functions should not be observed at all, for example because their input holds 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`. RBAC still applies on top: an observer needs permission to invoke both the
observed function and the target.

## 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 changes the [timer example](#example-a-timer-that-emits): its handler and index stay
as they were.

```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 before
    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 as before
      },
      onObserverRemoved(obs) {
        // ... use argsOf(obs) here too
      },
    },
  },
);
```

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

* The observer belongs to the worker that called `registerTrigger`; its id is the trigger id.
* It has no args and no filters, and `deliver: 'input'`: its target receives the event you emit,
  exactly as the old trigger handler received the fired payload.
* The trigger's metadata becomes the observer's metadata sidecar.
* Your hook sees `obs.legacy = { trigger_type, config }`. Validate the config as your old
  `TriggerHandler` did, then accept or reject it like any observer. The old worker receives one
  `TriggerRegistrationResult` with the outcome, your rejection included.
* `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`. Your worker must run a 0.25 SDK. A worker that still registers the trigger type
the old way always wins for new registrations, so stop calling `registerTriggerType` for the type in
the same release that adds the function.

## Against an older engine

Against an engine older than 0.25, a function with an `observe` descriptor registers as a plain
function: its hooks never run, and the SDK logs one warning. Local dispatch also stays off, and
every call goes through the engine as before.


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