# Long runs

A long run is a fixed list of steps that saves a checkpoint after each step. If the process dies, a new process resumes the run from the last checkpoint. The types live in `com.tnsai.reliability` in `tnsai-core`.

| Type | Role |
|---|---|
| `DefaultResumableRun` | Runs the steps, saves checkpoints and applies the `RunConfig` limits |
| `CheckpointStore` | Keeps the state after each step |
| `IdempotencyStore` | Keeps the result of each side-effecting step, so a resume does not repeat it |
| `ProgressSink` | Receives a `Progress` event for each step, checkpoint and terminal state |
| `RunConfig` | Duration, retry, cost and progress limits |
| `Outcome` | The result: completed, failed or aborted |

Use a long run when a job takes longer than a request should wait, when its steps have external side effects such as opening a pull request or charging a card, or when it needs a spend limit during the run. A single model call or a read-only job does not need it.

## Build and run

Each step is a `UnaryOperator` over the run's state. `Step.pure` marks a step that is safe to repeat. `Step.sideEffect` marks a step whose result is cached in the idempotency store under `<runId>:<stepIndex>:<stepName>` for 24 hours. The codec turns the state into bytes for the checkpoint.

<!-- java-contract: src/main/java/com/example/tnsai/docs/LongRunExample.java -->
```java
package com.example.tnsai.docs;

import com.tnsai.idempotency.IdempotencyStore;
import com.tnsai.quality.cost.CostBudgetStore;
import com.tnsai.quality.cost.CostScope;
import com.tnsai.reliability.CheckpointStore;
import com.tnsai.reliability.DefaultResumableRun;
import com.tnsai.reliability.Outcome;
import com.tnsai.reliability.Progress;
import com.tnsai.reliability.ProgressSink;
import com.tnsai.reliability.RunConfig;

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.function.UnaryOperator;

public final class LongRunExample {
    private LongRunExample() {}

    // A String state survives the JSON round trip of a Redis or Postgres
    // idempotency store unchanged.
    static final DefaultResumableRun.StateCodec<String> CODEC = new DefaultResumableRun.StateCodec<>() {
        @Override
        public byte[] encode(String state) {
            return state.getBytes(StandardCharsets.UTF_8);
        }

        @Override
        public String decode(byte[] bytes) {
            return new String(bytes, StandardCharsets.UTF_8);
        }
    };

    // Three steps. Only the side-effecting one consults the idempotency store.
    static DefaultResumableRun<String> build(CheckpointStore checkpoints,
                                             IdempotencyStore idempotency,
                                             ProgressSink progress,
                                             UnaryOperator<String> postPr) {
        return DefaultResumableRun.<String>builder()
            .codec(CODEC)
            .checkpointStore(checkpoints)
            .idempotencyStore(idempotency)
            .progressSink(progress)
            .step(DefaultResumableRun.Step.pure("plan", state -> state + "\n[plan]"))
            .step(DefaultResumableRun.Step.sideEffect("post-pr", postPr))
            .step(DefaultResumableRun.Step.pure("verify", state -> state + "\n[verified]"))
            .build();
    }

    static RunConfig config() {
        return RunConfig.builder()
            .maxDuration(Duration.ofHours(8))
            .maxRetries(3)
            .progressTimeout(Duration.ofMinutes(10))
            .build();
    }

    static String summarize(Outcome<String> outcome) {
        return switch (outcome) {
            case Outcome.Completed<String> c -> c.result();
            case Outcome.Failed<String> f -> "failed: " + f.reason();
            case Outcome.Aborted<String> a -> "aborted: " + a.reason();
            case Outcome.Suspended<String> s -> "suspended at " + s.checkpointId();
        };
    }

    // execute() creates the run id. Record it from the first event so that a
    // new process knows which run to resume.
    static final class RunIdRecorder implements ProgressSink {
        private final ProgressSink delegate;
        private final Consumer<String> onNewRun;

        RunIdRecorder(ProgressSink delegate, Consumer<String> onNewRun) {
            this.delegate = delegate;
            this.onNewRun = onNewRun;
        }

        @Override
        public void publish(String runId, Progress event) {
            if (event instanceof Progress.StepStarted started && started.stepIndex() == 0) {
                onNewRun.accept(runId);
            }
            delegate.publish(runId, event);
        }

        @Override
        public Subscription subscribe(String runId, Consumer<Progress> listener) {
            return delegate.subscribe(runId, listener);
        }
    }

    // In a new process: the same steps, the same stores, the saved run id.
    static Outcome<String> resumeAfterCrash(CheckpointStore checkpoints,
                                            IdempotencyStore idempotency,
                                            UnaryOperator<String> postPr,
                                            String runId) {
        return build(checkpoints, idempotency, null, postPr).resume(runId);
    }

    // Abort with BUDGET_EXHAUSTED once the tenant's spend over 24 hours reaches $10.
    static Outcome<String> capped(CheckpointStore checkpoints, CostBudgetStore costs, String tenantId) {
        DefaultResumableRun<String> run = DefaultResumableRun.<String>builder()
            .codec(CODEC)
            .checkpointStore(checkpoints)
            .costMonitor(() -> Optional.of(
                costs.currentSpend(CostScope.tenant(tenantId), Duration.ofHours(24))))
            .step(DefaultResumableRun.Step.pure("draft", state -> state + "\n[draft]"))
            .build();
        return run.execute("start", RunConfig.builder()
            .costCeilingUSD(new BigDecimal("10.00"))
            .build());
    }
}
```

`execute(input, config)` creates a new run id and returns an `Outcome`:

| Variant | When |
|---|---|
| `Outcome.Completed` | Every step finished. `result()` is the final state. |
| `Outcome.Failed` | A step threw on every attempt, or `resume` found no checkpoint. It carries `reason()` and `cause()`. |
| `Outcome.Aborted` | A limit or `abort(runId, reason)` stopped the run. It carries the `AbortReason`. |
| `Outcome.Suspended` | Declared, but `DefaultResumableRun` never returns it. |

A step gets `1 + maxRetries` attempts. When they all fail, the run returns `Outcome.Failed`, not `Outcome.Aborted`.

## Limits

`DefaultResumableRun` checks the limits before each step, never during one. A step that hangs is not interrupted.

| `RunConfig` field | Default | Effect |
|---|---|---|
| `maxDuration` | 8 hours | Abort with `MAX_DURATION_EXCEEDED` once this much time has passed since `execute` or `resume` started |
| `progressTimeout` | 10 minutes | Abort with `PROGRESS_TIMEOUT` when the previous step finished more than this long ago |
| `maxRetries` | 3 | Retries per step before `Outcome.Failed` |
| `costCeilingUSD` | none | With a cost monitor, abort with `BUDGET_EXHAUSTED` once spend reaches it |
| `checkpointInterval` | 5 minutes | Not implemented: nothing reads it. A checkpoint is saved after every step. |

`RunConfig.defaults()` returns these values.

`AbortReason` has five values: `USER_REQUESTED`, `BUDGET_EXHAUSTED`, `PROGRESS_TIMEOUT`, `MAX_DURATION_EXCEEDED` and `ERROR_LIMIT_EXCEEDED`. The run itself never uses `ERROR_LIMIT_EXCEEDED`. `abort(runId, reason)` stops the run before its next step with the reason you pass.

### Cost ceiling

A `CostMonitor` returns the current spend. The `capped` method above wires it to a `CostBudgetStore` from `tnsai-quality`. Before each step the run publishes a `Progress.CostUpdate`. When spend reaches the ceiling, it saves a checkpoint, publishes `Progress.RunAborted(BUDGET_EXHAUSTED)` and returns `Outcome.Aborted`. See [Tenant budgets](../llm/cost.md#tenant-budgets).

## Resume after a crash

`resume(runId)` loads the latest checkpoint, decodes the state and runs the next step. Build the run with the same steps and stores, as `resumeAfterCrash` does above.

- **The run id.** `execute` creates it and returns it only in the `Outcome`. To know it before a crash, record it from the `ProgressSink`, as `RunIdRecorder` does. `FilesystemCheckpointStore` also names one directory per run id.
- **The config.** A resumed run uses `RunConfig.defaults()`, not the config passed to `execute`.
- **No checkpoint.** `resume` returns `Outcome.Failed` with "No checkpoint found for run". It does not start a new run.

### Side-effecting steps

A side-effecting step stores its result before its checkpoint is saved. If the process dies between the two, `resume` finds the stored result and does not run the step again. If the process dies while the step is still running, nothing was stored and the step runs again.

This protection lasts only as long as the idempotency store does. `InMemoryIdempotencyStore` is lost with the process, so use `RedisIdempotencyStore` or `PostgresIdempotencyStore` for crash safety. Those stores keep the result as JSON, so a record state comes back as a `Map`. Use a `String`, `Map` or `List` state with them. See [Idempotency](../tools/idempotency.md).

## Checkpoint stores

| Store | Use |
|---|---|
| `InMemoryCheckpointStore` | Tests. Lost when the process exits. |
| `FilesystemCheckpointStore(Path)` | One host. One JSON file per checkpoint at `<root>/<runId>/<stepIndex>.json`, written to a temporary file, synced and renamed. |
| `SqliteCheckpointStore(Path)` | One host, one SQLite file. Needs `org.xerial:sqlite-jdbc`. |
| `RedisCheckpointStore(RedisClient)` | Several processes sharing Redis. Needs Jedis. |
| `S3CheckpointStore(S3Client, bucket)` | Runs that may resume on another host. One object per checkpoint at `s3://<bucket>/<keyPrefix><runId>/<stepIndex>.json`. Needs `software.amazon.awssdk:s3`. |

The three dependencies are optional in `tnsai-core`, so add the one your store needs.

## Progress events

`Progress` is a sealed interface. Subscribe with `sink.subscribe(runId, listener)` and call `cancel()` on the returned `Subscription` when done. `InMemoryProgressSink` is the built-in sink.

| Event | When |
|---|---|
| `StepStarted` | A step starts |
| `StepCompleted` | A step finishes. `took()` is its duration. |
| `CheckpointSaved` | A checkpoint is saved |
| `CostUpdate` | Before each step, when a cost monitor is set. `spentUSD()` and `remainingUSD()`. |
| `RunCompleted` | The run finished |
| `RunFailed` | A step failed on every attempt |
| `RunAborted` | A limit or `abort` stopped the run |
| `Heartbeat` | Declared, but `DefaultResumableRun` never publishes it |

## Side effects on tools

`@Tool` has two related fields for tools an agent calls, outside any long run:

- `sideEffect`: `NONE` (the default), `READ`, `WRITE` or `EXTERNAL`. An `EXTERNAL` tool without an `idempotencyHint` logs a warning when it is registered.
- `idempotencyHint`: `NONE` (the default), `OPTIONAL` or `REQUIRED`. A `REQUIRED` tool fails at dispatch unless it also has `@Idempotent` with the `HASH_INPUT` or `EXPLICIT` strategy and a positive `ttlSeconds`. With `OPTIONAL` or `NONE`, an `@Idempotent` policy still applies when present, and the hint adds no check.

<!-- java-contract: src/main/java/com/example/tnsai/docs/LongRunToolsExample.java -->
```java
package com.example.tnsai.docs;

import com.tnsai.annotations.IdempotencyHint;
import com.tnsai.annotations.SideEffect;
import com.tnsai.annotations.Tool;
import com.tnsai.annotations.ToolParam;
import com.tnsai.idempotency.Idempotent;
import com.tnsai.idempotency.KeyStrategy;

import java.util.List;

public final class LongRunToolsExample {
    private LongRunToolsExample() {}

    public static final class GithubTools {

        // REQUIRED: dispatch fails unless @Idempotent gives a stable key and a positive TTL.
        @Tool(name = "github_create_pr",
              description = "Open a pull request on a GitHub repository",
              sideEffect = SideEffect.EXTERNAL,
              idempotencyHint = IdempotencyHint.REQUIRED)
        @Idempotent(strategy = KeyStrategy.HASH_INPUT, ttlSeconds = 86_400)
        public String createPr(@ToolParam(description = "PR title") String title,
                               @ToolParam(description = "Branch to merge") String branch) {
            return "opened '" + title + "' from " + branch;
        }

        // READ: no idempotency hint needed.
        @Tool(name = "github_list_branches",
              description = "List the branches of a GitHub repository",
              sideEffect = SideEffect.READ)
        public List<String> listBranches(@ToolParam(description = "Repository as owner/name") String repo) {
            return List.of("main");
        }
    }
}
```

The agent's default idempotency store is in memory. For records that outlive a restart, pass a durable store to `AgentBuilder.idempotencyStore(...)`.

## Related

- [Idempotency](../tools/idempotency.md): the stores and `@Idempotent`.
- [Tenant budgets](../llm/cost.md#tenant-budgets): the spend ledger behind a cost monitor.
- [Resilience](resilience.md): retries and timeouts for a single action.
- [Exceptions](exceptions.md): what a failing step can throw.