# Run a worker in Python (experimental)

Source: https://docs.littlebigbrain.com/guides/python-worker/

A workflow worker runs your code for the messages of a workflow. Each message
is one **turn**. A turn's named **steps** save their outputs in LBB. After a
crash, a turn runs again: it replays the saved steps and does each effect
once.

`lbb.workflows` is the worker of the Python SDK. It follows the same rules as
the TypeScript `WorkflowWorker` ([Workflows](https://docs.littlebigbrain.com/guides/workflows/)). The server
needs the workflow engine (`--workflows`, or a hosted stack).

```sh
pip install littlebigbrain
```

## Start from a template

`npm create @littlebigbrain` creates a project from a
[workflow template](https://docs.littlebigbrain.com/guides/workflow-templates/). With `--lang python`, the
project is the template's Python port:

```sh
npm create @littlebigbrain -- --template values-extract --lang python my-values
cd my-values
python3 -m venv .venv
source .venv/bin/activate
pip install -e ".[test]"
pytest
```

`pytest` runs the workflow on the template's samples, with mock models and no
server. `python -m values_extract try` puts the samples on a graph of your
stack (`try-values-extract`) and creates the workflow there. Then
`LBB_GRAPH=try-values-extract python -m values_extract start` serves it. Both
commands read `LBB_URL` and `LBB_API_KEY`.

## Write a workflow

A workflow is an async handler with a name and a version. It gets the turn's
context, the saved state and the message. It returns the new state and a
reply.

```python
from lbb.workflows import WorkflowContext, WorkflowResult, workflow

@workflow(name="counter", version="v1", initial_state={"count": 0})
async def counter(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    amount = await ctx.step("add", lambda s: message["add"])
    total = state["count"] + amount
    return WorkflowResult(state={"count": total}, result=total)
```

Start an instance of the workflow and send it a message. Reuse the message id
when you send the same message again: the server admits it once.

```python
import asyncio
import os

from lbb import AsyncLbbClient

def connect() -> AsyncLbbClient:
    return AsyncLbbClient(
        os.environ.get("LBB_URL", "http://127.0.0.1:7400"),
        api_key=os.environ.get("LBB_API_KEY"),
        graph="main",
    )

async def send() -> None:
    async with connect() as client:
        handle = await counter.start(client, "counter-1")
        await handle.send({"add": 2}, id="request-1")

asyncio.run(send())
```

Then run the worker. It claims the turns of the workflows you register and
runs them.

```python
from lbb.workflows import WorkflowWorker

client = connect()
WorkflowWorker(client, [counter], worker="counter-1").run()
```

`run()` serves until the process gets SIGTERM or Ctrl-C.

## Steps and effect ids

Put I/O, clock reads and randomness inside `ctx.step(key, fn)`. The step's
output is saved as JSON before the turn goes on. Keys are unique in a turn.
Await the steps one after the other.

An `async` step function runs on the event loop. A sync one runs in a
thread. So a blocking call there (`requests`, the sync `LbbClient`) holds no
other turn. It can report `progress` from that thread.

A step's `effect_id` is `<effect prefix>/<key>`. It stays the same for the
instance, the message and the step on every attempt. Pass it to an external
service as its idempotency key.

The effect can succeed before its step is saved. The next attempt then sends
it again, with the same id.

The model activities take the effect id. LBB keeps the first answer under it,
and a repeat costs nothing.

```python
from lbb import models
from lbb.workflows import lbb_activities

lbb = lbb_activities(client)
SCHEMA = {"type": "object", "properties": {"code": {"type": "string"}}}

@workflow(name="define-codes", version="v1", initial_state={})
async def define_codes(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    answer = await ctx.step(
        "define",
        lambda s: lbb.reason(message["prompt"], SCHEMA, effect_id=s.effect_id, label="code"),
        model=models.ReasonActivityResponse,
    )
    return WorkflowResult(state={"code": answer.value["code"]})
```

A step returns JSON. A step that returns a Pydantic model needs `model=`. The
step then reads its saved JSON back as that model, also on a replay.

`lbb_activities(client)` has `reason`, `decide` and `embed`, on the generated
models of `lbb.models`. A call names the turn's run and the step that makes it
(`workflow.step`, the step's key). Its usage goes with the turn's completion.

A turn replays its saved steps by key and kind, in order. A change of the step
order fails the turn for good. Change the order only while no turn of the
version runs.

A step that reported [run results](https://docs.littlebigbrain.com/guides/workflows/#run-results) saves them
beside its value. An SDK from before run results does not take them out
again. On a replay, it gives your code the saved wrapper, not the step's
value. So roll workers forward only. Do not go back to an earlier SDK while
newer turns can still run.

## Waits, signals and continuations

`ctx.wait_for_signal` waits for an approval or a callback without holding a
worker. `ctx.sleep` saves a timer and releases the worker until it is due.

```python
@workflow(name="review-draft", version="v1", initial_state={"sent": 0})
async def review_draft(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    draft = await ctx.step("draft", lambda s: f"Draft for {message['task']}")
    review = await ctx.wait_for_signal("review", name="approval", timeout_ms=86_400_000)
    if not review.received:
        return WorkflowResult(state=state, result="approval timed out")
    if not review.value["approved"]:
        return WorkflowResult(state=state, result="rejected")
    await ctx.sleep("cool-down", 1_000)
    return WorkflowResult(state={"sent": state["sent"] + 1}, result=draft)
```

Send the signal to the turn from your app. The same event id is delivered
once.

```python
from lbb.workflows import WorkflowHandle

async def approve() -> None:
    async with connect() as client:
        handle = WorkflowHandle(client, "review-1")
        await handle.signal("approval", {"approved": True}, turn=1, id="review-event-1")

asyncio.run(approve())
```

A turn holds at most 64 steps. For a large message, return
`ctx.continue_(state, next_message)`. The state and the next message are saved
in one write. The next message joins the end of the inbox.

```python
@workflow(name="pages", version="v1", initial_state={"read": 0})
async def pages(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    batch, rest = message["pages"][:20], message["pages"][20:]
    read = await ctx.step("read", lambda s: len(batch))
    after = {"read": state["read"] + read}
    if rest:
        return ctx.continue_(after, {"pages": rest})
    return WorkflowResult(state=after)
```

## Ask a person

`lbb.wait_for_review` asks a person and waits for the answer. The worker is
free while the turn waits. The item is a `value`, a `choice`, a `grade` or an
`answer`.

```python
@workflow(name="rooms", version="v1", initial_state={})
async def rooms(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    room = message["room"]
    asked = await lbb.wait_for_review(
        ctx,
        "floor",
        {
            "kind": "choice",
            "key": f"floor:{room}",
            "title": f"Which floor covering does {room} have?",
            "question": {
                "options": [
                    {"id": "parquet", "label": "Parquet"},
                    {"id": "tiles", "label": "Tiles"},
                ]
            },
        },
        timeout_ms=7 * 24 * 3_600_000,
    )
    floor = asked.answer.option if asked.answer else None
    return WorkflowResult(state={"floor": floor})
```

The step `floor.ask` files the item, with this run as the one that waits.
The turn then waits for the answer in `floor.answer`. An answer that arrives
before the wait is kept, and the wait returns at once.

When the wait times out, the step `floor.read` reads the item again. An
answer whose signal did not arrive still counts.

A trigger shard serves many inputs, and a wait would hold them all. So
`wait_for_review` refuses to run there. File the item with `lbb.review`, and
take the answer with a review trigger. See
[Review in workflows](https://docs.littlebigbrain.com/guides/review-in-workflows/).

## Long steps

A long step reports progress and saves a resume cursor with `heartbeat`. The
next attempt of the step reads the cursor as `checkpoint`. `progress` reports
what the step does now; the console draws it as a bar.

```python
from lbb.workflows import WorkflowStepContext

async def crawl(s: WorkflowStepContext) -> dict:
    start = s.checkpoint or 0
    for page in range(start, 3):
        s.progress("reading pages", done=page, total=3, unit="pages")
        await s.heartbeat({"page": page}, checkpoint=page + 1)
    return {"pages": 3}

@workflow(name="crawler", version="v1", initial_state={})
async def crawler(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    read = await ctx.step("crawl", crawl, kind="tool", heartbeat_timeout_ms=30_000)
    return WorkflowResult(state=read)
```

A step has a timeout for each attempt, 5 minutes by default (`timeout_ms`).
With `heartbeat_timeout_ms`, a step that reports no progress that long ends its
attempt.

## Work on the instances of a class

A triggered workflow watches a class and sends its new and changed instances
to your workflow. `triggered_workflow` defines that workflow, and `write_owned`
writes what it owns on the instances.

```python
from lbb.workflows import OwnedWrite, entity_ref, triggered_workflow, write_owned

QUERY = """SELECT ?s ?text WHERE {{
  VALUES ?s {{ {} }} ?s <https://littlebigbrain.com/p/text> ?text }}"""

@triggered_workflow(name="word-count", version="v1")
async def word_count(ctx: WorkflowContext, message: models.WorkflowTriggerMessage, state: dict):
    async def count(s: WorkflowStepContext) -> dict:
        query = QUERY.format(" ".join(f"<{iri}>" for iri in message.instances))
        rows = await lbb.sparql(query, as_of_commit_seq=message.as_of_commit_seq)
        return {row["s"]: len(row["text"].split()) for row in rows.rows()}

    counts = await ctx.step("count", count)
    write = OwnedWrite(
        subjects=[entity_ref(iri, "Note") for iri in message.instances],
        properties=["word_count"],
        entity_properties=[
            {**entity_ref(iri, "Note"), "properties": {"word_count": words}}
            for iri, words in counts.items()
        ],
    )
    await write_owned(client, ctx, "write", write, graph=message.graph)
```

`write_owned` commits inside a step, with the step's effect id as the commit's
idempotency key. A turn that runs again sends the same key. The server answers
with the first commit.

The commit also carries the turn's fence. A worker whose lease ended gets
`409 workflow_fence_lost` and writes nothing.

Create the triggered workflow on the graph. It names your workflow type and
the properties it owns.

```python
async def watch() -> None:
    async with connect() as client:
        await client.use_workflow_starter(
            "workflow",
            name="notes",
            class_="Note",
            fields=["text"],
            workflow={"workflow_type": "word-count", "version": "v1"},
            owns={"properties": ["word_count"]},
        )

asyncio.run(watch())
```

An instance the turn cannot do goes back to the watch with `requeue_failed`.
It is sent again after 1, 4, then 16 minutes, at most 3 times.

## Test a workflow before it writes

`capture_activities` gives activities that write nothing. The models and the
reads run, and they cost what they cost live. Each `write_owned` is checked
with the commit's dry run and kept. Each review item is kept and filed
nowhere.

Make the workflow a function of its activities. A test then runs it on
chosen instances, under its own instance and version. No live turn runs.

```python
from lbb.workflows import LbbActivities, TriggeredWorkflowDefinition, capture_activities

def word_counter(lbb: LbbActivities, version: str = "v1") -> TriggeredWorkflowDefinition:
    @triggered_workflow(name="word-count", version=version)
    async def count(ctx: WorkflowContext, message: models.WorkflowTriggerMessage, state: dict):
        write = OwnedWrite(
            subjects=[entity_ref(iri, "Note") for iri in message.instances],
            properties=["word_count"],
            entity_properties=[
                {**entity_ref(iri, "Note"), "properties": {"word_count": 3}}
                for iri in message.instances
            ],
        )
        await lbb.write_owned(ctx, "write", write, graph=message.graph)

    return count

async def try_word_count(iris: list[str]) -> None:
    async with connect() as client:
        capture = capture_activities(lbb_activities(client), workflow="notes")
        test = word_counter(capture.activities, version="test")
        handle = await test.start(client, "word-count-test-1")
        await handle.send(
            {
                "type": "instances",
                "workflow": "notes",
                "graph": "main",
                "as_of_commit_seq": 1,
                "reason": "refresh",
                "instances": iris,
                "removed": [],
            },
            id="word-count-test-1",
        )
        await WorkflowWorker(client, [test], worker="test").run_once()
        for write in capture.result().writes:
            print(write.step_key, write.check, write.errors or "passes")

asyncio.run(try_word_count(["https://littlebigbrain.com/e/0123456789abcdef0123456789abcdef"]))
```

The model calls of a capture name the test (`workflow="notes"`). The call
log shows them under `tests/notes`, and model checks leave them out.

A `wait_for_review` ends at once, unanswered. On a trigger shard it fails, as
the live call fails. On an RDF-native graph, the dry run refuses the write,
as the commit would.

## Failures and waits

An exception fails the attempt. The turn runs again, up to its
`max_attempts` (3 by default). A `WorkflowError` fails the turn for good. A
`WorkflowWait` waits until a time and counts no attempt. A step that meets
`429 workflow_budget_exceeded` waits that way by itself.

```python
import time

from lbb.workflows import WorkflowError, WorkflowWait

@workflow(name="invoice", version="v1", initial_state={})
async def invoice(ctx: WorkflowContext, state: dict, message: dict) -> WorkflowResult[dict]:
    if "order" not in message:
        raise WorkflowError("an invoice message names its order")
    ready = await ctx.step("ready", lambda s: message.get("paid", False))
    if not ready:
        raise WorkflowWait(retry_at_ms=int(time.time() * 1000) + 60_000)
    return WorkflowResult(state={"last": message["order"]})
```

A turn that failed for good holds the inbox until a person retries or skips
it. Start the instance with `on_exhausted="skip"` to keep the failed turn in
the history and go on.

## Run, scale and stop

A worker runs at most `concurrency` turns at once (default 1). Each turn
runs in its own asyncio task. Run more worker processes for more.

On SIGTERM the worker stops claiming. Its held turns finish within
`drain_seconds` (default 30). Then it releases the rest.

A released turn fails with `retry_at_ms`. It counts no attempt and keeps its
saved steps. Another worker claims it a second later. Ctrl-C (SIGINT), or a
second SIGTERM, releases at once. A second Ctrl-C ends the process.

A claim can wait on the server when the stop comes. The worker lets it
answer, and gives a turn it claimed back at once. The signal handlers need the
main thread. Off it, `run()` installs none: call `stop()` from another thread.

`run_async()` serves inside your own event loop, for example beside a web
app. `stop()` ends it in the same way.

```python
async def serve() -> None:
    async with connect() as client:
        worker = WorkflowWorker(client, [counter, review_draft], worker="api-1", concurrency=4)
        serving = asyncio.create_task(worker.run_async())
        await asyncio.sleep(1)
        worker.stop()
        await serving

asyncio.run(serve())
```

A claim that meets a network failure, a `429` or a `5xx` waits and claims
again. Any other refusal of a claim ends the loop with its error.

## The rules of a worker

The TypeScript and Python workers follow the same rules. A suite of scripted
server exchanges in the SDKs' repositories (`conformance/workflow-worker`)
runs against both.

| Request | Rule |
| --- | --- |
| `turns/claim` | Sent once. A failed claim is not sent again. |
| `turns/heartbeat` | Every third of the lease. A failed heartbeat ends the attempt in this worker. |
| `turns/checkpoint`, `turns/complete` | Retried. A retried completion sends the same body. |
| `turns/fail` | Sent once. A wait sends `retry_at_ms`, at least a second ahead. |
| A `409` or `404` on a turn | The turn stops. The worker sends nothing more for it. |
| `409 workflow_fence_lost` | The turn stops. The turn's next holder writes. |

## Try the template on a local stack

`values-extract` fills properties of a class from the text of its instances.
Its Python port is in `examples/values_extract` of the Python SDK. One command
starts a local stack with mock models and the Python worker:

```sh
npm run dev:stack -- --worker python:values-extract
```

The worker reads the six sample records and writes their values, each with its
quote. It gives a person the values it cannot confirm, such as a size in
another unit. The [workflow templates](https://docs.littlebigbrain.com/guides/workflow-templates/) guide
describes the template.