# Flows

> For the complete documentation index, see [llms.txt](https://docs.lertha.com/llms.txt).
> Any documentation page is available as raw Markdown by appending `.md` to its URL.

> Writing a flow body: sequence, fan-out, signals, sleep, and catching a failed task.

## The definition

```kotlin
interface FlowDefinition<I : Any, O : Any> {
    val type: FlowType<I, O>
    fun execute(input: I, flow: FlowScope): O
}
```

Not `suspend`. The engine owns scheduling: the body blocks on `FlowScope` calls and the engine
parks it durably. Read [Determinism](/concepts/determinism) before writing one.

## Sequence

```kotlin
val first = flow.task(Echo.type, EchoInput("hi"), quick)
val second = flow.task(Echo.type, EchoInput(first.text + "!"), quick)
```

## Fan-out

```kotlin
val pending = statements.map { flow.startTask(Classify.type, it, options) }
val results = pending.map { it.await() }
```

## Waiting for a person

```kotlin
val approvals = flow.signals<Approval>("approval")
val approval = approvals.receive(30.minutes)
if (approval?.approved != true) { /* declined or timed out */ }
```

Sending, from the API or the CLI:

```kotlin
orchestrator.handle(AgentRun.type, RunId("agent-42")).signal("approval", Approval(true, "abiola"))
```

```bash
lertha flow signal --id agent-42 --name approval --input '{"approved":true,"by":"cli"}'
```

## Sleeping

```kotlin
flow.sleep(7.days)
```

A durable timer. The process can die and the timer still fires; a run can sleep for a year and
cost nothing while it does.

## Catching a failed task

```kotlin
val observation = try {
    flow.task(RunTool.type, ToolInput(plan.tool, plan.args), tooling).observation
} catch (e: TaskFailed) {
    "Tool ${plan.tool} failed: ${e.message}"
}
```

`e.state` says whether the task failed, timed out or was cancelled. Never catch `Throwable`
around a scope call: cancellation of the flow itself travels through that path.

## Starting a flow

```kotlin
val options = RunOptions(
    id = "agent-${conversationId}",
    queue = "lab",
    attemptTimeout = 1.hours,
    overallTimeout = 1.hours,
    retry = RetryPolicy.NONE,
)
val handle: FlowHandle<AgentOutput> = orchestrator.start(AgentRun.type, AgentInput(goal), options)
```

A `FlowHandle` is a `RunHandle` plus `signal()`.

## The agent shape

Plan step, tool step, a human before the consequential action, loop until the planner says done.
The planner and the tools are tasks; the loop and the waiting are the flow. The lab repository
ships this as `AgentRun`, with the planner scripted so the lab is deterministic and free. Swap
the script for a model call and nothing else changes.