A durable org.llm4s.agent.graph.Checkpointer in a PostgreSQL database: threads, their checkpoint histories, pending writes, event logs, effect records and run claims, shared by every runtime on every machine that opens a store over the same database and schema. It is the store for running one agent service on several machines (Stage 5, design 4.50).
Open it with PostgresCheckpointer.open, over a pool of its own, or PostgresCheckpointer.create, over a DataSource the caller already has; 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 transaction, which first locks the thread's row (SELECT ... FOR UPDATE) and then checks the claim's token, a Commit.fence's claim on another thread, and the parent checkpoint, before it allocates event numbers from the row and writes. A call that locks two threads' rows
a fenced commit or delete - locks them in the order of their ids, so two such calls never deadlock.
'''The database decides time.''' A claim's expiry and a retention policy's age are judged by the database server's clock (clock_timestamp(), read once the call holds its locks), never a host's: skew between the machines sharing the store changes nothing. Age is measured from the commit time the store stamps, by that clock, on every checkpoint and event row - never from the producer's Checkpoint.createdAt or event timestamp, which stay as the producer's record. The clock a runtime is given still schedules its renewals.
'''Fencing tokens''' come from one sequence for the whole schema, so they increase across restarts, releases, takeovers and deleteThread, and are never issued twice.
'''History, retention, deletes and children''' behave exactly as in the SQLite store: every superseded checkpoint is kept until prune, retractTurn or deleteThread removes it; a deleted thread leaves a tombstone of its event numbering; a parent's children are listed in the order their first commits landed.
'''Data only.''' Checkpoints, pending writes, events and effect records are stored as JSON text and decoded on read, so nothing executable is stored.
'''Change notification by polling''' for now: a subscription reads the log once per WatchPolicy.pollInterval (the Checkpointer.awaitEventsAfter default). LISTEN/NOTIFY replaces it in a later slice.
Reads run in a REPEATABLE READ, read-only transaction, so what one call reads is one snapshot; they never wait for a writer.
'''Every call returns within a bound''', as the Checkpointer contract requires (#1919): the pool's connection timeout plus PostgresCheckpointerConfig.callTimeout, plus the time a rollback takes. The call timeout is set on the server for the transaction (statement_timeout, lock_timeout, idle_in_transaction_session_timeout) and enforced on the client as well: a call that reaches it has its connection aborted and discarded from the pool, so a stalled network cannot hold it either. A call that reaches the bound fails with a ProcessingError. Like any networked store, a call that times out while its COMMIT is in flight cannot know whether the commit landed; the claim's fencing keeps that safe, as a later commit on a stale view is refused.
'''Interrupts.''' An interrupt flag already set when a call begins decides nothing: it is cleared for the call and set again when the call returns. An interrupt that lands during a call (on a virtual thread, it closes the connection's socket) ends that call with CancelledError or a failed result, and the connection is discarded, so the store stays usable by every other thread.
Every failure of the database itself is a ProcessingError, and so is any call after close. Safe to call from any number of threads: each call takes a connection of its own from the pool.
Up to limit of parent's direct children (design 4.41), oldest first: every thread whose latest checkpoint's Checkpoint.origin names parent, in the order each first committed a checkpoint with that origin, starting after child after when it is given, so a caller pages through them by passing the last id of each page. A thread deleteThread removed is not listed, nor is a thread whose latest checkpoint names another parent or none. An after the store does not know as a child of parent is Left(ValidationError); a child deleted since it was listed keeps its place as long as its tombstone does (deleteThread), so a page boundary survives retention deleting it. A thread with no children has none; limit below 1 reads none. Needs no claim, and reads no tenant: the runtime checks the parent's tenant (GraphRuntime.children).
Up to limit of parent's direct children (design 4.41), oldest first: every thread whose latest checkpoint's Checkpoint.origin names parent, in the order each first committed a checkpoint with that origin, starting after child after when it is given, so a caller pages through them by passing the last id of each page. A thread deleteThread removed is not listed, nor is a thread whose latest checkpoint names another parent or none. An after the store does not know as a child of parent is Left(ValidationError); a child deleted since it was listed keeps its place as long as its tombstone does (deleteThread), so a page boundary survives retention deleting it. A thread with no children has none; limit below 1 reads none. Needs no claim, and reads no tenant: the runtime checks the parent's tenant (GraphRuntime.children).
Claims threadId for request.holder, recording request.tenantId with the claim, 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(threadId, latest, holder)) with the thread's latest checkpoint id and nothing changed. holder is GraphError.BusyHolder.Run(holder) only when the live claim's tenant is request.tenantId (None and Some differ), and GraphError.BusyHolder.Unnamed otherwise - also for a claim that recorded no tenant, such as one persisted before claims carried one - so that a refusal never reveals another tenant's RunId; a store never returns GraphError.BusyHolder.Lapsed. A thread need not have a checkpoint to be claimed. A claim that replaces an expired one nobody released names that claim's holder as RunClaim.takenOverFrom, under the same rule: only if it was of request.tenantId, and None otherwise.
Claims threadId for request.holder, recording request.tenantId with the claim, 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(threadId, latest, holder)) with the thread's latest checkpoint id and nothing changed. holder is GraphError.BusyHolder.Run(holder) only when the live claim's tenant is request.tenantId (None and Some differ), and GraphError.BusyHolder.Unnamed otherwise - also for a claim that recorded no tenant, such as one persisted before claims carried one - so that a refusal never reveals another tenant's RunId; a store never returns GraphError.BusyHolder.Lapsed. A thread need not have a checkpoint to be claimed. A claim that replaces an expired one nobody released names that claim's holder as RunClaim.takenOverFrom, under the same rule: only if it was of request.tenantId, and None otherwise.
Closes the store, and its pool if it owns one. Later calls return a ProcessingError; closing again does nothing. A claim this store granted stays in the database until it is released through another store or expires.
Closes the store, and its pool if it owns one. Later calls return a ProcessingError; closing again does nothing. A claim this store granted stays in the database until it is released through another store or expires.
Applies commit atomically, fenced by its token, and returns its events with their sequence numbers, each recording that token. A Commit.fence of (other, otherToken) fences it by other's claim too, checked in the same transaction or under the same lock as the write, after the thread's own claim: otherwise Left(GraphError.StaleClaim(other, otherToken, otherCurrent)) with nothing written.
Applies commit atomically, fenced by its token, and returns its events with their sequence numbers, each recording that token. A Commit.fence of (other, otherToken) fences it by other's claim too, checked in the same transaction or under the same lock as the write, after the thread's own claim: otherwise Left(GraphError.StaleClaim(other, otherToken, otherCurrent)) with nothing written.
Deletes the thread in one transaction, which first locks its row (and a fence thread's, in id order) and checks that token is its current claim - so the check and the delete see one state of the thread. The row stays, emptied, as a tombstone once the thread has numbered an event, so that a new thread of the id numbers on past it.
Deletes the thread in one transaction, which first locks its row (and a fence thread's, in id order) and checks that token is its current claim - so the check and the delete see one state of the thread. The row stays, emptied, as a tombstone once the thread has numbered an event, so that a new thread of the id numbers on past it.
Up to limit events with seq > afterSeq, ascending. afterSeq0 reads from the thread's first event, as does the number just before it; one behind the replay floor is GraphError.ReplayUnavailable, and one inside the log of a deleted thread of the id (Checkpointer.deleted) is GraphError.ThreadDeleted.
Up to limit events with seq > afterSeq, ascending. afterSeq0 reads from the thread's first event, as does the number just before it; one behind the replay floor is GraphError.ReplayUnavailable, and one inside the log of a deleted thread of the id (Checkpointer.deleted) is GraphError.ThreadDeleted.
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 transaction holding the thread's row, so it sees and changes one state of the thread; it takes no claim, as the contract says. Age is judged by the database's clock against the time, by that clock, each checkpoint and event was committed - never the producer's createdAt or timestamp, which a writer with a skewed clock would make too old or too young (rule 1).
Removes what policy does not keep, in one transaction holding the thread's row, so it sees and changes one state of the thread; it takes no claim, as the contract says. Age is judged by the database's clock against the time, by that clock, each checkpoint and event was committed - never the producer's createdAt or timestamp, which a writer with a skewed clock would make too old or too young (rule 1).
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.