iii / worker
$worker

state

v0.1.1

Distributed key-value state management with reactive change triggers — registers the `state` trigger type and the `state::*` functions.

  • macOS: arm64 · x64
  • Linux: arm64 · armv7 · x64
  • Windows: arm64 · x64 · x86
agent-ready brief for v0.1.1
install + config + dependencies + readme + api reference, all in one place. fetch as agent-context.md for an llm to consume.

full markdown

open as agent-context.md
the same content rendered as discrete blocks below is exposed as a single markdown document at /workers/state.md?version=0.1.1. paste it into an llm prompt or pipe it through curl from a worker.

install

install
$iii worker add state@0.1.1

configuration

iii-config.yaml
- adapter:
    config:
      file_path: ./data/state_store.db
      store_method: file_based
    name: kv

dependencies

dependencies1
configuration^0.19.0

readme

README.md

state

Distributed key-value state storage with scope-based organization and reactive change triggers. Values are addressed by a scope (namespace) and a key, shared across every worker connected to the engine, and persisted through a pluggable adapter (kv or redis). Callers reach the store through six functions — state::set, state::get, state::delete, state::update, state::list, state::list_groups — and this worker also registers the state trigger type, which fires state:created, state:updated, or state:deleted after every successful mutation so downstream functions can react to data changes without polling. This worker is the standalone migration of the engine's built-in iii-state.

Install

iii worker add state

iii worker add fetches the binary, writes a config block into ~/.iii/config.yaml, and the engine starts the worker on the next iii start.

Functions

Function Input Returns Fires
state::set { scope, key, value } (data accepted as an alias for value) { old_value, new_value } state:created (new key) or state:updated
state::get { scope, key } the value, or null —
state::delete { scope, key } the deleted value, or null state:deleted (even when the key did not exist)
state::update { scope, key, ops } — ordered atomic ops: set, merge, increment, decrement, append, remove { old_value, new_value, errors } state:created or state:updated
state::list { scope } flat array of every value in the scope —
state::list_groups {} { groups } — sorted, deduplicated scope names —

Trigger type

This worker always registers the state trigger type. Bind a function to it with:

Field Required Default Description
scope no any scope Only fire for writes in this scope.
key no any key Only fire for writes to this key.
condition_function_id no — Function invoked first with the event; only an explicit false return skips the handler (null/no result passes, an error skips and logs).

Trigger delivery is asynchronous: handlers run after the write completes and a handler failure never rolls the write back. Duplicate trigger ids replace the previous binding silently (builtin parity).

iii.registerTrigger({
  type: 'state',
  function_id: 'orders::on-status-change',
  config: { scope: 'orders', key: 'status' },
})

The handler receives the event payload:

{
  "type": "state",
  "event_type": "state:updated",
  "scope": "orders",
  "key": "status",
  "old_value": { "status": "pending" },
  "new_value": { "status": "shipped" }
}

event_type is one of state:created, state:updated, state:deleted; old_value is null for created keys and new_value is null for deleted keys.

Configuration

Field Default Description
adapter kv Storage adapter: kv (in-process; store_method: in_memory or file_based with file_path and save_interval_ms) or redis (redis_url, default redis://localhost:6379). Restart-tier: a runtime change is logged and takes effect at the next worker start.
triggers_enabled true Globally enable/disable state change-trigger fan-out. Applied live.
max_value_bytes unset (no limit) Reject state::set writes whose JSON-serialized value exceeds this many bytes (VALUE_TOO_LARGE). Minimum 1. Applied live.
save_interval_ms 5000 Persistence flush cadence (ms) for the file-backed kv adapter. 100–3600000. Applied live by respawning the adapter's save loop (hot-retune; no-op for in-memory/redis).

Configuration is owned by the configuration worker — edit it from the console (Configuration → Workers → state) or seed it once via the worker's config block in config.yaml on first boot. triggers_enabled and max_value_bytes apply on the next write, save_interval_ms hot-retunes the save loop, and adapter takes effect on the next restart.

Requires removing the built-in iii-state worker

The built-in iii-state worker also owns the state trigger type and the state::* functions. Two owners of the same surface on one engine collide — whichever registers last wins — so this worker requires iii-state to be absent: omit it from the engine's config.yaml (a config that doesn't list a worker won't run it).

On boot, this worker queries the engine for connected workers and refuses to start with a clear error if iii-state is still active, so a stale config fails loudly instead of silently racing the built-in worker for ownership of state.

Store migration: if the builtin used a file-based kv store, point this worker's adapter.config.file_path at the builtin's existing directory (the default engine config used ./data/state_store.db). The on-disk format (one rkyv .bin file per scope) is identical, so the existing data loads as-is — no export/import step.

Latency

The builtin ran in-process inside the engine, so a state::* call cost microseconds; as a standalone worker every call crosses the engine⇄worker WebSocket, which puts round-trips in the low-millisecond range. That order-of-magnitude delta is inherent to the migration and applies to every externalized builtin. A formal benchmark was waived by project decision (2026-07-06).

Parity vs builtin

Behavior Builtin This worker
Function ids state::set/get/delete/update/list/list_groups same (exact)
set input {scope, key, value} (data alias) same
Events state:created/updated/deleted, payload {type:"state", event_type, scope, key, old_value, new_value} same
Trigger config {scope?, key?, condition_function_id?}; only explicit false blocks same
Duplicate trigger id silent replace same
Trigger metadata forwarded to handlers via call_with_metadata not forwarded (iii-sdk 0.20 TriggerRequest has no metadata field; same limitation as the http worker)
Store adapters kv (in_memory/file_based), redis, bridge kv, redis (bridge not ported — see docs/adr/0001)
kv on-disk format rkyv .bin per scope identical — builtin data loads as-is
save_interval_ms default 5000ms, floor 100ms, hot-retune same
max_value_bytes guards set only, VALUE_TOO_LARGE same (code is the message prefix)
Error codes coded ErrorBody (SET_ERROR, ...) code as message prefix (SDK handler errors carry a message)
Durability store dies with the ENGINE process store dies with the WORKER process (ADR 0001; file_based/redis unchanged)
Latency in-process µs engine⇄worker WS round-trip (low ms) — formal benchmark waived by project decision 2026-07-06
Telemetry (track_state_*) engine-internal counters none — out of scope for this migration; lands with the shared worker observability story

api reference (json)

agent-api-reference.json
{
  "functions": [
    {
      "description": "Delete a value from state",
      "metadata": {},
      "name": "state::delete",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "key": {
            "description": "Identifier for the value to delete within the scope.",
            "type": "string"
          },
          "scope": {
            "description": "Namespace that groups related keys.",
            "type": "string"
          }
        },
        "required": [
          "key",
          "scope"
        ],
        "title": "StateDeleteInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "The value that was deleted (read before delete), or null if it did not exist.",
        "title": "StateDeleteResponse",
        "type": [
          "string",
          "number",
          "boolean",
          "object",
          "array",
          "null"
        ]
      }
    },
    {
      "description": "Get a value from state",
      "metadata": {},
      "name": "state::get",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "key": {
            "description": "Identifier for the value within the scope.",
            "type": "string"
          },
          "scope": {
            "description": "Namespace that groups related keys.",
            "type": "string"
          }
        },
        "required": [
          "key",
          "scope"
        ],
        "title": "StateGetInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "The raw value stored at scope/key, or null if absent.",
        "title": "StateGetResponse",
        "type": [
          "string",
          "number",
          "boolean",
          "object",
          "array",
          "null"
        ]
      }
    },
    {
      "description": "Get a group from state",
      "metadata": {},
      "name": "state::list",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "scope": {
            "description": "Namespace whose keys should be listed as a group.",
            "type": "string"
          }
        },
        "required": [
          "scope"
        ],
        "title": "StateGetGroupInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "The values in the given scope, as a JSON array (or null on the rare serialization failure).",
        "title": "StateListResponse",
        "type": [
          "string",
          "number",
          "boolean",
          "object",
          "array",
          "null"
        ]
      }
    },
    {
      "description": "List all state groups",
      "metadata": {},
      "name": "state::list_groups",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "StateListGroupsInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "groups": {
            "items": {
              "type": "string"
            },
            "type": "array"
          }
        },
        "required": [
          "groups"
        ],
        "title": "StateListGroupsResult",
        "type": "object"
      }
    },
    {
      "description": "Internal: reload state configuration from the authoritative store on change.",
      "metadata": {},
      "name": "state::on-config-change",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "title": "ConfigChangeRequest",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "ok": {
            "type": "boolean"
          }
        },
        "required": [
          "ok"
        ],
        "title": "ConfigChangeAck",
        "type": "object"
      }
    },
    {
      "description": "Set a value in state",
      "metadata": {},
      "name": "state::set",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "key": {
            "description": "Identifier for the value within the scope.",
            "type": "string"
          },
          "scope": {
            "description": "Namespace that groups related keys (e.g. `users`, `orders`).",
            "type": "string"
          },
          "value": {
            "description": "Arbitrary JSON value to store. Replaces any existing value at `scope`/`key`."
          }
        },
        "required": [
          "key",
          "scope",
          "value"
        ],
        "title": "StateSetInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "properties": {
          "new_value": {
            "description": "The value after the update"
          },
          "old_value": {
            "description": "The value before the update (None if key didn't exist)"
          }
        },
        "required": [
          "new_value"
        ],
        "title": "StreamSetResult",
        "type": "object"
      }
    },
    {
      "description": "Update a value in state",
      "metadata": {},
      "name": "state::update",
      "request_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "MergePath": {
            "anyOf": [
              {
                "type": "string"
              },
              {
                "items": {
                  "type": "string"
                },
                "type": "array"
              }
            ],
            "description": "Path target for a [`UpdateOp::Merge`] operation. Accepts either a single string (legacy / first-level field) or an array of literal segments (nested path).\n\nPath normalization rules applied by the engine: - absent / `Single(\"\")` / `Segments(vec![])` → root merge - `Single(\"foo\")` is equivalent to `Segments(vec![\"foo\".into()])` - `Segments([\"a\", \"b\", \"c\"])` walks three literal keys, never interpreting dots specially. `Segments(vec![\"a.b\".into()])` is a single literal key named `\"a.b\"`.\n\n**Variant ordering is load-bearing.** `#[serde(untagged)]` tries variants in declaration order, `Single` MUST come before `Segments` so a JSON string deserializes into `Single` rather than failing the array match first."
          },
          "UpdateOp": {
            "description": "Operations that can be performed atomically on a stream value",
            "oneOf": [
              {
                "description": "Set a value at path (overwrite)",
                "properties": {
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "set"
                    ],
                    "type": "string"
                  },
                  "value": true
                },
                "required": [
                  "path",
                  "type"
                ],
                "type": "object"
              },
              {
                "description": "Merge object into existing value (object-only). Path may be omitted (root merge), a single first-level key, or an array of literal segments for nested merge. See [`MergePath`].",
                "properties": {
                  "path": {
                    "anyOf": [
                      {
                        "$ref": "#/definitions/MergePath"
                      },
                      {
                        "type": "null"
                      }
                    ]
                  },
                  "type": {
                    "enum": [
                      "merge"
                    ],
                    "type": "string"
                  },
                  "value": true
                },
                "required": [
                  "type",
                  "value"
                ],
                "type": "object"
              },
              {
                "description": "Increment numeric value",
                "properties": {
                  "by": {
                    "format": "int64",
                    "type": "integer"
                  },
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "increment"
                    ],
                    "type": "string"
                  }
                },
                "required": [
                  "by",
                  "path",
                  "type"
                ],
                "type": "object"
              },
              {
                "description": "Decrement numeric value",
                "properties": {
                  "by": {
                    "format": "int64",
                    "type": "integer"
                  },
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "decrement"
                    ],
                    "type": "string"
                  }
                },
                "required": [
                  "by",
                  "path",
                  "type"
                ],
                "type": "object"
              },
              {
                "description": "Append an element to an array or concatenate a string at the optional path. Path may be omitted (root append), a single first-level key, or an array of literal segments for nested append. See [`MergePath`] for the variant shape.",
                "properties": {
                  "path": {
                    "anyOf": [
                      {
                        "$ref": "#/definitions/MergePath"
                      },
                      {
                        "type": "null"
                      }
                    ]
                  },
                  "type": {
                    "enum": [
                      "append"
                    ],
                    "type": "string"
                  },
                  "value": true
                },
                "required": [
                  "type",
                  "value"
                ],
                "type": "object"
              },
              {
                "description": "Remove a field",
                "properties": {
                  "path": {
                    "type": "string"
                  },
                  "type": {
                    "enum": [
                      "remove"
                    ],
                    "type": "string"
                  }
                },
                "required": [
                  "path",
                  "type"
                ],
                "type": "object"
              }
            ]
          }
        },
        "properties": {
          "key": {
            "description": "Identifier for the value to update within the scope.",
            "type": "string"
          },
          "ops": {
            "description": "Ordered list of update operations applied atomically to the existing value.",
            "items": {
              "$ref": "#/definitions/UpdateOp"
            },
            "type": "array"
          },
          "scope": {
            "description": "Namespace that groups related keys.",
            "type": "string"
          }
        },
        "required": [
          "key",
          "ops",
          "scope"
        ],
        "title": "StateUpdateInput",
        "type": "object"
      },
      "response_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "UpdateOpError": {
            "description": "Per-op error reported by an atomic update operation.",
            "properties": {
              "code": {
                "description": "Stable error code, e.g. `\"merge.path.too_deep\"`.",
                "type": "string"
              },
              "doc_url": {
                "description": "Optional documentation URL for this error class.",
                "type": [
                  "string",
                  "null"
                ]
              },
              "message": {
                "description": "Human-readable description with concrete numbers when applicable.",
                "type": "string"
              },
              "op_index": {
                "description": "Index of the offending op within the original `ops` array.",
                "format": "uint",
                "minimum": 0,
                "type": "integer"
              }
            },
            "required": [
              "code",
              "message",
              "op_index"
            ],
            "type": "object"
          }
        },
        "description": "Result of an atomic update operation",
        "properties": {
          "errors": {
            "description": "Errors encountered while applying ops. Successfully applied ops are still reflected in `new_value`. Field is omitted from JSON when empty for backward compatibility.",
            "items": {
              "$ref": "#/definitions/UpdateOpError"
            },
            "type": "array"
          },
          "new_value": {
            "description": "The value after the update"
          },
          "old_value": {
            "description": "The value before the update (None if key didn't exist)"
          }
        },
        "required": [
          "new_value"
        ],
        "title": "StreamUpdateResult",
        "type": "object"
      }
    }
  ],
  "triggers": [
    {
      "description": "State trigger",
      "invocation_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "description": "Trigger config schema — field-parity with the engine's StateTriggerConfig (engine/src/workers/state/trigger.rs:18-23).",
        "properties": {
          "condition_function_id": {
            "description": "Optional function evaluated per event; only an explicit `false` skips.",
            "type": [
              "string",
              "null"
            ]
          },
          "key": {
            "description": "Only fire for writes to this key (any key when unset).",
            "type": [
              "string",
              "null"
            ]
          },
          "scope": {
            "description": "Only fire for writes in this scope (any scope when unset).",
            "type": [
              "string",
              "null"
            ]
          }
        },
        "title": "StateTriggerSpec",
        "type": "object"
      },
      "metadata": {},
      "name": "state",
      "return_schema": {
        "$schema": "http://json-schema.org/draft-07/schema#",
        "definitions": {
          "StateEventType": {
            "enum": [
              "state:created",
              "state:updated",
              "state:deleted"
            ],
            "type": "string"
          }
        },
        "properties": {
          "event_type": {
            "allOf": [
              {
                "$ref": "#/definitions/StateEventType"
              }
            ],
            "description": "Type of state change"
          },
          "key": {
            "description": "State key",
            "type": "string"
          },
          "new_value": {
            "description": "New value"
          },
          "old_value": {
            "description": "Previous value (null for created events)"
          },
          "scope": {
            "description": "State scope",
            "type": "string"
          },
          "type": {
            "description": "Always \"state\"",
            "type": "string"
          }
        },
        "required": [
          "event_type",
          "key",
          "new_value",
          "scope",
          "type"
        ],
        "title": "StateCallRequest",
        "type": "object"
      }
    }
  ]
}