Compaction Engine

Use BeamWeaver.Compaction when your application owns the transcript and provider calls. It produces a source-linked summary checkpoint and retained recent events. Your application decides how to render the next request and atomically activate the candidate. For an agent's built-in conversation, use Context Management instead.

Run this page’s examples in order in one IEx session; later blocks reuse the modules and values defined earlier.

A first compaction

Complete Installation . This example uses a real OpenAI summarizer. Repeated background text makes the sample large enough to exercise compaction with a small demonstration policy.

defmodule MyApp.TranscriptCompaction do
  alias BeamWeaver.Compaction.{Canonical, Semantic}
  alias BeamWeaver.Core.{ChatModel, ID, Message}

  def events(lane_id, rows) do
    for {{role, text}, seq} <- Enum.with_index(rows, 1) do
      provenance = %{"origin" => Atom.to_string(role)}

      %{
        event_id: ID.uuidv7(),
        chat_seq: seq,
        lane_event_ordinal: seq,
        run_agent_id: lane_id,
        checkpoint_namespace: "root",
        role: role,
        content: text,
        provenance: provenance,
        provenance_sha256: Canonical.hash(provenance),
        content_hash: Canonical.hash(%{content: text, tool: nil, artifact_ids: [], truncated: false})
      }
    end
  end

  def render(semantic, events) do
    summary = if semantic, do: JSON.encode!(Semantic.to_map(semantic)) <> "\n", else: ""
    {:ok, summary <> Enum.map_join(events, "\n", &"#{&1.role}: #{&1.content}")}
  end

  def summarize(payload) do
    model = BeamWeaver.Models.init_chat_model!("openai:gpt-5.4-mini", max_tokens: 2_048, timeout: 60_000)
    indexed = Enum.with_index(payload.events)
    sources = Map.new(indexed, fn {event, index} -> {index, event.event_id} end)
    input = Enum.map(indexed, fn {event, index} -> %{event_index: index, text: event.content} end)

    schema = %{
      "type" => "object",
      "additionalProperties" => false,
      "required" => ["facts"],
      "properties" => %{
        "facts" => %{
          "type" => "array",
          "items" => %{
            "type" => "object",
            "additionalProperties" => false,
            "required" => ["event_index", "text"],
            "properties" => %{"event_index" => %{"type" => "integer"}, "text" => %{"type" => "string"}}
          }
        }
      }
    }

    messages = [
      Message.system(
        "Extract at most six distinct order facts. Omit repetitive background. Each fact must cite the supplied event_index containing it."
      ),
      Message.user(JSON.encode!(input))
    ]

    with {:ok, reply} <- ChatModel.invoke(model, messages, response_format: %{name: "facts", schema: schema}) do
      notes =
        for fact <- reply.metadata.parsed["facts"],
            do: %{"text" => fact["text"], "source_event_ids" => [Map.fetch!(sources, fact["event_index"])]}

      earlier = if payload.previous_semantic, do: payload.previous_semantic.critical_context, else: []

      {:ok,
       %{
         "version" => 1,
         "objective" => [],
         "user_requests" => [],
         "constraints" => [],
         "decisions" => [],
         "progress" => %{"completed" => [], "active" => [], "blocked" => [], "pending" => []},
         "critical_context" => earlier ++ notes,
         "errors" => [],
         "artifact_refs" => [],
         "coverage" => payload.covered_range
       }, reply.usage_metadata}
    end
  end
end

alias BeamWeaver.Compaction
alias BeamWeaver.Compaction.{Canonical, Request}
alias BeamWeaver.Core.ID
lane_id = ID.uuidv7()
background = String.duplicate("The customer portal displays the same tracking details and contact instructions. ", 12)
rows = for n <- 1..6, role <- [:user, :assistant], do: {role, "Order A-100#{n} arrives Thursday. " <> background}
events = MyApp.TranscriptCompaction.events(lane_id, rows)
{:ok, rendered} = MyApp.TranscriptCompaction.render(nil, events)

{:ok, request} =
  Request.new(%{
    request_id: ID.uuidv7(),
    trigger: :manual,
    thread_id: "compaction-demo",
    run_id: ID.uuidv7(),
    root_turn_id: ID.uuidv7(),
    run_agent_id: lane_id,
    checkpoint_namespace: "root",
    provider_connection_id: "openai-main",
    destination_identity_hash: Canonical.hash(%{"provider" => "openai", "model" => "gpt-5.4-mini"}),
    model_profile: BeamWeaver.Models.init_chat_model!("openai:gpt-5.4-mini").profile,
    policy: %{recent_tail_min_tokens: 100, recent_tail_max_tokens: 500, minimum_total_reclaim_tokens: 100},
    rendered_request: rendered,
    events: events,
    rehydration_state: %{},
    render: &MyApp.TranscriptCompaction.render/2,
    summarize: &MyApp.TranscriptCompaction.summarize/1
  })

{:ok, result} = Compaction.compact(%{}, request)
:compacted = result.status
true = result.tokens_after < result.tokens_before
true = result.tokens_reclaimed > 0
true = length(result.retained_events) < length(events)
{:ok, next_request} = MyApp.TranscriptCompaction.render(result.compaction_checkpoint.semantic, result.retained_events)
true = next_request =~ "Thursday"

The callback asks the model for facts with integer source indices, then maps those indices to actual event IDs in code. The sample transcript contains order facts, so the other semantic collections stay empty; a general conversation summarizer must also preserve requests, constraints and progress.

A returned checkpoint is a candidate, not an activated conversation. This example validates source shape, summary coverage and actual reduction, but does not persist anything.

When to use it

The engine starts no processes and owns no database or provider client. Callbacks perform rendering and summarization. This separates context transformation from your application's storage, locking and job lifecycle. Use it when those are already controlled by your application.

How a compaction runs

The engine measures the rendered request, protects the recent tail and active inputs, projects oversized old tool results, summarizes the remaining older events, validates the summary, then renders again to verify useful reduction. It can return :pruned without summarization when projection alone suffices.

Default policy triggers automatic work at 85% of the context limit and targets 70% afterward. It keeps a tail around 10% of the window, bounded by 8,000 and 32,000 tokens, and normally requires at least 8,192 tokens and 15% reduction. The demonstration above deliberately overrides those large-workload defaults.

Source events

Each request covers one lane: one run_agent_id and checkpoint namespace. Events carry UUIDv7 IDs, increasing chat positions/ordinals, roles, content, provenance and canonical content/provenance hashes. Those hashes establish which source material a summary or projection refers to.

Tool-call/result pairs must remain associated. artifact_ids identify durable copies of large results; protected: true or active_input_event_ids protects current input from summarization. Preserve source IDs across retries and subsequent compactions.

The request

Supply the exact current rendered_request, source events, model context limits and a renderer that matches your actual next provider request. Identity fields bind the attempt to a thread, run, root user turn and model destination. For shared input/output windows, reserve requested output tokens. Unknown context limits fail instead of being guessed.

previous_semantic and parent_checkpoint_id connect later attempts to the active checkpoint. rehydration_state carries exact application data alongside the summary. focus can guide summarization without changing source facts. Context budgeting conservatively counts rendered bytes; it is not the provider's billing tokenizer. BeamWeaver.ContextBudget describes counting with a tokenizer or from provider-reported usage, and the byte count it falls back to when the profile has no tokenizer.

The summarizer

The payload includes events, previous semantic state, exact covered range and an output budget. Return a semantic map or JSON text, optionally with usage. Semantic.json_schema/0 describes objectives, user requests, constraints, decisions, progress, critical context, errors, artifact references and coverage.

Validation checks schema shape, source IDs, exact user excerpts, coverage, duplicates and size. Invalid output receives a bounded repair attempt. These checks do not prove every paraphrased fact is correct; evaluate summary quality against application requirements. Set provider timeouts inside your callback; the engine does not enforce a provider timeout for you.

Results

Results include token estimates, retained events, candidate checkpoint, provider usage and anti-thrashing state. Errors include ineffective reduction, invalid semantic output, missing artifacts and invalid/oversized requests. Inspect error details before retrying.

Save and activate the checkpoint

In one application transaction, save the candidate idempotently and update the conversation's active checkpoint only if its expected parent/revision still matches. Use an optimistic revision check or equivalent concurrency control. A stored candidate alone is not current state.

If another writer advanced the conversation, do not activate the stale candidate. Reload and construct a new request. If the same candidate is already active, the original attempt succeeded.

The following in-memory owner demonstrates an atomic revision check and idempotent activation. Its lifetime is the owner process; use a database transaction with the same invariants for durable conversations:

defmodule MyApp.CompactionHead do
  def activate(owner, expected_revision, checkpoint, anti_thrash) do
    Agent.get_and_update(owner, fn state ->
      cond do
        state.active_id == checkpoint.checkpoint_id ->
          {{:ok, state.revision}, state}

        state.revision != expected_revision ->
          {{:error, :stale_conversation}, state}

        true ->
          {:ok, recorded} = BeamWeaver.Compaction.State.record_success(anti_thrash, checkpoint)

          updated = %{
            state
            | revision: state.revision + 1,
              active_id: checkpoint.checkpoint_id,
              checkpoints:
                Map.put(
                  state.checkpoints,
                  checkpoint.checkpoint_id,
                  BeamWeaver.Compaction.Checkpoint.to_map(checkpoint)
                ),
              anti_thrash: recorded
          }

          {{:ok, updated.revision}, updated}
      end
    end)
  end
end

{:ok, owner} = Agent.start_link(fn -> %{revision: 0, active_id: nil, checkpoints: %{}, anti_thrash: nil} end)

try do
  {:ok, 1} = MyApp.CompactionHead.activate(owner, 0, result.compaction_checkpoint, result.anti_thrash)
  {:ok, 1} = MyApp.CompactionHead.activate(owner, 0, result.compaction_checkpoint, result.anti_thrash)
  1 = Agent.get(owner, &map_size(&1.checkpoints))
after
  Agent.stop(owner)
end

Compact again later

Render the active semantic checkpoint plus retained events and new events. Pass the active checkpoint as parent and carry its semantic state forward. Keep sequence numbers increasing; do not reassign event IDs or turn old summary text into fabricated source events.

Append new events with increasing sequence numbers and use the prior checkpoint as the parent. This performs a second real summary:

checkpoint = result.compaction_checkpoint
last_seq = Enum.max_by(events, & &1.chat_seq).chat_seq
new_rows = for n <- 7..12, role <- [:user, :assistant], do: {role, "Order B-#{n} arrives Friday. " <> background}

new_events =
  MyApp.TranscriptCompaction.events(lane_id, new_rows)
  |> Enum.map(fn event ->
    %{event | chat_seq: event.chat_seq + last_seq, lane_event_ordinal: event.lane_event_ordinal + last_seq}
  end)

lane = result.retained_events ++ new_events
{:ok, rendered} = MyApp.TranscriptCompaction.render(checkpoint.semantic, lane)

next = %{
  request
  | request_id: ID.uuidv7(),
    parent_checkpoint_id: checkpoint.checkpoint_id,
    previous_semantic: checkpoint.semantic,
    events: lane,
    rendered_request: rendered
}

{:ok, second} = Compaction.compact(%{}, next)
:compacted = second.status
true = second.compaction_checkpoint.parent_checkpoint_id == checkpoint.checkpoint_id

Automatic compaction

Keep Compaction.State per conversation lane. Pass it as anti_thrash, record success only after activation, and observe new tokens as context grows. The returned state or an ineffective-compaction error's state governs the next attempt. Suppression includes insufficient growth, a low request size, per-turn limits, rapid refill and pending cancellation. A new root turn resets the per-turn allowance; it does not erase the active summary.

After activating a successful candidate, retain the anti-thrashing state. An unchanged small lane needs no immediate automatic summary:

{:ok, anti_thrash} = BeamWeaver.Compaction.State.record_success(second.anti_thrash, second.compaction_checkpoint)
{:ok, skipped} = Compaction.compact(%{}, %{next | request_id: ID.uuidv7(), trigger: :auto, anti_thrash: anti_thrash})
:skipped = skipped.status
IO.inspect(skipped.provider_usage, label: "Automatic compaction suppression")

Large tool results

Old tool results larger than the projection threshold retain excerpts instead of full content. Their full contents must already have durable artifact IDs; otherwise compaction fails. The checkpoint's projection manifest plus ToolProjection.replay/2 reconstructs the same projection from originals and verifies source identity.

Recovery

Allocate a request ID once and reuse it for the same attempt. Reconcile ambiguous provider completion or transaction outcomes before making another summary request. Keep source events and candidate checkpoints until activation or abandonment is resolved. Compaction does not supply a scheduler.

Limits

Requests allow 10,000 events, up to 1 MiB per event, 128 artifact IDs per event, 1 MiB of rehydration state, 128 KiB of semantic output and 2,000 bytes of focus. Summarization is bounded to three chunks plus one repair. Portable compaction is the supported mode; provider-native context mechanisms are separate.