AsyncEmotionalMemory

emotional_memory.async_engine.AsyncEmotionalMemory(store: AsyncMemoryStore, embedder: AsyncEmbedder, appraisal_engine: AsyncAppraisalEngine | None = None, config: EmotionalMemoryConfig | None = None, state_store: AffectiveStateStore | None = None, query_classifier: QueryClassifier | None = None)

Async-native EmotionalMemory using the AFT pipeline.

All public methods are coroutines. An internal asyncio.Lock protects the affective state from interleaved mutations across concurrent encode / encode_batch calls. Synchronous helpers (set_affect, reset_state, load_state) are intended for setup before concurrent work begins and are not lock-protected — do not call them while encode or observe coroutines may be running.

When using a state_store backed by network I/O (e.g. RedisAffectiveStateStore), constructing the engine will briefly block the event loop to load the persisted snapshot. For fully non-blocking initialisation, pass state_store=None and explicitly call await engine.restore_persisted_state() after construction.

Source code in src/emotional_memory/async_engine.py

def __init__(
    self,
    store: AsyncMemoryStore,
    embedder: AsyncEmbedder,
    appraisal_engine: AsyncAppraisalEngine | None = None,
    config: EmotionalMemoryConfig | None = None,
    state_store: AffectiveStateStore | None = None,
    query_classifier: QueryClassifier | None = None,
) -> None:
    self._store = store
    self._embedder = embedder
    self._appraisal_engine = appraisal_engine
    self._config = config or EmotionalMemoryConfig()
    self._state_store = state_store
    self._state = self._load_initial_state()
    self._state_lock: asyncio.Lock = asyncio.Lock()
    qc_cfg = self._config.retrieval.query_classifier
    if qc_cfg is not None and qc_cfg.mode != "disabled":
        if qc_cfg.mode == "heuristic":
            self._query_classifier: QueryClassifier | None = HeuristicQueryClassifier()
        else:
            self._query_classifier = query_classifier
    else:
        self._query_classifier = None

clear_persisted_state() -> None async

Remove the persisted affective-state snapshot, if configured.

Source code in src/emotional_memory/async_engine.py

async def clear_persisted_state(self) -> None:
    """Remove the persisted affective-state snapshot, if configured."""
    if self._state_store is not None:
        await asyncio.to_thread(self._state_store.clear)

close() -> None async

Release resources held by store, state store, and appraisal engine.

Awaits store.close() when available; falls back to sync close() via asyncio.to_thread; silently skips collaborators without cleanup. The embedder is not owned by the engine and remains the caller's responsibility.

Source code in src/emotional_memory/async_engine.py

async def close(self) -> None:
    """Release resources held by store, state store, and appraisal engine.

    Awaits ``store.close()`` when available; falls back to sync ``close()``
    via ``asyncio.to_thread``; silently skips collaborators without
    cleanup. The embedder is not owned by the engine and remains the
    caller's responsibility.
    """
    close = getattr(self._store, "close", None)
    if callable(close):
        if inspect.iscoroutinefunction(close):
            await close()
        else:
            await asyncio.to_thread(close)
    state_close = getattr(self._state_store, "close", None)
    if callable(state_close):
        await asyncio.to_thread(state_close)
    appraisal_close = getattr(self._appraisal_engine, "close", None)
    if callable(appraisal_close):
        if inspect.iscoroutinefunction(appraisal_close):
            await appraisal_close()
        else:
            await asyncio.to_thread(appraisal_close)

count() -> int async

Return the number of memories in the store.

Source code in src/emotional_memory/async_engine.py

async def count(self) -> int:
    """Return the number of memories in the store."""
    return await self._store.count()

delete(memory_id: str) -> None async

Remove a memory from the store.

Source code in src/emotional_memory/async_engine.py

async def delete(self, memory_id: str) -> None:
    """Remove a memory from the store."""
    await self._store.delete(memory_id)

elaborate(memory_id: str) -> Memory | None async

Run full appraisal on a fast-path (pending) memory and blend core_affect.

Async version of EmotionalMemory.elaborate().

Parameters:

Name	Type	Description	Default
memory_id	str	ID of the memory to elaborate.	required

Returns:

Type	Description
Memory | None	The updated Memory, or None if the memory does not exist,
Memory | None	is not pending appraisal, or no appraisal engine is configured.

Source code in src/emotional_memory/async_engine.py

async def elaborate(self, memory_id: str) -> Memory | None:
    """Run full appraisal on a fast-path (pending) memory and blend core_affect.

    Async version of ``EmotionalMemory.elaborate()``.

    Args:
        memory_id: ID of the memory to elaborate.

    Returns:
        The updated Memory, or None if the memory does not exist,
        is not pending appraisal, or no appraisal engine is configured.
    """
    with traced_span("emotional_memory.elaborate", {"memory_id": memory_id}):
        mem = await self._store.get(memory_id)
        if mem is None:
            return None
        result = await self._elaborate_with_memory(mem)
    return result

elaborate_pending() -> list[Memory] async

Elaborate all memories with pending_appraisal=True (async).

Returns:

Type	Description
list[Memory]	List of updated Memory objects (only those that were elaborated).

Source code in src/emotional_memory/async_engine.py

async def elaborate_pending(self) -> list[Memory]:
    """Elaborate all memories with pending_appraisal=True (async).

    Returns:
        List of updated Memory objects (only those that were elaborated).
    """
    results = []
    for mem in await self._store.list_all():
        if mem.tag.pending_appraisal:
            updated = await self._elaborate_with_memory(mem)
            if updated is not None:
                results.append(updated)
    return results

encode(content: str, appraisal: AppraisalVector | None = None, metadata: dict[str, Any] | None = None) -> Memory async

Encode content into emotional memory (async).

Mirrors EmotionalMemory.encode with awaited I/O calls.

Source code in src/emotional_memory/async_engine.py

async def encode(
    self,
    content: str,
    appraisal: AppraisalVector | None = None,
    metadata: dict[str, Any] | None = None,
) -> Memory:
    """Encode content into emotional memory (async).

    Mirrors ``EmotionalMemory.encode`` with awaited I/O calls.
    """
    check_content_length(content, self._config.max_content_length)
    with traced_span("emotional_memory.encode", {"content_length": len(content)}):
        now = datetime.now(tz=UTC)
        logger.debug("encode start: content_len=%d", len(content))

        # Steps 1-3: resolve affect, update state, and build EmotionalTag.
        tag, _ = await self._build_tag(
            content,
            appraisal,
            metadata,
            now=now,
            allow_fast_path=True,
        )

        # Step 4: embed (async I/O)
        with traced_span("emotional_memory.embed", {"content_length": len(content)}):
            embedding = await self._embedder.embed(content)
        if embedding and bool(np.isnan(np.asarray(embedding)).any()):
            warnings.warn(
                f"Embedder returned NaN values for content (len={len(content)}). "
                "Semantic retrieval will be degraded for this memory.",
                stacklevel=2,
            )

        # Step 5: store (async I/O)
        memory = Memory.create(
            content=content, tag=tag, embedding=embedding, metadata=metadata
        )
        await self._store.save(memory)
        logger.debug(
            "encode stored: id=%s valence=%.3f arousal=%.3f cs=%.3f",
            memory.id,
            tag.core_affect.valence,
            tag.core_affect.arousal,
            tag.consolidation_strength,
        )

        # Step 6: resonance links — pre-filter when store is large (async I/O)
        if self._config.enable_resonance:
            resonance_limit = (
                self._config.resonance.max_links * self._config.resonance.candidate_multiplier
            )
            store_size = await self._store.count()
            if store_size > resonance_limit and memory.embedding is not None:
                with traced_span("emotional_memory.store.search_by_embedding"):
                    candidates = await self._store.search_by_embedding(
                        memory.embedding, resonance_limit
                    )
            else:
                candidates = await self._store.list_all()

            links = build_resonance_links(memory, candidates, self._config.resonance)
            if links:
                updated_tag = tag.model_copy(update={"resonance_links": links})
                memory = memory.model_copy(update={"tag": updated_tag})
                await self._store.update(memory)
                logger.debug("encode resonance: id=%s links=%d", memory.id, len(links))

                await self._add_bidirectional_links(memory, links)
    return memory

encode_batch(contents: list[str], metadata: list[dict[str, Any]] | None = None) -> list[Memory] async

Encode multiple contents using a single embed_batch call (async).

Source code in src/emotional_memory/async_engine.py

async def encode_batch(
    self,
    contents: list[str],
    metadata: list[dict[str, Any]] | None = None,
) -> list[Memory]:
    """Encode multiple contents using a single embed_batch call (async)."""
    if metadata is not None and len(metadata) != len(contents):
        raise ValueError(
            f"metadata length ({len(metadata)}) must match contents length ({len(contents)})"
        )
    with traced_span("emotional_memory.encode_batch", {"batch_size": len(contents)}):
        with traced_span(
            "emotional_memory.embed", {"content_length": sum(len(c) for c in contents)}
        ):
            embeddings = await self._embedder.embed_batch(contents)
        # Dual-path: skip appraisal on fast path. Appraise the whole batch
        # concurrently up front, then evolve state sequentially below.
        use_fast_path = self._config.dual_path_encoding and self._appraisal_engine is not None
        appraisals = await self._appraise_many(contents, metadata, use_fast_path)
        results = []
        for i, (content, embedding) in enumerate(zip(contents, embeddings, strict=True)):
            meta = metadata[i] if metadata else None
            now = datetime.now(tz=UTC)
            computed_a: AppraisalVector | GenericAppraisalVector | None = appraisals[i]

            new_affect = (
                computed_a.to_core_affect()
                if computed_a is not None
                else self._state.core_affect
            )
            # Persist inside the lock (same ordering guarantee as _build_tag).
            async with self._state_lock:
                self._state = self._state.update(
                    new_affect,
                    now=now,
                    mood_alpha=self._config.mood_alpha,
                    mood_decay=self._config.mood_decay,
                )
                _state_snap = self._state
                await self._persist_state_async(_state_snap)

            stored_a: AppraisalVector | None = (
                computed_a if isinstance(computed_a, AppraisalVector) else None
            )
            cs = consolidation_strength(new_affect.arousal, _state_snap.mood.arousal)
            tag = make_emotional_tag(
                core_affect=_state_snap.core_affect,
                momentum=_state_snap.momentum,
                mood=_state_snap.mood,
                consolidation_strength=cs,
                appraisal=stored_a,
            )

            # Mark pending_appraisal for fast-path memories
            if use_fast_path:
                tag = tag.model_copy(update={"pending_appraisal": True})

            # Auto-categorize: attach Plutchik EmotionLabel
            if self._config.auto_categorize:
                tag = label_tag(tag)

            if embedding and bool(np.isnan(np.asarray(embedding)).any()):
                warnings.warn(
                    f"Embedder returned NaN values for content[{i}] (len={len(content)}). "
                    "Semantic retrieval will be degraded for this memory.",
                    stacklevel=2,
                )
            memory = Memory.create(
                content=content, tag=tag, embedding=embedding, metadata=meta
            )
            await self._store.save(memory)

            # Layer-5 ablation: mirror the single-encode guard so batch-encoded
            # memories are not silently given resonance links when disabled.
            if self._config.enable_resonance:
                resonance_limit = (
                    self._config.resonance.max_links
                    * self._config.resonance.candidate_multiplier
                )
                store_size = await self._store.count()
                if store_size > resonance_limit and memory.embedding is not None:
                    candidates = await self._store.search_by_embedding(
                        memory.embedding, resonance_limit
                    )
                else:
                    candidates = await self._store.list_all()

                links = build_resonance_links(memory, candidates, self._config.resonance)
                if links:
                    updated_tag = tag.model_copy(update={"resonance_links": links})
                    memory = memory.model_copy(update={"tag": updated_tag})
                    await self._store.update(memory)

                    await self._add_bidirectional_links(memory, links)

            results.append(memory)
    return results

export_memories() -> list[dict[str, Any]] async

Export all memories as a list of JSON-serialisable dicts (async).

Source code in src/emotional_memory/async_engine.py

async def export_memories(self) -> list[dict[str, Any]]:
    """Export all memories as a list of JSON-serialisable dicts (async)."""
    memories = await self._store.list_all()
    return [m.model_dump(mode="json") for m in memories]

get(memory_id: str) -> Memory | None async

Look up a single memory by ID, or None if not found.

Source code in src/emotional_memory/async_engine.py

async def get(self, memory_id: str) -> Memory | None:
    """Look up a single memory by ID, or None if not found."""
    return await self._store.get(memory_id)

get_current_mood(now: datetime | None = None) -> MoodField

Return the mood field regressed to now without modifying state.

Source code in src/emotional_memory/async_engine.py

def get_current_mood(self, now: datetime | None = None) -> MoodField:
    """Return the mood field regressed to ``now`` without modifying state."""
    if now is None:
        now = datetime.now(tz=UTC)
    if self._config.mood_decay is not None:
        return self._state.mood.regress(now, self._config.mood_decay)
    return self._state.mood

get_state() -> AffectiveState

Return a copy of the current affective state.

Source code in src/emotional_memory/async_engine.py

def get_state(self) -> AffectiveState:
    """Return a copy of the current affective state."""
    return self._state.model_copy()

import_memories(data: list[dict[str, Any]], *, overwrite: bool = False) -> int async

Import memories from a list of dicts produced by export_memories() (async).

Parameters:

Name	Type	Description	Default
data	list[dict[str, Any]]	List of memory dicts (as returned by export_memories()).	required
overwrite	bool	When True, existing memories with the same ID are replaced. When False (default), duplicates are skipped silently.	False

Returns:

Type	Description
int	Number of memories actually written.

Source code in src/emotional_memory/async_engine.py

async def import_memories(self, data: list[dict[str, Any]], *, overwrite: bool = False) -> int:
    """Import memories from a list of dicts produced by ``export_memories()`` (async).

    Args:
        data:      List of memory dicts (as returned by ``export_memories()``).
        overwrite: When True, existing memories with the same ID are replaced.
                   When False (default), duplicates are skipped silently.

    Returns:
        Number of memories actually written.
    """
    written = 0
    for item in data:
        memory = Memory.model_validate(item)
        existing = await self._store.get(memory.id)
        if existing is not None:
            if not overwrite:
                continue
            await self._store.update(memory)
        else:
            await self._store.save(memory)
        written += 1
    return written

list_all() -> list[Memory] async

Return all memories in the store.

Source code in src/emotional_memory/async_engine.py

async def list_all(self) -> list[Memory]:
    """Return all memories in the store."""
    return await self._store.list_all()

load_state(data: dict[str, Any]) -> None

Restore a previously saved affective state.

Source code in src/emotional_memory/async_engine.py

def load_state(self, data: dict[str, Any]) -> None:
    """Restore a previously saved affective state."""
    self._state = AffectiveState.restore(data)
    self._persist_state_sync()

observe(content: str, appraisal: AppraisalVector | None = None, metadata: dict[str, Any] | None = None) -> EmotionalTag async

Update affective state from content without storing a retrievable memory.

Source code in src/emotional_memory/async_engine.py

async def observe(
    self,
    content: str,
    appraisal: AppraisalVector | None = None,
    metadata: dict[str, Any] | None = None,
) -> EmotionalTag:
    """Update affective state from content without storing a retrievable memory."""
    check_content_length(content, self._config.max_content_length)
    with traced_span("emotional_memory.observe", {"content_length": len(content)}):
        tag, _ = await self._build_tag(
            content,
            appraisal,
            metadata,
            now=datetime.now(tz=UTC),
            allow_fast_path=False,
        )
    return tag

persist_state() -> dict[str, Any] async

Return and, when configured, persist the current affective state.

Source code in src/emotional_memory/async_engine.py

async def persist_state(self) -> dict[str, Any]:
    """Return and, when configured, persist the current affective state."""
    await self._persist_state_async()
    return self.save_state()

prune(threshold: float = 0.05) -> int async

Remove memories whose effective strength has fallen below threshold (async).

Parameters:

Name	Type	Description	Default
threshold	float	Minimum acceptable strength in [0, 1]. Defaults to 0.05 (5%).	0.05

Returns:

Type	Description
int	Number of memories removed.

Source code in src/emotional_memory/async_engine.py

async def prune(self, threshold: float = 0.05) -> int:
    """Remove memories whose effective strength has fallen below *threshold* (async).

    Args:
        threshold: Minimum acceptable strength in [0, 1].  Defaults to 0.05 (5%).

    Returns:
        Number of memories removed.
    """
    from emotional_memory.decay import compute_effective_strength_batch

    validate_prune_threshold(threshold)
    with traced_span("emotional_memory.prune", {"threshold": threshold}):
        now = datetime.now(tz=UTC)
        mems = await self._store.list_all()
        if mems:
            strengths = compute_effective_strength_batch(
                [m.tag for m in mems], now, self._config.decay
            )
            to_delete = [m.id for m, s in zip(mems, strengths, strict=True) if s < threshold]
        else:
            to_delete = []
        for memory_id in to_delete:
            await self._store.delete(memory_id)
    return len(to_delete)

reset_state() -> None

Reset the runtime affective state to its initial baseline.

Source code in src/emotional_memory/async_engine.py

def reset_state(self) -> None:
    """Reset the runtime affective state to its initial baseline."""
    self._state = AffectiveState.initial()
    self._persist_state_sync()

restore_persisted_state() -> bool async

Load the last persisted affective state from the configured state store.

Source code in src/emotional_memory/async_engine.py

async def restore_persisted_state(self) -> bool:
    """Load the last persisted affective state from the configured state store."""
    if self._state_store is None:
        return False
    persisted = await asyncio.to_thread(self._state_store.load)
    if persisted is None:
        return False
    self._state = persisted.model_copy()
    return True

retrieve(query: str, top_k: int = 5, *, query_affect: CoreAffect | None = None, precomputed_weights: NDArray[np.float64] | None = None) -> list[Memory] async

Retrieve the top-k most relevant memories for the query (async).

Mirrors EmotionalMemory.retrieve with awaited I/O calls. query_affect overrides the affect-proximity signal source (default: runtime state).

Source code in src/emotional_memory/async_engine.py

async def retrieve(
    self,
    query: str,
    top_k: int = 5,
    *,
    query_affect: CoreAffect | None = None,
    precomputed_weights: NDArray[np.float64] | None = None,
) -> list[Memory]:
    """Retrieve the top-k most relevant memories for the query (async).

    Mirrors ``EmotionalMemory.retrieve`` with awaited I/O calls. ``query_affect``
    overrides the affect-proximity signal source (default: runtime state).
    """
    if top_k < 1:
        raise ValueError(f"top_k must be >= 1, got {top_k}")
    check_content_length(query, self._config.max_content_length, label="query")
    with traced_span(
        "emotional_memory.retrieve",
        {"query_length": len(query), "top_k": top_k},
    ):
        now = datetime.now(tz=UTC)
        store_count = await self._store.count()
        logger.debug(
            "retrieve start: query_len=%d top_k=%d store=%d", len(query), top_k, store_count
        )
        with traced_span("emotional_memory.embed", {"content_length": len(query)}):
            query_embedding = await self._embedder.embed(query)

        # G2 pre-filter (async)
        rc = self._config.retrieval
        candidate_limit = top_k * rc.candidate_multiplier
        if store_count > candidate_limit:
            with traced_span(
                "emotional_memory.store.search_by_embedding", {"limit": candidate_limit}
            ):
                candidates = await self._store.search_by_embedding(
                    query_embedding, candidate_limit
                )
        else:
            candidates = await self._store.list_all()

        if not candidates:
            return []

        _spreading_fn = (
            (lambda *a, **kw: {})
            if not self._config.enable_resonance
            else spreading_activation
        )
        # Single consistent snapshot: ``self._state`` is reassigned (not mutated)
        # under ``_state_lock`` by concurrent encodes, so read it once to avoid a
        # torn mix of core_affect / mood / momentum from different states.
        state = self._state
        plan = build_retrieval_plan(
            query_embedding=query_embedding,
            query_affect=query_affect or state.core_affect,
            current_mood=state.mood,
            current_momentum=state.momentum,
            candidates=candidates,
            top_k=top_k,
            now=now,
            decay_config=self._config.decay,
            retrieval_config=rc,
            propagation_hops=self._config.resonance.propagation_hops,
            spreading_activation_fn=_spreading_fn,
            precomputed_weights=(
                self._effective_retrieval_weights(query, mood=state.mood)
                if precomputed_weights is None
                else precomputed_weights
            ),
        )
        top = [item.memory for item in plan.pass2[:top_k]]
        result = await self._apply_retrieval_updates(top, now)
        logger.debug("retrieve done: returned=%d candidates=%d", len(result), len(candidates))
    return result

retrieve_query_gated(query: str, top_k: int = 5, *, tau: float | None = None) -> list[Memory] async

Retrieve with the Addendum Y query-affect gate (Branch A, async).

Mirrors EmotionalMemory.retrieve_query_gated.

Source code in src/emotional_memory/async_engine.py

async def retrieve_query_gated(
    self, query: str, top_k: int = 5, *, tau: float | None = None
) -> list[Memory]:
    """Retrieve with the Addendum Y query-affect gate (Branch A, async).

    Mirrors ``EmotionalMemory.retrieve_query_gated``.
    """
    if self._appraisal_engine is None:
        raise RuntimeError(
            "retrieve_query_gated requires an appraisal_engine; "
            "none is configured. Pass appraisal_engine=... to AsyncEmotionalMemory."
        )
    effective_tau = self._config.query_affect_gate_tau if tau is None else tau
    appraisal = await self._appraisal_engine.appraise(query)
    query_affect = appraisal.to_core_affect()
    if is_query_affect_neutral(query_affect.valence, effective_tau):
        return await self.retrieve(
            query, top_k=top_k, precomputed_weights=semantic_only_weights()
        )
    return await self.retrieve(query, top_k=top_k, query_affect=query_affect)

retrieve_with_explanations(query: str, top_k: int = 5, *, query_affect: CoreAffect | None = None, precomputed_weights: NDArray[np.float64] | None = None) -> list[RetrievalExplanation] async

Async retrieval variant that exposes the ranking breakdown.

query_affect overrides the affect-proximity signal source (default: runtime state), as in retrieve().

Source code in src/emotional_memory/async_engine.py

async def retrieve_with_explanations(
    self,
    query: str,
    top_k: int = 5,
    *,
    query_affect: CoreAffect | None = None,
    precomputed_weights: NDArray[np.float64] | None = None,
) -> list[RetrievalExplanation]:
    """Async retrieval variant that exposes the ranking breakdown.

    ``query_affect`` overrides the affect-proximity signal source (default:
    runtime state), as in ``retrieve()``.
    """
    if top_k < 1:
        raise ValueError(f"top_k must be >= 1, got {top_k}")
    check_content_length(query, self._config.max_content_length, label="query")
    now = datetime.now(tz=UTC)
    store_count = await self._store.count()
    logger.debug(
        "retrieve_with_explanations start: query_len=%d top_k=%d store=%d",
        len(query),
        top_k,
        store_count,
    )
    query_embedding = await self._embedder.embed(query)

    rc = self._config.retrieval
    candidate_limit = top_k * rc.candidate_multiplier
    if store_count > candidate_limit:
        candidates = await self._store.search_by_embedding(query_embedding, candidate_limit)
    else:
        candidates = await self._store.list_all()

    if not candidates:
        return []

    _spreading_fn = (
        (lambda *a, **kw: {}) if not self._config.enable_resonance else spreading_activation
    )
    # Single consistent snapshot (see ``retrieve``): avoid a torn read of the
    # affective state under concurrent encodes.
    state = self._state
    plan = build_retrieval_plan(
        query_embedding=query_embedding,
        query_affect=query_affect or state.core_affect,
        current_mood=state.mood,
        current_momentum=state.momentum,
        candidates=candidates,
        top_k=top_k,
        now=now,
        decay_config=self._config.decay,
        retrieval_config=rc,
        propagation_hops=self._config.resonance.propagation_hops,
        spreading_activation_fn=_spreading_fn,
        precomputed_weights=(
            self._effective_retrieval_weights(query, mood=state.mood)
            if precomputed_weights is None
            else precomputed_weights
        ),
    )
    top_ranked = plan.pass2[:top_k]
    result = await self._apply_retrieval_updates([item.memory for item in top_ranked], now)
    updated_by_id = {mem.id: mem for mem in result}
    pass1_ranks = {item.memory.id: index for index, item in enumerate(plan.pass1, start=1)}

    explanations = [
        RetrievalExplanation(
            memory=updated_by_id[item.memory.id],
            score=item.score,
            breakdown=item.breakdown,
            activation_level=plan.activation_map.get(item.memory.id, 0.0),
            pass1_rank=pass1_ranks.get(item.memory.id),
            pass2_rank=index,
            selected_as_seed=item.memory.id in plan.seed_ids,
            candidate_count=plan.candidate_count,
        )
        for index, item in enumerate(top_ranked, start=1)
    ]
    logger.debug(
        "retrieve_with_explanations done: returned=%d candidates=%d",
        len(explanations),
        len(candidates),
    )
    return explanations

retrieve_with_query_appraisal(query: str, top_k: int = 5) -> list[Memory] async

Retrieve using the query's own appraised affect (Addendum T, async).

Mirrors EmotionalMemory.retrieve_with_query_appraisal: appraises the query text and uses the resulting affect as query_affect without mutating runtime state. Pairs best with DIRECT_VAD_SCHEMA.

Raises:

Type	Description
RuntimeError	if no appraisal engine is configured.

Source code in src/emotional_memory/async_engine.py

async def retrieve_with_query_appraisal(self, query: str, top_k: int = 5) -> list[Memory]:
    """Retrieve using the query's own appraised affect (Addendum T, async).

    Mirrors ``EmotionalMemory.retrieve_with_query_appraisal``: appraises the
    query text and uses the resulting affect as ``query_affect`` without
    mutating runtime state. Pairs best with ``DIRECT_VAD_SCHEMA``.

    Raises:
        RuntimeError: if no appraisal engine is configured.
    """
    if self._appraisal_engine is None:
        raise RuntimeError(
            "retrieve_with_query_appraisal requires an appraisal_engine; "
            "none is configured. Pass appraisal_engine=... to AsyncEmotionalMemory, "
            "or call retrieve(query, query_affect=...) with an explicit affect."
        )
    appraisal = await self._appraisal_engine.appraise(query)
    return await self.retrieve(query, top_k=top_k, query_affect=appraisal.to_core_affect())

save_state() -> dict[str, Any]

Serialise the current affective state for persistence.

Source code in src/emotional_memory/async_engine.py

def save_state(self) -> dict[str, Any]:
    """Serialise the current affective state for persistence."""
    return self._state.snapshot()

set_affect(core_affect: CoreAffect) -> None

Manually inject a CoreAffect, updating state synchronously.

Source code in src/emotional_memory/async_engine.py

def set_affect(self, core_affect: CoreAffect) -> None:
    """Manually inject a CoreAffect, updating state synchronously."""
    self._state = self._state.update(
        core_affect,
        mood_alpha=self._config.mood_alpha,
        mood_decay=self._config.mood_decay,
    )
    self._persist_state_sync()

update_memory(memory_id: str, *, content: str | None = None, tag: EmotionalTag | None = None, metadata: dict[str, Any] | None = None) -> Memory async

Replace selected fields of an existing memory asynchronously.

Source code in src/emotional_memory/async_engine.py

async def update_memory(
    self,
    memory_id: str,
    *,
    content: str | None = None,
    tag: EmotionalTag | None = None,
    metadata: dict[str, Any] | None = None,
) -> Memory:
    """Replace selected fields of an existing memory asynchronously."""
    memory = await self._store.get(memory_id)
    if memory is None:
        raise KeyError(f"memory not found: {memory_id}")
    changes: dict[str, Any] = {}
    if content is not None:
        check_content_length(content, self._config.max_content_length)
        changes["content"] = content
        changes["embedding"] = await self._embedder.embed(content)
    if tag is not None:
        changes["tag"] = tag
    if metadata is not None:
        changes["metadata"] = metadata
    updated = memory.model_copy(update=changes)
    await self._store.update(updated)
    return updated