> ## Documentation Index
> Fetch the complete documentation index at: https://usefused.com/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# harnest.store

> Public Python signatures, fields, and methods.

[All Python modules](https://usefused.com/docs/harnest/reference/api/overview.md) · [Import and compatibility guidance](https://usefused.com/docs/harnest/build/python-imports.md)

## Exports

| Name | Kind |
| - | - |
| [`CustomStorage`](#customstorage) | Class |
| [`MemoryStore`](#memorystore) | Class |
| [`PostgresStore`](#postgresstore) | Class |
| [`RedisStore`](#redisstore) | Class |
| [`StorageRegistry`](#storageregistry) | Class |

## CustomStorage

Lifecycle-owned application storage exposed by a stable authored name.

```python theme={null}
CustomStorage(*args, **kwargs)
```

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import CustomStorage
>
> class MyCustomStorage(CustomStorage):
>     ...
> ```

### CustomStorage.start

```python theme={null}
async CustomStorage.start() -> None
```

Open the underlying storage resource.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import CustomStorage
>
> storage: CustomStorage
> await storage.start()
> ```

### CustomStorage.close

```python theme={null}
async CustomStorage.close() -> None
```

Release the underlying storage resource.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import CustomStorage
>
> storage: CustomStorage
> await storage.close()
> ```

## MemoryStore

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

```python theme={null}
MemoryStore() -> None
```

Fields (`—` means no declared default):

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

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store = MemoryStore()
> ```

### MemoryStore.start

```python theme={null}
async MemoryStore.start() -> None
```

Reject reopening because closed development state is intentionally lost.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> await store.start()
> ```

### MemoryStore.create

```python theme={null}
async MemoryStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord
```

Create one process-local tenant-scoped session.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `state` | `Mapping[str, Any]` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.create(session_id=..., user_id=..., state=...)
> ```

### MemoryStore.get

```python theme={null}
async MemoryStore.get(*, session_id: str, user_id: str) -> SessionRecord | None
```

Read one process-local tenant-scoped session.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.get(session_id=..., user_id=...)
> ```

### MemoryStore.list

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.list(user_id=...)
> ```

### MemoryStore.update

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `state_delta` | `Mapping[str, Any]` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.update(session_id=..., user_id=..., state_delta=...)
> ```

### MemoryStore.delete

```python theme={null}
async MemoryStore.delete(*, session_id: str, user_id: str) -> bool
```

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

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.delete(session_id=..., user_id=...)
> ```

### MemoryStore.acquire

```python theme={null}
MemoryStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]
```

Hold exclusive access to one process-local session.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> async with store.acquire(session_id=..., user_id=...):
>     ...
> ```

### MemoryStore.close

```python theme={null}
async MemoryStore.close() -> None
```

Close both session and checkpoint sides of the shared store.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> await store.close()
> ```

### MemoryStore.as\_langgraph\_checkpointer

```python theme={null}
MemoryStore.as_langgraph_checkpointer() -> Any
```

Return one stable native saver for advanced LangGraph construction.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = store.as_langgraph_checkpointer()
> ```

### MemoryStore.put\_a2a\_task

```python theme={null}
async MemoryStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord
```

Upsert one owner-scoped snapshot without allowing context reassignment.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `A2ATaskRecord` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.put_a2a_task(record=...)
> ```

### MemoryStore.get\_a2a\_task

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `task_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)
> ```

### MemoryStore.list\_a2a\_tasks

```python theme={null}
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.

> [accordion] 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` |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> page = await store.list_a2a_tasks(application_id=..., user_id=...)
> ```

### MemoryStore.delete\_a2a\_task

```python theme={null}
async MemoryStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool
```

Delete exactly one owner-scoped A2A snapshot.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `task_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)
> ```

### MemoryStore.begin\_run

```python theme={null}
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.

> [accordion] 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 |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.begin_run(
>     application_id=...,
>     user_id=...,
>     session_id=...,
>     run_id=...,
>     framework=...,
> )
> ```

### MemoryStore.get\_run

```python theme={null}
async MemoryStore.get_run(*, scope: RunScope) -> RunRecord | None
```

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

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.get_run(scope=...)
> ```

### MemoryStore.get\_checkpoint

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str \| None` | `None` |
> | `namespace` | `str` | `''` |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.get_checkpoint(scope=...)
> ```

### MemoryStore.list\_checkpoints

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `namespace` | `str \| None` | `None` |
> | `before` | `str \| None` | `None` |
> | `limit` | `int` | `20` |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> async for item in store.list_checkpoints(scope=...):
>     ...
> ```

### MemoryStore.put

```python theme={null}
async MemoryStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord
```

Persist a checkpoint only when its expected revision still matches.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `checkpoint` | `CheckpointRecord` | Required |
> | `scope` | `RunScope` | Required |
> | `expected_revision` | `int \| None` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.put(checkpoint=..., scope=..., expected_revision=...)
> ```

### MemoryStore.put\_writes

```python theme={null}
async MemoryStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None
```

Add pending writes idempotently by task and channel.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str` | Required |
> | `writes` | `Sequence[CheckpointWrite]` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> await store.put_writes(scope=..., checkpoint_id=..., writes=...)
> ```

### MemoryStore.get\_writes

```python theme={null}
async MemoryStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]
```

Read process-local pending writes for one owned checkpoint.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.get_writes(scope=..., checkpoint_id=...)
> ```

### MemoryStore.get\_writes\_batch

```python theme={null}
async MemoryStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]
```

Load writes for several checkpoints without per-checkpoint calls.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_ids` | `Sequence[str]` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.get_writes_batch(scope=..., checkpoint_ids=...)
> ```

### MemoryStore.transition

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `expected_status` | `RunStatus` | Required |
> | `status` | `RunStatus` | Required |
> | `pending_action` | `PendingAction \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> record = await store.transition(scope=..., expected_status=..., status=...)
> ```

### MemoryStore.suspend\_continuation

```python theme={null}
async MemoryStore.suspend_continuation(*, record: Any, external_id: str) -> Any
```

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

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `Any` | Required |
> | `external_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.suspend_continuation(record=..., external_id=...)
> ```

### MemoryStore.get\_continuation

```python theme={null}
async MemoryStore.get_continuation(*, scope: RunScope, continuation_id: str) -> Any | None
```

Return one record only when every run ownership field matches.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `continuation_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.get_continuation(scope=..., continuation_id=...)
> ```

### MemoryStore.get\_provider\_continuation

```python theme={null}
async MemoryStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> Any | None
```

Return an exact continuation together with its provider external ID.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.get_provider_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
> )
> ```

### MemoryStore.get\_continuation\_by\_external\_id

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `provider` | `str` | Required |
> | `external_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.get_continuation_by_external_id(
>     application_id=...,
>     provider=...,
>     external_id=...,
> )
> ```

### MemoryStore.list\_pending\_continuations

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `provider` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.list_pending_continuations(application_id=..., provider=...)
> ```

### MemoryStore.resolve\_continuation

```python theme={null}
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.

> [accordion] 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` |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.resolve_continuation(
>     scope=...,
>     provider=...,
>     external_id=...,
>     schema_id=...,
> )
> ```

### MemoryStore.cancel\_continuation

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
> | `failure` | `Any` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.cancel_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
>     failure=...,
> )
> ```

### MemoryStore.claim\_continuation

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.claim_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
> )
> ```

### MemoryStore.arm\_continuation

```python theme={null}
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.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> result = await store.arm_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
> )
> ```

### MemoryStore.delete\_run

```python theme={null}
async MemoryStore.delete_run(*, scope: RunScope) -> None
```

Delete a run and every private checkpoint owned by it.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
>
> ```python theme={null}
> from harnest.store import MemoryStore
>
> store: MemoryStore
> await store.delete_run(scope=...)
> ```

## PostgresStore

Durable combined store using asyncpg and one Harnest-owned schema.

The driver is imported lazily by `start`, keeping module discovery
independent from database connectivity.

```python theme={null}
PostgresStore(
    dsn: str,
    *,
    pool_options: Mapping[str, Any] | None = None,
    lease_pool_options: Mapping[str, Any] | None = None,
    setup_schema: bool = True,
) -> None
```

Fields (`—` means no declared default):

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

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `dsn` | `str` | Required |
> | `pool_options` | `Mapping[str, Any] \| None` | `None` |
> | `lease_pool_options` | `Mapping[str, Any] \| None` | `None` |
> | `setup_schema` | `bool` | `True` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store = PostgresStore(dsn=...)
> ```

### PostgresStore.as\_langgraph\_checkpointer

```python theme={null}
PostgresStore.as_langgraph_checkpointer() -> Any
```

Return one stable native saver for advanced LangGraph construction.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> result = store.as_langgraph_checkpointer()
> ```

### PostgresStore.start

```python theme={null}
async PostgresStore.start() -> None
```

Open the pool and create or validate the Harnest-owned schema.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> await store.start()
> ```

### PostgresStore.create

```python theme={null}
async PostgresStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord
```

Create one tenant-scoped session without overwriting conflicts.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `state` | `Mapping[str, Any]` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.create(session_id=..., user_id=..., state=...)
> ```

### PostgresStore.get

```python theme={null}
async PostgresStore.get(*, session_id: str, user_id: str) -> SessionRecord | None
```

Read one tenant-scoped session from PostgreSQL.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.get(session_id=..., user_id=...)
> ```

### PostgresStore.list

```python theme={null}
async PostgresStore.list(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]
```

Apply tenant filtering and optional keyset pagination in PostgreSQL.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> result = await store.list(user_id=...)
> ```

### PostgresStore.put\_a2a\_task

```python theme={null}
async PostgresStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord
```

Upsert an owner-scoped protobuf snapshot with immutable context.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `A2ATaskRecord` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.put_a2a_task(record=...)
> ```

### PostgresStore.get\_a2a\_task

```python theme={null}
async PostgresStore.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None
```

Read one A2A task using its complete application and owner scope.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `task_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)
> ```

### PostgresStore.list\_a2a\_tasks

```python theme={null}
async PostgresStore.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
```

Apply A2A filters, ordering, and keyset pagination in PostgreSQL.

> [accordion] 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` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> page = await store.list_a2a_tasks(application_id=..., user_id=...)
> ```

### PostgresStore.delete\_a2a\_task

```python theme={null}
async PostgresStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool
```

Delete one exact A2A owner record without revealing foreign tasks.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `task_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)
> ```

### PostgresStore.update

```python theme={null}
async PostgresStore.update(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None
```

Atomically merge a state delta in PostgreSQL.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `state_delta` | `Mapping[str, Any]` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.update(session_id=..., user_id=..., state_delta=...)
> ```

### PostgresStore.delete

```python theme={null}
async PostgresStore.delete(*, session_id: str, user_id: str) -> bool
```

Delete one tenant-scoped session and report whether it existed.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> result = await store.delete(session_id=..., user_id=...)
> ```

### PostgresStore.acquire

```python theme={null}
PostgresStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]
```

Hold a cross-replica advisory lock for one session execution.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> async with store.acquire(session_id=..., user_id=...):
>     ...
> ```

### PostgresStore.begin\_run

```python theme={null}
async PostgresStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord
```

Create an idempotent run while allowing one active run per session.

> [accordion] 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 |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.begin_run(
>     application_id=...,
>     user_id=...,
>     session_id=...,
>     run_id=...,
>     framework=...,
> )
> ```

### PostgresStore.get\_run

```python theme={null}
async PostgresStore.get_run(*, scope: RunScope) -> RunRecord | None
```

Read one PostgreSQL checkpoint run through its ownership scope.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.get_run(scope=...)
> ```

### PostgresStore.get\_checkpoint

```python theme={null}
async PostgresStore.get_checkpoint(
    *,
    scope: RunScope,
    checkpoint_id: str | None = None,
    namespace: str = '',
) -> CheckpointRecord | None
```

Load an exact or newest checkpoint entirely in PostgreSQL.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str \| None` | `None` |
> | `namespace` | `str` | `''` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.get_checkpoint(scope=...)
> ```

### PostgresStore.list\_checkpoints

```python theme={null}
PostgresStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]
```

Push checkpoint filtering, ordering, cursor, and limit into SQL.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `namespace` | `str \| None` | `None` |
> | `before` | `str \| None` | `None` |
> | `limit` | `int` | `20` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> async for item in store.list_checkpoints(scope=...):
>     ...
> ```

### PostgresStore.put

```python theme={null}
async PostgresStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord
```

Allocate a serialized revision and write only on matching CAS state.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `checkpoint` | `CheckpointRecord` | Required |
> | `scope` | `RunScope` | Required |
> | `expected_revision` | `int \| None` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.put(checkpoint=..., scope=..., expected_revision=...)
> ```

### PostgresStore.put\_writes

```python theme={null}
async PostgresStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None
```

Insert pending writes as one set and ignore existing identities.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str` | Required |
> | `writes` | `Sequence[CheckpointWrite]` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> await store.put_writes(scope=..., checkpoint_id=..., writes=...)
> ```

### PostgresStore.get\_writes

```python theme={null}
async PostgresStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]
```

Read pending writes for one owned PostgreSQL checkpoint.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> result = await store.get_writes(scope=..., checkpoint_id=...)
> ```

### PostgresStore.get\_writes\_batch

```python theme={null}
async PostgresStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]
```

Fetch pending writes for several checkpoints in one query.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_ids` | `Sequence[str]` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> result = await store.get_writes_batch(scope=..., checkpoint_ids=...)
> ```

### PostgresStore.suspend\_continuation

```python theme={null}
async PostgresStore.suspend_continuation(*, record: ContinuationRecord, external_id: str) -> ContinuationRecord
```

Atomically persist a provider wait and mark its owned run waiting.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `ContinuationRecord` | Required |
> | `external_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.suspend_continuation(record=..., external_id=...)
> ```

### PostgresStore.get\_continuation

```python theme={null}
async PostgresStore.get_continuation(*, scope: RunScope, continuation_id: str) -> ContinuationRecord | None
```

Load one continuation with its complete ownership predicate in SQL.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `continuation_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.get_continuation(scope=..., continuation_id=...)
> ```

### PostgresStore.get\_provider\_continuation

```python theme={null}
async PostgresStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> ProviderPendingContinuation | None
```

Load one exact provider wait and its private external identity.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> continuation = await store.get_provider_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
> )
> ```

### PostgresStore.get\_continuation\_by\_external\_id

```python theme={null}
async PostgresStore.get_continuation_by_external_id(
    *,
    application_id: str,
    provider: str,
    external_id: str,
) -> ProviderPendingContinuation | None
```

Use the unique provider key so replica callbacks never scan waits.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `provider` | `str` | Required |
> | `external_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> continuation = await store.get_continuation_by_external_id(
>     application_id=...,
>     provider=...,
>     external_id=...,
> )
> ```

### PostgresStore.list\_pending\_continuations

```python theme={null}
async PostgresStore.list_pending_continuations(
    *,
    application_id: str,
    provider: str,
    after: str | None = None,
    limit: int = 100,
) -> Sequence[ProviderPendingContinuation]
```

Push provider reconciliation filtering and pagination into PostgreSQL.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `provider` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> result = await store.list_pending_continuations(application_id=..., provider=...)
> ```

### PostgresStore.resolve\_continuation

```python theme={null}
async PostgresStore.resolve_continuation(
    *,
    scope: RunScope,
    provider: str,
    external_id: str,
    schema_id: str,
    result: Any = None,
    failure: ContinuationFailure | None = None,
) -> ContinuationRecord
```

Commit one validated provider outcome with an indexed SQL CAS.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `external_id` | `str` | Required |
> | `schema_id` | `str` | Required |
> | `result` | `Any` | `None` |
> | `failure` | `ContinuationFailure \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.resolve_continuation(
>     scope=...,
>     provider=...,
>     external_id=...,
>     schema_id=...,
> )
> ```

### PostgresStore.claim\_continuation

```python theme={null}
async PostgresStore.claim_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord
```

Claim an outcome and resume its run in one PostgreSQL statement.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.claim_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
> )
> ```

### PostgresStore.cancel\_continuation

```python theme={null}
async PostgresStore.cancel_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
    failure: ContinuationFailure,
) -> ContinuationRecord
```

Fail a pending continuation and cancel its waiting run atomically.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
> | `failure` | `ContinuationFailure` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.cancel_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
>     failure=...,
> )
> ```

### PostgresStore.arm\_continuation

```python theme={null}
async PostgresStore.arm_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord
```

Arm one wait with a SQL CAS after native checkpoint persistence.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.arm_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
> )
> ```

### PostgresStore.transition

```python theme={null}
async PostgresStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord
```

Apply a validated status transition with one SQL compare-and-swap.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `expected_status` | `RunStatus` | Required |
> | `status` | `RunStatus` | Required |
> | `pending_action` | `PendingAction \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> record = await store.transition(scope=..., expected_status=..., status=...)
> ```

### PostgresStore.delete\_run

```python theme={null}
async PostgresStore.delete_run(*, scope: RunScope) -> None
```

Delete a run and cascade its private checkpoint data.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> await store.delete_run(scope=...)
> ```

### PostgresStore.close

```python theme={null}
async PostgresStore.close() -> None
```

Close only pools created and owned by this store.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import PostgresStore
>
> store: PostgresStore
> await store.close()
> ```

## RedisStore

Combined Redis store with atomic CAS and renewable execution leases.

Production durability depends on Redis persistence, replication, and
failover configuration; Postgres remains the reference durable backend.

```python theme={null}
RedisStore(
    url: str,
    *,
    prefix: str = 'harnest',
    session_ttl_seconds: int | None = None,
    checkpoint_ttl_seconds: int = 604800,
    lease_seconds: int = 60,
    lease_wait_seconds: float = 30,
    client_options: Mapping[str, Any] | None = None,
) -> None
```

Fields (`—` means no declared default):

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

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `url` | `str` | Required |
> | `prefix` | `str` | `'harnest'` |
> | `session_ttl_seconds` | `int \| None` | `None` |
> | `checkpoint_ttl_seconds` | `int` | `604800` |
> | `lease_seconds` | `int` | `60` |
> | `lease_wait_seconds` | `float` | `30` |
> | `client_options` | `Mapping[str, Any] \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store = RedisStore(url=...)
> ```

### RedisStore.as\_langgraph\_checkpointer

```python theme={null}
RedisStore.as_langgraph_checkpointer() -> Any
```

Return one stable native saver for advanced LangGraph construction.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> result = store.as_langgraph_checkpointer()
> ```

### RedisStore.start

```python theme={null}
async RedisStore.start() -> None
```

Open the client and establish the compatible schema marker.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> await store.start()
> ```

### RedisStore.create

```python theme={null}
async RedisStore.create(*, session_id: str, user_id: str, state: Mapping[str, Any]) -> SessionRecord
```

Create one tenant-scoped session with an atomic Lua mutation.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `state` | `Mapping[str, Any]` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.create(session_id=..., user_id=..., state=...)
> ```

### RedisStore.get

```python theme={null}
async RedisStore.get(*, session_id: str, user_id: str) -> SessionRecord | None
```

Read one tenant-scoped session from Redis.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.get(session_id=..., user_id=...)
> ```

### RedisStore.list

```python theme={null}
async RedisStore.list(*, user_id: str, after: str | None = None, limit: int | None = None) -> Sequence[SessionRecord]
```

Read one lexicographically ordered keyset page with one MGET.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `user_id` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> result = await store.list(user_id=...)
> ```

### RedisStore.put\_a2a\_task

```python theme={null}
async RedisStore.put_a2a_task(record: A2ATaskRecord) -> A2ATaskRecord
```

Atomically update one A2A record and all of its query indexes.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `A2ATaskRecord` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.put_a2a_task(record=...)
> ```

### RedisStore.get\_a2a\_task

```python theme={null}
async RedisStore.get_a2a_task(*, application_id: str, user_id: str, task_id: str) -> A2ATaskRecord | None
```

Read one task through its digested application and owner key.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `task_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.get_a2a_task(application_id=..., user_id=..., task_id=...)
> ```

### RedisStore.list\_a2a\_tasks

```python theme={null}
async RedisStore.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
```

Use one prefiltered sorted index and one batched record read.

> [accordion] 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` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> page = await store.list_a2a_tasks(application_id=..., user_id=...)
> ```

### RedisStore.delete\_a2a\_task

```python theme={null}
async RedisStore.delete_a2a_task(*, application_id: str, user_id: str, task_id: str) -> bool
```

Delete a record and every selected-list index with optimistic CAS.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `task_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> result = await store.delete_a2a_task(application_id=..., user_id=..., task_id=...)
> ```

### RedisStore.update

```python theme={null}
async RedisStore.update(*, session_id: str, user_id: str, state_delta: Mapping[str, Any]) -> SessionRecord | None
```

Atomically merge state while preserving session expiry policy.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
> | `state_delta` | `Mapping[str, Any]` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.update(session_id=..., user_id=..., state_delta=...)
> ```

### RedisStore.delete

```python theme={null}
async RedisStore.delete(*, session_id: str, user_id: str) -> bool
```

Delete a session and its user index entry atomically.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> result = await store.delete(session_id=..., user_id=...)
> ```

### RedisStore.acquire

```python theme={null}
RedisStore.acquire(*, session_id: str, user_id: str) -> AsyncIterator[SessionLease]
```

Hold and renew a token-bound distributed lease for one execution.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `session_id` | `str` | Required |
> | `user_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> async with store.acquire(session_id=..., user_id=...):
>     ...
> ```

### RedisStore.begin\_run

```python theme={null}
async RedisStore.begin_run(
    *,
    application_id: str,
    user_id: str,
    session_id: str,
    run_id: str,
    framework: Literal['adk', 'langgraph'],
) -> RunRecord
```

Create an idempotent run while allowing one active run per session.

> [accordion] 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 |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.begin_run(
>     application_id=...,
>     user_id=...,
>     session_id=...,
>     run_id=...,
>     framework=...,
> )
> ```

### RedisStore.get\_run

```python theme={null}
async RedisStore.get_run(*, scope: RunScope) -> RunRecord | None
```

Read one Redis checkpoint run through its ownership scope.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.get_run(scope=...)
> ```

### RedisStore.get\_checkpoint

```python theme={null}
async RedisStore.get_checkpoint(*, scope: RunScope, checkpoint_id: str | None = None, namespace: str = '') -> CheckpointRecord | None
```

Resolve an exact or newest checkpoint through its Redis index.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str \| None` | `None` |
> | `namespace` | `str` | `''` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.get_checkpoint(scope=...)
> ```

### RedisStore.list\_checkpoints

```python theme={null}
RedisStore.list_checkpoints(
    *,
    scope: RunScope,
    namespace: str | None = None,
    before: str | None = None,
    limit: int = 20,
) -> AsyncIterator[CheckpointRecord]
```

Read bounded checkpoint history from sorted-set indexes.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `namespace` | `str \| None` | `None` |
> | `before` | `str \| None` | `None` |
> | `limit` | `int` | `20` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> async for item in store.list_checkpoints(scope=...):
>     ...
> ```

### RedisStore.put

```python theme={null}
async RedisStore.put(checkpoint: CheckpointRecord, *, scope: RunScope, expected_revision: int | None) -> CheckpointRecord
```

Persist checkpoint data and revision indexes with one Lua CAS.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `checkpoint` | `CheckpointRecord` | Required |
> | `scope` | `RunScope` | Required |
> | `expected_revision` | `int \| None` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.put(checkpoint=..., scope=..., expected_revision=...)
> ```

### RedisStore.put\_writes

```python theme={null}
async RedisStore.put_writes(*, scope: RunScope, checkpoint_id: str, writes: Sequence[CheckpointWrite]) -> None
```

Add pending writes idempotently with one Lua operation.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str` | Required |
> | `writes` | `Sequence[CheckpointWrite]` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> await store.put_writes(scope=..., checkpoint_id=..., writes=...)
> ```

### RedisStore.get\_writes

```python theme={null}
async RedisStore.get_writes(*, scope: RunScope, checkpoint_id: str) -> Sequence[CheckpointWrite]
```

Read pending writes for one owned Redis checkpoint.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> result = await store.get_writes(scope=..., checkpoint_id=...)
> ```

### RedisStore.get\_writes\_batch

```python theme={null}
async RedisStore.get_writes_batch(*, scope: RunScope, checkpoint_ids: Sequence[str]) -> Mapping[str, Sequence[CheckpointWrite]]
```

Fetch several pending-write hashes in one Redis round trip.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `checkpoint_ids` | `Sequence[str]` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> result = await store.get_writes_batch(scope=..., checkpoint_ids=...)
> ```

### RedisStore.suspend\_continuation

```python theme={null}
async RedisStore.suspend_continuation(*, record: ContinuationRecord, external_id: str) -> ContinuationRecord
```

Atomically persist a provider wait and move its run to waiting.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `record` | `ContinuationRecord` | Required |
> | `external_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.suspend_continuation(record=..., external_id=...)
> ```

### RedisStore.get\_continuation

```python theme={null}
async RedisStore.get_continuation(*, scope: RunScope, continuation_id: str) -> ContinuationRecord | None
```

Read one private record only when all ownership fields match.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `continuation_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.get_continuation(scope=..., continuation_id=...)
> ```

### RedisStore.get\_provider\_continuation

```python theme={null}
async RedisStore.get_provider_continuation(*, scope: RunScope, provider: str, continuation_id: str) -> ProviderPendingContinuation | None
```

Load an exact provider envelope only while its owned run exists.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> continuation = await store.get_provider_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
> )
> ```

### RedisStore.get\_continuation\_by\_external\_id

```python theme={null}
async RedisStore.get_continuation_by_external_id(*, application_id: str, provider: str, external_id: str) -> ProviderPendingContinuation | None
```

Resolve a callback through its provider-owned Redis index.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `provider` | `str` | Required |
> | `external_id` | `str` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> continuation = await store.get_continuation_by_external_id(
>     application_id=...,
>     provider=...,
>     external_id=...,
> )
> ```

### RedisStore.list\_pending\_continuations

```python theme={null}
async RedisStore.list_pending_continuations(
    *,
    application_id: str,
    provider: str,
    after: str | None = None,
    limit: int = 100,
) -> Sequence[ProviderPendingContinuation]
```

Load one indexed provider page with a single batched value read.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `application_id` | `str` | Required |
> | `provider` | `str` | Required |
> | `after` | `str \| None` | `None` |
> | `limit` | `int` | `100` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> result = await store.list_pending_continuations(application_id=..., provider=...)
> ```

### RedisStore.resolve\_continuation

```python theme={null}
async RedisStore.resolve_continuation(
    *,
    scope: RunScope,
    provider: str,
    external_id: str,
    schema_id: str,
    result: Any = None,
    failure: ContinuationFailure | None = None,
) -> ContinuationRecord
```

Resolve an indexed provider callback with an atomic status CAS.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `external_id` | `str` | Required |
> | `schema_id` | `str` | Required |
> | `result` | `Any` | `None` |
> | `failure` | `ContinuationFailure \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.resolve_continuation(
>     scope=...,
>     provider=...,
>     external_id=...,
>     schema_id=...,
> )
> ```

### RedisStore.claim\_continuation

```python theme={null}
async RedisStore.claim_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord
```

Claim a provider outcome and resume its run with one Lua CAS.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.claim_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
> )
> ```

### RedisStore.cancel\_continuation

```python theme={null}
async RedisStore.cancel_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
    failure: ContinuationFailure,
) -> ContinuationRecord
```

Cancel one exact waiting run and fail its continuation in one script.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
> | `failure` | `ContinuationFailure` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.cancel_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
>     failure=...,
> )
> ```

### RedisStore.arm\_continuation

```python theme={null}
async RedisStore.arm_continuation(
    *,
    scope: RunScope,
    provider: str,
    continuation_id: str,
    expected_revision: int,
) -> ContinuationRecord
```

Arm one persisted wait after the framework commits its checkpoint.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `provider` | `str` | Required |
> | `continuation_id` | `str` | Required |
> | `expected_revision` | `int` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.arm_continuation(
>     scope=...,
>     provider=...,
>     continuation_id=...,
>     expected_revision=...,
> )
> ```

### RedisStore.transition

```python theme={null}
async RedisStore.transition(
    *,
    scope: RunScope,
    expected_status: RunStatus,
    status: RunStatus,
    pending_action: PendingAction | None = None,
) -> RunRecord
```

Apply one status compare-and-swap and release terminal ownership.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
> | `expected_status` | `RunStatus` | Required |
> | `status` | `RunStatus` | Required |
> | `pending_action` | `PendingAction \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> record = await store.transition(scope=..., expected_status=..., status=...)
> ```

### RedisStore.delete\_run

```python theme={null}
async RedisStore.delete_run(*, scope: RunScope) -> None
```

Make run data unreachable immediately and let private blobs expire.

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `scope` | `RunScope` | Required |
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> await store.delete_run(scope=...)
> ```

### RedisStore.close

```python theme={null}
async RedisStore.close() -> None
```

Close only clients created and owned by this store.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import RedisStore
>
> store: RedisStore
> await store.close()
> ```

## StorageRegistry

One normalized storage boundary consumed by compiler and runtime adapters.

```python theme={null}
StorageRegistry(
    sessions: SessionStore | ADKStore | None = None,
    checkpoints: CheckpointAuthority | None = None,
    assets: Mapping[str, AssetStore] = ...,
    custom: Mapping[str, CustomStorage] = ...,
    tasks: TaskStore | None = None,
    cron: CronStore | None = None,
    memory: MemoryStore | None = None,
) -> None
```

Fields (`—` means no declared default):

| Field | Type | Default |
| - | - | - |
| `sessions` | `SessionStore \| ADKStore \| None` | `None` |
| `checkpoints` | `CheckpointAuthority \| None` | `None` |
| `assets` | `Mapping[str, AssetStore]` | `dict()` |
| `custom` | `Mapping[str, CustomStorage]` | `dict()` |
| `tasks` | `TaskStore \| None` | `None` |
| `cron` | `CronStore \| None` | `None` |
| `memory` | `MemoryStore \| None` | `None` |

> [accordion] Parameters and example
>
> | Parameter | Type | Default |
> | - | - | - |
> | `sessions` | `SessionStore \| ADKStore \| None` | `None` |
> | `checkpoints` | `CheckpointAuthority \| None` | `None` |
> | `assets` | `Mapping[str, AssetStore]` | `...` |
> | `custom` | `Mapping[str, CustomStorage]` | `...` |
> | `tasks` | `TaskStore \| None` | `None` |
> | `cron` | `CronStore \| None` | `None` |
> | `memory` | `MemoryStore \| None` | `None` |
>
> ```python theme={null}
> from harnest.store import StorageRegistry
>
> registry = StorageRegistry()
> ```

### StorageRegistry.default\_assets

Return the explicitly named default without choosing by insertion order.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import StorageRegistry
>
> registry: StorageRegistry
> value = registry.default_assets
> ```

### StorageRegistry.owned\_resources

```python theme={null}
StorageRegistry.owned_resources() -> tuple[Any, ...]
```

Return each lifecycle-owned object once in deterministic role order.

> [accordion] Example
>
> ```python theme={null}
> from harnest.store import StorageRegistry
>
> registry: StorageRegistry
> result = registry.owned_resources()
> ```


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.