Plan — CompactionService + `COMPACTING` state

Extract compaction decision AND execution from GeneratorAgent into a dedicated CompactionService. Add an explicit COMPACTING state to the generator state machine.

Motivation

GeneratorAgent currently handles two distinct concerns under compaction:

The decision — should this context be compacted? (watches token count vs threshold)
The execution — summarize history, replace it, publish result

Neither belongs in a generation agent. The decision is a monitoring concern. The execution is a history-maintenance operation. Both belong in CompactionService.

Serialization constraint: Compaction execution must be serialized with _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.

Additional fix: the generator currently runs compaction while appearing IDLE to the state machine and to GeneratorWindow. Adding a COMPACTING state makes the actual runtime state visible and honest.

Responsibility Split

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()`

Architecture After This Change

response.generation ───▶ CompactionService._check() │ prompt_tokens > threshold * num_ctx? │ yes ▼ compaction.request ───▶ GeneratorAgent._handle_compaction() │ IDLE → COMPACTING │ ▼ CompactionService.compact() (reads history, calls Ollama, replaces history, publishes result) │ COMPACTING → IDLE

User-triggered compaction (Compact button in the web UI) continues to publish compaction.request directly via the gateway — no change to that path.

CompactionService (revised)

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,
        )

Generator State Machine Changes

generator_states.py

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"

generator_actions.py

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

generator_transitions.py

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.

generator_agent.py — _handle_compaction()

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

generator_agent.py — init()

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

Config Change

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

Startup Change

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 Map

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

Bus Subjects

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.

Acceptance Criteria

After a generation that crosses the threshold, compaction.request appears on the bus from compaction_service without any user action
After auto-compaction, compaction.result appears on the bus with valid tokens_before and tokens_after
User-triggered compact button still works via the existing gateway path
Setting compaction_threshold: 0 disables auto-compaction — no compaction.request published by the service
During compaction, GeneratorWindow shows state compacting (not idle)
If a query arrives during compaction, the IDLE check rejects it with an error result and the generator returns to COMPACTING cleanly
All existing tests pass

Plan — CompactionService + COMPACTING state