Observers

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 whose filters match. The function does not need an observe descriptor or any other change. This is auto delivery, the default. See 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);
check for observers, and run code when an observer is added or removed (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);
accept the old registerTrigger calls of a trigger type the function replaces (Accept old triggers).

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.

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 of the call. Here the source, the target, and the observer are in one worker. They can also be in different workers.

Node / TypeScript
Python
Rust
Go

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 } });

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}})

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?;

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

Node / TypeScript
Python
Rust
Go

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 });

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"}})

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?;

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.

Node / TypeScript
Python
Rust
Go

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 });
}

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})

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?;
}

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

Node / TypeScript
Python
Rust
Go

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

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"),
    ),
)

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)),
);

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

Use owner delivery when different observers of the function must get different events, and 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:

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.
The producer calls the function when something happens.
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 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: 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.

Node / TypeScript
Python
Rust
Go

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

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

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

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:

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.

Node / TypeScript
Python
Rust
Go

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

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

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

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:

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:

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.

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.

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 with one timer per interval instead of one poller per path. Only the parts that differ for old triggers are shown.

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

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.

Getting Started

Using iii

Understanding iii

Creating Workers

Upgrading

Observers

Observing functions

Publish your own events

Know when a function is observed

When to use owner delivery

Advanced function observation with owner delivery

Example: a file watcher that emits

Example: a command router that calls

Errors of `obs.call`

Reject an observer later

Refuse observers

Accept old triggers

The `observe` descriptor

Hook rules

The `Observer` object

Against an older engine

​Observing functions

​Publish your own events

​Know when a function is observed

​When to use owner delivery

​Advanced function observation with owner delivery

​Example: a file watcher that emits

​Example: a command router that calls

​Errors of obs.call

​Reject an observer later

​Refuse observers

​Accept old triggers

​The observe descriptor

​Hook rules

​The Observer object

​Against an older engine

Observing functions

Publish your own events

Know when a function is observed

When to use owner delivery

Advanced function observation with owner delivery

Example: a file watcher that emits

Example: a command router that calls

Errors of `obs.call`

Reject an observer later

Refuse observers

Accept old triggers

The `observe` descriptor

Hook rules

The `Observer` object

Against an older engine