A durable org.llm4s.agent.graph.Checkpointer in one SQLite database file: threads, their checkpoint histories, pending writes, event logs and run claims survive a restart, and several runtimes - in this process or in others on the same host - can share the file, each through a store of its own.
Open it with SqliteCheckpointer.open and close it with close. It meets the Checkpointer contract (and passes CheckpointerContract from llm4s-agent-testkit):
'''Atomic, fenced commits.''' Every call that writes is one BEGIN IMMEDIATE transaction: a Commit's checkpoint, pending writes and events land together or not at all, and the claim's token and the thread's version are checked inside the same transaction, so two stores over one file cannot both pass the checks.
'''Claims expire by clock''', the clock this store was opened with: the clock of the process that grants or renews a claim decides when it lapses. Stores in processes on different hosts sharing one file therefore judge expiry by different clocks, and skew between them changes a claim's effective ttl (see docs/guide/agents/durable-checkpointers.md); fencing keeps writes safe regardless.
'''History and retention.''' Every checkpoint a commit supersedes stays in the file, readable with history (newest first, paged through the (thread, position) primary key) and checkpoint, until prune, a commit's retractTurn or deleteThread removes it; each of those is one write transaction too.
'''Fencing tokens''' come from one counter for the whole file, stored in it, so they keep increasing across restarts and deleteThread, and are never issued twice.
'''Data only.''' Checkpoints, pending writes and events are stored as JSON - the checkpoint through Checkpoint.toJson, which records its format version - and decoded on read, so nothing executable is stored.
'''Change notification by polling.''' SQLite cannot tell a connection that another has committed, so this store keeps Checkpointer.awaitEventsAfter's default: a subscription watching it reads the event log once per WatchPolicy.pollInterval, and sees commits made through other stores on the file that long after they land.
The database runs in WAL mode with synchronous = FULL: readers never block the writer, and a commit that returned survives a crash or a power loss. SQLite admits one writer at a time per file; a write that finds another connection writing waits up to SqliteCheckpointerConfig.busyTimeout, then fails with a ProcessingError, changing nothing. WAL needs shared memory between the processes that open the file, so keep the file on a local disk - not on a network file system.
One store is one connection, and its calls are serialised: it is safe to call from any number of threads. Every failure of the database itself is a ProcessingError, and so is any call after close.
Claims threadId for request.holder until request.ttl from now, by the store's clock, if no other claim on it is live, and returns the claim with a new FencingToken. A live claim - whoever holds it, the same holder included - is Left(GraphError.ThreadBusy) naming its holder, with nothing changed. A thread need not have a checkpoint to be claimed.
Claims threadId for request.holder until request.ttl from now, by the store's clock, if no other claim on it is live, and returns the claim with a new FencingToken. A live claim - whoever holds it, the same holder included - is Left(GraphError.ThreadBusy) naming its holder, with nothing changed. A thread need not have a checkpoint to be claimed.
Closes the connection. Later calls return a ProcessingError; closing again does nothing. A claim this store granted stays in the file until it is released through another store or expires.
Closes the connection. Later calls return a ProcessingError; closing again does nothing. A claim this store granted stays in the file until it is released through another store or expires.
Removes everything stored for the thread - its checkpoint history, its pending writes, its event log and its claim - so that its id names a new thread again, whose events are numbered from 1; tokens issued after it are still greater than every token issued before. An unknown thread is Right(()). Use GraphRuntime.deleteThread, which claims the thread first, so it refuses a thread with a live run in any runtime, and checks the tenant; the store itself checks neither.
Removes everything stored for the thread - its checkpoint history, its pending writes, its event log and its claim - so that its id names a new thread again, whose events are numbered from 1; tokens issued after it are still greater than every token issued before. An unknown thread is Right(()). Use GraphRuntime.deleteThread, which claims the thread first, so it refuses a thread with a live run in any runtime, and checks the tenant; the store itself checks neither.
Up to limit of the thread's checkpoints, newest first - the latest is the first of the first page - starting after checkpoint before when it is given, so a caller pages through the history by passing the last id of each page. A before that is not in the history is Left(GraphError.CheckpointNotFound); a thread with no checkpoint has an empty history.
Up to limit of the thread's checkpoints, newest first - the latest is the first of the first page - starting after checkpoint before when it is given, so a caller pages through the history by passing the last id of each page. A before that is not in the history is Left(GraphError.CheckpointNotFound); a thread with no checkpoint has an empty history.
Removes what policy does not keep, in one write transaction, so it sees and changes one state of the thread whatever other stores on the file are doing; it takes no claim, as the contract says.
Removes what policy does not keep, in one write transaction, so it sees and changes one state of the thread whatever other stores on the file are doing; it takes no claim, as the contract says.
Ends the claim whose token is token, so the thread can be claimed again at once. A token that is not the thread's current claim changes nothing and is Right(()): a run that lost its claim must never release its successor's.
Ends the claim whose token is token, so the thread can be claimed again at once. A token that is not the thread's current claim changes nothing and is Right(()): a run that lost its claim must never release its successor's.
Extends the claim whose token is token to ttl from now, by the store's clock, and returns it. Left(GraphError.StaleClaim) if token is not the thread's current claim (another run took it over, it was released, or the thread was deleted). An expired claim nobody replaced is renewed.
Extends the claim whose token is token to ttl from now, by the store's clock, and returns it. Left(GraphError.StaleClaim) if token is not the thread's current claim (another run took it over, it was released, or the thread was deleted). An expired claim nobody replaced is renewed.
eventsAfter, waiting for a commit when there is nothing to return yet: up to limit events with seq > afterSeq, ascending, at once if the log holds any; otherwise those of a commit that lands within timeout, or an empty Vector once it has passed. It is how a subscription learns of commits made through another GraphRuntime or process sharing the store (see WatchPolicy): each subscription calls it in a loop, from the last event it has queued.
eventsAfter, waiting for a commit when there is nothing to return yet: up to limit events with seq > afterSeq, ascending, at once if the log holds any; otherwise those of a commit that lands within timeout, or an empty Vector once it has passed. It is how a subscription learns of commits made through another GraphRuntime or process sharing the store (see WatchPolicy): each subscription calls it in a loop, from the last event it has queued.
The default polls. It reads once and, finding nothing, waits timeout and returns empty, so the caller's next call reads again: a commit is seen up to timeout after it lands, at the cost of one eventsAfter per timeout per subscription. A store that can be told of commits overrides it to return as soon as one lands, without reading meanwhile - InMemoryCheckpointer wakes on its own commits, and a database store can use its notification channel (PostgreSQL's LISTEN/NOTIFY). Such a store must wake for commits made through every store over the same storage, not only its own, and still return within about timeout when none lands.
Errors are eventsAfter's - GraphError.ReplayUnavailable when afterSeq is behind the replay floor. Interrupting the waiting thread ends the wait with Left(CancelledError), the thread's interrupt flag set again.
'''The reading thread may be interrupted at any point.''' A subscription's watch calls this on a thread of its own, and Subscription.cancel - or the subscription's end - interrupts that thread whether it is waiting or in the middle of a read, so the interrupt can land inside the store's own I/O (eventsAfter included). A store must take that as the end of this one call, never as damage to itself: release what the call held (a pooled connection, a lock), and stay usable by other threads and later calls. Beware drivers whose I/O goes through a java.nio.channels.InterruptibleChannel (a FileChannel or SocketChannel): an interrupt during a read or write closes the channel, with ClosedByInterruptException, so a store that shares one channel - one connection - between threads would lose it for all of them. Such a store reads on a connection of its own for the waiting call, or discards and replaces a connection an interrupt has closed.