osyra_ome.adapters.langgraph.checkpoint
Private pre-release / network boundary: The currently usable customer surface is offline verification of an existing signed artifact against an out-of-band trust anchor. Memory/network RPC APIs and host snippets on this page describe protocol and interface shape; the public Edge bridge, package publication, and release acceptance remain pending. Do not treat
api.osyra.ai:443examples as a released customer route.
LangGraph :class:`BaseCheckpointSaver` backed by the OME memory namespace.
LangGraph's checkpoint mechanism persists the state of a stateful graph between executions so the run can pause, resume, and time-travel. The abstract :class:`langgraph.checkpoint.base.BaseCheckpointSaver` defines the contract; common in-tree implementations include MemorySaver, SqliteSaver, and PostgresSaver. :class:`OmeCheckpointSaver` adds a cryptographically-attested option backed by OME claims.
Each ``put()`` writes one OME :class:`~osyra_ome.Claim` carrying the serialized checkpoint payload (config + checkpoint dict + metadata). Each ``get_tuple()`` replays the most recent claim by re-fetching it through :meth:`osyra_ome.OmeClient.memory.get_claim`.
.. warning:: **EXPERIMENTAL / PREVIEW in v0.1 — write works, read-back does not.** The v0.1 read wire (``ListClaims`` projection) does NOT transport claim bodies, so a fresh process resuming a thread cannot reconstruct the checkpoint it wrote. Rather than silently report "no checkpoint" (which would discard the thread's history — ADAPT2-1 / ENG-2225), the read methods (``get_tuple`` / ``list`` / ``aget_tuple`` / ``alist``) FAIL CLOSED with :class:`~osyra_ome.errors.OmeBodyUnavailableError` when a known claim resolves with an empty body. Durable resume/time-travel is deferred to v0.2 via a dedicated ``GetClaimBody`` gRPC RPC. Until then, treat this saver as write-only attestation; do not depend on read-back.
.. note:: **ENG-2413 verify-on-read body binding is PENDING for this adapter.** The body reads here call ``get_claim_body(..., verify_on_read=False)`` — they do NOT bind the returned body to a claim signature yet. The OLD ``SHA-256(body) == response.content_hash`` check was REMOVED because that ``content_hash`` is UNSIGNED and server-provided (forgeable) — it gave false assurance. ``verify_on_read=False`` is the HONEST opt-out: the forgeable field is no longer consulted at all. Full signature-bound verify-on-read (the langchain adapter pattern: cache the :class:`~osyra_ome.types.SignedClaim` from ``write_claim_signed`` and bind each read to its attested ``ch``) requires either same-process SignedClaim caching OR a v0.2 ``GetClaimBody`` that carries the signing preimage — complicated here by ``adopt_claim_ids`` cross-process replay (an adopted id has no in-process preimage source). Tracked as an ENG-2413 follow-up.
Thread isolation: the LangGraph ``thread_id`` from ``config["configurable"] ["thread_id"]`` is encoded into the OME ``subject_id`` (``did:web:thread.osyra.ai/<thread_id>``) so all checkpoints belonging to one thread are queryable together.
Example::
from osyra_ome import OmeClient
from osyra_ome.adapters.langgraph import OmeCheckpointSaver
from langgraph.graph import StateGraph
saver = OmeCheckpointSaver(client=OmeClient(...), workspace_id="ws-prod")
graph_builder = StateGraph(MyState)
# ... define nodes/edges ...
graph = graph_builder.compile(checkpointer=saver)
graph.invoke({"input": "x"}, {"configurable": {"thread_id": "t-42"}})Classes
class OmeAsyncCheckpointSaver
Bases: BaseCheckpointSaver
Async LangGraph checkpoint saver backed by OME.
Wraps :class:`osyra_ome.OmeAsyncClient`. Mirrors the sync class method-for-method via aput / aget_tuple / alist / aput_writes.
Methods
__init__
def __init__(self, *, client: OmeAsyncClient, workspace_id: str | None = None) -> Noneadelete_thread (async)
async def adelete_thread(self, thread_id: str) -> Noneadopt_claim_ids
def adopt_claim_ids(self, thread_id: str, claim_ids: Sequence[str]) -> Noneaget_tuple (async)
async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | Nonealist (async)
async def alist(self, config: RunnableConfig | None, *, filter: dict[str, Any] | None = None, before: RunnableConfig | None = None, limit: int | None = None) -> AsyncIterator[CheckpointTuple]aput (async)
async def aput(self, config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: dict[str, Any]) -> RunnableConfigaput_writes (async)
async def aput_writes(self, config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str, task_path: str = '') -> Nonethread_claim_ids
def thread_claim_ids(self, thread_id: str) -> tuple[str, ...]workspace_id @property
def workspace_id(self) -> strclass OmeCheckpointSaver
Bases: BaseCheckpointSaver
LangGraph checkpoint saver backed by the OME memory namespace.
Args:
client: a :class:osyra_ome.OmeClient.
workspace_id: OME workspace UUID-like id; if omitted, defers
to the workspace_id bound on the wrapped client.
State model:
The saver keeps a per-thread_id list of claim ids in
memory. :meth:get_tuple returns the most recent
checkpoint; :meth:list returns all checkpoints for the
thread (newest first). Cross-process replay is supported via
the OME service — but a fresh saver instance does NOT
automatically know about claims written by a previous
instance for the same thread. Customers running multi-
process inference farms should wire their own
thread_id → claim_ids mapping (typically Redis) and
seed the saver with :meth:adopt_claim_ids.
Methods
__init__
def __init__(self, *, client: OmeClient, workspace_id: str | None = None) -> Noneadopt_claim_ids
def adopt_claim_ids(self, thread_id: str, claim_ids: Sequence[str]) -> NoneSeed the saver with externally-known CHECKPOINT claim ids.
Used to bridge multi-process inference where a prior worker wrote checkpoints and a new worker resuming the thread needs to know about them. Order matters: pass the claim ids in insertion order (oldest first) so :meth:`get_tuple` returns the most recent one as the last element.
ADAPT2-7: merges (append-and-dedupe in insertion order) rather than overwriting, so seeding externally-known ids does NOT drop checkpoints this saver instance already wrote for the thread.
delete_thread
def delete_thread(self, thread_id: str) -> Noneget_tuple
def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | Nonelist
def list(self, config: RunnableConfig | None, *, filter: dict[str, Any] | None = None, before: RunnableConfig | None = None, limit: int | None = None) -> Iterator[CheckpointTuple]put
def put(self, config: RunnableConfig, checkpoint: Checkpoint, metadata: CheckpointMetadata, new_versions: dict[str, Any]) -> RunnableConfigput_writes
def put_writes(self, config: RunnableConfig, writes: Sequence[tuple[str, Any]], task_id: str, task_path: str = '') -> Nonethread_claim_ids
def thread_claim_ids(self, thread_id: str) -> tuple[str, ...]workspace_id @property
def workspace_id(self) -> str