Long-Running Runs

The com.tnsai.reliability package in tnsai-core ships the primitives:

• Checkpoint + CheckpointStore — durable state at step boundaries
• ResumableRun + DefaultResumableRun — execution driver that consumes the stores
• Progress + ProgressSink — observable mid-run state for live dashboards
• RunConfig — duration / retries / cost / progress-timeout policy bundle
• Outcome + AbortReason — terminal-result sealed type

Pairs with the idempotency layer (tnsai-core.idempotency) so side-effecting steps don't double-execute on retry, and with cost governance so a runaway loop hits a configured ceiling instead of burning unbounded spend.

When to use

You probably want this if your agent run:

• takes longer than the longest interactive session you'd let the user wait on (rule of thumb: \> 5 minutes)
• calls external systems with non-trivial side effects (creates issues, posts messages, runs payments, kicks off CI builds)
• has a per-run budget worth enforcing during execution rather than after the fact
• needs to be inspectable from a separate process / dashboard while it's running

You probably don't need it for:

• single-LLM-call request handlers
• read-only research agents that always finish in seconds
• one-shot CLI utilities without resumption needs

Quick start

import com.tnsai.reliability.*;
import com.tnsai.idempotency.InMemoryIdempotencyStore;
import java.time.Duration;
import java.math.BigDecimal;

// 1. Pick stores. InMemory for tests; Filesystem / SQLite for persistence.
CheckpointStore checkpoints  = new FilesystemCheckpointStore(Path.of("/var/tns/checkpoints"));
var idempotency               = new InMemoryIdempotencyStore();
ProgressSink progress         = new InMemoryProgressSink();

// 2. Define a state codec — encode/decode whatever your run threads
//    through its steps. JSON is the recommended default; here we use UTF-8
//    bytes for a string state.
DefaultResumableRun.StateCodec<String> codec =
    new DefaultResumableRun.StateCodec<>() {
        public byte[] encode(String s) { return s.getBytes(StandardCharsets.UTF_8); }
        public String decode(byte[] b) { return new String(b, StandardCharsets.UTF_8); }
    };

// 3. Compose the run. Each step is a UnaryOperator<S> with a name and a
//    side-effecting flag. Side-effecting steps consult the idempotency
//    store so a retry doesn't double-execute.
DefaultResumableRun<String> run = 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",
            state -> { githubClient.createPr(...); return state + "\n[posted]"; }))
    .step(DefaultResumableRun.Step.pure("verify", state -> state + "\n[verified]"))
    .build();

// 4. Execute with a config bound by duration / retries / cost / progress timeout.
RunConfig config = RunConfig.builder()
    .maxDuration(Duration.ofHours(8))
    .costCeilingUSD(new BigDecimal("5.00"))
    .checkpointInterval(Duration.ofMinutes(5))
    .maxRetries(3)
    .progressTimeout(Duration.ofMinutes(10))
    .build();

Outcome<String> result = run.execute("start", config);

The orchestrator returns one of four Outcome variants when the run terminates.

Outcome shape

Outcome<T> is a sealed interface — pattern-match exhaustively:

Outcome<String> result = run.execute(input, config);
String final = switch (result) {
    case Outcome.Completed<String> c -> c.result();
    case Outcome.Failed<String> f -> { logger.error("Failed: {}", f.reason()); yield ""; }
    case Outcome.Aborted<String> a -> { alert("Aborted: " + a.reason()); yield ""; }
    case Outcome.Suspended<String> s -> { /* resume later via run.resume(s.runId()) */ yield ""; }
};

Resume after a crash

After a process crash, runtime upgrade, or OOMError, the run is recoverable as long as the configured CheckpointStore survives:

// New process, same store backend.
DefaultResumableRun<String> run = /* ... same builder, same store paths ... */;

Outcome<String> result = run.resume(savedRunId);
// Resume loads the latest checkpoint, decodes the state via the codec,
// and continues from the next step.

If run.resume(runId) finds no checkpoint for the id, it returns Outcome.Failed with a clear "no such run" message rather than silently starting a fresh run — confusing one for the other was the bug pattern this primitive prevents.

Side-effect safety: the kill-and-resume invariant

The unit test that defines the contract:

A side-effecting step that runs once and crashes mid-execution must not run a second time on resume — its cached result is replayed instead.

Implementation: DefaultResumableRun keys side-effecting steps by runId:stepIndex:stepName against the configured IdempotencyStore. On the original run the orchestrator records the step's return value via IdempotencyEntry.forSuccess(...); on resume the orchestrator checks the store before invoking the body. Cached hit → replay; miss → execute + record.

Concretely: a step that posts a GitHub PR, captured the PR id in its return state, then crashed before saving its checkpoint, will on resume:

• find the cached PR id in the idempotency store under <runId>:<stepIndex>:post-pr
• skip the body (no second POST /pulls to GitHub)
• thread the cached state forward into the next step

This is what makes hour-scale runs against rate-limited / cost-bearing external APIs feasible.

CheckpointStore selection

Picking between distributed stores

Use Redis for fast resume cycles (operator-paused runs picked up within minutes); use S3 for archival / cross-region durability. The two are not mutually exclusive — a layered stack with a Redis hot path and an S3 cold path is a reasonable production shape, but ships as a consumer-side composite rather than a built-in.

Cost ceiling enforcement

DefaultResumableRun.Builder#costMonitor is optional. Provide one and pair it with RunConfig.costCeilingUSD to make the orchestrator check spend before every step:

// Wire to the cost-governance store from tnsai-quality.
CostBudgetStore costStore = ...; // your spend ledger
DefaultResumableRun.CostMonitor monitor = () ->
    Optional.of(costStore.currentSpend(
        CostScope.tenant(tenantId), Duration.ofHours(24)));

DefaultResumableRun<S> run = DefaultResumableRun.<S>builder()
    // ...
    .costMonitor(monitor)
    .build();

run.execute(input, RunConfig.builder()
    .costCeilingUSD(new BigDecimal("10.00"))   // $10 cap
    .build());

When currentSpend ≥ ceiling, the orchestrator:

1. Saves a final checkpoint at the current state
2. Publishes Progress.RunAborted(BUDGET_EXHAUSTED)
3. Returns Outcome.Aborted

The consumer can resume the run on a higher ceiling once it's available — the saved checkpoint preserves work-done-so-far.

Progress events

Progress is a sealed interface with eight variants. Subscribe to a run's events for live dashboards or progress-timeout enforcement:

ProgressSink.Subscription sub = sink.subscribe(runId, event -> {
    switch (event) {
        case Progress.StepStarted s   -> dashboard.markStepRunning(s.stepIndex());
        case Progress.StepCompleted s -> dashboard.markStepDone(s.stepIndex(), s.took());
        case Progress.CheckpointSaved c -> dashboard.checkpointAt(c.checkpointId());
        case Progress.CostUpdate c    -> dashboard.spendTick(c.spentUSD());
        case Progress.Heartbeat h     -> dashboard.heartbeat(h.currentStep());
        case Progress.RunCompleted r  -> dashboard.markDone();
        case Progress.RunFailed r     -> dashboard.markFailed(r.reason());
        case Progress.RunAborted r    -> dashboard.markAborted(r.reason());
    }
});
// Cancel when the dashboard tab closes.
sub.cancel();

InMemoryProgressSink is the default. KafkaProgressSink and similar fan-out adapters are tracked as follow-ups; the SPI accepts third-party impls today.

RunConfig defaults

Use RunConfig.defaults() for an overnight-run-friendly starting point, or RunConfig.builder() for finer control.

Tool annotations

Tools that participate in resumable runs benefit from declaring their effect classification + idempotency expectation. Two annotations were added to @Tool for this:

import com.tnsai.annotations.*;

public class GithubTools {

    @Tool(name = "github.create_pr",
          description = "Open a pull request on a GitHub repo",
          sideEffect = SideEffect.EXTERNAL,
          idempotencyHint = IdempotencyHint.REQUIRED)
    public PullRequest createPr(@ToolParam("title") String title, ...) { ... }

    @Tool(name = "github.list_branches",
          description = "List branches on a GitHub repo",
          sideEffect = SideEffect.READ)            // idempotencyHint defaults to NONE
    public List<Branch> listBranches(...) { ... }
}

SideEffect:

IdempotencyHint:

Both fields default to safe values so existing tools retain their current behaviour without modification — the annotations are purely additive metadata.

What's not in this layer

Out of scope for the framework primitives, tracked separately:

• Distributed transaction coordination (sagas) — checkpointing is local; cross-system consistency is a different problem.
• Run inspection UI — the ProgressSink API is enough for now; UI is downstream.
• Backfill for runs without checkpointing — additive feature, not retroactive.
• Harness execution-loop refactor — the existing AgentExecutor will move onto ResumableRun as part of the harness work (TNS-291).

See Also

• Idempotency — the SPI consumed by side-effecting steps
• Cost Governance — the spend ledger the cost monitor wires into
• Resilience — companion retry / fallback / circuit-breaker layer for individual operations
• Error Handling — how step exceptions surface and propagate