Stateful workflows | Patterns

Besides conversational memory, SurrealDB Agent Memory's 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.

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.
Write deltas - each write returns the attributes, corrections and uncertainties it produced, so progress can be counted without 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. Entity types come from a fixed lower-case set (person, organisation, project, location, topic, product, policy, concept, event and other), and any other type is stored as other. A research workflow might look like:

Entity type	Example entities	Key attributes
`project`	`market-analysis-q1-2025`	`web_search_status`, `summarise_status`, `draft_report_status`
`other`	`result_001`	`query`, `url`, `summary`, `relevance_score`
`other`	`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.memory import AsyncMemory

memory = AsyncMemory(context="workflows", api_key="sp-...")

# 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 { AgentMemory } from "@surrealdb/memory";

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

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 attributes on the relevant entities, and memory_category="context" marks them as short-lived workflow state.

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",
        memory_category="context",
    )

    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", memoryCategory: "context" },
    );

    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.

Free text leaves the entity and key names to extraction, so record each completion as a structured fact with infer="triples". The step then has a fixed place in memory, and one entity read finds it:

from surrealdb.memory import MemoryNotFoundError

async def mark_step_done(session, workflow_id: str, step_name: str) -> None:
    await memory.remember(
        infer="triples",
        session_id=session.id,
        memory_category="context",
        triples=[{
            "entity": {"type": "project", "name": workflow_id},
            "key": f"{step_name}_status",
            "value": "completed",
        }],
    )

async def should_run_step(workflow_id: str, step_name: str) -> bool:
    try:
        read = await memory.entities.get("project", workflow_id)
    except MemoryNotFoundError:
        return True

    return not any(
        attr.key == f"{step_name}_status" and attr.value == "completed"
        for attr in read.attributes
    )

import { NotFoundError } from "@surrealdb/memory";

async function markStepDone(session: Session, workflowId: string, stepName: string) {
    await memory.remember(undefined, {
        infer: "triples",
        sessionId: session.id,
        memoryCategory: "context",
        triples: [{
            entity: { type: "project", name: workflowId },
            key: `${stepName}_status`,
            value: "completed",
        }],
    });
}

async function shouldRunStep(workflowId: string, stepName: string): Promise<boolean> {
    try {
        const read = await memory.entities.get("project", workflowId);
        return !read.attributes.some(
            (attr) => attr.key === `${stepName}_status` && attr.value === "completed",
        );
    } catch (err) {
        if (err instanceof NotFoundError) return true;
        throw err;
    }
}

Step 5 - Expiring step results

Some workflow steps have results that go stale - a price lookup, a news summary, a resource availability check. A write has no per-fact time-to-live. Instead, facts written with memory_category="context" expire together: POST /api/v1/{context_id}/lifecycle/expire ends every context-category attribute older than seven days, and needs the memory:forget grant.

POST /api/v1/{context_id}/lifecycle/expire

An expired attribute no longer appears in reads, so should_run_step returns true again and the step runs again. For a shorter lifetime, store the time of the result as its own attribute, such as web_search_fetched_at, and compare it with the current time in should_run_step.

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):
    workflow_id = f"research-{topic.replace(' ', '-')}"
    scope = [f"org/acme/project/{workflow_id}"]
    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(workflow_id, step_name):
            await step_fn()
            await mark_step_done(session, workflow_id, step_name)
        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 workflowId = `research-${topic.replace(/ /g, "-")}`;
    const scope = [`org/acme/project/${workflowId}`];
    const session = await memory.sessions.create({ scopes: 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(workflowId, name)) {
            await fn();
            await markStepDone(session, workflowId, name);
        } else {
            console.log(`Skipping ${name} - already completed.`);
        }
    }

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