Upgrading from 0.25.x to 0.26.x

from iii import TriggerAction

order = iii.trigger({"function_id": "orders::create", "payload": {"sku": "A1", "qty": 2}})
await iii.trigger_async(
    {"function_id": "audit::log", "payload": entry, "action": TriggerAction.Void()}
)

from iii import CallAction

order = iii.call({"function_id": "orders::create", "payload": {"sku": "A1", "qty": 2}})
await iii.call_async(
    {"function_id": "audit::log", "payload": entry, "action": CallAction.Void()}
)

use iii_sdk::TriggerAction;
use iii_sdk::protocol::TriggerRequest;
use serde_json::json;

let order = iii
    .trigger(TriggerRequest::new("orders::create", json!({ "sku": "A1", "qty": 2 })))
    .await?;
iii.trigger(TriggerRequest {
    function_id: "audit::log".into(),
    payload: entry,
    action: Some(TriggerAction::Void),
    timeout_ms: None,
})
.await?;

use iii_sdk::CallAction;
use iii_sdk::protocol::CallRequest;
use serde_json::json;

let order = iii
    .call(CallRequest::new("orders::create", json!({ "sku": "A1", "qty": 2 })))
    .await?;
iii.call(CallRequest {
    function_id: "audit::log".into(),
    payload: entry,
    action: Some(CallAction::Void),
    timeout_ms: None,
})
.await?;

order, err := client.Trigger(ctx, iii.TriggerRequest{
    FunctionID: "orders::create",
    Payload:    json.RawMessage(`{"sku":"A1","qty":2}`),
})
if err != nil {
    return err
}
_, err = client.Trigger(ctx, iii.TriggerRequest{
    FunctionID: "audit::log",
    Payload:    entry,
    Action:     iii.VoidAction(),
})

order, err := client.Call(ctx, iii.CallRequest{
    FunctionID: "orders::create",
    Payload:    json.RawMessage(`{"sku":"A1","qty":2}`),
})
if err != nil {
    return err
}
_, err = client.Call(ctx, iii.CallRequest{
    FunctionID: "audit::log",
    Payload:    entry,
    Action:     iii.VoidAction(),
})

iii trigger orders::create --json '{"sku":"A1","qty":2}'

iii call orders::create --json '{"sku":"A1","qty":2}'

trigger = iii.register_trigger({
    "type": "cron",
    "function_id": "reports::daily-summary",
    "config": {"expression": "0 0 9 * * * *"},
    "metadata": {"team": "reports"},
})
# later
trigger.unregister()

observer = iii.observe(
    {"function_id": "cron::on-tick", "args": {"expression": "0 0 9 * * * *"}, "capture": ["input"]},
    "reports::daily-summary",
    {"metadata": {"team": "reports"}},
)
observer.ready()  # or: await observer.ready_async()
# later
observer.dispose()

use iii_sdk::protocol::RegisterTriggerInput;
use serde_json::json;

let trigger = iii.register_trigger(
    RegisterTriggerInput::new(
        "cron",
        "reports::daily-summary",
        json!({ "expression": "0 0 9 * * * *" }),
    )
    .with_metadata(json!({ "team": "reports" })),
)?;
// later
trigger.unregister();

use iii_sdk::{Capture, ObserverOptions, Source};
use serde_json::json;

let observer = iii.observe_with(
    Source::function("cron::on-tick")
        .args(json!({ "expression": "0 0 9 * * * *" }))
        .capture([Capture::Input]),
    "reports::daily-summary",
    ObserverOptions::default().metadata(json!({ "team": "reports" })),
);
observer.ready().await?;
// later
observer.dispose().await?;

trigger, err := client.RegisterTrigger(iii.RegisterTriggerInput{
    Type:       "cron",
    FunctionID: "reports::daily-summary",
    Config:     json.RawMessage(`{"expression":"0 0 9 * * * *"}`),
    Metadata:   json.RawMessage(`{"team":"reports"}`),
})
if err != nil {
    log.Fatal(err)
}
// later
trigger.Unregister()

observer, err := client.Observe(iii.Source{
    FunctionID: "cron::on-tick",
    Args:       iii.Args{"expression": "0 0 9 * * * *"},
    Capture:    []iii.Capture{iii.CaptureInput},
}, "reports::daily-summary", iii.ObserverOptions{Metadata: json.RawMessage(`{"team":"reports"}`)})
if err != nil {
    log.Fatal(err)
}
if err := observer.Ready(ctx); err != nil {
    log.Fatal(err)
}
// later
observer.Dispose(ctx)

import threading
import time

from iii.trigger import TriggerConfig, TriggerHandler

bindings: dict[str, threading.Event] = {}  # trigger id → stop flag

class TimerHandler(TriggerHandler):
    async def register_trigger(self, config: TriggerConfig) -> None:
        every_ms = config.config["every_ms"]
        stop = bindings[config.id] = threading.Event()

        def run() -> None:
            while not stop.wait(every_ms / 1000):
                iii.trigger({
                    "function_id": config.function_id,
                    "payload": {"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)},
                })

        threading.Thread(target=run, daemon=True).start()

    async def unregister_trigger(self, config: TriggerConfig) -> None:
        stop = bindings.pop(config.id, None)
        if stop:
            stop.set()

iii.register_trigger_type(
    {"id": "timer", "description": "Fires every every_ms milliseconds"},
    TimerHandler(),
)

import threading
import time

from iii import CallAction

# every_ms → (stop flag, number of observers that share the timer)
timers: dict[int, tuple[threading.Event, int]] = {}

def on_tick(tick: dict) -> dict:
    return tick

# by id, before registering, so the hooks see every observer
@iii.on_observer_added("timer::on-tick")
def start_timer(obs):
    every_ms = obs.args["every_ms"]
    if every_ms in timers:
        stop, count = timers[every_ms]
        timers[every_ms] = (stop, count + 1)
        return
    stop = threading.Event()
    timers[every_ms] = (stop, 1)

    def run() -> None:
        while not stop.wait(every_ms / 1000):
            iii.call({
                "function_id": "timer::on-tick",
                "payload": {"every_ms": every_ms, "fired_at_ms": int(time.time() * 1000)},
                "action": CallAction.Void(),
            })

    threading.Thread(target=run, daemon=True).start()

@iii.on_observer_removed("timer::on-tick")
def stop_timer(obs):
    every_ms = obs.args["every_ms"]
    stop, count = timers[every_ms]
    if count == 1:
        stop.set()
        del timers[every_ms]
    else:
        timers[every_ms] = (stop, count - 1)

iii.on_observation_match(
    "timer::on-tick",
    lambda obs, call: obs.args["every_ms"] == call.input["every_ms"],
)

iii.register_function(
    "timer::on-tick",
    on_tick,
    description="One call per tick. Observe it with args { every_ms }.",
    args_format={
        "type": "object",
        "additionalProperties": False,
        "required": ["every_ms"],
        "properties": {"every_ms": {"type": "integer", "minimum": 100}},
    },
    private=True,
)

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use iii_sdk::protocol::TriggerRequest;
use iii_sdk::trigger::{TriggerConfig, TriggerHandler};
use iii_sdk::{Error, IIIClient, RegisterTriggerType};
use serde_json::json;

#[derive(Clone)]
struct TimerHandler {
    iii: IIIClient,
    tasks: Arc<Mutex<HashMap<String, tokio::task::JoinHandle<()>>>>, // trigger id → its task
}

#[async_trait::async_trait]
impl TriggerHandler for TimerHandler {
    async fn register_trigger(&self, config: TriggerConfig) -> Result<(), Error> {
        let every_ms = config.config["every_ms"].as_u64().unwrap_or(1_000);
        let (iii, function_id) = (self.iii.clone(), config.function_id.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).unwrap().as_millis() as u64;
                let tick = json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms });
                let _ = iii.trigger(TriggerRequest::new(function_id.clone(), tick)).await;
            }
        });
        self.tasks.lock().unwrap().insert(config.id, task);
        Ok(())
    }

    async fn unregister_trigger(&self, config: TriggerConfig) -> Result<(), Error> {
        if let Some(task) = self.tasks.lock().unwrap().remove(&config.id) {
            task.abort();
        }
        Ok(())
    }
}

iii.register_trigger_type(RegisterTriggerType::new(
    "timer",
    "Fires every every_ms milliseconds",
    TimerHandler { iii: iii.clone(), tasks: Default::default() },
));

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use iii_sdk::protocol::CallRequest;
use iii_sdk::{
    AddedHook, Error, IIIClient, Match, ObserveRejection, Observer, RegisterFunction, RemovedHook,
};
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 → its task and the number of observers that share it
#[derive(Clone)]
struct Timers {
    iii: IIIClient,
    tasks: Arc<Mutex<HashMap<u64, (tokio::task::JoinHandle<()>, usize)>>>,
}

impl AddedHook for Timers {
    async fn on_added(&self, obs: Observer) -> Result<(), ObserveRejection> {
        let TimerArgs { every_ms } = obs.args()?;
        let mut tasks = self.tasks.lock().unwrap();
        if let Some((_, observers)) = tasks.get_mut(&every_ms) {
            *observers += 1;
            return Ok(());
        }
        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).unwrap().as_millis() as u64;
                let tick = json!({ "every_ms": every_ms, "fired_at_ms": fired_at_ms });
                let _ = iii.call(CallRequest::new("timer::on-tick", tick)).await;
            }
        });
        tasks.insert(every_ms, (task, 1));
        Ok(())
    }
}

impl RemovedHook for Timers {
    async fn on_removed(&self, obs: Observer) {
        let Ok(TimerArgs { every_ms }) = obs.args() else { return };
        let mut tasks = self.tasks.lock().unwrap();
        let last = match tasks.get_mut(&every_ms) {
            Some((_, observers)) => {
                *observers -= 1;
                *observers == 0
            }
            None => false,
        };
        if last {
            if let Some((task, _)) = tasks.remove(&every_ms) {
                task.abort();
            }
        }
    }
}

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

// set the hooks by id before registering, so they run for every observer
iii.on_observer_added_for("timer::on-tick", timers.clone());
iii.on_observer_removed_for("timer::on-tick", timers);
iii.on_observation_match_for(
    "timer::on-tick",
    Match::predicate(|obs, call| obs.args_value()["every_ms"] == call.input["every_ms"]),
);

iii.register_function(
    "timer::on-tick",
    RegisterFunction::new(|tick: Tick| Ok::<_, Error>(tick))
        .description("One call per tick. Observe it with args { every_ms }.")
        .args_format::<TimerArgs>()
        .private(),
);

type timerHandler struct {
    client *iii.Client
    mu     sync.Mutex
    stops  map[string]chan struct{} // trigger id → stop
}

func (h *timerHandler) RegisterTrigger(ctx context.Context, cfg iii.TriggerConfig) error {
    var config struct {
        EveryMs int64 `json:"every_ms"`
    }
    if err := json.Unmarshal(cfg.Config, &config); err != nil {
        return err
    }
    stop := make(chan struct{})
    h.mu.Lock()
    h.stops[cfg.ID] = stop
    h.mu.Unlock()
    go func() {
        ticker := time.NewTicker(time.Duration(config.EveryMs) * time.Millisecond)
        defer ticker.Stop()
        for {
            select {
            case <-stop:
                return
            case now := <-ticker.C:
                data, _ := json.Marshal(map[string]int64{"every_ms": config.EveryMs, "fired_at_ms": now.UnixMilli()})
                h.client.Trigger(context.Background(), iii.TriggerRequest{FunctionID: cfg.FunctionID, Payload: data, Namespace: cfg.Namespace})
            }
        }
    }()
    return nil
}

func (h *timerHandler) UnregisterTrigger(ctx context.Context, cfg iii.TriggerConfig) error {
    h.mu.Lock()
    defer h.mu.Unlock()
    if stop, ok := h.stops[cfg.ID]; ok {
        close(stop)
        delete(h.stops, cfg.ID)
    }
    return nil
}

client.RegisterTriggerType(
    iii.RegisterTriggerTypeInput{ID: "timer", Description: "Fires every every_ms milliseconds"},
    &timerHandler{client: client, stops: map[string]chan struct{}{}},
)

type timerArgs struct {
    EveryMs int64 `json:"every_ms"`
}

type timer struct {
    stop      chan struct{}
    observers int
}

var (
    mu     sync.Mutex
    timers = map[int64]*timer{} // every_ms → its ticker and the observers that share it
)

func registerTimer(client *iii.Client) error {
    // set the hooks by id before registering, so they run for every observer
    client.OnObserverAddedFor("timer::on-tick", func(ctx context.Context, obs *iii.Observer) error {
        var args timerArgs
        if err := obs.DecodeArgs(&args); err != nil {
            return err
        }
        mu.Lock()
        defer mu.Unlock()
        if t, ok := timers[args.EveryMs]; ok {
            t.observers++
            return nil
        }
        t := &timer{stop: make(chan struct{}), observers: 1}
        timers[args.EveryMs] = t
        go func() {
            ticker := time.NewTicker(time.Duration(args.EveryMs) * time.Millisecond)
            defer ticker.Stop()
            for {
                select {
                case <-t.stop:
                    return
                case now := <-ticker.C:
                    tick, _ := json.Marshal(map[string]int64{"every_ms": args.EveryMs, "fired_at_ms": now.UnixMilli()})
                    client.Call(context.Background(), iii.CallRequest{FunctionID: "timer::on-tick", Payload: tick, Action: iii.VoidAction()})
                }
            }
        }()
        return nil
    })
    client.OnObserverRemovedFor("timer::on-tick", func(ctx context.Context, obs *iii.Observer) {
        var args timerArgs
        if obs.DecodeArgs(&args) != nil {
            return
        }
        mu.Lock()
        defer mu.Unlock()
        if t, ok := timers[args.EveryMs]; ok {
            t.observers--
            if t.observers == 0 {
                close(t.stop)
                delete(timers, args.EveryMs)
            }
        }
    })
    client.OnObservationMatchFor("timer::on-tick", iii.MatchFunc(func(obs *iii.Observer, call *iii.Call) bool {
        var args, tick timerArgs // the tick carries every_ms too
        obs.DecodeArgs(&args)
        json.Unmarshal(call.Input, &tick)
        return args.EveryMs == tick.EveryMs
    }))

    _, err := client.RegisterFunction("timer::on-tick", iii.Identity, iii.FunctionOptions{
        Description: "One call per tick. Observe it with args { every_ms }.",
        ArgsFormat:  json.RawMessage(`{"type":"object","additionalProperties":false,"required":["every_ms"],"properties":{"every_ms":{"type":"integer","minimum":100}}}`),
        Private:     true,
    })
    return err
}

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

iii.register_function(
    "chat::on-command",
    on_command,
    description="One invocation per chat command. Observe it with args { command }.",
    observe=CommandHooks(),
)

from iii import ObserveRejection

commands: dict = {}  # command → Observer

def on_command(req: dict) -> dict:
    return req

# by id, before registering, so the hooks see every observer
@iii.on_observer_added("chat::on-command")
def add_command(obs):
    command = obs.args["command"]
    if command in commands:
        raise ObserveRejection("COMMAND_TAKEN", f"/{command} already has a handler")
    commands[command] = obs

@iii.on_observer_removed("chat::on-command")
def remove_command(obs):
    command = obs.args["command"]
    if command in commands and commands[command].id == obs.id:
        del commands[command]

@iii.on_observation_deliver("chat::on-command")
async def route(call, observers):
    command = call.input["command"]
    obs = commands.get(command)
    if obs is None:
        return {"reply": f"unknown command /{command}"}
    if not obs.matches(call):
        return {"reply": f"/{command} is not available"}
    # what this hook returns is what the caller gets back
    return await iii.call_async({"observer": obs, "payload": call.input})

iii.register_function(
    "chat::on-command",
    on_command,
    description="One call per chat command. Observe it with args { command }.",
    args_format={
        "type": "object",
        "additionalProperties": False,
        "required": ["command"],
        "properties": {"command": {"type": "string"}},
    },
    private=True,
)

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

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