Stateful workflows | Patterns

SurrealDB Agent Memory is not limited to conversational memory. Its entity-attribute model and temporal validity system make it a natural fit for tracking the state of long-running, multi-step agent workflows - processes that span hours or days, survive restarts, and need to avoid repeating completed steps.

Note

Spectron was the project name for SurrealDB Agent Memory. These type names
will be renamed in a future release.

Why use SurrealDB Agent Memory for workflow state

Traditional approaches to workflow state - Redis keys, database records, task queue metadata - require purpose-built state management. SurrealDB Agent Memory adds:

Scoped isolation - workflow state for project A cannot interfere with project B.
Temporal validity - step results can expire, triggering re-execution.
Provenance - every state change traces back to the turn or operation that caused it.
State diff - query what changed since a checkpoint without building your own diffing logic.
Observability - inspect the full state of any workflow at any point in time.

Step 1 - Design your state model

Map your workflow steps to entity types and attributes. A research workflow might look like:

Entity type	Example entities	Key attributes
`ResearchTask`	`market_analysis_q1`	`status`, `assigned_model`, `deadline`
`SearchResult`	`result_001`	`query`, `url`, `summary`, `relevance_score`
`Report`	`market_report_draft`	`status`, `word_count`, `sections_complete`

Step 2 - Scope workflow state to a project identifier

Use a project or task scope dimension to isolate each workflow's state:

from surrealdb import Spectron

memory = Spectron(context="workflows", api_key="sk-...")

# Each workflow run gets its own scope
workflow_scope = ["org/acme/project/market-analysis-q1-2025"]

session = await memory.sessions.create(scopes=workflow_scope)

import { Spectron } from "@surrealdb/spectron";

const memory = new Spectron({ context: "workflows", apiKey: "sk-..." });

const workflowScope = ["org/acme/project/market-analysis-q1-2025"];
const session = await memory.sessions.create({ scopes: workflowScope });

Step 3 - Record step completions as turns

Use turns to record what each step did. The extraction pipeline stores the results as Context-category attributes on the relevant entities.

async def run_search_step(session, query: str) -> list[dict]:
    results = await your_search_api(query)

    # Record the step as an agent turn - extraction captures the results
    await memory.remember(
        f"Completed search for '{query}'. Found {len(results)} results. "
        f"Top result: {results[0]['url']} - {results[0]['summary'][:200]}",
        session_id=session.id,
        role="assistant",
    )

    return results

async function runSearchStep(session: Session, query: string): Promise<SearchResult[]> {
    const results = await yourSearchApi(query);

    await memory.remember(
        `Completed search for '${query}'. Found ${results.length} results. `
            + `Top result: ${results[0].url} - ${results[0].summary.slice(0, 200)}`,
        { sessionId: session.id, role: "assistant" },
    );

    return results;
}

Step 4 - Check state before each step

Before running a step, check whether it has already been completed. This makes the workflow idempotent - safe to restart without duplicating work.

async def should_run_step(memory, lens: list[str], step_name: str) -> bool:
    ctx = await memory.context(
        lens=lens,
        query=f"Has the {step_name} step been completed?",
        k=3,
    )

    # Check if there is an existing completion record for this step
    for item in ctx.items:
        if item.entity_type == "WorkflowStep" and item.attributes.get("name") == step_name:
            if item.attributes.get("status") == "completed":
                return False

    return True

async function shouldRunStep(
    memory: Memory,
    lens: string[],
    stepName: string,
): Promise<boolean> {
    const ctx = await memory.context({
        lens,
        query: `Has the ${stepName} step been completed?`,
        k: 3,
    });

    for (const item of ctx.items) {
        if (
            item.entityType === "WorkflowStep"
            && item.attributes.name === stepName
            && item.attributes.status === "completed"
        ) {
            return false;
        }
    }

    return true;
}

Step 5 - Using temporal validity for step expiry

Some workflow steps have results that go stale - a price lookup, a news summary, a resource availability check. Set valid_until on the turn to give the extracted attributes a time-to-live:

POST /api/v1/{context_id}/sessions/{session_id}/turns
Content-Type: application/json

{
  "role": "assistant",
  "content": "Fetched current gold price: $2,340/oz",
  "metadata": {
    "valid_until": "2025-11-16T00:00:00Z"
  }
}

When the validity period expires, the attribute no longer appears in context retrievals and should_run_step returns true again, triggering a re-fetch.

Step 6 - Tracking progress

Each step's write returns its own delta, so accumulate those rather than polling
state. /state returns the whole Context, and a workflow's Context only grows -
reading it twice per step gets slower as the run goes on.

progress = {"completed": 0, "revised": 0, "blocked": 0}

for step_name, step_fn in steps:
    result = await step_fn()                     # each step calls memory.remember(...)
    progress["completed"] += len(result.extraction.attributes)
    progress["revised"] += len(result.extraction.corrections)
    progress["blocked"] += len(result.extraction.uncertainties)

print(progress)

const progress = { completed: 0, revised: 0, blocked: 0 };

for (const { name, fn } of steps) {
    const result = await fn();                   // each step calls memory.remember(...)
    progress.completed += result.extraction.attributes.length;
    progress.revised += result.extraction.corrections.length;
    progress.blocked += result.extraction.uncertainties.length;
}

console.log(progress);

The diff is useful for:

Progress bars - count completed steps vs total expected.
Stall detection - alert if no state changes have occurred in N minutes.
Audit trails - record what changed during a workflow run for compliance.

Putting it together

async def run_research_workflow(user_id: str, topic: str):
    scope = [f"org/acme/project/research-{topic.replace(' ', '-')}"]
    session = await memory.sessions.create(scopes=scope)

    steps = [
        ("web_search", lambda: run_search_step(session, topic)),
        ("summarise", lambda: run_summarise_step(session, topic)),
        ("draft_report", lambda: run_draft_step(session, topic)),
    ]

    for step_name, step_fn in steps:
        if await should_run_step(memory, scope, step_name):
            await step_fn()
        else:
            print(f"Skipping {step_name} - already completed.")

    # Final state summary
    state = await memory.state()
    return state

async function runResearchWorkflow(userId: string, topic: string) {
    const scope = [`org/acme/project/research-${topic.replace(/ /g, "-")}`];
    const session = await memory.sessions.create({ scope });

    const steps = [
        { name: "web_search", fn: () => runSearchStep(session, topic) },
        { name: "summarise", fn: () => runSummariseStep(session, topic) },
        { name: "draft_report", fn: () => runDraftStep(session, topic) },
    ];

    for (const { name, fn } of steps) {
        if (await shouldRunStep(memory, scope, name)) {
            await fn();
        } else {
            console.log(`Skipping ${name} - already completed.`);
        }
    }

    const state = await memory.state();
    return state;
}