Checkpoints & State

A revision turns a stale write into a conflict

Two workers read revision 4. Writer A saves first and the checkpoint advances to revision 5; writer B's save against revision 4 returns checkpoint_conflict, and B rereads before it proposes anything else.

In a worker, the pattern reads the checkpoint, does one step, and saves under the revision it read. The application's slot sits beside adapter-owned properties in the same value, and the read route and path must be in the host's signed manifest and this process's grant.

checkpoint = client.inspect()["checkpoint"]
value = checkpoint["value"] or {}
progress = value.get("report.v1", {"version": 1, "completed": []})
if "notes.txt" not in progress["completed"]:
    result = client.invoke(
        "read-source-17", "source", "read", {"path": "notes.txt"}
    )
    if (
        result["verdict"] != "allow"
        or result["terminal_state"]["state"] != "completed"
    ):
        raise RuntimeError("Source read did not complete")
    # Validate the source tool's documented output before advancing.
    if result["output"]["kind"] != "value":
        raise RuntimeError("Expected a source value")
    next_progress = {
        **progress,
        "completed": [*progress["completed"], "notes.txt"],
    }
    client.checkpoint(
        checkpoint["revision"], {**value, "report.v1": next_progress}
    )

A crash after the tool call and before the save repeats this code, and the same read-source-17 key resolves the original kernel operation. A conflict means rereading state and deciding which progress still holds; the old key keeps its original request.

Phases name the work still to do

Give recovery a small, versioned state machine. prepared means the original operation still needs resolution; completed means its validated result is recorded.

{
  "report.v1": {
    "version": 1,
    "phase": "prepared",
    "request": {
      "key": "publish-report-42",
      "server": "reports",
      "tool": "publish",
      "arguments": {
        "report": "Three inputs total ten."
      },
      "known_outcome_only": true
    },
    "completed": []
  }
}

Reject an unsupported state version instead of reading it as empty. A missing completion marker says nothing about whether the tool ran; the saved key and frozen request carry that, as Calls and recovery shows.

The value is shared. The AI SDK model journal owns chio.ai-sdk.journal.v1, and cooperative child waits keep their own slot, so a replacement checkpoint carries every property forward. The revision protects the record as a whole; for the meaning of each property, keep one writer per application phase or define an explicit merge protocol.

A lost acknowledgement is resolved by rereading

A compare-and-swap reply can be lost after the commit. Reread before changing anything.

If your slot matches the value you proposed, the progress is retained. If the revision is unchanged, the exact same save can be sent again. If another writer advanced the state, stop and reconcile what it means; a loop that overwrites the newest value with a stale calculation undoes what the revision protects.

from chio_process import WorkerError


def commit_report(client, checkpoint, proposed_report):
    previous = checkpoint["value"] or {}
    next_value = {**previous, "report.v1": proposed_report}
    try:
        return client.checkpoint(checkpoint["revision"], next_value)
    except WorkerError as error:
        if error.code not in {
            "checkpoint_conflict",
            "transport_error",
            "truncated_response",
        }:
            raise
        observed = client.inspect()["checkpoint"]
        if (observed["value"] or {}).get("report.v1") == proposed_report:
            return observed
        raise RuntimeError(
            "Checkpoint changed or commit remains unconfirmed; reconcile"
        ) from error

Large results live as immutable, process-owned blobs

Store the bytes first and keep their digest and byte count, then compare-and-swap that reference into the next checkpoint revision.

body = result_bytes
stored = client.put_blob(body)
checkpoint = client.inspect()["checkpoint"]
client.checkpoint(
    checkpoint["revision"],
    {
        **(checkpoint["value"] or {}),
        "report_sha256": stored["sha256"],
        "report_bytes": stored["bytes"],
    },
)
recovered = client.read_blob(stored["sha256"])
assert recovered == body

A blob is addressed by SHA-256 within the authenticated process, so knowing the digest gives another process nothing. Writes are immutable, and the client checks the digest and byte count on every read. A crash between the two writes leaves a charged blob with no reference; reread the checkpoint before retrying, because a lost reply can hide a commit, and write the same bytes again to recover the reference. Blobs have no delete or collection operation.

Snapshots and blobs draw on one storage budget

JsonSnapshot splits supported JSON into immutable blobs and rebuilds it with integrity and size checks, reusing process-owned bytes for repeated fragments.

from chio_process.snapshot import JsonSnapshot

snapshots = JsonSnapshot(client)
reference = snapshots.write(report_document)
cp = client.inspect()["checkpoint"]
value = cp["value"] or {}
client.checkpoint(
    cp["revision"], {**value, "report.snapshot.v1": reference}
)
restored = snapshots.read(reference)
assert restored == report_document

The core checkpoint is limited to 1 MiB, and so is each blob. The default root-tree blob budget is 64 MiB and 4,096 records. A snapshot can spread an 8 MiB document across blobs, and it spends the same bytes and records doing so; retained history, including blobs whose checkpoint never committed, can exhaust the quota. Watch the process's reported storage usage before a long model session fills it.

Progress and effects are separate records

Each record answers its own question, and recovery reads them together.

• Checkpoint. Application progress at one revision. An external effect and the checkpoint commit separately.
• Process blob. Private immutable bytes for the same process, with no claim about tool authenticity or provider completion.
• Kernel receipt. The signed decision and retained operation evidence; the correctness of each application output is the application's to check.
• Runner journal. OS attempt and dependency observations; whether a published artifact satisfies the task is checked separately.

A model response that contains tool-call IDs is saved before those tools run when recovery resumes the same turn. The AI SDK process adapter supplies that response journal; a bare ProcessClient leaves it to you. A missing or corrupt snapshot is a recovery failure, and replacing a lost model response with a fresh provider call can name different work under the same tool IDs. Model-call recovery covers the provider boundary.

chio process state --state .chio --process consumer

{
  "schema": "chio.process.application-state.v1",
  "process": "consumer",
  "data": {
    "checkpoint": {
      "revision": "1",
      "value": {
        "item_count": 3,
        "total": 10
      }
    }
  }
}

An operator reads the retained checkpoint after the worker's authority ends. The offline reader authenticates as no worker, starts no MCP server, and renews no authority, so a diagnostic stays a read and never becomes execution.

Every worker language keeps the same process-scoped state

These recipes from the three-language recipe project keep unrelated checkpoint fields, recognize work already done, and leave a conflict visible to the caller. The blob round trip checks the exact bytes, and a change of worker language grants no access to a sibling's blobs.

def retain_progress(client):
    cp = client.inspect()["checkpoint"]
    previous = cp["value"] or {}
    if "report.v1" in previous:
        if previous["report.v1"] != {"completed": ["report-42"]}:
            raise ValueError("Checkpoint belongs to different work")
        return cp
    # A CAS conflict requires rereading and reconciling, not blind overwrite.
    return client.checkpoint(
        cp["revision"],
        {
            **previous,
            "report.v1": {"completed": ["report-42"]},
        },
    )

export async function retainProgress(client: ProcessClient) {
  const cp = (await client.inspect()).checkpoint;
  const previous = cp.value === null ? {} : object(cp.value);
  const progress = { completed: ["report-42"] };
  if (previous["report.v1"] !== undefined) {
    assert.deepEqual(previous["report.v1"], progress);
    return cp;
  }
  // A CAS conflict requires rereading and reconciling, not blind overwrite.
  return client.checkpoint(cp.revision, { ...previous, "report.v1": progress });
}

async fn retain_progress(client: &ProcessClient) -> Result<Checkpoint> {
    let cp = client.inspect().await?.checkpoint;
    let mut previous = if cp.value.is_null() {
        serde_json::Map::new()
    } else {
        cp.value
            .as_object()
            .ok_or("Checkpoint must be an object")?
            .clone()
    };
    let progress = json!({"completed":["report-42"]});
    if let Some(saved) = previous.get("report.v1") {
        if saved != &progress {
            return Err("Checkpoint belongs to different work".into());
        }
        return Ok(cp);
    }
    previous.insert("report.v1".into(), progress);
    // A CAS conflict requires rereading and reconciling, not blind overwrite.
    Ok(client
        .checkpoint(&cp.revision, Value::Object(previous))
        .await?)
}

def round_trip_blob(client):
    original = b"Retained report body"
    reference = client.put_blob(original)
    restored = client.read_blob(reference["sha256"])
    if restored != original:
        raise RuntimeError("Blob mismatch")
    return reference

export async function roundTripBlob(client: ProcessClient) {
  const original = new TextEncoder().encode("Retained report body");
  const reference = await client.putBlob(original);
  assert.deepEqual(await client.readBlob(reference.sha256), original);
  return reference;
}

async fn round_trip_blob(client: &ProcessClient) -> Result<()> {
    let original = b"Retained report body";
    let reference = client.put_blob(original).await?;
    if client.read_blob(&reference.sha256).await? != original {
        return Err("Blob mismatch".into());
    }
    Ok(())
}

Test each interruption point on its own fresh host, with the original operation keys. Stop the consumer after blob storage, after the checkpoint commit, and after acknowledgement; on restart, assert exactly one completed item, the same checkpointed calculation, and the original receipt identity. For the revision check, read one revision twice, commit writer A, then submit writer B with the stale revision: B fails and a reread shows A's state intact.

Next: hand work between processes