/iii-observe

name: iii-observe
description: >-
  Use when observing iii functions with iii.observe, writing filters, reacting to engine events,
  writing observable functions with owner delivery (onObserverAdded, obs.emit, obs.call), tuning
  local dispatch, or migrating from the deprecated registerTrigger / registerTriggerType across
  TypeScript, browser, Python, Rust, and Go.

Observe

Since iii 0.25 every function is observable. `iii.observe(source, target, options?)` connects a source function to a target function: each event of the source that passes the observer's filters invokes the target. It replaces `registerTrigger`, and an `observe` descriptor on `registerFunction` replaces `registerTriggerType`. Both old APIs still work in 0.25 (see [Migrating from triggers](#migrating-from-triggers)).

| Part | Fields | Notes | | --- | --- | --- | | `source` (left side) | `function_id`, `namespace?`, `args?`, `filters?`, `on?`, `capture?`, `deliver?` | `args` only for owner-delivered functions. `on` (auto only): `success` (default), `error`, `complete`. `capture` (auto and `obs.emit`): default `['input', 'output', 'caller']`, `[]` allowed. `deliver`: `envelope` (default), `input`, `output` | | `target` (right side) | a registered function id (or `{ function_id, namespace }`) | No inline handler. The observer stays `pending` until the target is registered | | `options` | `id?`, `max_fires?`, `expires_in_ms?`, `max_in_flight?`, `metadata?` | `id` defaults to a UUID kept across replays. `metadata` is the sidecar of every activation |

The source namespace resolves like `trigger()`: the worker's namespace, except `engine::*` and `configuration::*`, which resolve in `default`.

Observe a function

Register the target first, then observe. This alerts on failed charges of 500 or more:

import { registerWorker, type Observed } from "iii-sdk";

const iii = registerWorker(process.env.III_URL ?? "ws://localhost:49134");

iii.registerFunction("alerts::payment-failed", async (event: Observed<{ amount: number }>) => {
  console.log(event.function_id, event.error?.code, event.input?.amount, event.caller?.worker_name);
  return null; // ignored for auto delivery
});

const failures = iii.observe(
  {
    function_id: "payments::charge",
    on: "error",
    filters: { "input.amount": { $gte: 500 } },
    capture: ["input", "error", "caller"],
  },
  "alerts::payment-failed",
  { max_in_flight: 4 },
);
await failures.ready(); // rejects with ObserveRejection (error.code) on error or disposed
// later: await failures.dispose()

iii.register_function("alerts::payment-failed", lambda event: print(event["error"]["code"]))

failures = iii.observe(
    {"function_id": "payments::charge", "on": "error",
     "filters": {"input.amount": {"$gte": 500}}, "capture": ["input", "error", "caller"]},
    "alerts::payment-failed",
    {"max_in_flight": 4},
)
failures.ready()  # blocks; in async code: await failures.ready_async()

use iii_sdk::{Capture, Error, ObserverOptions, Observed, On, RegisterFunction, Source};
use serde_json::{json, Value};

iii.register_function(
    "alerts::payment-failed",
    RegisterFunction::new(|event: Observed<Value>| {
        println!("{} failed: {:?}", event.function_id, event.error);
        Ok::<_, Error>(())
    }),
);
let failures = iii.observe_with(
    Source::function("payments::charge")
        .on(On::Error)
        .filters(json!({ "input.amount": { "$gte": 500 } }))
        .capture([Capture::Input, Capture::Error, Capture::Caller]),
    "alerts::payment-failed",
    ObserverOptions::default().max_in_flight(4),
);
failures.ready().await?;

client.RegisterFunction("alerts::payment-failed", func(ctx context.Context, data json.RawMessage) (any, error) {
	var ev iii.Observed
	if err := json.Unmarshal(data, &ev); err != nil {
		return nil, err
	}
	log.Printf("%s failed: %s", ev.FunctionID, ev.Error.Code)
	return nil, nil
})
failures, err := client.Observe(iii.Source{
	FunctionID: "payments::charge",
	On:         iii.OnError,
	Filters:    iii.Filters{"input.amount": iii.Op{"$gte": 500}},
	Capture:    []iii.Capture{iii.CaptureInput, iii.CaptureError, iii.CaptureCaller},
}, "alerts::payment-failed", iii.ObserverOptions{MaxInFlight: 4})
if err == nil {
	err = failures.Ready(ctx) // *iii.ObserveRejectionError carries the code
}

The browser SDK (`iii-browser-sdk`) has the same `observe` API as Node.

What the target receives

• **Envelope** (default `deliver`): `{ observer_id, activation_id, function_id, namespace,

invocation_id, status, input?, output?, error?, caller?, metadata?, started_at_ms, duration_ms }`. Only captured fields are present. `deliver: 'input'` or `'output'` sends that value alone.

• The observer's `options.metadata` arrives as the handler's metadata argument (Node and browser

second argument, Python `metadata` parameter, the `Option<Value>` of a Rust `Fn(Value, Option<Value>)` handler, Go `iii.MetadataFromContext(ctx)`).

• Activations are fire-and-forget and at most once; the target's return value is discarded, except

for `obs.call` (below). Drops are counted, never silent.

Handle and states

`pending` (no owner, no ack, or no target yet) → `active` or `error`. `suspended` while the target is away, or while the observing worker is disconnected (30 s grace, then `disposed`). `disposed` after `dispose()`, `max_fires`, or expiry. `observe()` returns synchronously, works before the connection is up, and is replayed with the same id on reconnect. `onStateChange(listener)` (Python `on_state_change`, Go `OnStateChange`, Rust `on_state_change`) reports every transition with a reason code.

Filters

Filters are declarative JSON, evaluated by the same `iii-filter` language in the engine and every SDK. Full reference: https://iii.dev/docs/reference/filters.

• A clause is `"<path>": { "$op": operand }`. Explicit operators only: `{ "input.op": "insert" }`

is `INVALID_FILTER`; write `{ "input.op": { "$eq