harnest.memory

All Python modules · Import and compatibility guidance

Retention and limits

Setting	Behavior
`expires_at`	Optional future UTC epoch timestamp. Expired records disappear from reads; purge them for physical cleanup.
Session lifetime	Deleting a session or expiring a checkpoint does not delete memories.
Record metadata	Creation/update times, revision, and source session/invocation/agent; updates replace content and metadata, not append history.
Size limits	Key: 256 UTF-8 bytes. Content: 64 KiB. Metadata and provenance: 16 KiB combined.
Page size	1–100 records.

Storage maintenance

Provider method	Scope and result
`delete_all(scope)`	Physically remove live and expired records in one namespace; return the removed count
`delete_user(application_id, user_id)`	Remove that application’s user across every namespace; return the removed count
`purge_expired(scope, limit=100, after=None)`	Inspect a bounded page and return `MemoryCleanupPage(deleted, next_cursor)`

Exports

Name	Kind
`InMemoryStore`	Class
`MemoryCleanupPage`	Class
`MemoryConflictError`	Class
`MemoryPage`	Class
`MemoryRecord`	Class
`MemoryScope`	Class
`MemoryStorageError`	Class
`MemoryStore`	Class

InMemoryStore

Implement memory in process-local storage that is lost on process exit.

InMemoryStore() -> None

Example

from harnest.memory import InMemoryStore

store = InMemoryStore()

InMemoryStore.delete_all

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.delete_user

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.purge_expired

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.put

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.get

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.list

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.search

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.delete

async InMemoryStore.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.memory import InMemoryStore

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

InMemoryStore.start

async InMemoryStore.start() -> None

The process-local reference provider needs no external resources.

Example

from harnest.memory import InMemoryStore

store: InMemoryStore
await store.start()

InMemoryStore.close

async InMemoryStore.close() -> None

Release no external resources; retained test records stay inspectable.

Example

from harnest.memory import InMemoryStore

store: InMemoryStore
await store.close()

MemoryCleanupPage

Physical removals in a bounded sweep; continue even when deleted is zero.

MemoryCleanupPage(deleted: int, next_cursor: str | None = None) -> None

Fields (— means no declared default):

Field	Type	Default
`deleted`	`int`	`—`
`next_cursor`	`str \| None`	`None`

Parameters and example

Parameter	Type	Default
`deleted`	`int`	Required
`next_cursor`	`str \| None`	`None`

from harnest.memory import MemoryCleanupPage

page = MemoryCleanupPage(deleted=...)

MemoryConflictError

A conditional write refers to a stale or missing memory revision.

Example

from harnest.memory import MemoryConflictError

try:
    ...
except MemoryConflictError as error:
    ...

MemoryPage

A bounded page; pass next_cursor as after, even for an empty page.

MemoryPage(items: tuple[MemoryRecord, ...], next_cursor: str | None = None) -> None

Fields (— means no declared default):

Field	Type	Default
`items`	`tuple[MemoryRecord, ...]`	`—`
`next_cursor`	`str \| None`	`None`

Parameters and example

Parameter	Type	Default
`items`	`tuple[MemoryRecord, ...]`	Required
`next_cursor`	`str \| None`	`None`

from harnest.memory import MemoryPage

page = MemoryPage(items=...)

MemoryRecord

One deliberate memory; timestamps are UTC epoch seconds. Providers assign opaque revisions and timestamps on writes. Source records provenance, not instructions or authorization. Payloads must be JSON-safe.

MemoryRecord(
    scope: MemoryScope,
    key: str,
    content: str,
    metadata: Mapping[str, Any] = ...,
    source: Mapping[str, str] = ...,
    revision: str = '',
    created_at: float = 0.0,
    updated_at: float = 0.0,
    expires_at: float | None = None,
) -> None

Fields (— means no declared default):

Field	Type	Default
`scope`	`MemoryScope`	`—`
`key`	`str`	`—`
`content`	`str`	`—`
`metadata`	`Mapping[str, Any]`	`dict()`
`source`	`Mapping[str, str]`	`dict()`
`revision`	`str`	`''`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`
`expires_at`	`float \| None`	`None`

Parameters and example

Parameter	Type	Default
`scope`	`MemoryScope`	Required
`key`	`str`	Required
`content`	`str`	Required
`metadata`	`Mapping[str, Any]`	`...`
`source`	`Mapping[str, str]`	`...`
`revision`	`str`	`''`
`created_at`	`float`	`0.0`
`updated_at`	`float`	`0.0`
`expires_at`	`float \| None`	`None`

from harnest.memory import MemoryRecord

record = MemoryRecord(scope=..., key=..., content=...)

MemoryScope

Trusted application/user isolation, with an optional logical namespace.

MemoryScope(application_id: str, user_id: str, namespace: str = 'default') -> None

Fields (— means no declared default):

Field	Type	Default
`application_id`	`str`	`—`
`user_id`	`str`	`—`
`namespace`	`str`	`'default'`

Parameters and example

Parameter	Type	Default
`application_id`	`str`	Required
`user_id`	`str`	Required
`namespace`	`str`	`'default'`

from harnest.memory import MemoryScope

scope = MemoryScope(application_id=..., user_id=...)

MemoryStorageError

Memory persistence failed without exposing private provider details.

Example

from harnest.memory import MemoryStorageError

try:
    ...
except MemoryStorageError as error:
    ...

MemoryStore

Async persistence contract for explicit cross-session memories. Every operation must enforce the complete scope in the datastore. Expired records are invisible. Writes and revision checks are atomic. Reads return detached records. No method may invoke models or extract memories implicitly.

MemoryStore(*args, **kwargs)

Example

from harnest.memory import MemoryStore

class MyMemoryStore(MemoryStore):
    ...

MemoryStore.start

async MemoryStore.start() -> None

Open owned resources and initialize or validate storage.

Example

from harnest.memory import MemoryStore

store: MemoryStore
await store.start()

MemoryStore.close

async MemoryStore.close() -> None

Release only resources owned by this provider.

Example

from harnest.memory import MemoryStore

store: MemoryStore
await store.close()

MemoryStore.put

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

Upsert atomically, preserving creation time and assigning a new revision. None allows an unconditional write. A supplied revision must match a live record, otherwise raise MemoryConflictError, including after deletion.

Parameters and example

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

from harnest.memory import MemoryStore

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

MemoryStore.get

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

Read one live record by its fully scoped key.

Parameters and example

Parameter	Type	Default
`scope`	`MemoryScope`	Required
`key`	`str`	Required

from harnest.memory import MemoryStore

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

MemoryStore.list

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

Page live records in ascending key order; limit is 1 through 100.

Parameters and example

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

from harnest.memory import MemoryStore

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

MemoryStore.search

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

Page literal, case-sensitive content matches, not semantic similarity. Apply scope, query, expiry, ordering and limits in the datastore. Bounded scans may yield empty pages with a continuation cursor. Cursors are not snapshots: concurrent edits may change subsequent results.

Parameters and example

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

from harnest.memory import MemoryStore

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

MemoryStore.delete

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

Atomically remove the record and index references; return false if absent. A supplied stale revision raises MemoryConflictError. Deletion cannot erase copies already returned to callers or retained in database backups.

Parameters and example

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

from harnest.memory import MemoryStore

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

MemoryStore.delete_all

async MemoryStore.delete_all(scope: MemoryScope) -> int

Physically erase a namespace, including expired records; return count. Trusted application maintenance only: no HTTP route or model tool is registered. Revoke writes first when using this for erasure, because concurrent new writes may survive. Backups and returned copies remain.

Parameters and example

Parameter	Type	Default
`scope`	`MemoryScope`	Required

from harnest.memory import MemoryStore

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

MemoryStore.delete_user

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

Physically erase all namespaces for one application’s user. Trusted application code must authorize the owner and revoke their writes first. Cleanup may be batched, not one global transaction; on failure retry to remove remaining records. Return the removed count.

Parameters and example

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

from harnest.memory import MemoryStore

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

MemoryStore.purge_expired

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

Physically sweep at most limit keys in ascending order for expiry. Continue while a cursor is present, even when deleted is zero. Repeat sweeps for records that expire after the cursor passes. This maintenance operation is provider-only, not automatically scheduled or model-facing.

Parameters and example

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

from harnest.memory import MemoryStore

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

Overview

Get started

Harnest Studio

Guides

API reference

harnest.memory

Retention and limits

Storage maintenance

Exports

InMemoryStore

InMemoryStore.delete_all

InMemoryStore.delete_user

InMemoryStore.purge_expired

InMemoryStore.put

InMemoryStore.get

InMemoryStore.list

InMemoryStore.search

InMemoryStore.delete

InMemoryStore.start

InMemoryStore.close

MemoryCleanupPage

MemoryConflictError

MemoryPage

MemoryRecord

MemoryScope

MemoryStorageError

MemoryStore

MemoryStore.start

MemoryStore.close

MemoryStore.put

MemoryStore.get

MemoryStore.list

MemoryStore.search

MemoryStore.delete

MemoryStore.delete_all

MemoryStore.delete_user

MemoryStore.purge_expired

​Retention and limits

​Storage maintenance

​Exports

​InMemoryStore

​InMemoryStore.delete_all

​InMemoryStore.delete_user

​InMemoryStore.purge_expired

​InMemoryStore.put

​InMemoryStore.get

​InMemoryStore.list

​InMemoryStore.search

​InMemoryStore.delete

​InMemoryStore.start

​InMemoryStore.close

​MemoryCleanupPage

​MemoryConflictError

​MemoryPage

​MemoryRecord

​MemoryScope

​MemoryStorageError

​MemoryStore

​MemoryStore.start

​MemoryStore.close

​MemoryStore.put

​MemoryStore.get

​MemoryStore.list

​MemoryStore.search

​MemoryStore.delete

​MemoryStore.delete_all

​MemoryStore.delete_user

​MemoryStore.purge_expired

Retention and limits

Storage maintenance

Exports

InMemoryStore

InMemoryStore.delete_all

InMemoryStore.delete_user

InMemoryStore.purge_expired

InMemoryStore.put

InMemoryStore.get

InMemoryStore.list

InMemoryStore.search

InMemoryStore.delete

InMemoryStore.start

InMemoryStore.close

MemoryCleanupPage

MemoryConflictError

MemoryPage

MemoryRecord

MemoryScope

MemoryStorageError

MemoryStore

MemoryStore.start

MemoryStore.close

MemoryStore.put

MemoryStore.get

MemoryStore.list

MemoryStore.search

MemoryStore.delete

MemoryStore.delete_all

MemoryStore.delete_user

MemoryStore.purge_expired