platform-state — shared-state seam (memory reference + Redis)

@demystify/platform-state (TS) · demystify-platform-state (Py) · v0.2.0 · MIT

What it is / when to use it

The cross-instance shared-state seam for the platform. Five ports, each with a semantically faithful in-memory reference adapter (the keyless/offline default) and a Redis adapter (opt-in, integration-marked). Backs gateway multi-instance correctness (KV, atomic budget counters, priority queue) and the voice session registry across instances (leased registry), plus event/health fan-out (pub/sub).

Leaf package: no module dependency; ioredis (TS) / redis (Py) are optional integration-only deps, imported lazily.

Ports

| port | methods | backs | |---|---|---| | KvStore | get · set(key,val,ttlMs?) · del/delete · ttl | caches, flags | | AtomicCounter | get · incrby · reserve(key,amount,limit) · reconcile(res,actual) · release(res) | budgets / rate limits | | PriorityQueueStore | enqueue(q,item,priority,ts?) · drain(q,max) · peek · depth | gateway request queue | | LeasedRegistry | register(id,val,ttlMs) · renew · list · get · evict | voice session registry | | PubSub | publish(ch,msg) · subscribe(ch,handler) → unsubscribe | events + provider-health |

reserve → reconcile (the budget protocol gateway needs)

res = counter.reserve("budget:tenant", estimatedMicroUsd, ceilingMicroUsd)
if (res === null)  → over budget, reject the request
... make the provider call ...
counter.reconcile(res, actualMicroUsd)   // adjusts by (actual - estimate)
// or counter.release(res)               // cancel (actual = 0)

reserve atomically books the estimate only if current + amount <= limit; reconcile/release are idempotent per reservation. The Redis adapter does the check-and-add in a Lua script so it is atomic across instances.

PriorityQueueStore ordering

drain returns items priority desc, then ts asc, then insertion order. The Redis adapter encodes score = -priority*1e13 + ts and uses ZPOPMIN.

LeasedRegistry (voice sessions across instances)

Entries self-expire; list returns only unexpired leases. An instance registers a session lease and renews it on a heartbeat; a crashed instance's sessions disappear after the TTL. Memory expiry is driven by an injectable clock (ManualClock) so tests are deterministic.

SDK usage

import { createMemoryState, ManualClock } from "@demystify/platform-state";
const state = createMemoryState();                       // keyless, offline default
await state.queue.enqueue("gateway", { id, payload }, priority);
// opt-in Redis (integration): const state = await createRedisState(process.env.REDIS_URL!);

from demystify_platform_state import create_memory_state
state = create_memory_state()
await state.queue.enqueue("gateway", QueueItemInput(id=id, payload=payload), priority)
# opt-in Redis: state = await create_redis_state(os.environ["REDIS_URL"])

Note: the KV delete method is del in TS and delete in Python (del is a Python keyword); every other method name is identical across the two ports.

Architecture & ports/adapters

| port | memory adapter (default) | redis adapter (integration) | selection | |---|---|---|---| | all five | Memory* / create_memory_state | Redis* / create_redis_state(url) | memory unless a REDIS_URL is wired |

Testing

make -C packages/platform-state setup lint test build. Unit tests exercise every memory port (TTL expiry, reserve/reconcile, priority order, lease expiry/eviction, pubsub delivery). Redis adapter tests are integration-marked and skip without REDIS_URL (TS: INTEGRATION=1 + REDIS_URL; Py: pytest -m integration) — never in CI. Parity: TS and Py replay fixtures/priority-order.json and fixtures/counter-reserve.json and must produce identical results.

v1 scope

Five ports, memory reference + Redis. No cluster/streams/consumer-groups (events owns durable queues via pgmq); this is the lightweight coordination seam. Redis adapter is opt-in and never the default.