Developer docs Guides

State & memory

Three persistence layers with different lifetimes: task memories (write-once lessons), project status (shared mutable state), and reactive stores (typed state with change tracking).

Pick the right layer

Layer	Lifetime	Write model	Use for
Task memory	Permanent, per meta-task	Written by the platform after each run	Lessons and patterns future runs should retrieve
Project status	Life of the project	Direct dict writes through a reactive proxy	Cross-run flags, counters, and coordination state
Reactive store	Run-scoped, optionally persisted	Typed fields, actions, patches	Structured state with change watchers inside a run

Reactive stores

Declare fields with class annotations, derive with @getter, mutate with @action. Every mutation emits a typed event that subscribers can observe:

from allybuild_sdk.reactive import defineStore, getter, action, field

@defineStore("review-queue")
class ReviewQueue:
    items: list = field(default_factory=list)

    @getter
    def pending(self) -> int:
        return sum(1 for i in self.items if not i["done"])

    @action
    def add(self, path: str) -> None:
        self.items = self.items + [{"path": path, "done": False}]

queue = ReviewQueue()
unsub = queue["$subscribe"](lambda m, s: print(m.type, len(s["items"])))
queue.add("src/auth.py")     # prints: direct 1
print(queue.pending)          # 1

Dollar-sign API

Method	Purpose
$patch(dict \| fn)	Batch update producing one "patch" mutation
$reset()	Restore declared defaults, emits "reset"
$subscribe(cb)	Watch mutations; returns an unsubscribe fn
$on_action(cb)	Action lifecycle hooks: after / on_error
$atomic(fn)	Explicit single-event multi-field update
$state	Deep-copied read-only snapshot
$dispose()	Detach subscribers and remove from registry

HTTP persistence

from allybuild_sdk.reactive import HttpBackend, create_reactive

backend = HttpBackend(
    api_url=ALLYBUILD_API_URL,
    jwt=TASK_JWT,
    project_id=PROJECT_ID,
    timeout=5.0,
)
create_reactive(redis=backend)

Memories in practice

Memories are written for you: a meta-task with memory extraction enabled distills each completed run into lessons — agent runs from the conversation trajectory, script runs from stdout. For sub-task trees, extraction fires once for the whole tree, governed by the root meta-task’s toggle. The high-value pattern is retrieving them at the start of the next run to ground decisions, and searching adversarially with order="least" to surface novel failure modes:

from allybuild_sdk import MemoriesApi

memories = MemoriesApi(client)
grounding = memories.list_task_type_memories(
    "incident-triage", query="database failover", k=5
)
novel = memories.list_task_type_memories(
    "incident-triage", query="database failover", k=5, order="least"
)

Project status across runs

# project_status is pre-wired in the workflow module
project_status["last_run"] = TASK_ID
project_status["deploy_count"] = project_status.get("deploy_count", 0) + 1

for key, value in project_status.items():
    print(key, "=", value)