# Stateful workflows

Diff-friendly state for workflow UIs.

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:

```python
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)
```

```typescript
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.

```python
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
```

```typescript
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.

```python
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
```

```typescript
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:

```http
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.

```python
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)
```

```typescript
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

```python
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
```

```typescript
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;
}
```