Iii Architecture PatternsSAFE

Effortlessly compose, extend, and observe every service in real-time for the first time ever.

The instruction file, verbatim from the audited commit — this is the text the model reads, and the surface the audit's instruction layer examines. Quoted here so you can judge it without cloning anything.

---
name: iii-architecture-patterns
description: >-
  Use when composing iii primitives into backend architectures: durable workflows,
  reactive backends, observer side effects, agentic pipelines, event-driven CQRS, effect
  pipelines, and trigger-transform-action automation.
---

# Architecture Patterns

Use this skill when the task is larger than one function or trigger. Pick the pattern from the
requirement, then implement it with `Function`, `Trigger`, `Worker`, observers, state, queues,
streams, and pub/sub.

## Pattern Map

| Requirement | Pattern | iii shape |
| --- | --- | --- |
| Sequential work with retries, DLQ, step tracking | Durable workflow | Functions chained through named queues, progress in state |
| Keep views, metrics, cache, or clients in sync | Reactive backend | State triggers plus stream/pubsub side effects |
| React to another function's calls, results, or failures without changing it | Observer side effect | `iii.observe({ function_id, on, filters }, target)` (`iii-observe`) |
| Specialized AI agents hand work to each other | Agentic backend | One function per agent, queue handoffs, shared state |
| Commands publish events and projections update independently | Event-driven CQRS | Command functions, event log in state, subscribe triggers |
| Pure, traceable composition | Effect pipeline | Small functions composed synchronously with `trigger()` |
| Webhook/cron automation chains | Low-code automation | Trigger, transform, action nodes chained by enqueue |

## Durable Workflow

Use named queues when steps need different retry, FIFO, or concurrency policy.

### TypeScript

```typescript
import { registerWorker, TriggerAction } from "iii-sdk";

const iii = registerWorker("ws://localhost:49134", { workerName: "order-workflow" });

async function track(orderId: string, step: string, status: string) {
  await iii.trigger({
    function_id: "state::update",
    payload: { scope: "orders", key: orderId, ops: [{ type: "merge", path: ["steps"], value: { [step]: status } }] },
  });
}

iii.registerFunction("orders::validate", async (order) => {
  await track(order.id, "validate", "done");
  return iii.trigger({
    function_id: "orders::charge",
    payload: order,
    action: TriggerAction.Enqueue({ queue: "order-payment" }),
  });
});

iii.registerFunction("orders::charge", async (order) => {
  await track(order.id, "payment", "done");
  return iii.trigger({
    function_id: "orders::ship",
    payload: order,
    action: TriggerAction.Enqueue({ queue: "order-ship" }),
  });
});
```

### Python

```python
from iii import register_worker

iii = register_worker("ws://localhost:49134")

def track(order_id, step, status):
    iii.trigger({
        "function_id": "state::update",
        "payload": {
            "scope": "orders",
            "key": order_id,
            "ops": [{"type": "merge", "path": ["steps"], "value": {step: status}}],
        },
    })

def validate(order):
    track(order["id"], "validate", "done")
    return iii.trigger({
        "function_id": "orders::charge",
        "payload": order,
        "action": {"type": "enqueue", "queue": "order-payment"},
    })

iii.register_function("orders::validate", validate)
```

### Rust

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

async fn enqueue_charge(iii: iii_sdk::IIIClient, order: serde_json::Value) -> Result<serde_json::Value, iii_sdk::Error> {
    iii.trigger(TriggerRequest {
        function_id: "state::update".into(),
        payload: json!({
            "scope": "orders",
            "key": order["id"],
            "ops": [{ "type": "merge", "path": ["steps"], "value": { "validate": "done" } }]
        }),
        action: None,
        timeout_ms: None,
    }).await?;

    iii.trigger(TriggerRequest {
        function_id: "orders::charge".into(),
        payload: order,
        action: Some(TriggerAction::Enqueue { queue: "order-payment".into() }),
        timeout_ms: None,
    }).await
}
```

## Reactive Backend

Use state triggers when the requirement says "after create/update, do this", "avoid polling", or
"push live updates".

```typescript
iii.registerFunction("todos::on-change", async (event) => {
  await iii.trigger({
    function_id: "stream::send",
    payload: { stream_name: "todos-live", group_id: "default", data: event.new_value },
    action: TriggerAction.Void(),
  });
});

iii.registerTrigger({
  type: "state",
  function_id: "todos::on-change",
  config: { scope: "todos" },
});
```

## Observer Side Effects

Use an observer when "after X runs (or fails), also do Y" and X should not know about Y: audit
logs, alerts, projections, cache invalidation. Every function is observable since 0.25, and an
observer failure never affects the observed call.

```typescript
import type { Observed } from "iii-sdk";

iii.registerFunction("audit::order-created", async (event: Observed) => {
  await iii.trigger({
    function_id: "state::set",
    payload: { scope: "audit", key: event.invocation_id, value: { order: event.input, by: event.caller?.worker_name } },
  });
  return null;
});

// every successful orders::create, except the replay tool's own calls
iii.observe(
  { function_id: "orders::create", filters: { "caller.worker_name": { $ne: "order-replayer" } } },
  "audit::order-created",
  { max_in_flight: 8 },
);
```

```python
def notify(event):  # the Observed envelope: input, error, caller, ...
    print("charge failed", event["input"], event["error"]["code"])

iii.register_function("alerts::payment-failed", notify)
iii.observe(
    {"function_id": "payments::charge", "on": "error", "filters": {"input.amount": {"$gte": 500}}},
    "alerts::payment-failed",
)
```

Activations are at most once and fire-and-forget. When the side effect must complete with retries,
have the target enqueue the work (`TriggerAction.Enqueue`). Filter on `caller` or
`caller.parent` to stop echo loops; the engine also drops activations deeper than
`observers.max_observe_d

SAFEgrade B · trust 89/100 Nothing in the source contradicts what it says it does. Grade A is reserved for packages that have also passed the behavioural sandbox.

What the source does

Filesystem

none-observed

Network

none-observed

Shell

none-observed

Dependencies

pinned

Secrets in source

none-found

Findings (1)

sdk/packages/node/iii/local.yml

Gates applied: no_behavioural_pass.

[![ModelsAtlas audit: Iii Architecture Patterns](https://modelsatlas.com/skills/iii-hq/iii-architecture-patterns/badge.svg)](https://modelsatlas.com/skills/iii-hq/iii-architecture-patterns/)

Not your listing, or something out of date? Claim or update your listing.