`mnesis.compaction`¶

compaction ¶

Mnesis compaction components.

CompactionEngine ¶

CompactionEngine(
    store: ImmutableStore,
    dag_store: SummaryDAGStore,
    token_estimator: TokenEstimator,
    event_bus: EventBus,
    config: MnesisConfig,
    id_generator: Any = None,
    session_model: str | None = None,
    model_info: ModelInfo | None = None,
    context_measure: Callable[[str], Awaitable[int]]
    | None = None,
)

Orchestrates the full compaction protocol (summarise → condense → loop).

Guarantees: - run_compaction() never raises on failure — errors are caught and Level 3 runs. Only external task cancellation (CancelledError) propagates. - The resulting summary always fits within the token budget. - Level 3 (deterministic) is the final fallback and always succeeds. - Atomic SQLite commit per round: partial failures leave no inconsistent state. - The EventBus receives COMPACTION_COMPLETED (or COMPACTION_FAILED).

Threshold semantics:

Soft threshold (soft_threshold_fraction, default 60 %): triggers early background compaction via check_and_trigger(). This keeps the context lean well before the hard limit.
Hard threshold (100 % of usable): checked inside session.send() before the LLM call; if exceeded, the caller should await wait_for_pending() to block until compaction completes.

Both thresholds take the size of the current context window in tokens (see :attr:mnesis.context.builder.BuiltContext.context_tokens), never lifetime usage. At most one compaction is in flight per engine: :meth:check_and_trigger is a no-op while a previous task is running, and :meth:compact_exclusive (manual compaction) waits for it first.

Example::

engine = CompactionEngine(store, dag_store, estimator, event_bus, config)
if engine.is_soft_overflow(tokens, model_info):
    engine.check_and_trigger(session_id, tokens, model_info)  # non-blocking
if engine.is_hard_overflow(tokens, model_info):
    await engine.wait_for_pending()  # block if must compact before LLM call

stalled `property` ¶

stalled: bool

True while background auto-compaction is paused as futile.

Set when a run ended at/above the soft threshold with nothing left to summarise or condense; cleared by :meth:note_user_turn or by a run that gets under the soft threshold. Never affects the blocking hard-limit path.

more_to_compact `property` ¶

more_to_compact: bool

True if the last run ended with work a further run could still do.

in_flight `property` ¶

in_flight: bool

True while a compaction task (background or manual) is running.

has_pending `property` ¶

has_pending: bool

True if a task handle is held, running or finished but not yet reaped.

note_user_turn ¶

note_user_turn() -> None

Record that a new user turn was appended, re-arming a paused trigger.

A new user turn can make older messages summarisable (they leave the protected tail), so a pause based on the previous history no longer holds.

set_context_measure ¶

set_context_measure(
    measure: Callable[[str], Awaitable[int]] | None,
) -> None

Set the callable that returns the session's current context size.

is_soft_overflow ¶

is_soft_overflow(
    tokens: TokenUsage | int, model: ModelInfo
) -> bool

Return True if the soft threshold has been crossed.

The soft threshold triggers early background compaction (non-blocking).

Parameters:

Name	Type	Description	Default
`tokens`	`TokenUsage \| int`	Size of the current context window in tokens (an `int`, e.g. `BuiltContext.context_tokens`). A `TokenUsage` is also accepted and compared by its `effective_total()`.	required
`model`	`ModelInfo`	Model metadata providing context limit.	required

Returns:

Type	Description
`bool`	True if tokens reach `soft_threshold_fraction * usable`.

is_hard_overflow ¶

is_hard_overflow(
    tokens: TokenUsage | int, model: ModelInfo
) -> bool

Return True if the hard threshold has been crossed.

The hard threshold means the current context must be compacted before the next LLM call to avoid an over-limit request.

Parameters:

Name	Type	Description	Default
`tokens`	`TokenUsage \| int`	Size of the current context window in tokens (see :meth:`is_soft_overflow`).	required
`model`	`ModelInfo`	Model metadata providing context limit.	required

Returns:

Type	Description
`bool`	True if tokens reach the full usable budget.

is_overflow ¶

is_overflow(
    tokens: TokenUsage | int, model: ModelInfo
) -> bool

Return True if compaction should be triggered (soft threshold check).

Preserved for backwards compatibility; delegates to :meth:is_soft_overflow.

Parameters:

Name	Type	Description	Default
`tokens`	`TokenUsage \| int`	Size of the current context window in tokens.	required
`model`	`ModelInfo`	Model metadata providing context limit.	required

Returns:

Type	Description
`bool`	True if tokens exceed the soft threshold.

check_and_trigger ¶

check_and_trigger(
    session_id: str,
    tokens: TokenUsage | int,
    model: ModelInfo,
    abort: Event | None = None,
    force: bool = False,
    full_drain: bool = False,
) -> bool

Check for overflow and trigger async compaction if needed.

Non-blocking — schedules a background task and returns immediately. The task handle is stored in self._pending_task so callers can await or cancel it during shutdown. At most one compaction is in flight: while a previous task is still running this is a no-op, so rapid overflowing turns cannot run concurrent compactions (duplicate summaries / conflicting context-item swaps) or orphan the tracked task.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to compact.	required
`tokens`	`TokenUsage \| int`	Size of the current context window in tokens.	required
`model`	`ModelInfo`	Model info for overflow detection.	required
`abort`	`Event \| None`	Optional event to signal early termination.	`None`
`force`	`bool`	The caller is about to block on the result (the hard-limit path): bypass the stall pause and stop the run once the context is under the soft threshold, leaving the rest of the drain to the next background run.	`False`
`full_drain`	`bool`	With `force`, run to the normal condensation target instead of stopping at the soft threshold.	`False`

Returns:

Type	Description
`bool`	True if a new compaction task was scheduled, False if the context
`bool`	is under the soft threshold, a compaction is already in flight, or
`bool`	background compaction is paused (see below).

A run that ends still at/above the soft threshold with nothing left to summarise or condense (typically: the protected last two user turns alone exceed it) cannot be improved by running again. Re-triggering on every turn would burn LLM calls without progress, so the background trigger is paused until a new user turn arrives (:meth:note_user_turn), which can make older messages summarisable. A warning is logged once when the engine pauses. The pause never applies with force=True: the blocking hard-limit path always re-evaluates. A futile run makes no summarisation calls; it can only repeat a condensation for a set of live summaries that is new since the last condensation that failed to shrink them even at the deterministic level.

wait_for_pending `async` ¶

wait_for_pending() -> CompactionResult | None

Await any in-flight background compaction task to completion, then release it.

The handle is released unless a newer run replaced it while waiting (that run stays tracked; loop on :attr:has_pending to drain).

Cancelling the caller does not cancel the compaction: it is shielded, so the run continues in the background, the handle is kept, and a later call (or close()) collects it. asyncio.CancelledError still propagates to the cancelled caller.

Returns:

Name	Type	Description
`The`	`CompactionResult \| None`	class:`CompactionResult` of the pending task (whether it was
	`CompactionResult \| None`	still running or had already finished), or `None` if no task was
	`CompactionResult \| None`	pending or it was cancelled. :meth:`run_compaction` never raises:
	`CompactionResult \| None`	a failed run is returned as a stub result with `level_used == 0`
	`CompactionResult \| None`	and an empty `summary_message_id`, so callers should check
	`CompactionResult \| None`	`level_used` before treating the result as a real summary.

compact_exclusive `async` ¶

compact_exclusive(
    session_id: str, abort: Event | None = None
) -> CompactionResult

Run a compaction as the tracked in-flight task, after any pending one.

Used for manual compaction so that it never overlaps a background run (duplicate summaries / conflicting context-item swaps). abort ends the run early with the stub result, as for :meth:run_compaction.

run_compaction `async` ¶

run_compaction(
    session_id: str,
    abort: Event | None = None,
    model_override: str | None = None,
    until_under_hard: bool = False,
) -> CompactionResult

Run the full compaction protocol. Never raises on failure.

Setting abort ends the run with the stub failure result (level_used == 0) and a COMPACTION_FAILED event whose payload has aborted=True. External task cancellation (Task.cancel()) is different: it is not swallowed and asyncio.CancelledError propagates to the caller.

Note

The two cases are told apart with Task.cancelling(). If the calling task earlier swallowed a cancellation without calling uncancel(), its stale count makes an abort propagate as a cancellation instead of returning the stub. Mnesis always runs compaction in a fresh task, so it cannot hit this; callers awaiting run_compaction(abort=...) directly from a long-lived task should uncancel() after swallowing a cancel.

Steps (per round, up to max_compaction_rounds): 1. Run tool output pruner (reduce input size first). 2. Summarise raw messages: level 1 → level 2 → level 3. 3. Condense accumulated summary nodes if still over budget: lvl 1→2→3. 4. If no progress was made, break early to avoid spinning.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to compact.	required
`abort`	`Event \| None`	Optional asyncio.Event — checked before each level attempt.	`None`
`model_override`	`str \| None`	Override compaction model (for testing).	`None`
`until_under_hard`	`bool`	Stop draining once the context is under the soft threshold (used when a caller blocks on the run) instead of continuing down to the condensation target.	`False`

Returns:

Type	Description
`CompactionResult`	CompactionResult describing what happened.

SummaryCandidate `dataclass` ¶

SummaryCandidate(
    text: str,
    token_count: int,
    span_start_message_id: str,
    span_end_message_id: str,
    compaction_level: int,
    messages_covered: int,
)

A candidate compaction summary before it is committed to the store.

ToolOutputPruner ¶

ToolOutputPruner(
    store: ImmutableStore,
    estimator: TokenEstimator,
    config: MnesisConfig,
)

Reclaims context space by tombstoning stale tool outputs.

The protect window ensures the most recent tool outputs (within the last prune_protect_tokens worth of content) are never pruned. Only completed tool calls outside this window are candidates.

No LLM call is required — this is entirely deterministic.

Example::

pruner = ToolOutputPruner(store, estimator, config)
result = await pruner.prune("sess_01JXYZ...")
print(f"Pruned {result.pruned_count} tool outputs ({result.pruned_tokens:,} tokens)")

prune `async` ¶

prune(session_id: str) -> PruneResult

Run a prune pass for the given session.

Algorithm: 1. Fetch all messages (including summaries) in chronological order. 2. Pre-fetch all raw parts to build a (message_id, tool_call_id) → part_id map. 3. Walk backward, tracking user turn count. 4. Stop at any is_summary message (compaction boundary). 5. Skip the most recent 2 user turns (protect recent content). 6. For each completed, non-protected, non-already-pruned tool part: - Accumulate output tokens. - Once outside the protect window (>40K tokens), add to candidates. 7. If total prunable volume < minimum threshold (20K): no-op. 8. Apply tombstones in a single batch UPDATE.

Parameters:

Name	Type	Description	Default
`session_id`	`str`	The session to prune.	required

Returns:

Type	Description
`PruneResult`	PruneResult with counts of pruned parts and tokens.

level1_summarise `async` ¶

level1_summarise(
    messages: list[MessageWithParts],
    model: str,
    budget: ContextBudget,
    estimator: TokenEstimator,
    llm_call: Any,
    compaction_prompt: str | None = None,
    model_context_limit: int = 200000,
    model_max_output_tokens: int = 0,
    compaction_estimator: TokenEstimator | None = None,
) -> SummaryCandidate | None

Attempt Level 1 (selective) summarisation via LLM.

File IDs found in the input messages are automatically appended to the summary via a [LCM File IDs: ...] footer.

Input messages are capped at 75 % of the compaction model's context window to prevent the summarisation call itself from overflowing.

Parameters:

Name	Type	Description	Default
`messages`	`list[MessageWithParts]`	All non-summary messages in the session.	required
`model`	`str`	Model string to use for compaction.	required
`budget`	`ContextBudget`	Token budget for validation.	required
`estimator`	`TokenEstimator`	Token estimator for result validation.	required
`llm_call`	`Any`	Async callable `(model, messages, max_tokens) -> str`.	required
`compaction_prompt`	`str \| None`	Custom system prompt override.	`None`
`model_context_limit`	`int`	Context window of the compaction model.	`200000`
`model_max_output_tokens`	`int`	Output limit of the compaction model (0 = unknown).	`0`
`compaction_estimator`	`TokenEstimator \| None`	Estimator in the compaction model's units, used to size the input against its window (defaults to estimator).	`None`

Returns:

Type	Description
`SummaryCandidate \| None`	SummaryCandidate if successful and fits budget, or None to escalate.

level2_summarise `async` ¶

level2_summarise(
    messages: list[MessageWithParts],
    model: str,
    budget: ContextBudget,
    estimator: TokenEstimator,
    llm_call: Any,
    compaction_prompt: str | None = None,
    model_context_limit: int = 200000,
    model_max_output_tokens: int = 0,
    compaction_estimator: TokenEstimator | None = None,
) -> SummaryCandidate | None

Attempt Level 2 (aggressive) summarisation via LLM.

Uses a more compressed prompt format and drops reasoning details. File IDs found in the input messages are propagated to the summary. Input is capped at 75 % of the compaction model's context window.

Parameters:

Name	Type	Description	Default
`messages`	`list[MessageWithParts]`	All non-summary messages in the session.	required
`model`	`str`	Model string to use for compaction.	required
`budget`	`ContextBudget`	Token budget for validation.	required
`estimator`	`TokenEstimator`	Token estimator for result validation.	required
`llm_call`	`Any`	Async callable.	required
`compaction_prompt`	`str \| None`	Custom system prompt override.	`None`
`model_context_limit`	`int`	Context window of the compaction model.	`200000`
`model_max_output_tokens`	`int`	Output limit of the compaction model (0 = unknown).	`0`
`compaction_estimator`	`TokenEstimator \| None`	Estimator in the compaction model's units, used to size the input against its window (defaults to estimator).	`None`

Returns:

Type	Description
`SummaryCandidate \| None`	SummaryCandidate if successful and fits budget, or None to escalate.

level3_deterministic ¶

level3_deterministic(
    messages: list[MessageWithParts],
    budget: ContextBudget,
    estimator: TokenEstimator,
) -> SummaryCandidate

Level 3 deterministic fallback (no LLM required).

Keeps the most recent messages that fit within 85% of the usable budget, prefixed with a truncation notice. This always produces a valid result.

File IDs found in all input messages (not just kept ones) are preserved in a [LCM File IDs: ...] footer even when their surrounding context is truncated — this is the lossless guarantee.

Precedence when the budget is tight: file IDs are reserved first, then prose fills what remains, so messages are dropped before any file ID is. Only if the header plus the ID footer cannot fit in 85% of budget.usable is the footer truncated, keeping the most recently referenced IDs (by last occurrence; the footer itself stays in first-occurrence order). A warning with the dropped count is logged; the raw files stay addressable in the immutable store. Truncating there is the only way to keep the "always fits the budget" guarantee that lets compaction make progress. If budget.usable is smaller than the minimal header, no output can fit, so all IDs are kept (lossless wins).

Parameters:

Name	Type	Description	Default
`messages`	`list[MessageWithParts]`	All non-summary messages in the session.	required
`budget`	`ContextBudget`	Token budget — result is guaranteed to fit within usable.	required
`estimator`	`TokenEstimator`	Token estimator.	required

Returns:

Type	Description
`SummaryCandidate`	SummaryCandidate that fits within budget.usable (barring a budget too
`SummaryCandidate`	small for even the minimal header), as measured by estimator.

mnesis.compaction¶

compaction ¶

CompactionEngine ¶

stalled property ¶

more_to_compact property ¶

in_flight property ¶

has_pending property ¶

note_user_turn ¶

set_context_measure ¶

is_soft_overflow ¶

is_hard_overflow ¶

is_overflow ¶

check_and_trigger ¶

wait_for_pending async ¶

compact_exclusive async ¶

run_compaction async ¶

SummaryCandidate dataclass ¶

ToolOutputPruner ¶

prune async ¶

level1_summarise async ¶

level2_summarise async ¶

level3_deterministic ¶

`mnesis.compaction`¶

stalled `property` ¶

more_to_compact `property` ¶

in_flight `property` ¶

has_pending `property` ¶

wait_for_pending `async` ¶

compact_exclusive `async` ¶

run_compaction `async` ¶

SummaryCandidate `dataclass` ¶

prune `async` ¶

level1_summarise `async` ¶

level2_summarise `async` ¶