Behaviour and default implementation for context compaction in Planck.Agent.
Behaviour
Use use Planck.Agent.Hooks.Compactor to implement a custom compaction strategy
in a sidecar. Two callbacks are required, compact?/3 and compact/3;
compact_timeout/0 has a default of 600000 ms.
defmodule MySidecar.Compactors.Builder do
use Planck.Agent.Hooks.Compactor
@impl true
def compact?(state, context, _recent) do
Planck.AI.Context.estimate_tokens(context) >= state.model.context_window * 0.8
end
@impl true
def compact(_state, _context, recent) do
summary = Message.new({:custom, :summary}, [{:text, summarise(recent)}])
kept = Enum.take(recent, -5)
{:compact, summary, kept}
end
@impl true
def compact_timeout, do: 60_000
endcompact?/3 is checked first; compact/3 — the potentially slow part, an
LLM call for the built-in strategy — is only ever called when it returns
true. Splitting these means the dispatcher (compact/4 below), not
each implementation, can wrap the slow part with a progress announcement
correctly for any compactor, without needing to predict anything — see
"Why a separate compact?/3, and why the dispatcher wraps compact/3" below.
Dispatch
Planck.Agent calls compact/4 before every LLM turn:
Hooks.Compactor.compact(state, context, recent, opts)state— the full agent state (model, compactor, sidecar_node, ...).context— thePlanck.AI.Context.t()built forrecent(system prompt, tool schemas, andrecentitself, already estimated as one whole — seePlanck.AI.Context.estimate_tokens/1). Passed through rather than justrecentalone so a custom compactor (typically running on a sidecar node, which has no other way to see the agent's system prompt or tool list) can make an informed decision too, not only the built-in one.recent—state.messagessince the last{:custom, :summary}checkpoint (or all of them, if there isn't one yet).opts—:on_compactingand:on_compacted, both zero-arity functions, both optional. Called by this dispatch function, aroundcompact/3— never by acompact/3implementation itself, which never receivesoptsat all.Planck.Agentsupplies these so the UI can be told compaction is in progress without any compactor needing to know anything aboutPlanck.Agent's own PubSub topics or event shapes.state.compactor: nil— usesPlanck.Agent.Hooks.Compactor.Default, the built-in strategy (see its own moduledoc). Not special-cased beyond this:Defaultsatisfies the same behaviour a custom module would.state.compactorset,state.sidecar_node: nil— callsmodule.compact?/3, then, only if that'strue,module.compact/3, in-process.state.compactorset,state.sidecar_nodeset — calls the module on the remote node via RPC (same two-call shape); falls back toDefaulton:badrpcfrom either call.
Why a separate compact?/3, and why the dispatcher wraps compact/3
Planck.Agent calls this dispatcher on every turn, and can't know in
advance whether a given call will actually compact — that decision
belongs to the compactor, and for a custom one, its criteria are opaque to
Planck.Agent entirely. Broadcasting "compacting" unconditionally around
every call (clearing it right after) would flash it on every ordinary
turn, not just the rare one that actually compacts. Predicting the
outcome from Planck.Agent's side (e.g. reapplying the built-in ratio)
would only be accurate for the built-in compactor. Splitting the decision
(compact?/3, always cheap — no LLM call for the built-in strategy) from
the work (compact/3, potentially slow) lets the dispatcher check the
decision first and only announce progress around the part that's
genuinely slow — accurate for any compactor, without Planck.Agent
needing to predict anything or any compactor needing to call back into
Planck.Agent itself.
Summary
Types
:on_compacting/:on_compacted — both zero-arity, both optional (neither
given just means nothing tells the UI compaction is in progress, not an
error).
Callbacks
Compact the conversation. Only ever called when compact?/3 (checked by
the dispatcher, not called here) already returned true.
Cheap decision: would compact/3 actually do anything right now? Must not
itself do anything slow (no LLM call) — see the moduledoc for why.
RPC call timeout in milliseconds when this compactor is invoked remotely.
Functions
Dispatch compaction for the given agent state, its built request context, and the messages since the last summary — see this module's own moduledoc.
Default RPC timeout used when a compactor module omits compact_timeout/0.
Types
Callbacks
@callback compact( state :: Planck.Agent.t(), context :: Planck.AI.Context.t(), recent :: [Planck.Agent.Message.t()] ) :: compact_result()
Compact the conversation. Only ever called when compact?/3 (checked by
the dispatcher, not called here) already returned true.
Return {:compact, summary_msg, kept} to replace older messages with a
summary, or :skip to leave the list unchanged — a compactor is free to
still decide against compacting here even after saying true to
compact?/3 (e.g. nothing old enough left worth summarizing).
@callback compact?( state :: Planck.Agent.t(), context :: Planck.AI.Context.t(), recent :: [Planck.Agent.Message.t()] ) :: boolean()
Cheap decision: would compact/3 actually do anything right now? Must not
itself do anything slow (no LLM call) — see the moduledoc for why.
@callback compact_timeout() :: pos_integer()
RPC call timeout in milliseconds when this compactor is invoked remotely.
Defaults to 600000 ms.
Functions
@spec compact( Planck.Agent.t(), Planck.AI.Context.t(), [Planck.Agent.Message.t()], compact_opts() ) :: compact_result()
Dispatch compaction for the given agent state, its built request context, and the messages since the last summary — see this module's own moduledoc.
Returns :skip or {:compact, summary_msg, kept}.
@spec default_compact_timeout() :: pos_integer()
Default RPC timeout used when a compactor module omits compact_timeout/0.