Shahzad Bhatti Welcome to my ramblings and rants!

// Cloudflare DO — JavaScript
const count = await this.ctx.storage.get("count") ?? 0;
await this.ctx.storage.put("count", count + 1);

// Batch operations (storage.get([keys]) / storage.put(map))
const [history, meta] = await this.ctx.storage.get(["room:history", "room:meta"]);
await this.ctx.storage.put(new Map([["room:history", data], ["room:meta", index]]));

// PlexSpaces TypeScript — examples/typescript/apps/migrating_cloudflare_workers/guild_chat_actor.ts
protected override onInit(config: Record<string, unknown>): void {
    this.state.room_id = String(config.actor_id ?? "");

    // blockConcurrencyWhile() equivalent — batch-fetch persisted state before first message
    const keys = [
        "room:" + this.state.room_id + ":history",
        "room:" + this.state.room_id + ":meta",
    ];
    const values = host.kv.multiGet(keys);   // like DO storage.get(["k1","k2"])
    const [historyRaw] = values;
    if (historyRaw) {
        const msgs = JSON.parse(historyRaw);
        if (Array.isArray(msgs)) {
            this.state.messages = msgs;
            this.state.message_seq = msgs[msgs.length - 1]?.seq ?? 0;
        }
    }
}

private persistHistory(): void {
    // Batch write — like DO storage.put(new Map([["k1",v1],["k2",v2]]))
    host.kv.multiPut({
        ["room:" + this.state.room_id + ":history"]: JSON.stringify(this.state.messages),
        ["room:" + this.state.room_id + ":meta"]: JSON.stringify({
            message_seq: this.state.message_seq,
            last_updated: host.nowMs(),
        }),
    });
}

# PlexSpaces Python — examples/python/apps/migrating_cloudflare_workers/guild_chat.py
def _load_history(self) -> None:
    room_id = self._room_id()
    index_raw = host.kv.get(f"room:{room_id}:seq_index")
    if index_raw:
        seqs = json.loads(index_raw)
        keys = [f"room:{room_id}:msg:{seq}" for seq in seqs]
        values = host.kv.multi_get(keys)   # DO storage.get([k1,k2,...]) equivalent
        self.messages = [json.loads(v) for v in values if v]
        if self.messages:
            self.msg_seq = self.messages[-1]["seq"]

def _persist_history(self) -> None:
    room_id = self._room_id()
    entries = {}
    seqs = []
    for msg in self.messages:
        entries[f"room:{room_id}:msg:{msg['seq']}"] = json.dumps(msg)
        seqs.append(msg["seq"])
    entries[f"room:{room_id}:seq_index"] = json.dumps(seqs)
    host.kv.multi_put(entries)   # DO storage.put({k:v,...}) equivalent

// PlexSpaces TypeScript — from RateLimiterActor in guild_chat_actor.ts
// Atomic distributed counter — equivalent to Cloudflare KV atomic increment
const distributedCount = host.kv.increment(windowKey, 1);

// Compare-and-swap — equivalent to DO transactional storage read-modify-write
const applied = await host.kvCas("lock_key", expectedValue, newValue);

// KV with TTL — equivalent to DO storage.put with metadata expiration
await host.kvPutWithTtl("session_token", token, 3600);

# PlexSpaces Python — examples/python/apps/migrating_cloudflare_workers/guild_chat.py
@handler("check")
def check(self, user_id: str = "") -> dict:
    # Atomic increment — survives actor restarts, equivalent to Cloudflare KV atomic
    host.kv.increment(f"rate:{user_id}:total", 1)

    # CAS — idempotent token slot reservation, like DO transactional put
    cas_key = f"rate:{user_id}:window"
    current_val = host.kv.get(cas_key) or ""
    host.kv.cas(cas_key, current_val, str(host.now_ms()))

    # Token bucket logic operates on in-memory state (safe: single-threaded actor)
    allowed = self.buckets[user_id]["tokens"] > 0
    if allowed:
        self.buckets[user_id]["tokens"] -= 1
    return {"allowed": allowed, "remaining": self.buckets[user_id]["tokens"]}

// Cloudflare DO
await this.ctx.storage.setAlarm(Date.now() + 10_000);

async alarm() {
    const count = await this.ctx.storage.get("count");
    await this.ctx.storage.delete("count");
    console.log(`Processing ${count} batched requests`);
}

// PlexSpaces TypeScript — examples/typescript/apps/migrating_cloudflare_workers/guild_chat_actor.ts
onEnqueue(_payload: Record<string, unknown>): Record<string, unknown> {
    this.state.queued++;
    if (this.state.queued === 1) {
        // First item — equivalent to: await this.state.storage.setAlarm(Date.now() + 30_000)
        host.alarm.set(host.nowMs() + 30_000);
    }
    return { status: "ok", queued: this.state.queued };
}

// Fires when the scheduled timestamp is reached
// Equivalent to Cloudflare DO: async alarm() { ... }
on__alarm__(_payload: Record<string, unknown>): Record<string, unknown> {
    const processed = this.state.queued;
    this.state.processed += processed;
    this.state.queued = 0;
    this.state.total_alarm_fires++;
    return { status: "ok", processed };
}

onStatus(_payload: Record<string, unknown>): Record<string, unknown> {
    const alarmAt = host.alarm.get();   // DO: await this.state.storage.getAlarm()
    return { ...this.state, alarm_at: alarmAt, alarm_set: alarmAt > 0 };
}

# PlexSpaces Python — examples/python/apps/migrating_cloudflare_workers/guild_chat.py
@handler("start")
def start(self, delay_ms: int = 30000) -> dict:
    # Equivalent to: this.state.storage.setAlarm(Date.now() + delay_ms)
    host.alarm.set(host.now_ms() + delay_ms)
    return {"status": "ok", "fire_at_ms": host.now_ms() + delay_ms}

@handler("__alarm__")
def on_alarm(self) -> dict:
    # Equivalent to Cloudflare DO: async alarm() { ... }
    processed = len(self.pending_requests)
    self.total_processed += processed
    self.pending_requests = []
    return {"status": "ok", "processed": processed}

@handler("cancel")
def cancel(self) -> dict:
    host.alarm.delete()   # DO: this.state.storage.deleteAlarm()
    return {"status": "ok", "action": "alarm_cancelled"}

// Cloudflare DO — needs a binding in wrangler.toml
const id = env.CHAT_ROOM.idFromName(roomId);
const room = env.CHAT_ROOM.get(id);
await room.fetch("/send", { method: "POST", body: JSON.stringify(msg) });

// PlexSpaces TypeScript
import { getActorRef } from "@plexspaces/sdk";
const room = getActorRef("ChatRoomActor", roomId, "default");
const reply = await room.ask("send_message", { user_id: userId, content: msg });

# PlexSpaces Python
from plexspaces.host import get_actor_ref
room = get_actor_ref("ChatRoomActor", room_id, "default")
reply = room.ask("send_message", {"user_id": user_id, "content": msg}, timeout_ms=5000)

// proto/plexspaces/v1/actors/actor_runtime.proto
message ListActorsRequest {
    string actor_type = 3;
    ActorState state  = 4;
    string node_id    = 5;
    // Namespace for tenant isolation — only actors in this namespace are returned
    string namespace  = 6;
    PageRequest page_request = 2;
}

message ListActorsResponse {
    repeated Actor actors        = 2;
    PageResponse page_response   = 3;
}

# HTTP API (equivalent to Cloudflare's list-objects endpoint)
curl "http://localhost:8080/api/v1/actors/default/ChatRoomActor?state=active"

// Go SDK
actors, err := client.ListActors(ctx, &ListActorsRequest{
    ActorType: "ChatRoomActor",
    Namespace: "default",
    State:     ActorStateActive,
})

// Cloudflare DO — standard WebSocket
async fetch(request) {
    const [client, server] = Object.values(new WebSocketPair());
    this.ctx.acceptWebSocket(server);
    return new Response(null, { status: 101, webSocket: client });
}

async webSocketMessage(ws, message) {
    for (const peer of this.ctx.getWebSockets()) {
        peer.send(`broadcast: ${message}`);
    }
}

async webSocketClose(ws, code) {
    ws.close(code, "connection closed");
}

// PlexSpaces TypeScript — examples/typescript/apps/ws_chat_room/chat_server_actor.ts
onSend(payload: SendPayload): unknown {
    const senderUsername = this.state.members[payload.sender_actor_id] ?? payload.sender_actor_id;
    const ts = host.nowMs();
    this.state.history.push({
        senderActorId: payload.sender_actor_id,
        sender: senderUsername,
        text: payload.text,
        ts,
    });
    if (this.state.history.length > MAX_HISTORY) {
        this.state.history = this.state.history.slice(-MAX_HISTORY);
    }

    const event = {
        sender: payload.sender_actor_id,
        sender_username: senderUsername,
        text: payload.text,
        room_id: this.state.roomId,
        ts,
    };
    // Fan out to all members including sender (delivery confirmation)
    // host.send() routes each tell through WsActorTransportClient ? WsRegistry ? thin-node WS session
    const memberIds = Object.keys(this.state.members);
    for (const actorId of memberIds) {
        host.send(actorId, "chat_message", event);
    }
    return { success: true, members_notified: memberIds.length };
}

// PlexSpaces TypeScript — examples/typescript/apps/ws_chat_room/chat_server_actor.ts
onOnline(_payload: OnlinePayload): unknown {
    this.state.online = true;
    this.state.last_seen = host.nowMs();
    host.kv.putJson(`presence:${this.state.userId}`, { online: true, last_seen: this.state.last_seen });
    host.sendAfter(60_000, "timeout_check", {});   // durable reminder, survives restart
    return { success: true, online: true };
}

onTimeout_check(): unknown {
    const idleSince = host.nowMs() - this.state.last_seen;
    if (idleSince > 55_000) {
        this.state.online = false;
        host.kv.putJson(`presence:${this.state.userId}`, {
            online: false, last_seen: this.state.last_seen,
        });
    }
    return { checked: true, idle_ms: idleSince };
}

# PlexSpaces Python — examples/python/apps/migrating_cloudflare_workers/guild_chat.py
@handler("send_message")
def send_message(self, user_id: str = "", content: str = "") -> dict:
    msg = self._add_message(user_id, content, host.now_ms())

    # Fan-out: fire-and-forget to each member actor
    # Mirrors Discord's Manifold pattern for distributed fan-out
    # In Cloudflare DO, this would be individual fetch() calls — expensive
    fan_out_count = 0
    for member_id in list(self.members.keys()):
        if member_id != user_id:
            host.send(member_id, "receive_message", {
                "room_id": self._room_id(),
                "seq": msg["seq"],
                "from": user_id,
                "content": content,
            })
            fan_out_count += 1

    self._persist_history()   # batch multiPut — one KV call for the whole room
    return {"status": "ok", "seq": msg["seq"], "fan_out": fan_out_count}

# PlexSpaces Python — examples/python/apps/chat_agent/chat_agent.py
@handler("chat")
def chat(self, message: str = "") -> dict:
    # Load history — equivalent to: await this.storage.get('history')
    history = host.kv.get_json("history") or []

    history.append({"role": "user", "content": message, "timestamp": host.now_ms()})

    assistant_reply = self._call_llm(history)

    history.append({"role": "assistant", "content": assistant_reply, "timestamp": host.now_ms()})

    # Persist — equivalent to: await this.storage.put('history', history)
    host.kv.put_json("history", history)
    self.total_messages += 1

    # Schedule summarization alarm once history is long enough
    if len(history) > _ALARM_THRESHOLD and host.alarm.get() == 0:
        host.alarm.set(host.now_ms() + _ALARM_DELAY_MS)

    return {"status": "ok", "reply": assistant_reply, "history_length": len(history)}

@handler("__alarm__")
def on_alarm(self) -> dict:
    # Durable alarm callback — equivalent to Cloudflare Agents SDK onAlarm()
    history = host.kv.get_json("history") or []
    summary = self._call_llm([{
        "role": "user",
        "content": f"Summarize this conversation (2-3 sentences): {json.dumps(history)}"
    }])
    host.kv.put("summary", summary)
    host.kv.delete("history")   # clear after summarizing
    return {"status": "ok", "action": "summarized", "messages_summarized": len(history)}

// Cloudflare Agents SDK — locked to Cloudflare's AI gateway
const response = await this.env.AI.run("@cf/meta/llama-3-8b-instruct", { messages });

# PlexSpaces Python — examples/python/apps/chat_agent/chat_agent.py
def _call_llm(self, messages):
    http = ServiceHttpClient("llm-link")
    body = {
        "model": "claude-3-5-haiku-20241022",
        "max_tokens": 1024,
        "messages": [{"role": m["role"], "content": m["content"]} for m in messages],
    }
    resp = http.post("/v1/messages", body)
    # Parse Anthropic response
    if isinstance(resp, dict):
        content = resp.get("content", [])
        if content and isinstance(content, list):
            return content[0].get("text", "")
    return "[LLM unavailable]"

# Local development — Ollama
[[service_links]]
name = "llm-link"
url  = "http://localhost:11434"

# Production — swap without touching actor code
[[service_links]]
name = "llm-link"
url  = "https://api.anthropic.com"
headers = { "x-api-key" = "${ANTHROPIC_API_KEY}" }

// PlexSpaces TypeScript — examples/typescript/apps/chat_agent/
const resp = host.httpClient("llm-link").post("/v1/messages", {
    model: "claude-3-5-haiku-20241022",
    max_tokens: 1024,
    messages,
});

// PlexSpaces Go — examples/go/apps/migrating_cadence/payment_workflow.go
// PaymentWorkflow implements WorkflowActor for idempotent payment processing.
// Steps: validate ? authorize (with retry) ? capture ? settle.
// Signals: refund, cancel.  Queries: status, payment_id.
type PaymentWorkflow struct {
    plexspaces.BaseActor
    PaymentID       string        `json:"payment_id"`
    Status          string        `json:"status"` // pending ? validated ? authorized ? captured ? settled
    Steps           []PaymentStep `json:"steps"`
    RefundRequested bool          `json:"refund_requested"`
}

func (p *PaymentWorkflow) Run(payloadJSON string) string {
    // Each step checkpoints via getState/setState before proceeding.
    // If the node crashes mid-run, the workflow resumes from the last checkpoint.
    p.Status = "validating"
    if err := p.validatePayment(); err != nil {
        p.Status = "failed"
        return marshal(map[string]any{"error": err.Error()})
    }
    p.addStep("validate")

    // Authorize with retries (idempotency key prevents double-charge)
    p.Status = "authorizing"
    for attempt := 0; attempt < 3; attempt++ {
        if err := p.authorizePayment(); err == nil {
            break
        }
    }
    p.addStep("authorize")
    // ... capture, settle
    return marshal(map[string]any{"status": p.Status, "payment_id": p.PaymentID})
}

func (p *PaymentWorkflow) Signal(name, _ string) {
    switch name {
    case "refund":
        p.RefundRequested = true
    case "cancel":
        p.Status = "cancelled"
    }
}

func (p *PaymentWorkflow) Query(name, _ string) string {
    return marshal(map[string]any{"status": p.Status, "payment_id": p.PaymentID})
}

// PlexSpaces Go — examples/go/apps/miniclaw/orchestrator.go
func (o *OrchestratorActor) Run(payloadJSON string) string {
    task := stringVal(parsePayload(payloadJSON), "task", "")
    taskID := fmt.Sprintf("orch-%d", host.NowMs())

    o.Status = "running"
    o.TaskID = taskID

    // Discover available agents via process group membership
    agentID, err := pgFirst("svc:agent")
    if err != nil {
        return marshal(map[string]any{"error": "no agents in svc:agent process group"})
    }

    // Decompose and delegate sub-tasks
    subTasks := decompose(task)
    for i, subTask := range subTasks {
        o.Progress = (i + 1) * 100 / len(subTasks)
        result, err := host.Ask(agentID, "chat", map[string]any{
            "message":    subTask,
            "session_id": fmt.Sprintf("orch-%s-%d", taskID, i),
        }, 30000)
        if err != nil {
            return marshal(map[string]any{"error": "sub-task failed: " + err.Error()})
        }
        // Store result in TupleSpace for aggregation
        host.TS().Write([]any{"orch_result", taskID, i, result})
    }

    o.Status = "completed"
    return marshal(map[string]any{"task_id": taskID, "status": "completed", "sub_tasks": len(subTasks)})
}

# PlexSpaces Python — examples/python/apps/minipi/approval_gate.py
@fsm_actor(states=["idle", "awaiting_approval", "approved", "rejected"], initial="idle")
class ApprovalGateActor:
    """
    FSM states: idle ? awaiting_approval ? approved / rejected ? idle

    Key insight: the agent can wait for days. DurabilityFacet preserves all state
    durably — no polling, no timeouts burning tokens.
    """
    fsm_state: str = state(default="idle")
    pending_request: dict = state(default_factory=dict)
    pending_agent_id: str = state(default="")

    @handler("request_approval")
    def request_approval(self, agent_id: str = "", action: str = "", context: dict = None) -> dict:
        """An agent requests human approval for a high-stakes action."""
        if self.fsm_state != "idle":
            return {"status": "busy", "current_agent": self.pending_agent_id}

        self.fsm_state = "awaiting_approval"
        self.pending_agent_id = agent_id
        self.pending_request = {"action": action, "context": context or {}, "requested_at_ms": host.now_ms()}

        # Store request for external review (dashboard, Slack notification, etc.)
        host.kv.put(f"approval_request:{self.actor_id}", json.dumps(self.pending_request))

        return {"status": "pending", "gate_id": self.actor_id}

    @handler("approve")
    def approve(self, approver: str = "", comment: str = "") -> dict:
        """Human approves — signals the suspended agent to resume."""
        agent_id = self.pending_agent_id
        self.fsm_state = "approved"
        self.decision_history.append({
            "action": self.pending_request.get("action"),
            "decision": "approved",
            "approver": approver,
            "decided_at_ms": host.now_ms(),
        })

        # Signal the waiting agent to resume with the decision
        host.send(agent_id, "workflow_signal:resume", {
            "decision": "approved",
            "approver": approver,
            "comment": comment,
        })

        self.fsm_state = "idle"
        self.pending_agent_id = ""
        return {"status": "approved", "agent_id": agent_id}

    @handler("reject")
    def reject(self, approver: str = "", reason: str = "") -> dict:
        """Human rejects — signals the agent with the rejection."""
        agent_id = self.pending_agent_id
        host.send(agent_id, "workflow_signal:resume", {
            "decision": "rejected",
            "approver": approver,
            "reason": reason,
        })
        self.fsm_state = "idle"
        return {"status": "rejected", "agent_id": agent_id}

@actor
class ChatAgentActor:
    """Minimal chat agent: conversation in KV, LLM via service link, alarm for summarization."""

    actor_id: str = state(default="")
    total_messages: int = state(default=0)
    total_summarizations: int = state(default=0)

    @init_handler
    def on_init(self, config: dict) -> None:
        self.actor_id = config.get("actor_id", "")
        host.info(f"ChatAgentActor init actor_id={self.actor_id}")


    @handler("__alarm__")
    def on_alarm(self) -> dict:
        """Durable alarm callback — equivalent to Cloudflare Agents SDK onAlarm().

        Summarizes conversation history and stores a summary KV key,
        then clears history.
        """
        host.info("ChatAgentActor: alarm fired — summarizing history")

        history = host.kv.get_json("history") or []
        if not history:
            return {"status": "ok", "action": "no_history_to_summarize"}

        # Summarize via LLM
        summary_prompt = (
            f"Summarize this conversation concisely (2-3 sentences): "
            f"{json.dumps([{'role': m['role'], 'content': m['content']} for m in history])}"
        )
        summary = self._call_llm([{"role": "user", "content": summary_prompt}])

        # Persist summary, clear history — equivalent to: storage.put('summary', s); storage.delete('history')
        host.kv.put("summary", summary)
        host.kv.delete("history")

        self.total_summarizations += 1

        host.info(f"ChatAgentActor: summarized {len(history)} messages")
        return {
            "status": "ok",
            "action": "summarized",
            "messages_summarized": len(history),
        }

// Create a pool of 20 workers, hash-partitioned
let pool_id = client.create_worker_pool(
    "worker-pool-1", "worker", 20,
    PartitionStrategy::Hash, HashMap::new(),
).await?;

// Bulk update: 10,000 messages routed to the right shard by key
client.parallel_update(&pool_id, updates, ConsistencyLevel::Eventual, false).await?;

// Parallel map: query every shard simultaneously
let results = client.parallel_map(&pool_id, json!({ "action": "get_total_count" })).await?;

// Parallel reduce: aggregate stats across all shards
let stats = client.parallel_reduce(
    &pool_id, json!({ "action": "stats" }),
    ShardGroupAggregationStrategy::Concat, 20,
).await?;

// PlexSpaces TypeScript — examples/typescript/apps/mersenne_prime/mersenne_actor.ts
onResult(payload: ResultPayload): unknown {
    const item = this.state.work[String(payload?.p)];
    item.status = 'done';
    item.is_prime = Boolean(payload.is_prime);
    item.duration_ms = payload.duration_ms ?? 0;

    // Record every completed candidate in TupleSpace and bump a Prometheus counter
    host.ts.write(['result', String(payload.p), item.is_prime ? 'true' : 'false',
        String(item.duration_ms), payload.actor_id ?? 'unknown']);
    host.incrCounter('ts-mersenne-prime', item.is_prime ? 'primes_found' : 'composites_found');

    // Immediately hand the same worker the next pending candidate
    const next = this._nextPending(this.state.workers[payload.actor_id!]?.cpu_cores ?? 1);
    if (next) {
        next.status = 'assigned';
        host.send(payload.actor_id!, 'assign_work', { p: next.p, done: false });
    }
    return { ok: true };
}

# Local development — exact production behavior
plexspaces-node start --config app-config.toml

# Docker — same binary, same config
docker run -v $(pwd):/app plexspaces/node start --config /app/app-config.toml

# Kubernetes — same config via Helm
helm install my-app plexspaces/app-chart \
  --set config.path=app-config.toml \
  --set persistence.storage=postgres

# Multi-cloud — nodes in GCP + AWS joined over a gRPC mesh; actors route transparently