harnest_redis

All Python modules · Import and compatibility guidance

Exports

Name	Kind
`RedisMemoryStore`	Class
`RedisStore`	Class

RedisMemoryStore

Store explicit memories in Redis; durability requires persistent Redis.

RedisMemoryStore(url: str, *, prefix: str = 'harnest-memory') -> None

Parameters and example

Parameter	Type	Default
`url`	`str`	Required
`prefix`	`str`	`'harnest-memory'`

from harnest_redis import RedisMemoryStore

store = RedisMemoryStore(url=...)

RedisMemoryStore.delete_all

async RedisMemoryStore.delete_all(scope: harnest.memory.MemoryScope) -> int

Erase one explicit namespace through trusted application code only.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
result = await store.delete_all(scope=...)

RedisMemoryStore.delete_user

async RedisMemoryStore.delete_user(application_id: str, user_id: str) -> int

Erase an explicitly authorized owner without registering any endpoint.

Parameters and example

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

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
result = await store.delete_user(application_id=..., user_id=...)

RedisMemoryStore.purge_expired

async RedisMemoryStore.purge_expired(
    scope: harnest.memory.MemoryScope,
    *,
    limit: int = 100,
    after: str | None = None,
) -> harnest.memory.MemoryCleanupPage

Sweep a bounded key page without exposing another user’s collection.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`limit`	`int`	`100`
`after`	`str \| None`	`None`

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
page = await store.purge_expired(scope=...)

RedisMemoryStore.put

async RedisMemoryStore.put(record: harnest.memory.MemoryRecord, *, expected_revision: str | None = None) -> harnest.memory.MemoryRecord

Validate before atomically persisting a deliberate memory write.

Parameters and example

Parameter	Type	Default
`record`	`harnest.memory.MemoryRecord`	Required
`expected_revision`	`str \| None`	`None`

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
record = await store.put(record=...)

RedisMemoryStore.get

async RedisMemoryStore.get(scope: harnest.memory.MemoryScope, key: str) -> harnest.memory.MemoryRecord | None

Read only an explicitly scoped live record.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`key`	`str`	Required

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
record = await store.get(scope=..., key=...)

RedisMemoryStore.list

async RedisMemoryStore.list(
    scope: harnest.memory.MemoryScope,
    *,
    limit: int = 20,
    after: str | None = None,
) -> harnest.memory.MemoryPage

Page records without loading an unbounded provider collection.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`limit`	`int`	`20`
`after`	`str \| None`	`None`

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
page = await store.list(scope=...)

RedisMemoryStore.search

async RedisMemoryStore.search(
    scope: harnest.memory.MemoryScope,
    query: str,
    *,
    limit: int = 20,
    after: str | None = None,
) -> harnest.memory.MemoryPage

Retrieve explicit literal text matches without model or embedding calls.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`query`	`str`	Required
`limit`	`int`	`20`
`after`	`str \| None`	`None`

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
page = await store.search(scope=..., query=...)

RedisMemoryStore.delete

async RedisMemoryStore.delete(
    scope: harnest.memory.MemoryScope,
    key: str,
    *,
    expected_revision: str | None = None,
) -> bool

Remove memory and its indexes with an optional atomic revision fence.

Parameters and example

Parameter	Type	Default
`scope`	`harnest.memory.MemoryScope`	Required
`key`	`str`	Required
`expected_revision`	`str \| None`	`None`

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
result = await store.delete(scope=..., key=...)

RedisMemoryStore.start

async RedisMemoryStore.start() -> None

Open an owned async client only when the selected provider starts.

Example

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
await store.start()

RedisMemoryStore.close

async RedisMemoryStore.close() -> None

Release only this provider’s Redis client.

Example

from harnest_redis import RedisMemoryStore

store: RedisMemoryStore
await store.close()

RedisStore

Combined storage with atomic task leases and cron occurrence handoff. Tasks have no expiration: durable retention follows explicit lifecycle operations, independent of the session/checkpoint TTL configuration. Configure Redis persistence and replication for the required durability.

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'`

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`

from harnest_redis import RedisStore

store = RedisStore(url=...)

RedisStore.start

async RedisStore.start() -> None

Open the client and establish the compatible schema marker.

Example

from harnest_redis import RedisStore

store: RedisStore
await store.start()

RedisStore.close

async RedisStore.close() -> None

Close only clients created and owned by this store.

Example

from harnest_redis import RedisStore

store: RedisStore
await store.close()

RedisStore.create_cron

async RedisStore.create_cron(record: CronRecord) -> CronRecord

Atomically persist schedule identity, owner index, and due index.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required

from harnest_redis import RedisStore

store: RedisStore
record = await store.create_cron(record=...)

RedisStore.get_cron

async RedisStore.get_cron(*, application_id: str, user_id: str, schedule_id: str) -> CronRecord | None

Read one schedule using an owner predicate evaluated by Redis.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
record = await store.get_cron(application_id=..., user_id=..., schedule_id=...)

RedisStore.list_crons

async RedisStore.list_crons(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Return a bounded lexicographic owner-index page in one Redis call.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
result = await store.list_crons(application_id=..., user_id=...)

RedisStore.update_cron

async RedisStore.update_cron(record: CronRecord, *, expected_revision: int) -> CronRecord

Replace mutable schedule fields with revision and terminal fencing.

Parameters and example

Parameter	Type	Default
`record`	`CronRecord`	Required
`expected_revision`	`int`	Required

from harnest_redis import RedisStore

store: RedisStore
record = await store.update_cron(record=..., expected_revision=...)

RedisStore.delete_cron

async RedisStore.delete_cron(*, application_id: str, user_id: str, schedule_id: str) -> bool

Remove one owner’s record and its identity/query indexes atomically.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
result = await store.delete_cron(application_id=..., user_id=..., schedule_id=...)

RedisStore.list_due_crons

async RedisStore.list_due_crons(
    *,
    application_id: str,
    now: float,
    after: tuple[float, str] | None = None,
    limit: int = 100,
) -> tuple[CronRecord, ...]

Read due schedules from a time/ID index with stable keyset bounds.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`now`	`float`	Required
`after`	`tuple[float, str] \| None`	`None`
`limit`	`int`	`100`

from harnest_redis import RedisStore

store: RedisStore
result = await store.list_due_crons(application_id=..., now=...)

RedisStore.commit_cron_occurrence

async RedisStore.commit_cron_occurrence(
    *,
    application_id: str,
    user_id: str,
    schedule_id: str,
    expected_revision: int,
    due_at: float,
    next_run_at: float,
    task: TaskRecord,
) -> TaskRecord | None

Enqueue and advance a fenced occurrence within one Lua transaction.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`schedule_id`	`str`	Required
`expected_revision`	`int`	Required
`due_at`	`float`	Required
`next_run_at`	`float`	Required
`task`	`TaskRecord`	Required

from harnest_redis import RedisStore

store: RedisStore
record = await store.commit_cron_occurrence(
    application_id=...,
    user_id=...,
    schedule_id=...,
    expected_revision=...,
    due_at=...,
    next_run_at=...,
    task=...,
)

RedisStore.enqueue_task

async RedisStore.enqueue_task(record: TaskRecord) -> TaskRecord

Persist a job once per scoped key, retaining immutable fingerprints.

Parameters and example

Parameter	Type	Default
`record`	`TaskRecord`	Required

from harnest_redis import RedisStore

store: RedisStore
record = await store.enqueue_task(record=...)

RedisStore.get_task

async RedisStore.get_task(*, application_id: str, job_id: str, user_id: str | None = None) -> TaskRecord | None

Apply owner scope inside Redis before returning a single record.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`user_id`	`str \| None`	`None`

from harnest_redis import RedisStore

store: RedisStore
record = await store.get_task(application_id=..., job_id=...)

RedisStore.claim_tasks

async RedisStore.claim_tasks(
    *,
    application_id: str,
    queues: tuple[str, ...],
    now: float,
    lease_seconds: float,
    limit: int = 1,
) -> tuple[TaskRecord, ...]

Claim bounded indexed work and recover expired attempts atomically.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`queues`	`tuple[str, ...]`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required
`limit`	`int`	`1`

from harnest_redis import RedisStore

store: RedisStore
result = await store.claim_tasks(
    application_id=...,
    queues=...,
    now=...,
    lease_seconds=...,
)

RedisStore.renew_task_lease

async RedisStore.renew_task_lease(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    lease_seconds: float,
) -> bool

Renew only the matching unexpired attempt’s fenced queue lease.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`lease_seconds`	`float`	Required

from harnest_redis import RedisStore

store: RedisStore
result = await store.renew_task_lease(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    lease_seconds=...,
)

RedisStore.finish_task

async RedisStore.finish_task(
    *,
    application_id: str,
    job_id: str,
    lease_token: str,
    now: float,
    status: str,
    result: Any = None,
    failure_code: str | None = None,
    retry_at: float | None = None,
) -> bool

Commit a fenced result or delayed retry and scrub terminal payloads.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`job_id`	`str`	Required
`lease_token`	`str`	Required
`now`	`float`	Required
`status`	`str`	Required
`result`	`Any`	`None`
`failure_code`	`str \| None`	`None`
`retry_at`	`float \| None`	`None`

from harnest_redis import RedisStore

store: RedisStore
result = await store.finish_task(
    application_id=...,
    job_id=...,
    lease_token=...,
    now=...,
    status=...,
)

RedisStore.cancel_task

async RedisStore.cancel_task(*, application_id: str, user_id: str, job_id: str, now: float) -> bool

Cancel scoped work and invalidate running claims in one mutation.

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`job_id`	`str`	Required
`now`	`float`	Required

from harnest_redis import RedisStore

store: RedisStore
result = await store.cancel_task(application_id=..., user_id=..., job_id=..., now=...)

RedisStore.as_langgraph_checkpointer

RedisStore.as_langgraph_checkpointer() -> Any

Return one stable native saver for advanced LangGraph construction.

Example

from harnest_redis import RedisStore

store: RedisStore
result = store.as_langgraph_checkpointer()

RedisStore.create

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

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

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.get

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

Read one tenant-scoped session from Redis.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.list

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.put_a2a_task

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

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

Parameters and example

Parameter	Type	Default
`record`	`A2ATaskRecord`	Required

from harnest_redis import RedisStore

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

RedisStore.get_a2a_task

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.list_a2a_tasks

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.

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_redis import RedisStore

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

RedisStore.delete_a2a_task

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.update

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

Atomically merge state while preserving session expiry policy.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.delete

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

Delete a session and its user index entry atomically.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.acquire

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

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

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.begin_run

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.

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_redis import RedisStore

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

RedisStore.get_run

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

Read one Redis checkpoint run through its ownership scope.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest_redis import RedisStore

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

RedisStore.get_checkpoint

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.list_checkpoints

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.put

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

Persist checkpoint data and revision indexes with one Lua CAS.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.put_writes

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

Add pending writes idempotently with one Lua operation.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.get_writes

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

Read pending writes for one owned Redis checkpoint.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.get_writes_batch

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.suspend_continuation

async RedisStore.suspend_continuation(*, record: ContinuationRecord, external_id: str) -> ContinuationRecord

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

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
record = await store.suspend_continuation(record=..., external_id=...)

RedisStore.get_continuation

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

Read one private record only when all ownership fields match.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
record = await store.get_continuation(scope=..., continuation_id=...)

RedisStore.get_provider_continuation

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.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
continuation = await store.get_provider_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
)

RedisStore.get_continuation_by_external_id

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.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
continuation = await store.get_continuation_by_external_id(
    application_id=...,
    provider=...,
    external_id=...,
)

RedisStore.list_pending_continuations

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.resolve_continuation

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.

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`

from harnest_redis import RedisStore

store: RedisStore
record = await store.resolve_continuation(
    scope=...,
    provider=...,
    external_id=...,
    schema_id=...,
)

RedisStore.claim_continuation

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.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
record = await store.claim_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

RedisStore.cancel_continuation

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.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
record = await store.cancel_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
    failure=...,
)

RedisStore.arm_continuation

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.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
record = await store.arm_continuation(
    scope=...,
    provider=...,
    continuation_id=...,
    expected_revision=...,
)

RedisStore.transition

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.

Parameters and example

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

from harnest_redis import RedisStore

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

RedisStore.delete_run

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

Make run data unreachable immediately and let private blobs expire.

Parameters and example

Parameter	Type	Default
`scope`	`RunScope`	Required

from harnest_redis import RedisStore

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

RedisStore.list_task_metadata

async RedisStore.list_task_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Index legacy jobs incrementally and project one owner page inside Redis.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
result = await store.list_task_metadata(application_id=..., user_id=...)

RedisStore.list_cron_metadata

async RedisStore.list_cron_metadata(
    *,
    application_id: str,
    user_id: str,
    after: str | None = None,
    limit: int = 100,
) -> dict[str, Any]

Project one indexed owner page in Redis without transferring private arguments.

Parameters and example

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

from harnest_redis import RedisStore

store: RedisStore
result = await store.list_cron_metadata(application_id=..., user_id=...)

​Exports

​RedisMemoryStore

​RedisMemoryStore.delete_all

​RedisMemoryStore.delete_user

​RedisMemoryStore.purge_expired

​RedisMemoryStore.put

​RedisMemoryStore.get

​RedisMemoryStore.list

​RedisMemoryStore.search

​RedisMemoryStore.delete

​RedisMemoryStore.start

​RedisMemoryStore.close

​RedisStore

​RedisStore.start

​RedisStore.close

​RedisStore.create_cron

​RedisStore.get_cron

​RedisStore.list_crons

​RedisStore.update_cron

​RedisStore.delete_cron

​RedisStore.list_due_crons

​RedisStore.commit_cron_occurrence

​RedisStore.enqueue_task

​RedisStore.get_task

​RedisStore.claim_tasks

​RedisStore.renew_task_lease

​RedisStore.finish_task

​RedisStore.cancel_task

​RedisStore.as_langgraph_checkpointer

​RedisStore.create

​RedisStore.get

​RedisStore.list

​RedisStore.put_a2a_task

​RedisStore.get_a2a_task

​RedisStore.list_a2a_tasks

​RedisStore.delete_a2a_task

​RedisStore.update

​RedisStore.delete

​RedisStore.acquire

​RedisStore.begin_run

​RedisStore.get_run

​RedisStore.get_checkpoint

​RedisStore.list_checkpoints

​RedisStore.put

​RedisStore.put_writes

​RedisStore.get_writes

​RedisStore.get_writes_batch

​RedisStore.suspend_continuation

​RedisStore.get_continuation

​RedisStore.get_provider_continuation

​RedisStore.get_continuation_by_external_id

​RedisStore.list_pending_continuations

​RedisStore.resolve_continuation

​RedisStore.claim_continuation

​RedisStore.cancel_continuation

​RedisStore.arm_continuation

​RedisStore.transition

​RedisStore.delete_run

​RedisStore.list_task_metadata

​RedisStore.list_cron_metadata

Exports

RedisMemoryStore

RedisMemoryStore.delete_all

RedisMemoryStore.delete_user

RedisMemoryStore.purge_expired

RedisMemoryStore.put

RedisMemoryStore.get

RedisMemoryStore.list

RedisMemoryStore.search

RedisMemoryStore.delete

RedisMemoryStore.start

RedisMemoryStore.close

RedisStore

RedisStore.start

RedisStore.close

RedisStore.create_cron

RedisStore.get_cron

RedisStore.list_crons

RedisStore.update_cron

RedisStore.delete_cron

RedisStore.list_due_crons

RedisStore.commit_cron_occurrence

RedisStore.enqueue_task

RedisStore.get_task

RedisStore.claim_tasks

RedisStore.renew_task_lease

RedisStore.finish_task

RedisStore.cancel_task

RedisStore.as_langgraph_checkpointer

RedisStore.create

RedisStore.get

RedisStore.list

RedisStore.put_a2a_task

RedisStore.get_a2a_task

RedisStore.list_a2a_tasks

RedisStore.delete_a2a_task

RedisStore.update

RedisStore.delete

RedisStore.acquire

RedisStore.begin_run

RedisStore.get_run

RedisStore.get_checkpoint

RedisStore.list_checkpoints

RedisStore.put

RedisStore.put_writes

RedisStore.get_writes

RedisStore.get_writes_batch

RedisStore.suspend_continuation

RedisStore.get_continuation

RedisStore.get_provider_continuation

RedisStore.get_continuation_by_external_id

RedisStore.list_pending_continuations

RedisStore.resolve_continuation

RedisStore.claim_continuation

RedisStore.cancel_continuation

RedisStore.arm_continuation

RedisStore.transition

RedisStore.delete_run

RedisStore.list_task_metadata

RedisStore.list_cron_metadata