Events

Overview

Every significant operation in Mnesis publishes an event to an in-process EventBus. The bus enables external monitoring — compaction progress, message creation, doom loop detection, operator fan-out — without polling or modifying core logic.

Each MnesisSession owns its own EventBus instance. Operators (LLMMap, AgenticMap) do not create a bus by default: they only emit MAP_* events when an EventBus is explicitly passed (for example, by injecting the session bus at construction time to unify all events on one subscriber).

Key properties of the bus:

• Sync handlers are called inline within publish(). For a given event, handlers registered with subscribe() run first (in their registration order), followed by handlers registered with subscribe_all() (also in their registration order).
• Async handlers are scheduled as background tasks (asyncio.create_task), non-blocking. A handler may return any awaitable (a coroutine, an asyncio.Future, or an object implementing __await__); non-coroutine awaitables are wrapped and awaited. With no running event loop, async handlers are skipped.
• Async handlers follow the same per-event-then-global ordering as sync handlers.
• Handler exceptions are logged (event_handler_error) and swallowed — they never propagate to the publisher. For async handlers the error is logged when the handler task finishes; the bus keeps a strong reference to each in-flight handler task so it cannot be garbage-collected mid-flight.
• unsubscribe() is a silent no-op if the handler is not registered.

---

Subscribing to Events

import asyncio
from mnesis import MnesisSession, MnesisConfig
from mnesis.events.bus import MnesisEvent
from mnesis.events.payloads import CompactionCompletedPayload, MessageCreatedPayload

async def main() -> None:
    session = await MnesisSession.create(model="openai/gpt-4o")

    # Sync handler — called inline, no await needed
    def on_message(event: MnesisEvent, payload: MessageCreatedPayload) -> None:
        print(f"[{event}] role={payload['role']} id={payload['message_id']}")

    # Async handler — scheduled as a background task
    async def on_compaction(event: MnesisEvent, payload: CompactionCompletedPayload) -> None:
        print(
            f"[{event}] level={payload['level_used']} "
            f"tokens {payload['tokens_before']} → {payload['tokens_after']}"
        )

    session.event_bus.subscribe(MnesisEvent.MESSAGE_CREATED, on_message)  # type: ignore[arg-type]
    session.event_bus.subscribe(MnesisEvent.COMPACTION_COMPLETED, on_compaction)  # type: ignore[arg-type]

    async with session:
        result = await session.send("Hello, world!")
        print(result.text)

    # Clean up when done
    session.event_bus.unsubscribe(MnesisEvent.MESSAGE_CREATED, on_message)

To subscribe to every event type with a single handler, use subscribe_all:

def log_all(event: MnesisEvent, payload: dict) -> None:  # type: ignore[type-arg]
    print(f"EVENT {event}: {payload}")

session.event_bus.subscribe_all(log_all)

---

Event Reference

Currently Published Events

Event	Value	When it fires	Payload class
SESSION_CREATED	session.created	Session is initialised	SessionCreatedPayload
SESSION_CLOSED	session.closed	Session context manager exits	SessionClosedPayload
MESSAGE_CREATED	message.created	A user or assistant message is persisted	MessageCreatedPayload
COMPACTION_TRIGGERED	compaction.triggered	Token usage crosses the soft or hard threshold	CompactionTriggeredPayload
COMPACTION_COMPLETED	compaction.completed	A compaction pass finishes (success or partial)	CompactionCompletedPayload
COMPACTION_FAILED	compaction.failed	Compaction raises an unhandled exception, or is aborted (aborted=True)	CompactionFailedPayload
PRUNE_COMPLETED	prune.completed	A compaction run's pruning step tombstoned at least one tool output (not published when nothing was pruned)	PruneCompletedPayload
DOOM_LOOP_DETECTED	doom_loop.detected	The same tool call repeats past the threshold	DoomLoopDetectedPayload
LLM_RETRY	llm.retry	send() or a compaction LLM call is about to retry after a transient LLM error	LlmRetryPayload
MAP_STARTED	map.started	An operator begins processing its item list	MapStartedPayload
MAP_ITEM_COMPLETED	map.item_completed	One item finishes (success or failure)	MapItemCompletedPayload
MAP_COMPLETED	map.completed	All items have been processed	MapCompletedPayload

Reserved (Defined but Not Yet Published)

Event	Value	Notes
SESSION_UPDATED	session.updated	Reserved for future use
SESSION_DELETED	session.deleted	Reserved for future use
MESSAGE_UPDATED	message.updated	Reserved for future use
PART_CREATED	part.created	Reserved for streaming granularity
PART_UPDATED	part.updated	Reserved for streaming granularity

---

Payload Schemas

Each event carries a typed TypedDict payload. Import the payload classes from mnesis.events.payloads for IDE auto-complete and static type checking.

SessionCreatedPayload

Fired by: MnesisEvent.SESSION_CREATED

Field	Type	Description
session_id	str	The newly created session's ID
model	str	The LLM model string the session was created with

SessionClosedPayload

Fired by: MnesisEvent.SESSION_CLOSED

Field	Type	Description
session_id	str	The session that was closed

MessageCreatedPayload

Fired by: MnesisEvent.MESSAGE_CREATED

Field	Type	Description
message_id	str	The newly persisted message's ID
role	str	"user" or "assistant"

CompactionTriggeredPayload

Fired by: MnesisEvent.COMPACTION_TRIGGERED

Field	Type	Description
session_id	str	The session whose compaction was triggered
tokens	int	Size of the current context window (system prompt, live summaries and raw messages) that crossed the soft threshold. Earlier versions reported lifetime cumulative tokens.

CompactionCompletedPayload

Fired by: MnesisEvent.COMPACTION_COMPLETED

Field	Type	Description
session_id	str	The session that was compacted
summary_message_id	str	ID of the newly created summary message, or "" on failure
level_used	int	Escalation level used: 1 = selective LLM, 2 = aggressive LLM, 3 = deterministic
compacted_message_count	int	Number of messages folded into summaries, cumulative across the run's summarisation passes
summary_token_count	int	Token count of the summary content produced: the sum over the run's leaf passes, or the condensed node's size after condensation
tokens_before	int	Size of the active context window before compaction (the same measure the thresholds use)
tokens_after	int	Size of the active context window after compaction, older live summaries included
elapsed_ms	float	Wall-clock time the compaction pass took
pruned_tool_outputs	int	Number of tool outputs tombstoned during this run
pruned_tokens	int	Tokens reclaimed by pruning tool outputs

CompactionFailedPayload

Fired by: MnesisEvent.COMPACTION_FAILED

Field	Type	Description
session_id	str	The session whose compaction failed
error	str	Human-readable error description
aborted	bool	True only for an abort: the abort event passed to run_compaction() / check_and_trigger(), or session.close(abort_compaction=True); a plain session.close() waits for the run instead

PruneCompletedPayload

Fired by: MnesisEvent.PRUNE_COMPLETED

Published by the compaction engine's pruning step, only when at least one tool output was tombstoned.

Field	Type	Description
session_id	str	The session whose tool outputs were pruned
pruned_count	int	Number of tool output parts tombstoned in this prune pass
pruned_tokens	int	Estimated tokens reclaimed by this prune pass

DoomLoopDetectedPayload

Fired by: MnesisEvent.DOOM_LOOP_DETECTED

Field	Type	Description
session_id	str	The session where the doom loop was detected
tool	str	Name of the tool call that was repeated past the threshold

LlmRetryPayload

Fired by: MnesisEvent.LLM_RETRY

Published on each retry attempt, immediately before the backoff sleep begins, for send() turns and for compaction LLM calls. Only fires when RetryConfig.max_retries > 0 and the error is retryable. source tells the two apart; stage and compaction_level are present only for compaction.

Field	Type	Description
session_id	str	The session retrying the LLM call
attempt	int	Current attempt number (1-based: first retry is 1)
max_retries	int	Maximum retries configured for this session
error_type	str	Fully-qualified exception class name (e.g. "litellm.exceptions.RateLimitError")
error_message	str	Human-readable error message from the exception
delay_seconds	float	Seconds the session will sleep before the next attempt
source	str (optional)	"send" for a turn's LLM call, "compaction" for a compaction call. Absent on payloads from versions before this key existed: treat a missing value as "send"
stage	str (optional)	Compaction only: "summarisation" or "condensation"
compaction_level	int (optional)	Compaction only: escalation level (1 or 2) of the call being retried

---

Operator Events

MAP_STARTED, MAP_ITEM_COMPLETED, and MAP_COMPLETED are emitted by operators only when they are constructed with an event_bus=... argument, and the events are published on that provided bus. To receive them via the session bus, pass session.event_bus when constructing the operator:

from mnesis.operators import LLMMap, AgenticMap
from mnesis.events.payloads import MapStartedPayload, MapItemCompletedPayload, MapCompletedPayload

# Inject the session bus so operator events appear on it
llm_map = LLMMap(config.operators, event_bus=session.event_bus)
agentic_map = AgenticMap(config.operators, event_bus=session.event_bus)

MapStartedPayload

Fired by: MnesisEvent.MAP_STARTED

Field	Type	Present for	Description
total	int	All operators	Total number of items to process
model	str (optional)	LLMMap only	LLM model string
type	str (optional)	AgenticMap only	Operator type string ("agentic")

MapItemCompletedPayload

Fired by: MnesisEvent.MAP_ITEM_COMPLETED (once per item)

Field	Type	Description
completed	int	Number of items completed so far (including this one)
total	int	Total number of items
success	bool	Whether this item succeeded

MapCompletedPayload

Fired by: MnesisEvent.MAP_COMPLETED

Field	Type	Present for	Description
total	int	All operators	Total number of items processed
completed	int (optional)	LLMMap only	Number of items that have finished, including failures (not a success count; use MapItemCompletedPayload.success or the yielded MapResult.success per item). Always equal to total when published

AgenticMap omits completed

AgenticMap publishes only total in its MAP_COMPLETED payload. LLMMap includes both total and completed.

Not published on early close

MAP_COMPLETED is published only when every item has finished. If the generator is closed early, it is not published.

---

Complete Example

The following snippet wires up multiple event handlers, runs a session, and then uses an operator — all events on a single bus:

import asyncio
from pydantic import BaseModel

from mnesis import MnesisConfig, MnesisSession
from mnesis.events.bus import MnesisEvent
from mnesis.events.payloads import (
    CompactionCompletedPayload,
    DoomLoopDetectedPayload,
    MapCompletedPayload,
    MapItemCompletedPayload,
    MessageCreatedPayload,
    SessionCreatedPayload,
)
from mnesis.operators import LLMMap


class Sentiment(BaseModel):
    label: str  # "positive" | "negative" | "neutral"
    confidence: float


async def main() -> None:
    config = MnesisConfig()
    session = await MnesisSession.create(model="openai/gpt-4o", config=config)

    # ── event handlers ────────────────────────────────────────────────────────

    def on_session_created(event: MnesisEvent, payload: SessionCreatedPayload) -> None:
        print(f"Session started: {payload['session_id']} using {payload['model']}")

    def on_message(event: MnesisEvent, payload: MessageCreatedPayload) -> None:
        print(f"  Message persisted: role={payload['role']} id={payload['message_id']}")

    def on_compaction(event: MnesisEvent, payload: CompactionCompletedPayload) -> None:
        print(
            f"  Compacted at level {payload['level_used']}: "
            f"{payload['tokens_before']} → {payload['tokens_after']} tokens "
            f"({payload['elapsed_ms']:.0f} ms)"
        )

    def on_doom_loop(event: MnesisEvent, payload: DoomLoopDetectedPayload) -> None:
        print(f"  WARNING: doom loop on tool '{payload['tool']}'")

    def on_map_item(event: MnesisEvent, payload: MapItemCompletedPayload) -> None:
        status = "OK" if payload["success"] else "FAIL"
        print(f"  Map item {payload['completed']}/{payload['total']}: {status}")

    def on_map_done(event: MnesisEvent, payload: MapCompletedPayload) -> None:
        print(f"  Map complete: {payload['total']} items processed")

    # ── register handlers ─────────────────────────────────────────────────────

    session.event_bus.subscribe(MnesisEvent.SESSION_CREATED, on_session_created)  # type: ignore[arg-type]
    session.event_bus.subscribe(MnesisEvent.MESSAGE_CREATED, on_message)  # type: ignore[arg-type]
    session.event_bus.subscribe(MnesisEvent.COMPACTION_COMPLETED, on_compaction)  # type: ignore[arg-type]
    session.event_bus.subscribe(MnesisEvent.DOOM_LOOP_DETECTED, on_doom_loop)  # type: ignore[arg-type]
    session.event_bus.subscribe(MnesisEvent.MAP_ITEM_COMPLETED, on_map_item)  # type: ignore[arg-type]
    session.event_bus.subscribe(MnesisEvent.MAP_COMPLETED, on_map_done)  # type: ignore[arg-type]

    # ── run session ───────────────────────────────────────────────────────────

    async with session:
        result = await session.send("What is the capital of France?")
        print(f"Response: {result.text}")

        # Operator events flow to the session bus because we injected it
        llm_map = LLMMap(config.operators, event_bus=session.event_bus)
        reviews = ["Great product!", "Terrible quality.", "It is okay."]
        batch = await llm_map.run_all(
            inputs=reviews,
            prompt_template="Classify the sentiment of: {{ item }}",
            output_schema=Sentiment,
            model="openai/gpt-4o",
        )
        for result in batch.successes:
            sentiment = result.output
            if isinstance(sentiment, Sentiment):
                print(f"  {result.input!r} → {sentiment.label} ({sentiment.confidence:.2f})")


if __name__ == "__main__":
    asyncio.run(main())

Async handlers and task lifetime

Async handlers are scheduled via asyncio.create_task and are not awaited by the EventBus. If your program exits immediately after the last await, pending handler tasks may be cancelled before they complete. In long-running services this is rarely an issue; in short scripts, keep the event loop alive long enough for handlers to run, or have your handlers manage any long-running work they spawn if you need deterministic completion.