Durable Runs

OverviewCopy Anchor Link

An agent run lives in its process. When the process crashes, is redeployed or a worker shuts down halfway through, the run is gone: the thread keeps what was logged, but nobody finishes the task. A durable run keeps a checkpoint after every model turn and every batch of tool calls. Another process, or the same one after a restart, finds the runs left running with cog.recoverRuns() and continues each from its last step.

• Turning it on: durable for the runtime, an agent or one run
• Recovering runs: recoverRuns(), recoverRun(), discardRun()
• What a recovered run sees: finished calls kept, running calls "outcome unknown"
• Leases: one process holds a run at a time
• Stores: process memory, the memory adapter's threads, Redis
• Workers: queue jobs that go on from their last step

Without durable nothing changes: runs keep no checkpoint while they go on, as before.

Turning It OnCopy Anchor Link

import { Cogitator } from '@cogitator-ai/core';

const cog = new Cogitator({
  memory: { adapter: 'postgres', postgres: { connectionString: process.env.DATABASE_URL! } },
  durable: { owner: process.env.POD_NAME, takeoverOnRestart: true },
});

durable: true makes every run durable with the default store, an object sets it up:

An agent's durable and a run's own durable win over the runtime's setting:

const agent = new Agent({ name: 'reporter', model, instructions, durable: true });
await cog.run(agent, { input: 'Prepare the Q3 report', threadId: 'q3', durable: false });

A run started from a tool call of another run (a subagent, agentAsTool) is never durable on its own: its caller is, and the call that started it gets "outcome unknown" when the caller is recovered while it ran. Background subagents (run_in_background) live in their process too: the checkpoint names the ones going on, and the recovered run reads that they were lost, as failed with an unknown outcome, in task_output, send_message and its stop notice, so the model knows to check what they did.

What a Step WritesCopy Anchor Link

The checkpoint is a RunCheckpoint with status: 'running', the same shape a paused run returns: counters, usage and cost, the turn going on, permission mode and rules, budgets, hook state, project instructions, the command the input named, and the messages queued on the run's handle. It is kept small:

• The conversation's start (system prompt, history, input) is written once, as the record's base. Without a memory log the messages a step adds are added to the base, never written twice.
• Messages the thread's log already holds are written as references to their log entries (checkpoint.compact), not copied. The session log stays the source of truth.
• Tool calls are kept by their place in those messages, not with their arguments again.
• Between steps only small marks go to the record's journal: a tool call that starts, one that returns, the queue when it changes, a queued message the run delivered with what it added to the conversation.

A checkpoint is written after every model turn, the final answer included, so a run that answered before its process stopped ends on recovery without asking the model again. When the stop of that answer had not settled (follow-ups queued for the stop, beforeStop middleware and Stop hooks), the recovered run settles it first, and a run that reached maxIterations asks for its closing answer as it would have.

A step writes what is new since the last one, not the whole conversation.

Recovering RunsCopy Anchor Link

Call recoverRuns() at startup with the agents the runs were started with:

const { resumed, skipped } = await cog.recoverRuns({ agents: [reporter, support] });

for (const { run, handle } of resumed) {
  handle.result.then((result) => notifyUser(run.threadId, result.output));
}
for (const { run, reason } of skipped) console.warn(`Not recovered: ${run.runId}: ${reason}`);

Each recovered run comes back as a RunHandle, as cog.start() gives it: steer(), followUp(), inject(), abort() and result work on it. Runs a live process holds are left alone.

The options a run was started with that only lived in its process (callbacks, deps, the workspace, signal) are not in the checkpoint: pass them in options. Everything the run decided (its model, tool execution settings, permission mode and rules, budgets) comes back from the checkpoint.

To decide run by run:

const { found } = await cog.recoverRuns({ agents: [], resume: false });

for (const run of found) {
  if (await stillWanted(run.threadId)) {
    const handle = await cog.recoverRun(reporter, run.runId, { onToken });
    await handle.result;
  } else {
    await cog.discardRun(run.runId);
  }
}

cog.durableRuns(filter) lists the kept runs with held (a live process has them). discardRun() answers the calls of the run's last step that have no answer in the thread and forgets the run, so the next run of the thread reads a whole history. Both recoverRun() and discardRun() fail with RUN_NOT_FOUND for a run that is not kept and RUN_LEASED for one a live process holds.

What a Recovered Run SeesCopy Anchor Link

The conversation comes back from the base, the inline messages and the thread's log. Entries the run wrote after its last checkpoint are taken in too, so a crash between a log write and the checkpoint loses nothing and repeats nothing: a model turn that was logged is not asked for again, a final answer that was logged ends the run without another model call.

The tool calls of the step the process stopped in are settled from the journal:

So a tool with side effects never runs twice behind the model's back: the model reads that the outcome is unknown and checks before it calls again. Messages queued on the run's handle and not delivered yet are delivered by the recovered run.

LeasesCopy Anchor Link

A process holds each durable run it runs under a lease: it takes the lease when the run starts, renews it at every step and a third of the way through leaseMs while a long tool runs, and lets it go when the run ends. A lease belongs to the runtime's owner and to the instance of the Cogitator, so another live runtime never takes a run one holds, even with the same owner. recoverRuns() takes only runs whose lease ran out, and a runtime reserves a run before it takes its lease, so recovering one run twice at once, in one process or two, continues it once. With takeoverOnRestart: true it also takes at once a run whose lease names its own owner and an instance no runtime of its process has. That is a process that restarted only when no other live process has the same owner: with a host name several processes share, they would take each other's runs, so give each process an owner of its own (a pod name) or leave the option off and wait for the lease.

A process that lost its lease writes nothing more: every write to the thread first confirms the lease, so no tool result, no late model answer, no log entry and no afterRun hook comes from it. Every write is fenced: a checkpoint, its journal and its base change only while the writer holds the lease. A process that lost its lease (it froze longer than leaseMs and another process took the run over) finds out at its next step and stops the run with RUN_LEASED before it starts another tool or writes again, it never overwrites the checkpoint of the process that took over.

Writes to the thread's log are fenced by lease epochs. Every new holder of a run gets the next epoch (DurableRunLease.epoch), the run's log entries name it (metadata.durableEpoch), and a process that takes a run over writes a fence entry for its epoch before it reads the log, and notes it in the thread's metadata (durableFence.<runId>, one key per run), where every reader sees it. A memory adapter cannot refuse a write that was already on its way when the lease changed hands, so readers do: recovery, the history a later run reads and the context builder drop an entry of a run whose epoch is older than an epoch of that run that came before it in the log (a fence, or an entry of the newer holder). The guarantee holds in the order the adapter keeps entries (its sequence, else its timestamps), and an entry the former holder wrote before the fence counts as written before the takeover.

A store that fails is tried again (storeRetries, with a growing pause). When it keeps failing, the run is suspended with RUN_SUSPENDED and its last good checkpoint stays for a recovery: a storage error never deletes a checkpoint. With onStoreError: 'continue' the run goes on without checkpoints instead (a recovery may then repeat what it did since the last one).

A process that stops on purpose hands its runs over: abort a run with a CogitatorError of code RUN_SUSPENDED and it keeps its checkpoint and lets its lease go at once, for another process to recover. The worker pool does this when it shuts down.

import { CogitatorError, ErrorCode } from '@cogitator-ai/types';

controller.abort(new CogitatorError({ message: 'redeploy', code: ErrorCode.RUN_SUSPENDED }));

A run forgets its checkpoint only for an end of its own: it completed, paused, failed in its loop or was cancelled there. Every other end keeps it for a recovery: a hand-over, an error of the store, a lost lease, a configuration error or a recovery that gave up before the run got going. Forgetting or redacting entries of a thread (cog.forget(), cog.redactEntries()) drops the durable runs of that thread too, since their checkpoints may hold the text that is gone: a run going on there loses its lease and stops. A paused run waits in the paused-run store as before, resume() makes it durable again.

StoresCopy Anchor Link

A store implements DurableRunStore (acquire, renew, release, save, append, load, list, delete), its writes taking the holder of the lease.

import { Cogitator, RedisDurableRunStore } from '@cogitator-ai/core';
import { createRedisClient } from '@cogitator-ai/redis';

const client = await createRedisClient({ mode: 'standalone', url: process.env.REDIS_URL! });
const cog = new Cogitator({
  memory,
  durable: { store: new RedisDurableRunStore({ client }) },
});

The conversation's log references resolve through the runtime's memory adapter, so a recovering process needs the same memory as the one that ran. The Redis store keeps its keys under {cogitator}:durable-runs by default: a prefix of your own needs a {hash tag} in Redis Cluster, since one script writes all keys of a run and its indexes.

WorkersCopy Anchor Link

WorkerPool with durable: true runs agent jobs as durable runs named after the job (job:<jobId>). The attempt after a crash or a shutdown, on this worker or another, finds the run the stopped attempt left and goes on from its last step instead of running the whole job again. On shutdown the pool aborts its runs with RUN_SUSPENDED, so they hand over at once.

import { WorkerPool } from '@cogitator-ai/worker';

const pool = new WorkerPool({ redis, tools, durable: true });

When BullMQ hands a job on while the run of its stalled attempt is still held (its lease has not run out yet), the job waits until the lease runs out, without using an attempt. The pool's own Cogitator keeps the runs in its Redis. A Cogitator you pass needs a store all workers share, such as durable: { store: new RedisDurableRunStore({ client }), enabled: false } (enabled: false keeps runs outside the queue as they were).

StreamsCopy Anchor Link

A recovered run has no client attached. To let a client that lost its connection catch up with a run that is still going, use resumable streams in the server adapters.

ExampleCopy Anchor Link

examples/core/30-durable-runs.ts starts a run in a child process, kills it with SIGKILL while a tool runs and recovers the run in the parent.