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?(identity, ai_context, _recent) do
Planck.AI.Context.estimate_tokens(ai_context) >= identity.model.context_window * 0.8
end
@impl true
def compact(_identity, _ai_context, recent, _args) 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/4 — 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/4" below.
Dispatch
Planck.Agent calls compact/4 before every LLM turn:
Hooks.Compactor.compact(identity, hooks, ai_context, recent, opts)identity— the identity of the agent.hooks— the hooks with the compactor.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—agent.context.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.:args— a map (default%{prompt: nil}) forwarded to the compactor'scompact/4callback as its 4th parameter; used by/compact [prompt]to steer the summarization. The dispatcher reads the bypass flag fromargs.force(truefor/compact,falsefor auto-compaction) before forwarding. Called by this dispatch function, aroundcompact/4— never by acompact/4implementation itself, which never receivesoptsat all.Planck.Agentsupplieson_compacting/on_compactedso 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.hooks.compactorset,hooks.sidecar_node: nil— callsmodule.compact?/3, then, only if that'strue,module.compact/3, in-process.hooks.compactorset,hooks.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). :args — a map forwarded to the compactor's compact/4 callback;
defaults to %{prompt: nil} when absent (auto-compaction). The dispatcher
reads the bypass flag from args.force, not from opts: true (set by the
/compact slash-command path) bypasses compact?/3 and calls compact/4
directly; false means auto-compaction.
Callbacks
Compact the conversation. Only ever called when compact?/3 (checked by
the dispatcher, not called here) already returned true, or directly by
the forced /compact slash-command path (which bypasses compact?/3).
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 the agent identity and hooks, 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
@type compact_opts() :: [compact_opt()]
:on_compacting/:on_compacted — both zero-arity, both optional (neither
given just means nothing tells the UI compaction is in progress, not an
error). :args — a map forwarded to the compactor's compact/4 callback;
defaults to %{prompt: nil} when absent (auto-compaction). The dispatcher
reads the bypass flag from args.force, not from opts: true (set by the
/compact slash-command path) bypasses compact?/3 and calls compact/4
directly; false means auto-compaction.
Callbacks
@callback compact( identity :: Planck.Agent.Identity.t(), ai_context :: Planck.AI.Context.t(), recent :: [Planck.Agent.Message.t()], args :: compact_args() ) :: compact_result()
Compact the conversation. Only ever called when compact?/3 (checked by
the dispatcher, not called here) already returned true, or directly by
the forced /compact slash-command path (which bypasses compact?/3).
args is a map with :prompt — a user-supplied string (from
/compact [prompt]) that can steer the summarization, nil when absent
(auto-compaction) — and :force, true when invoked via the /compact
slash command, false for auto-compaction. A forced call should compact
even a short history rather than skipping; only skip when there is
genuinely nothing to summarize.
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?( identity :: Planck.Agent.Identity.t(), ai_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.Identity.t(), Planck.Agent.Hooks.t(), Planck.AI.Context.t(), [Planck.Agent.Message.t()], compact_opts() ) :: compact_result()
Dispatch compaction for the given agent the agent identity and hooks, 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.