PocketFlow & SharedStore

OpenFlows is built on two tightly coupled primitives: PocketFlow, a directed graph execution engine, and SharedStore, a typed state contract that every node reads and writes. PocketFlow decides when each agent runs. SharedStore decides what they know about the world. Together they make the system deterministic, observable, and recoverable even though the individual agents are non-deterministic LLMs.

The flow is the program. The store is the memory. Agents are the functions. This separation lets NEXUS reason about the pipeline as a state machine rather than a conversation. If a node crashes, the flow can retry it. If the state becomes inconsistent, NEXUS can reconcile it. If a ticket needs a human, the flow can suspend it cleanly and resume later.

PocketFlow: The Directed Graph Engine

PocketFlow models a pipeline as a directed graph of nodes connected by actions. A node is a unit of work, usually corresponding to one agent invocation in one Coder workspace. An action is a typed edge that carries the node from one state to the next. The graph is static; the state is dynamic. This means the overall shape of a ticket's lifecycle is fixed, but the data flowing through it is specific to the ticket.

Each node follows a strict lifecycle: prep, exec, post, and fallback. The prep phase reads the SharedStore and formulates a plan. The exec phase does the work. The post phase writes the result back to the SharedStore and selects the next action. The fallback phase handles errors and retries. This pattern is borrowed from resilient async programming and is what makes a node crash-recoverable.

use openflows::pocketflow::{Node, Action, SharedStore};

pub struct PlanNode {
    pub ticket_id: TicketId,
}

#[async_trait]
impl Node for PlanNode {
    async fn prep(&self, store: &SharedStore) -> anyhow::Result<PlanInput> {
        let ticket = store.get_ticket(&self.ticket_id).await?;
        let repo = store.get_repo_context(&ticket.repo).await?;
        Ok(PlanInput { ticket, repo })
    }

    async fn exec(&self, input: PlanInput) -> anyhow::Result<PlanOutput> {
        // FORGE writes PLAN.md inside the Coder workspace.
        let plan = agent::write_plan(input).await?;
        Ok(PlanOutput { plan })
    }

    async fn post(&self, store: &mut SharedStore, output: PlanOutput) -> anyhow::Result<Action> {
        store.set_plan(&self.ticket_id, &output.plan).await?;
        store.set_status(&self.ticket_id, TicketStatus::AwaitingReview).await?;
        Ok(Action::Route("review_plan"))
    }

    async fn fallback(&self, store: &mut SharedStore, err: &Error) -> anyhow::Result<Action> {
        tracing::error!(ticket = %self.ticket_id, "plan node failed: {}", err);
        store.push_event(Event::NodeFailed { ticket: self.ticket_id, error: err.to_string() }).await?;
        Ok(Action::Retry { max_retries: 3 })
    }
}

The Node Lifecycle

SharedStore: Typed State Contracts

SharedStore is not a key-value dumping ground. It is a typed contract that every node agrees to. In production, the store is backed by Redis; in development and testing, it runs in-memory. The schema is the same in both modes, so code that works locally works in production with no configuration change.

The contracts are defined in Rust and serialised through Redis with JSON or MessagePack depending on the value size. The type system prevents nodes from writing malformed state. For example, a ticket status is an enum, not a string, so a node cannot accidentally set it to "pr_open" instead of PrOpened. NEXUS enforces the valid transitions: a ticket cannot move from Open to Merged without passing through the intermediate states.

#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TicketStatus {
    Open,
    Assigned,
    InProgress,
    AwaitingReview,
    AwaitingHuman,
    Completed,
    PrOpened,
    Merged,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Ticket {
    pub id: TicketId,
    pub issue_number: u64,
    pub repo: RepoId,
    pub title: String,
    pub status: TicketStatus,
    pub assigned_worker: Option<WorkerId>,
    pub plan: Option<Plan>,
    pub contract: Option<ContractOutcome>,
    pub final_review: Option<ReviewOutcome>,
    pub pr: Option<PrEntry>,
}

SharedStore Keys

Redis Keyspace

In production, every SharedStore key is prefixed with openflows: to avoid collisions with other Redis users. The keyspace is intentionally flat and explicit: a human operator can connect with redis-cli, list the keys, and understand the current system state without reading source code. This is important for incident response and for the TUI dashboard.

Harness CLI

The Harness CLI is the operator's window into the flow and the store. It is used for local development, debugging, and production incident response. It can run a full flow, step through a single node, inspect keys, reset a stuck ticket, or tail the event ring.

The harness is especially useful when building new skills or custom nodes. You can run a single node against a known SharedStore snapshot, inspect the output, and iterate without provisioning a full Coder workspace. The --dry-run flag executes prep and exec but skips post, so the store is not mutated.

# Inspect the current state of a ticket
openflows harness inspect --key tickets --id 42

# Run a single node against the live store without writing back
openflows harness step --node review_plan --ticket 42 --dry-run

# Tail events while a ticket is in progress
openflows harness events --ticket 42

Why a Typed Store Matters

LLM agents are free-form text producers. If they communicated through raw prompts or unstructured files, the orchestrator would have to parse their output to decide what to do next. That parsing is fragile and creates a hidden coupling between the agent and the controller. SharedStore removes that coupling by requiring every node to write structured, typed state. The LLM can still write human-readable plans and reviews, but the machine-readable outcome is a schema-defined value.

This typed contract is what makes self-healing possible. NEXUS can look at the store and know, without asking an LLM, whether a ticket is blocked, completed, or ready to merge. It can detect that a worker slot is assigned to a ticket that has already been merged, or that a PR has been open for an hour without CI completing. Those decisions are made against the store, not against a chat log.

Dual-Backend Operation

SharedStore is implemented behind a trait that abstracts over in-memory and Redis backends. In local mode, the store lives in a single Tokio task and is reset on restart. In production, Redis provides durability, horizontal scaling, and the ability to run the TUI on a different host from the orchestrator. Switching between the two is controlled by the REDIS_URL environment variable: if it is absent, the system starts in-memory.

# In-memory mode (default for tests and local dev)
openflows orchestrator

# Redis-backed production mode
REDIS_URL=redis://redis.internal:6379 openflows orchestrator