COMPACTING stateExtract compaction decision AND execution from GeneratorAgent into a dedicated CompactionService. Add an explicit COMPACTING state to the generator state machine.
GeneratorAgent currently handles two distinct concerns under compaction:
Neither belongs in a generation agent. The decision is a monitoring concern. The execution is a history-maintenance operation. Both belong in CompactionService.
_handle_query() — both mutate conversation history and both must not interleave. The generator's single-threaded event loop is the serialization mechanism. CompactionService provides the logic; GeneratorAgent calls into it from within its dispatch loop, under the state machine gate.
COMPACTING state makes the actual runtime state visible and honest.
| Concern | Owner | Why |
|---|---|---|
| Decision: should we compact? | CompactionService (bus listener) | Observes response.generation, checks threshold — no generation knowledge needed |
| Execution: summarize + replace history | CompactionService (compact() method) | All compaction logic in one place; called by generator under its state gate |
| State gate: IDLE check + transitions | GeneratorAgent | Only the generator knows when it's safe to mutate history; enforces serialization with _handle_query() |
compaction.request directly via the gateway — no change to that path.
File: src/local/services/compaction_service.py
Two responsibilities in one service: (1) the bus listener that watches response.generation and fires compaction.request when threshold is crossed; (2) the compact() method called synchronously by GeneratorAgent from within its dispatch loop.
class CompactionService:
"""Auto-compaction decision and execution for GeneratorAgent.
Bus listener: subscribes to response.generation, publishes compaction.request
when prompt_tokens crosses the threshold.
Executor: compact() is called synchronously by GeneratorAgent under its
IDLE gate — runs in the generator's thread, serialized with _handle_query().
"""
AGENT_ID = "compaction_service"
def __init__(self, conversation_service: ConversationService, model: str, options: dict) -> None:
self._conv = conversation_service
self._model = model
self._options = options
self._pub, self._sub = make_participant_bus([RESPONSE_GENERATION])
def run(self) -> None:
logger.info("CompactionService ready")
while True:
try:
envelope = self._sub.receive()
except Exception as exc:
logger.error("CompactionService: receive error: %s", exc)
continue
self._check(envelope)
def _check(self, envelope: MessageEnvelope) -> None:
cfg = get_config("generator") or {}
threshold = cfg.get("compaction_threshold", 0.8)
if not threshold:
return
num_ctx = cfg.get("num_ctx", 32000)
prompt_tokens: int = envelope.payload.get("prompt_tokens", 0)
session_id: str | None = envelope.payload.get("session_id")
if prompt_tokens >= threshold * num_ctx:
logger.info(
"CompactionService: auto-compacting session %s (%d / %d tokens, threshold %.0f%%)",
(session_id or "")[:8], prompt_tokens, num_ctx, threshold * 100,
)
self._pub.publish(MessageEnvelope.create(
message_type="compaction_request",
subject=COMPACTION_REQUEST,
sender_id=self.AGENT_ID,
payload={"session_id": session_id, "auto": True},
))
def compact(self, envelope: MessageEnvelope, publisher, make_envelope) -> None:
"""Summarize a session's history and replace it with a summary + tail turns.
Called synchronously by GeneratorAgent._handle_compaction() while the
generator is in COMPACTING state. Runs in the generator's thread.
"""
import ollama
session_id: str | None = envelope.payload.get("session_id")
cfg = get_config("generator") or {}
tail_turns: int = cfg.get("compaction_tail_turns", 4)
history = self._conv.get_history(session_id)
tokens_before = self._conv.get_token_count(session_id)
if not history:
publisher.publish(make_envelope(
COMPACTION_RESULT, "compaction",
{"error": "no history to compact", "session_id": session_id},
envelope.correlation_id or str(uuid.uuid4()), None,
))
return
convo_text = []
for m in history:
role = m.get("role", "")
content = m.get("content") or ""
if role in ("user", "assistant") and content:
convo_text.append(f"{role.upper()}: {content}")
summary_input = "\n\n".join(convo_text)
compaction_system = cfg.get(
"compaction_system_prompt",
"Summarize this conversation concisely, preserving key facts and decisions.",
).strip()
resp = ollama.chat(
model=self._model,
messages=[
{"role": "system", "content": compaction_system},
{"role": "user", "content": summary_input},
],
stream=False,
options=self._options,
)
summary_text = (resp.message.content or "").strip()
tail_messages: list[dict] = []
pairs_collected = 0
i = len(history) - 1
while i >= 1 and pairs_collected < tail_turns:
if history[i].get("role") == "assistant" and history[i - 1].get("role") == "user":
tail_messages = history[i - 1: i + 1] + tail_messages
pairs_collected += 1
i -= 2
else:
i -= 1
new_messages = [{"role": "assistant", "content": f"[SUMMARY] {summary_text}"}] + tail_messages
total_chars = sum(len(m.get("content") or "") for m in new_messages)
tokens_estimated_after = total_chars // 4
self._conv.replace_messages(session_id, new_messages)
self._conv.set_token_count(session_id, tokens_estimated_after)
publisher.publish(make_envelope(
COMPACTION_RESULT, "compaction",
{"session_id": session_id,
"tokens_before": tokens_before,
"tokens_after": tokens_estimated_after,
"summary": summary_text},
envelope.correlation_id or str(uuid.uuid4()), None,
))
logger.info(
"CompactionService: compacted session %s — %d → ~%d tokens",
(session_id or "")[:8], tokens_before, tokens_estimated_after,
)
class GeneratorState(Enum):
IDLE = "idle"
RECEIVING = "receiving"
GENERATING = "generating"
DISPATCHING_TOOL = "dispatching_tool"
WAITING_FOR_TOOL = "waiting_for_tool"
PUBLISHING = "publishing"
COMPACTING = "compacting" # ← add
ERROR = "error"
class GeneratorAction(Enum):
RECEIVE = "receive"
START_GENERATION = "start_generation"
DISPATCH_TOOL = "dispatch_tool"
AWAIT_RESULT = "await_result"
TOOL_RESULT = "tool_result"
TOOL_TIMEOUT = "tool_timeout"
PUBLISH = "publish"
RESET = "reset"
FAIL = "fail"
START_COMPACTION = "start_compaction" # ← add
COMPLETE_COMPACTION = "complete_compaction" # ← add
Add two transitions alongside the existing table:
(S.IDLE, A.START_COMPACTION): S.COMPACTING,
(S.COMPACTING, A.COMPLETE_COMPACTION): S.IDLE,
The existing FAIL loop already covers COMPACTING → ERROR because it iterates over all non-IDLE states.
Replace the existing implementation with a thin gate that delegates to CompactionService:
def _handle_compaction(self, envelope: MessageEnvelope) -> None:
if self._sm.state != GeneratorState.IDLE:
self._pub.publish(self._make_envelope(
COMPACTION_RESULT, "compaction",
{"error": "generator busy — try again after current query finishes",
"session_id": envelope.payload.get("session_id")},
envelope.correlation_id or str(uuid.uuid4()), None,
))
return
self._do_transition(GeneratorAction.START_COMPACTION) # IDLE → COMPACTING
try:
self._compaction_service.compact(envelope, self._pub, self._make_envelope)
finally:
self._do_transition(GeneratorAction.COMPLETE_COMPACTION) # COMPACTING → IDLE
Accept compaction_service as an injected parameter (like conversation_service):
def __init__(
self,
model: str | None = None,
temperature: float | None = None,
conversation_service=None,
compaction_service=None,
) -> None:
...
self._compaction_service = compaction_service # set before run.py wires it
Add to config/generator.yaml and src/local/defaults/generator.yaml:
# Fraction of num_ctx at which auto-compaction triggers (0 = disabled)
compaction_threshold: 0.8
File: src/local/run.py
Create CompactionService with the shared dependencies, start its bus-listener thread, inject it into GeneratorAgent:
from local.services.compaction_service import CompactionService
# After shared_conv is created, before generator starts:
shared_compaction = CompactionService(
conversation_service=shared_conv,
model=args.model or cfg.get("model", "gemma4:e4b"),
options={"num_ctx": cfg.get("num_ctx", 128000), "temperature": cfg.get("temperature", 0.1)},
)
threading.Thread(target=shared_compaction.run, daemon=True, name="compaction_service").start()
# Pass into generator:
gen_thread = threading.Thread(
target=_start_generator,
args=(args.model,),
kwargs={"conversation_service": shared_conv, "compaction_service": shared_compaction},
...
)
| File | Change |
|---|---|
src/local/services/compaction_service.py | new — decision + execution logic |
src/local/agents/generator_states.py | add COMPACTING |
src/local/agents/generator_actions.py | add START_COMPACTION, COMPLETE_COMPACTION |
src/local/agents/generator_transitions.py | add IDLE→COMPACTING, COMPACTING→IDLE transitions |
src/local/agents/generator_agent.py | inject compaction_service; _handle_compaction() becomes a thin gate |
src/local/run.py | construct + start CompactionService; inject into generator |
config/generator.yaml | add compaction_threshold: 0.8 |
src/local/defaults/generator.yaml | same |
No new subjects. CompactionService subscribes to response.generation (existing) and publishes compaction.request (existing). The generator publishes compaction.result (existing) — via CompactionService.compact() which receives the publisher.
compaction.request appears on the bus from compaction_service without any user actioncompaction.result appears on the bus with valid tokens_before and tokens_aftercompaction_threshold: 0 disables auto-compaction — no compaction.request published by the servicecompacting (not idle)