harnest.checkpoint

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`A2ATaskConflictError`	Class
`A2ATaskCursorError`	Class
`A2ATaskPage`	Class
`A2ATaskPersistence`	Class
`A2ATaskRecord`	Class
`ADKStore`	Class
`CheckpointAuthority`	Class
`CheckpointConflictError`	Class
`CheckpointError`	Class
`CheckpointRecord`	Class
`CheckpointStore`	Class
`CheckpointWrite`	Class
`HarnestStore`	Class
`LangGraphStore`	Class
`MemoryStore`	Class
`PendingAction`	Class
`RunRecord`	Class
`RunScope`	Class
`RunStatus`	Value or type alias
`checkpoint_metadata`	Function

A2ATaskConflictError

An A2A task mutation conflicts with its durable ownership.

Example

from harnest.checkpoint import A2ATaskConflictError

try:
    ...
except A2ATaskConflictError as error:
    ...

A2ATaskCursorError

An A2A list cursor does not identify a task in the selected scope.

Example

from harnest.checkpoint import A2ATaskCursorError

try:
    ...
except A2ATaskCursorError as error:
    ...

A2ATaskPage

One bounded durable task page plus its pre-pagination total.

A2ATaskPage(records: tuple[A2ATaskRecord, ...], total_size: int) -> None

Fields (— means no declared default):

Field	Type	Default
`records`	`tuple[A2ATaskRecord, ...]`	`—`
`total_size`	`int`	`—`

Parameters and example

Parameter	Type	Default
`records`	`tuple[A2ATaskRecord, ...]`	Required
`total_size`	`int`	Required

from harnest.checkpoint import A2ATaskPage

page = A2ATaskPage(records=..., total_size=...)

A2ATaskPersistence

Optional indexed A2A persistence implemented by built-in stores.

A2ATaskPersistence(*args, **kwargs)

Example

from harnest.checkpoint import A2ATaskPersistence

class MyA2ATaskPersistence(A2ATaskPersistence):
    ...

A2ATaskPersistence.put_a2a_task

async A2ATaskPersistence.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord

Create or update one owner-scoped A2A task snapshot.

Parameters and example

Parameter	Type	Default
`record`	`A2ATaskRecord`	Required

from harnest.checkpoint import A2ATaskPersistence

persistence: A2ATaskPersistence
record = await persistence.put_a2a_task(record=...)

A2ATaskPersistence.get_a2a_task

async A2ATaskPersistence.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None

Read one A2A task through its complete ownership key.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.checkpoint import A2ATaskPersistence

persistence: A2ATaskPersistence
record = await persistence.get_a2a_task(application_id=..., user_id=..., task_id=...)

A2ATaskPersistence.list_a2a_tasks

async A2ATaskPersistence.list_a2a_tasks(
    *,
    application_id: str,
    user_id: str,
    context_id: str | None = None,
    status: int | None = None,
    status_timestamp_after: str | None = None,
    cursor_task_id: str | None = None,
    limit: int = 51,
) -> A2ATaskPage

Return a filtered, cursor-based page of owner-scoped A2A tasks.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`context_id`	`str \| None`	`None`
`status`	`int \| None`	`None`
`status_timestamp_after`	`str \| None`	`None`
`cursor_task_id`	`str \| None`	`None`
`limit`	`int`	`51`

from harnest.checkpoint import A2ATaskPersistence

persistence: A2ATaskPersistence
page = await persistence.list_a2a_tasks(application_id=..., user_id=...)

A2ATaskPersistence.delete_a2a_task

async A2ATaskPersistence.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool

Delete one owner-scoped A2A task and report whether it existed.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.checkpoint import A2ATaskPersistence

persistence: A2ATaskPersistence
result = await persistence.delete_a2a_task(application_id=..., user_id=..., task_id=...)

A2ATaskRecord

Transport-neutral durable snapshot of one protobuf A2A task.

A2ATaskRecord(
    application_id: str,
    user_id: str,
    task_id: str,
    context_id: str,
    status: int,
    payload: bytes,
    status_timestamp: str | None = None,
    created_at: str = ...,
    updated_at: str = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`application_id`	`str`	`—`
`user_id`	`str`	`—`
`task_id`	`str`	`—`
`context_id`	`str`	`—`
`status`	`int`	`—`
`payload`	`bytes`	`—`
`status_timestamp`	`str \| None`	`None`
`created_at`	`str`	`<lambda>()`
`updated_at`	`str`	`<lambda>()`

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required
`context_id`	`str`	Required
`status`	`int`	Required
`payload`	`bytes`	Required
`status_timestamp`	`str \| None`	`None`
`created_at`	`str`	`...`
`updated_at`	`str`	`...`

from harnest.checkpoint import A2ATaskRecord

record = A2ATaskRecord(
    application_id=...,
    user_id=...,
    task_id=...,
    context_id=...,
    status=...,
    payload=...,
)

ADKStore

Explicit lifecycle ownership wrapper for native ADK persistence.

ADKStore(session_service: Any) -> None

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'adk'`
`framework`	`str \| None`	`'adk'`
`schema_id`	`str`	`'adk-native'`

Parameters and example

Parameter	Type	Default
`session_service`	`Any`	Required

from harnest.checkpoint import ADKStore

store = ADKStore(session_service=...)

ADKStore.start

async ADKStore.start() -> None

Initialize the wrapped ADK session service when supported.

Example

from harnest.checkpoint import ADKStore

store: ADKStore
await store.start()

ADKStore.close

async ADKStore.close() -> None

Release resources owned by the wrapped ADK session service.

Example

from harnest.checkpoint import ADKStore

store: ADKStore
await store.close()

CheckpointAuthority

Typed lifecycle value identifying one checkpoint authority.

CheckpointAuthority()

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`—`
`framework`	`str \| None`	`—`
`schema_id`	`str`	`—`

Example

from harnest.checkpoint import CheckpointAuthority

authority = CheckpointAuthority()

CheckpointConflictError

A compare-and-swap or run exclusivity condition failed.

Example

from harnest.checkpoint import CheckpointConflictError

try:
    ...
except CheckpointConflictError as error:
    ...

CheckpointError

Base error for checkpoint persistence and ownership.

Example

from harnest.checkpoint import CheckpointError

try:
    ...
except CheckpointError as error:
    ...

CheckpointRecord

Opaque framework checkpoint with portable lineage and CAS revision.

CheckpointRecord(
    run_id: str,
    checkpoint_id: str,
    namespace: str,
    framework: Literal['adk', 'langgraph'],
    type_name: str,
    payload: bytes,
    metadata_type: str,
    metadata: bytes,
    versions_type: str,
    versions: bytes,
    parent_checkpoint_id: str | None = None,
    revision: int = 0,
    created_at: str = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`run_id`	`str`	`—`
`checkpoint_id`	`str`	`—`
`namespace`	`str`	`—`
`framework`	`Literal['adk', 'langgraph']`	`—`
`type_name`	`str`	`—`
`payload`	`bytes`	`—`
`metadata_type`	`str`	`—`
`metadata`	`bytes`	`—`
`versions_type`	`str`	`—`
`versions`	`bytes`	`—`
`parent_checkpoint_id`	`str \| None`	`None`
`revision`	`int`	`0`
`created_at`	`str`	`<lambda>()`

Parameters and example

Parameter	Type	Default
`run_id`	`str`	Required
`checkpoint_id`	`str`	Required
`namespace`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required
`type_name`	`str`	Required
`payload`	`bytes`	Required
`metadata_type`	`str`	Required
`metadata`	`bytes`	Required
`versions_type`	`str`	Required
`versions`	`bytes`	Required
`parent_checkpoint_id`	`str \| None`	`None`
`revision`	`int`	`0`
`created_at`	`str`	`...`

from harnest.checkpoint import CheckpointRecord

record = CheckpointRecord(
    run_id=...,
    checkpoint_id=...,
    namespace=...,
    framework=...,
    type_name=...,
    payload=...,
    metadata_type=...,
    metadata=...,
    versions_type=...,
    versions=...,
)

CheckpointStore

Storage contract for resumable, in-progress framework execution.

CheckpointStore(*args, **kwargs)

Example

from harnest.checkpoint import CheckpointStore

class MyCheckpointStore(CheckpointStore):
    ...

CheckpointStore.start

async CheckpointStore.start() -> None

Initialize resources required by the checkpoint store.

Example

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
await store.start()

CheckpointStore.begin_run

async CheckpointStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord

Create or recover the active checkpoint run for a session.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
record = await store.begin_run(
    application_id=...,
    user_id=...,
    session_id=...,
    run_id=...,
    framework=...,
)

CheckpointStore.get_run

async CheckpointStore.get_run(*, scope: RunScope) -> RunRecord | None

Read a checkpoint run through its complete ownership scope.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
record = await store.get_run(scope=...)

CheckpointStore.get_checkpoint

async CheckpointStore.get_checkpoint(
    *,
    scope: RunScope,
    checkpoint_id: str | None = None,
    namespace: str = '',
) -> CheckpointRecord | None

Read an exact or latest checkpoint within an owned run.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str \| None`	`None`
`namespace`	`str`	`''`

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
record = await store.get_checkpoint(scope=...)

CheckpointStore.list_checkpoints

CheckpointStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]

Stream a bounded checkpoint history for an owned run.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`namespace`	`str \| None`	`None`
`before`	`str \| None`	`None`
`limit`	`int`	`20`

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
result = store.list_checkpoints(scope=...)

CheckpointStore.put

async CheckpointStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord

Persist a checkpoint when its expected revision still matches.

Parameters and example

Parameter	Type	Default
`checkpoint`	`CheckpointRecord`	Required
`scope`	`RunScope`	Required
`expected_revision`	`int \| None`	Required

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
record = await store.put(checkpoint=..., scope=..., expected_revision=...)

CheckpointStore.put_writes

async CheckpointStore.put_writes(
    *,
    scope: RunScope,
    checkpoint_id: str,
    writes: Sequence[CheckpointWrite],
) -> None

Persist pending task writes for one checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required
`writes`	`Sequence[CheckpointWrite]`	Required

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
await store.put_writes(scope=..., checkpoint_id=..., writes=...)

CheckpointStore.get_writes

async CheckpointStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]

Read pending writes for one owned checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
result = await store.get_writes(scope=..., checkpoint_id=...)

CheckpointStore.get_writes_batch

async CheckpointStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]

Read pending writes for several owned checkpoints in one call.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_ids`	`Sequence[str]`	Required

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
result = await store.get_writes_batch(scope=..., checkpoint_ids=...)

CheckpointStore.transition

async CheckpointStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord

Apply one compare-and-swap run-state transition.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`expected_status`	`RunStatus`	Required
`status`	`RunStatus`	Required
`pending_action`	`PendingAction \| None`	`None`

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
record = await store.transition(scope=..., expected_status=..., status=...)

CheckpointStore.delete_run

async CheckpointStore.delete_run(*, scope: RunScope) -> None

Delete an owned run and all of its checkpoint state.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
await store.delete_run(scope=...)

CheckpointStore.close

async CheckpointStore.close() -> None

Release resources owned by the checkpoint store.

Example

from harnest.checkpoint import CheckpointStore

store: CheckpointStore
await store.close()

CheckpointWrite

Opaque framework write attached to a checkpoint task.

CheckpointWrite(task_id: str, channel: str, type_name: str, payload: bytes, task_path: str = '') -> None

Fields (— means no declared default):

Field	Type	Default
`task_id`	`str`	`—`
`channel`	`str`	`—`
`type_name`	`str`	`—`
`payload`	`bytes`	`—`
`task_path`	`str`	`''`

Parameters and example

Parameter	Type	Default
`task_id`	`str`	Required
`channel`	`str`	Required
`type_name`	`str`	Required
`payload`	`bytes`	Required
`task_path`	`str`	`''`

from harnest.checkpoint import CheckpointWrite

write = CheckpointWrite(task_id=..., channel=..., type_name=..., payload=...)

HarnestStore

Base for framework-neutral stores owned by the Harnest runtime.

HarnestStore()

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'harnest'`
`framework`	`str \| None`	`None`
`schema_id`	`str`	`'harnest-checkpoint/v1'`

Example

from harnest.checkpoint import HarnestStore

store = HarnestStore()

HarnestStore.as_langgraph_checkpointer

HarnestStore.as_langgraph_checkpointer() -> Any

Return one stable native saver for advanced LangGraph construction.

Example

from harnest.checkpoint import HarnestStore

store: HarnestStore
result = store.as_langgraph_checkpointer()

LangGraphStore

Explicit lifecycle ownership wrapper for a native LangGraph saver.

LangGraphStore(checkpointer: Any) -> None

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'langgraph'`
`framework`	`str \| None`	`'langgraph'`
`schema_id`	`str`	`'langgraph-native'`

Parameters and example

Parameter	Type	Default
`checkpointer`	`Any`	Required

from harnest.checkpoint import LangGraphStore

store = LangGraphStore(checkpointer=...)

LangGraphStore.config_specs

Expose the wrapped saver configuration fields to LangGraph.

Example

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
value = store.config_specs

LangGraphStore.get_tuple

LangGraphStore.get_tuple(config: Any) -> Any

Delegate synchronous checkpoint lookup to the wrapped saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
result = store.get_tuple(config=...)

LangGraphStore.list

LangGraphStore.list(config: Any, **options: Any) -> Any

Delegate synchronous checkpoint listing to the wrapped saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required
`**options`	`Any`	Optional

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
result = store.list(config=...)

LangGraphStore.put

LangGraphStore.put(config: Any, checkpoint: Any, metadata: Any, new_versions: Any) -> Any

Delegate synchronous checkpoint persistence to the wrapped saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required
`checkpoint`	`Any`	Required
`metadata`	`Any`	Required
`new_versions`	`Any`	Required

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
result = store.put(config=..., checkpoint=..., metadata=..., new_versions=...)

LangGraphStore.put_writes

LangGraphStore.put_writes(
    config: Any,
    writes: Sequence[tuple[str, Any]],
    task_id: str,
    task_path: str = '',
) -> None

Delegate synchronous pending-write persistence to the wrapped saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required
`writes`	`Sequence[tuple[str, Any]]`	Required
`task_id`	`str`	Required
`task_path`	`str`	`''`

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
store.put_writes(config=..., writes=..., task_id=...)

LangGraphStore.delete_thread

LangGraphStore.delete_thread(thread_id: str) -> None

Delete all checkpoints for a LangGraph thread.

Parameters and example

Parameter	Type	Default
`thread_id`	`str`	Required

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
store.delete_thread(thread_id=...)

LangGraphStore.aget_tuple

async LangGraphStore.aget_tuple(config: Any) -> Any

Delegate asynchronous checkpoint lookup to the wrapped saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
result = await store.aget_tuple(config=...)

LangGraphStore.alist

LangGraphStore.alist(config: Any, **options: Any) -> AsyncIterator[Any]

Stream checkpoints from the wrapped asynchronous saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required
`**options`	`Any`	Optional

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
async for item in store.alist(config=...):
    ...

LangGraphStore.aput

async LangGraphStore.aput(config: Any, checkpoint: Any, metadata: Any, new_versions: Any) -> Any

Delegate asynchronous checkpoint persistence to the wrapped saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required
`checkpoint`	`Any`	Required
`metadata`	`Any`	Required
`new_versions`	`Any`	Required

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
result = await store.aput(config=..., checkpoint=..., metadata=..., new_versions=...)

LangGraphStore.aput_writes

async LangGraphStore.aput_writes(
    config: Any,
    writes: Sequence[tuple[str, Any]],
    task_id: str,
    task_path: str = '',
) -> None

Delegate asynchronous pending-write persistence to the wrapped saver.

Parameters and example

Parameter	Type	Default
`config`	`Any`	Required
`writes`	`Sequence[tuple[str, Any]]`	Required
`task_id`	`str`	Required
`task_path`	`str`	`''`

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
await store.aput_writes(config=..., writes=..., task_id=...)

LangGraphStore.adelete_thread

async LangGraphStore.adelete_thread(thread_id: str) -> None

Asynchronously delete all checkpoints for a LangGraph thread.

Parameters and example

Parameter	Type	Default
`thread_id`	`str`	Required

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
await store.adelete_thread(thread_id=...)

LangGraphStore.get_next_version

LangGraphStore.get_next_version(current: Any, channel: Any) -> Any

Ask the wrapped saver to derive the next channel version.

Parameters and example

Parameter	Type	Default
`current`	`Any`	Required
`channel`	`Any`	Required

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
result = store.get_next_version(current=..., channel=...)

LangGraphStore.start

async LangGraphStore.start() -> None

Initialize the wrapped saver when it owns a start hook.

Example

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
await store.start()

LangGraphStore.close

async LangGraphStore.close() -> None

Release resources owned by the wrapped saver.

Example

from harnest.checkpoint import LangGraphStore

store: LangGraphStore
await store.close()

MemoryStore

Process-local session, checkpoint, and continuation store with atomic CAS.

MemoryStore() -> None

Fields (— means no declared default):

Field	Type	Default
`owner`	`CheckpointOwner`	`'harnest'`
`framework`	`str \| None`	`None`
`schema_id`	`str`	`'harnest-checkpoint/v1'`

Example

from harnest.checkpoint import MemoryStore

store = MemoryStore()

MemoryStore.start

async MemoryStore.start() -> None

Reject reopening because closed development state is intentionally lost.

Example

from harnest.checkpoint import MemoryStore

store: MemoryStore
await store.start()

MemoryStore.create

async MemoryStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord

Create one process-local tenant-scoped session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state`	`Mapping[str, Any]`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.create(session_id=..., user_id=..., state=...)

MemoryStore.get

async MemoryStore.get(*, session_id: str, user_id: str) -> SessionRecord | None

Read one process-local tenant-scoped session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.get(session_id=..., user_id=...)

MemoryStore.list

async MemoryStore.list(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]

List an ordered keyset page without materializing unrelated users.

Parameters and example

Parameter	Type	Default
`user_id`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int \| None`	`None`

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.list(user_id=...)

MemoryStore.update

async MemoryStore.update(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None

Merge a state delta into a process-local session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required
`state_delta`	`Mapping[str, Any]`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.update(session_id=..., user_id=..., state_delta=...)

MemoryStore.delete

async MemoryStore.delete(*, session_id: str, user_id: str) -> bool

Delete one process-local session after its active lease exits.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.delete(session_id=..., user_id=...)

MemoryStore.acquire

MemoryStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]

Hold exclusive access to one process-local session.

Parameters and example

Parameter	Type	Default
`session_id`	`str`	Required
`user_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
async with store.acquire(session_id=..., user_id=...):
    ...

MemoryStore.close

async MemoryStore.close() -> None

Close both session and checkpoint sides of the shared store.

Example

from harnest.checkpoint import MemoryStore

store: MemoryStore
await store.close()

MemoryStore.as_langgraph_checkpointer

MemoryStore.as_langgraph_checkpointer() -> Any

Return one stable native saver for advanced LangGraph construction.

Example

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = store.as_langgraph_checkpointer()

MemoryStore.put_a2a_task

async MemoryStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord

Upsert one owner-scoped snapshot without allowing context reassignment.

Parameters and example

Parameter	Type	Default
`record`	`A2ATaskRecord`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.put_a2a_task(record=...)

MemoryStore.get_a2a_task

async MemoryStore.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None

Read one snapshot only through its complete ownership key.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)

MemoryStore.list_a2a_tasks

async MemoryStore.list_a2a_tasks(
    *,
    application_id: str,
    user_id: str,
    context_id: str | None = None,
    status: int | None = None,
    status_timestamp_after: str | None = None,
    cursor_task_id: str | None = None,
    limit: int = 51,
) -> A2ATaskPage

Return a stable inclusive-cursor page for the process-local backend.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`context_id`	`str \| None`	`None`
`status`	`int \| None`	`None`
`status_timestamp_after`	`str \| None`	`None`
`cursor_task_id`	`str \| None`	`None`
`limit`	`int`	`51`

from harnest.checkpoint import MemoryStore

store: MemoryStore
page = await store.list_a2a_tasks(application_id=..., user_id=...)

MemoryStore.delete_a2a_task

async MemoryStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool

Delete exactly one owner-scoped A2A snapshot.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`task_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)

MemoryStore.begin_run

async MemoryStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord

Claim the session for one idempotent invocation run.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.begin_run(
    application_id=...,
    user_id=...,
    session_id=...,
    run_id=...,
    framework=...,
)

MemoryStore.get_run

async MemoryStore.get_run(*, scope: RunScope) -> RunRecord | None

Read one process-local run through its complete ownership scope.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.get_run(scope=...)

MemoryStore.get_checkpoint

async MemoryStore.get_checkpoint(
    *,
    scope: RunScope,
    checkpoint_id: str | None = None,
    namespace: str = '',
) -> CheckpointRecord | None

Load an exact checkpoint or the newest checkpoint in a namespace.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str \| None`	`None`
`namespace`	`str`	`''`

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.get_checkpoint(scope=...)

MemoryStore.list_checkpoints

MemoryStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]

Yield a bounded newest-first checkpoint history.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`namespace`	`str \| None`	`None`
`before`	`str \| None`	`None`
`limit`	`int`	`20`

from harnest.checkpoint import MemoryStore

store: MemoryStore
async for item in store.list_checkpoints(scope=...):
    ...

MemoryStore.put

async MemoryStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord

Persist a checkpoint only when its expected revision still matches.

Parameters and example

Parameter	Type	Default
`checkpoint`	`CheckpointRecord`	Required
`scope`	`RunScope`	Required
`expected_revision`	`int \| None`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.put(checkpoint=..., scope=..., expected_revision=...)

MemoryStore.put_writes

async MemoryStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None

Add pending writes idempotently by task and channel.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required
`writes`	`Sequence[CheckpointWrite]`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
await store.put_writes(scope=..., checkpoint_id=..., writes=...)

MemoryStore.get_writes

async MemoryStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]

Read process-local pending writes for one owned checkpoint.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.get_writes(scope=..., checkpoint_id=...)

MemoryStore.get_writes_batch

async MemoryStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]

Load writes for several checkpoints without per-checkpoint calls.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`checkpoint_ids`	`Sequence[str]`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.get_writes_batch(scope=..., checkpoint_ids=...)

MemoryStore.transition

async MemoryStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord

Apply one valid compare-and-swap run-state transition.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`expected_status`	`RunStatus`	Required
`status`	`RunStatus`	Required
`pending_action`	`PendingAction \| None`	`None`

from harnest.checkpoint import MemoryStore

store: MemoryStore
record = await store.transition(scope=..., expected_status=..., status=...)

MemoryStore.suspend_continuation

async MemoryStore.suspend_continuation(*, record: Any, external_id: str) -> Any

Atomically register provider work and move its owned run to waiting.

Parameters and example

Parameter	Type	Default
`record`	`Any`	Required
`external_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.suspend_continuation(record=..., external_id=...)

MemoryStore.get_continuation

async MemoryStore.get_continuation(*, scope: RunScope, continuation_id: str) -> Any | None

Return one record only when every run ownership field matches.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`continuation_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.get_continuation(scope=..., continuation_id=...)

MemoryStore.get_provider_continuation

async MemoryStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> Any | None

Return an exact continuation together with its provider external ID.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.get_provider_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
)

MemoryStore.get_continuation_by_external_id

async MemoryStore.get_continuation_by_external_id(*, application_id: str, provider: str, external_id: str) -> Any | None

Resolve one provider identity through the same indexed ownership key.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`external_id`	`str`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.get_continuation_by_external_id(
    application_id=...,
    provider=...,
    external_id=...,
)

MemoryStore.list_pending_continuations

async MemoryStore.list_pending_continuations(
    *,
    application_id: str,
    provider: str,
    after: str | None = None,
    limit: int = 100,
) -> Sequence[Any]

Return a bounded provider page for local reconciliation.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`provider`	`str`	Required
`after`	`str \| None`	`None`
`limit`	`int`	`100`

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.list_pending_continuations(application_id=..., provider=...)

MemoryStore.resolve_continuation

async MemoryStore.resolve_continuation(
    *,
    scope: RunScope,
    provider: str,
    external_id: str,
    schema_id: str,
    result: Any = None,
    failure: Any = None,
) -> Any

Commit a provider result or failure with one status compare-and-swap.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`external_id`	`str`	Required
`schema_id`	`str`	Required
`result`	`Any`	`None`
`failure`	`Any`	`None`

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.resolve_continuation(
    scope=...,
    provider=...,
    external_id=...,
    schema_id=...,
)

MemoryStore.cancel_continuation

async MemoryStore.cancel_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
    failure: Any,
) -> Any

Fail one pending continuation and cancel its exact waiting run.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required
`failure`	`Any`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.cancel_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
    failure=...,
)

MemoryStore.claim_continuation

async MemoryStore.claim_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> Any

Claim an outcome and return the waiting run to running atomically.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.claim_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

MemoryStore.arm_continuation

async MemoryStore.arm_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> Any

Arm a provider wait after its native suspension is checkpointed.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required
`provider`	`str`	Required
`continuation_id`	`str`	Required
`expected_revision`	`int`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
result = await store.arm_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

MemoryStore.delete_run

async MemoryStore.delete_run(*, scope: RunScope) -> None

Delete a run and every private checkpoint owned by it.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest.checkpoint import MemoryStore

store: MemoryStore
await store.delete_run(scope=...)

PendingAction

Portable description of a durable wait without customer payloads.

PendingAction(
    type: Literal['human_approval', 'client_tool', 'external_continuation'],
    action_id: str,
    capability: str,
) -> None

Fields (— means no declared default):

Field	Type	Default
`type`	`Literal['human_approval', 'client_tool', 'external_continuation']`	`—`
`action_id`	`str`	`—`
`capability`	`str`	`—`

Parameters and example

Parameter	Type	Default
`type`	`Literal['human_approval', 'client_tool', 'external_continuation']`	Required
`action_id`	`str`	Required
`capability`	`str`	Required

from harnest.checkpoint import PendingAction

action = PendingAction(type=..., action_id=..., capability=...)

RunRecord

One framework execution associated with a committed public session.

RunRecord(
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
    status: RunStatus = 'running',
    revision: int = 0,
    pending_action: PendingAction | None = None,
    created_at: str = ...,
    updated_at: str = ...,
) -> None

Fields (— means no declared default):

Field	Type	Default
`application_id`	`str`	`—`
`user_id`	`str`	`—`
`session_id`	`str`	`—`
`run_id`	`str`	`—`
`framework`	`Literal['adk', 'langgraph']`	`—`
`status`	`RunStatus`	`'running'`
`revision`	`int`	`0`
`pending_action`	`PendingAction \| None`	`None`
`created_at`	`str`	`<lambda>()`
`updated_at`	`str`	`<lambda>()`

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required
`framework`	`Literal['adk', 'langgraph']`	Required
`status`	`RunStatus`	`'running'`
`revision`	`int`	`0`
`pending_action`	`PendingAction \| None`	`None`
`created_at`	`str`	`...`
`updated_at`	`str`	`...`

from harnest.checkpoint import RunRecord

record = RunRecord(
    application_id=...,
    user_id=...,
    session_id=...,
    run_id=...,
    framework=...,
)

RunRecord.scope

Return the complete ownership key required for later operations.

Example

from harnest.checkpoint import RunRecord

record: RunRecord
value = record.scope

RunScope

Complete ownership key required at the store boundary for one run.

RunScope(application_id: str, user_id: str, session_id: str, run_id: str) -> None

Fields (— means no declared default):

Field	Type	Default
`application_id`	`str`	`—`
`user_id`	`str`	`—`
`session_id`	`str`	`—`
`run_id`	`str`	`—`

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`session_id`	`str`	Required
`run_id`	`str`	Required

from harnest.checkpoint import RunScope

scope = RunScope(application_id=..., user_id=..., session_id=..., run_id=...)

RunStatus

RunStatus = Literal['running', 'waiting', 'completed', 'failed', 'cancelled']

Example

from harnest.checkpoint import RunStatus

checkpoint_metadata

Return immutable, non-secret ownership metadata for an artifact.

checkpoint_metadata(authority: CheckpointAuthority) -> dict[str, str]

Parameters and example

Parameter	Type	Default
`authority`	`CheckpointAuthority`	Required

from harnest.checkpoint import checkpoint_metadata

result = checkpoint_metadata(authority=...)

​Exports

​A2ATaskConflictError

​A2ATaskCursorError

​A2ATaskPage

​A2ATaskPersistence

​A2ATaskPersistence.put_a2a_task

​A2ATaskPersistence.get_a2a_task

​A2ATaskPersistence.list_a2a_tasks

​A2ATaskPersistence.delete_a2a_task

​A2ATaskRecord

​ADKStore

​ADKStore.start

​ADKStore.close

​CheckpointAuthority

​CheckpointConflictError

​CheckpointError

​CheckpointRecord

​CheckpointStore

​CheckpointStore.start

​CheckpointStore.begin_run

​CheckpointStore.get_run

​CheckpointStore.get_checkpoint

​CheckpointStore.list_checkpoints

​CheckpointStore.put

​CheckpointStore.put_writes

​CheckpointStore.get_writes

​CheckpointStore.get_writes_batch

​CheckpointStore.transition

​CheckpointStore.delete_run

​CheckpointStore.close

​CheckpointWrite

​HarnestStore

​HarnestStore.as_langgraph_checkpointer

​LangGraphStore

​LangGraphStore.config_specs

​LangGraphStore.get_tuple

​LangGraphStore.list

​LangGraphStore.put

​LangGraphStore.put_writes

​LangGraphStore.delete_thread

​LangGraphStore.aget_tuple

​LangGraphStore.alist

​LangGraphStore.aput

​LangGraphStore.aput_writes

​LangGraphStore.adelete_thread

​LangGraphStore.get_next_version

​LangGraphStore.start

​LangGraphStore.close

​MemoryStore

​MemoryStore.start

​MemoryStore.create

​MemoryStore.get

​MemoryStore.list

​MemoryStore.update

​MemoryStore.delete

​MemoryStore.acquire

​MemoryStore.close

​MemoryStore.as_langgraph_checkpointer

​MemoryStore.put_a2a_task

​MemoryStore.get_a2a_task

​MemoryStore.list_a2a_tasks

​MemoryStore.delete_a2a_task

​MemoryStore.begin_run

​MemoryStore.get_run

​MemoryStore.get_checkpoint

​MemoryStore.list_checkpoints

​MemoryStore.put

​MemoryStore.put_writes

​MemoryStore.get_writes

​MemoryStore.get_writes_batch

​MemoryStore.transition

​MemoryStore.suspend_continuation

​MemoryStore.get_continuation

​MemoryStore.get_provider_continuation

​MemoryStore.get_continuation_by_external_id

​MemoryStore.list_pending_continuations

​MemoryStore.resolve_continuation

​MemoryStore.cancel_continuation

​MemoryStore.claim_continuation

​MemoryStore.arm_continuation

Exports

A2ATaskConflictError

A2ATaskCursorError

A2ATaskPage

A2ATaskPersistence

A2ATaskPersistence.put_a2a_task

A2ATaskPersistence.get_a2a_task

A2ATaskPersistence.list_a2a_tasks

A2ATaskPersistence.delete_a2a_task

A2ATaskRecord

ADKStore

ADKStore.start

ADKStore.close

CheckpointAuthority

CheckpointConflictError

CheckpointError

CheckpointRecord

CheckpointStore

CheckpointStore.start

CheckpointStore.begin_run

CheckpointStore.get_run

CheckpointStore.get_checkpoint

CheckpointStore.list_checkpoints

CheckpointStore.put

CheckpointStore.put_writes

CheckpointStore.get_writes

CheckpointStore.get_writes_batch

CheckpointStore.transition

CheckpointStore.delete_run

CheckpointStore.close

CheckpointWrite

HarnestStore

HarnestStore.as_langgraph_checkpointer

LangGraphStore

LangGraphStore.config_specs

LangGraphStore.get_tuple

LangGraphStore.list

LangGraphStore.put

LangGraphStore.put_writes

LangGraphStore.delete_thread

LangGraphStore.aget_tuple

LangGraphStore.alist

LangGraphStore.aput

LangGraphStore.aput_writes

LangGraphStore.adelete_thread

LangGraphStore.get_next_version

LangGraphStore.start

LangGraphStore.close

MemoryStore

MemoryStore.start

MemoryStore.create

MemoryStore.get

MemoryStore.list

MemoryStore.update

MemoryStore.delete

MemoryStore.acquire

MemoryStore.close

MemoryStore.as_langgraph_checkpointer

MemoryStore.put_a2a_task

MemoryStore.get_a2a_task

MemoryStore.list_a2a_tasks

MemoryStore.delete_a2a_task

MemoryStore.begin_run

MemoryStore.get_run

MemoryStore.get_checkpoint

MemoryStore.list_checkpoints

MemoryStore.put

MemoryStore.put_writes

MemoryStore.get_writes

MemoryStore.get_writes_batch

MemoryStore.transition

MemoryStore.suspend_continuation

MemoryStore.get_continuation

MemoryStore.get_provider_continuation

MemoryStore.get_continuation_by_external_id

MemoryStore.list_pending_continuations

MemoryStore.resolve_continuation

MemoryStore.cancel_continuation

MemoryStore.claim_continuation

MemoryStore.arm_continuation