Iproved chat functionality

README.md

@@ -165,18 +165,31 @@ enabled = true                              # audit/YYYY-MM-DD.jsonl
 dir     = "audit"
 
 [llm]                                       # local llama.cpp (OpenAI-compatible)
-enabled              = true
-endpoint             = "http://localhost:9931/v1/chat/completions"
-temperature          = 0.7
-max_tokens           = 1024                 # reasoning models spend tokens before the answer; keep generous
-timeout_s            = 120.0
-cache_prompt         = true                 # reuse the KV prefix (append-only history keeps it warm)
-max_prompt_tokens    = 64000                # ~half of the 131072 context
-trim_block_turns     = 4                    # trim history in blocks, not per-turn, to limit re-prefill
-max_tool_iterations  = 3
-system_prompt_chat   = "You are a concise voice assistant. Reply in 1-3 sentences."
-system_prompt_command = "You translate a spoken request into a single tool call when one applies."
-```
+enabled                  = true
+endpoint                 = "http://localhost:9931/v1/chat/completions"
+temperature              = 0.7
+max_tokens               = 1024             # max output per request; reasoning models spend tokens before the answer, so keep generous
+timeout_s                = 120.0
+cache_prompt             = true             # reuse the KV prefix (append-only history keeps it warm)
+max_context_tokens       = 131072           # the server's --ctx-size (llama.cpp `GET /props` -> n_ctx); the prompt budget derives from it
+compaction_safety_margin = 256              # tokens reserved for the token estimate error + template/tool overhead
+compaction_enabled       = true             # summarize old turns instead of deleting them
+compaction_trigger_ratio = 0.8              # compact when the prompt reaches this fraction of the budget
+compaction_keep_turns    = 6                # recent turns kept verbatim
+compaction_max_tokens    = 512              # summary length cap
+compaction_timeout_s     = 30.0             # summarizer timeout, so a compaction cannot hold the inference lane
+compaction_input_tokens  = 8192             # summarizer input cap before chunk-and-fold
+compaction_system_prompt = "You compress a conversation. Preserve names, numbers, decisions, user preferences, and unresolved questions. Omit pleasantries. Output only the summary."
+inference_queue_size     = 8                # bounded single-lane inference queue
+trim_block_turns         = 4                # fallback block trimming, not per-turn, to limit re-prefill
+max_tool_iterations      = 3
+system_prompt_chat       = "You are a concise voice assistant. Reply in 1-3 sentences."
+system_prompt_command    = "You translate a spoken request into a single tool call when one applies."
+```
+
+`max_prompt_tokens` was removed: the prompt budget is derived as
+`max_context_tokens - max_tokens - compaction_safety_margin`, so the window and the trim budget
+can never disagree. A legacy `max_prompt_tokens` key is ignored with a warning.
 
 ### Audio Source Setup
 
@@ -556,6 +569,34 @@ Only **actions** are persisted, to `audit/YYYY-MM-DD.jsonl`:
 
 Commands and chat history are volatile — held in memory and discarded.
 
+### Conversation memory (compaction)
+
+Chat history lives in memory only. To avoid overflowing the model's context, the assistant
+**compacts** rather than deletes: when the prompt reaches `compaction_trigger_ratio` of the
+derived budget (`max_context_tokens - max_tokens - compaction_safety_margin`), the oldest turns
+are summarized by the same endpoint (using `compaction_system_prompt`, capped at
+`compaction_max_tokens`) and replaced by one rolling summary, while the most recent
+`compaction_keep_turns` stay verbatim. Large evictions are summarized in chunks and folded.
+
+The summary is injected as a second `system` message, so the prefix (`system` + summary) is
+**stable between compactions** — a compaction invalidates the server's prefix cache *once*, and
+subsequent turns are append-only and cache-friendly. If summarization fails (error, empty reply,
+or a truncated summary) the assistant falls back to deleting the oldest turns in blocks
+(`trim_block_turns`); compaction is never the reason a request exceeds the window. Each
+compaction emits a `chat_message` with `role: "note"` so the pane shows why the context changed —
+the note is UI-only and never sent to the model.
+
+A reply cut off by the output limit (`finish_reason: "length"`) surfaces as
+`error{code:"truncated"}` instead of disappearing silently; an empty truncated turn is not stored.
+
+### Inference queue
+
+Every LLM call — chat turns, command executions, and the compaction summarizer — runs through a
+single worker thread, bounded by `inference_queue_size`. This serializes requests to the one
+llama.cpp server. Commands jump ahead of chat turns; when the queue is full an execute is refused
+with `error{code:"busy"}` and the oldest queued chat turn is dropped. Leaving chat mode discards
+queued chat turns.
+
 ### TUI keys
 
 `a` arm · `Enter` execute · `Esc` cancel · `e` edit · `t` chat toggle · `q` quit

config.example.toml

@@ -68,14 +68,23 @@ enabled = true
 dir     = "audit"
 
 [llm]
-enabled               = true
-endpoint              = "http://localhost:9931/v1/chat/completions"
-temperature           = 0.7
-max_tokens            = 1024
-timeout_s             = 120.0
-cache_prompt          = true
-max_prompt_tokens     = 64000
-trim_block_turns      = 4
-max_tool_iterations   = 3
-system_prompt_chat    = "You are a concise voice assistant. Reply in 1-3 sentences."
-system_prompt_command = "You translate a spoken request into a single tool call when one applies."
+enabled                  = true
+endpoint                 = "http://localhost:9931/v1/chat/completions"
+temperature              = 0.7
+max_tokens               = 1024
+timeout_s                = 120.0
+cache_prompt             = true
+max_context_tokens       = 131072
+compaction_safety_margin = 256
+compaction_enabled       = true
+compaction_trigger_ratio = 0.8
+compaction_keep_turns    = 6
+compaction_max_tokens    = 512
+compaction_timeout_s     = 30.0
+compaction_input_tokens  = 8192
+compaction_system_prompt = "You compress a conversation. Preserve names, numbers, decisions, user preferences, and unresolved questions. Omit pleasantries. Output only the summary."
+inference_queue_size     = 8
+trim_block_turns         = 4
+max_tool_iterations      = 3
+system_prompt_chat       = "You are a concise voice assistant. Reply in 1-3 sentences."
+system_prompt_command    = "You translate a spoken request into a single tool call when one applies."

src/stt_daemon/assistant/control.py

@@ -17,6 +17,7 @@ import time
 from pathlib import Path
 
 from .. import events
+from .service import EXECUTE_BUSY
 
 logger = logging.getLogger(__name__)
 
@@ -264,7 +265,11 @@ class ControlServer:
         if self.cfg.assistant.require_viewer and self.viewer_count() == 0:
             self._send(client, "error", code="no_viewer", message="no viewer attached")
             return
-        if self.service.execute() is None:
+        result = self.service.execute()
+        if result is EXECUTE_BUSY:
+            self._send(client, "error", code="busy", message="inference queue full")
+            return
+        if result is None:
             self._send(client, "error", code="not_ready", message="command is not ready")
             return
         self._send(client, "ack", command=command, ok=True)

src/stt_daemon/assistant/executor.py

@@ -67,6 +67,14 @@ class LlmExecutor:
                 return ExecutionResult(ok=False, result=f"llm error: {e}")
 
             if not response.tool_calls:
+                if response.finish_reason == "length":
+                    emit("error", code="truncated", message="response truncated")
+                    return ExecutionResult(
+                        ok=False,
+                        result=response.content or "response truncated",
+                        tool=last_tool,
+                        args=last_args,
+                    )
                 return ExecutionResult(
                     ok=True, result=response.content, tool=last_tool, args=last_args
                 )

src/stt_daemon/assistant/inference.py

@@ -0,0 +1,138 @@
+"""Single-lane inference queue for serialized LLM work.
+
+All LLM calls (chat, command execution, compaction) run through one worker thread,
+so there are no concurrent requests to the server.
+"""
+
+import collections
+import logging
+import threading
+from dataclasses import dataclass
+from typing import Callable
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class InferenceJob:
+    """A unit of work for the inference queue."""
+
+    name: str  # "chat" | "command" | "summary"
+    run: Callable[[], None]  # the work to do
+    priority: int = 0  # lower values run first; commands use -1
+
+
+class InferenceQueue:
+    """Single worker thread owning every LLM inference call."""
+
+    def __init__(self, size: int = 8):
+        """Create the queue.
+
+        Args:
+            size: Max queued jobs (bounded backpressure).
+        """
+        self.size = size
+        self._queue: collections.deque[InferenceJob] = collections.deque()
+        self._in_flight: InferenceJob | None = None
+        self._lock = threading.Lock()
+        self._cond = threading.Condition(self._lock)
+        self._stop = threading.Event()
+        self._worker: threading.Thread | None = None
+
+    def start(self) -> None:
+        """Spawn the worker thread."""
+        if self._worker is not None:
+            return
+        self._worker = threading.Thread(
+            target=self._worker_loop, name="assistant-inference", daemon=True
+        )
+        self._worker.start()
+
+    def stop(self) -> None:
+        """Discard queued jobs, let in-flight finish, join the worker."""
+        with self._cond:
+            self._queue.clear()
+            self._stop.set()
+            self._cond.notify_all()
+        if self._worker is not None:
+            self._worker.join(timeout=1.0)
+
+    def submit(self, job: InferenceJob) -> bool:
+        """Queue a job for execution.
+
+        Returns:
+            False only when a priority job (priority < 0) is rejected due to overflow.
+            Normal jobs drop the oldest normal job on overflow and return True.
+        """
+        with self._cond:
+            if self._stop.is_set():
+                return False
+
+            # Capacity includes the in-flight job (if any) and the queue
+            capacity_used = (1 if self._in_flight else 0) + len(self._queue)
+
+            # Check if at capacity
+            if capacity_used >= self.size:
+                if job.priority < 0 or not self._drop_oldest_normal():
+                    return False
+
+            # Insert the job: priority jobs at front, normal jobs at back
+            if job.priority < 0:
+                self._queue.appendleft(job)
+            else:
+                self._queue.append(job)
+
+            self._cond.notify_all()
+        return True
+
+    def drop(self, predicate: Callable[[InferenceJob], bool]) -> int:
+        """Remove queued jobs matching the predicate.
+
+        Does not touch the in-flight job. Returns the count of jobs removed.
+        """
+        with self._cond:
+            to_drop = [j for j in self._queue if predicate(j)]
+            for job in to_drop:
+                self._queue.remove(job)
+            return len(to_drop)
+
+    def depth(self) -> int:
+        """Return the number of queued jobs (not counting in-flight)."""
+        with self._lock:
+            return len(self._queue)
+
+    def _drop_oldest_normal(self) -> bool:
+        """Remove the oldest normal (priority >= 0) job; False if none exists.
+
+        Assumes the lock is held.
+        """
+        for i, job in enumerate(self._queue):
+            if job.priority >= 0:
+                del self._queue[i]
+                return True
+        return False
+
+    def _worker_loop(self) -> None:
+        """Worker thread: dequeue and execute jobs until stop."""
+        while True:
+            with self._cond:
+                while not self._queue and not self._stop.is_set():
+                    self._cond.wait()
+
+                if self._stop.is_set() and not self._queue:
+                    break
+
+                if not self._queue:
+                    continue
+
+                job = self._queue.popleft()
+                self._in_flight = job
+
+            # Release the lock during execution so a concurrent drop() can proceed.
+            try:
+                job.run()
+            except Exception as e:  # noqa: BLE001 - one bad job must not kill the lane
+                logger.error("Inference job %r failed: %s", job.name, e)
+            finally:
+                with self._lock:
+                    self._in_flight = None

src/stt_daemon/assistant/llm.py

@@ -6,6 +6,7 @@ by field name — never by scraping it out of `content`.
 
 import json
 import logging
+import threading
 import urllib.error
 import urllib.request
 from dataclasses import dataclass, field
@@ -67,12 +68,28 @@ class LlmClient:
         self.cfg = cfg
         self._opener = opener or _default_opener
 
-    def chat(self, messages: list[dict], tools: list[dict] | None = None) -> LlmResponse:
-        """Send one completion request and parse the reply."""
+    def chat(
+        self,
+        messages: list[dict],
+        tools: list[dict] | None = None,
+        max_tokens: int | None = None,
+        timeout_s: float | None = None,
+    ) -> LlmResponse:
+        """Send one completion request and parse the reply.
+
+        Args:
+            messages: The conversation messages to send.
+            tools: Optional tool schemas.
+            max_tokens: Per-request output cap; uses cfg.max_tokens when None.
+            timeout_s: Per-request timeout; uses cfg.timeout_s when None.
+        """
+        effective_max_tokens = self.cfg.max_tokens if max_tokens is None else max_tokens
+        effective_timeout   = self.cfg.timeout_s  if timeout_s is None else timeout_s
+
         body: dict[str, Any] = {
             "messages": messages,
             "temperature": self.cfg.temperature,
-            "max_tokens": self.cfg.max_tokens,
+            "max_tokens": effective_max_tokens,
             "cache_prompt": self.cfg.cache_prompt,
             "stream": False,
         }
@@ -82,7 +99,7 @@ class LlmClient:
 
         try:
             payload = self._opener(
-                {"__url__": self.cfg.endpoint, "body": body}, self.cfg.timeout_s
+                {"__url__": self.cfg.endpoint, "body": body}, effective_timeout
             )
         except (urllib.error.URLError, OSError, json.JSONDecodeError) as e:
             raise LlmError(str(e)) from e
@@ -114,45 +131,219 @@ class LlmClient:
 
 
 class ChatSession:
-    """Append-only conversation history with block trimming.
+    """Conversation history with rolling-summary compaction and block-trim fallback.
 
-    The system prompt is fixed (never rebuilt) and history only ever grows,
-    except when it must be trimmed to fit `max_prompt_tokens`; trimming drops
-    the oldest turns in blocks so the cached prefix is invalidated rarely.
+    State:
+        summary  — compressed text of evicted turns (empty string when none).
+        history  — verbatim recent window (append-only between compactions).
+
+    Prompt assembly:
+        [system: system_prompt_chat]
+        [system: "Earlier conversation summary:\\n{summary}"]  # only when summary is set
+        *history
+
+    Lock discipline: self._lock (RLock) guards history/summary mutations and reads.
+    It is NEVER held across an HTTP call.
     """
 
-    def __init__(self, cfg, client: LlmClient):
-        self.cfg = cfg
+    def __init__(self, cfg, client: LlmClient, notify: Callable[..., None] | None = None):
+        self.cfg    = cfg
         self.client = client
         self.system = cfg.system_prompt_chat
+        self.notify = notify
+
         self.history: list[dict] = []
-        self._generation = 0
+        self.summary: str        = ""
+        self._generation         = 0
+        self._lock               = threading.RLock()
 
     def reset(self) -> None:
         """Forget the conversation (volatile by design)."""
-        self.history = []
-        self._generation += 1
+        with self._lock:
+            self.history     = []
+            self.summary     = ""
+            self._generation += 1
 
     def send(self, user_text: str) -> str:
         """Append a user turn, ask the model, append and return the reply."""
-        generation = self._generation
-        self.history.append({"role": "user", "content": user_text})
-        self._trim()
-        messages = [{"role": "system", "content": self.system}] + self.history
+        with self._lock:
+            generation = self._generation
+            self.history.append({"role": "user", "content": user_text})
+
+        self._maybe_compact()
+
+        with self._lock:
+            messages = [{"role": "system", "content": self.system}]
+            if self.summary:
+                messages.append(
+                    {"role": "system", "content": f"Earlier conversation summary:\n{self.summary}"}
+                )
+            messages += self.history
+
+        # HTTP call — no lock held.
         response = self.client.chat(messages)
-        reply = response.content
-        if generation != self._generation:
-            # The session was reset while the model was thinking: the reply belongs
-            # to a conversation that no longer exists, so drop it.
-            return ""
-        self.history.append({"role": "assistant", "content": reply})
+        reply    = response.content
+        truncated = response.finish_reason == "length"
+
+        with self._lock:
+            if generation != self._generation:
+                # The session was reset while the model was thinking: the reply belongs
+                # to a conversation that no longer exists, so drop it.
+                return ""
+            if reply:
+                self.history.append({"role": "assistant", "content": reply})
+
+        if truncated and self.notify is not None:
+            self.notify("truncated", message="response truncated")
+
         return reply
 
     def _estimate_tokens(self) -> int:
-        characters = len(self.system) + sum(len(m["content"]) for m in self.history)
+        # Approximate: 4 chars ≈ 1 token (good enough for budget gating).
+        characters = (
+            len(self.system)
+            + len(self.summary)
+            + sum(len(m["content"]) for m in self.history)
+        )
         return characters // 4
 
     def _trim(self) -> None:
         block = max(1, self.cfg.trim_block_turns) * 2
-        while self._estimate_tokens() > self.cfg.max_prompt_tokens and self.history:
+        while self._estimate_tokens() > self.cfg.prompt_budget and self.history:
             del self.history[: min(block, len(self.history))]
+
+    def _maybe_compact(self) -> None:
+        """Compact history when the prompt approaches the budget.
+
+        Compaction path (compaction_enabled=True):
+          1. Return early if still under the trigger threshold.
+          2. If nothing to evict (window alone is over budget), fall back to _trim().
+          3. Snapshot evicted/recent under the lock, release, call the summarizer.
+          4. Re-acquire, re-check generation; apply or fall back to _trim().
+
+        Legacy path (compaction_enabled=False):
+          Trim under the lock (history stays bounded).
+        """
+        cfg = self.cfg
+
+        if not cfg.compaction_enabled:
+            with self._lock:
+                self._trim()
+            return
+
+        with self._lock:
+            if self._estimate_tokens() <= cfg.compaction_trigger_ratio * cfg.prompt_budget:
+                return
+
+            keep = max(0, cfg.compaction_keep_turns) * 2
+            if len(self.history) <= keep:
+                # The recent window alone is over budget; no turns to evict.
+                # Fall back to block trimming (Oracle R5).
+                self._trim()
+                return
+
+            evicted    = list(self.history[:-keep])
+            recent     = list(self.history[-keep:])
+            generation = self._generation
+            previous   = self.summary
+
+        # Summarizer HTTP call — no lock held.
+        new_summary = self._summarize(previous, evicted)
+
+        with self._lock:
+            if generation != self._generation:
+                # A reset() intervened; the session is already clean.
+                return
+
+            if new_summary is None:
+                # Summarizer failed; fall back to block trimming (Oracle R7/R8).
+                self._trim()
+                return
+
+            self.summary = new_summary
+            self.history = recent
+
+        # Notify outside the lock to avoid re-entry issues.
+        if self.notify is not None:
+            self.notify("compacted", turns=len(evicted))
+
+    def _summarize(self, previous: str, evicted: list[dict]) -> str | None:
+        """Summarize evicted turns, optionally folding over a previous summary.
+
+        Bounded input: if the combined text exceeds compaction_input_tokens * 4
+        characters, split evicted into chronological chunks and fold sequentially.
+
+        Returns the summary string, or None on any failure.
+        """
+        cfg = self.cfg
+        char_limit = cfg.compaction_input_tokens * 4
+
+        # Build chunks that each fit within the character limit.
+        chunks: list[list[dict]] = []
+        current_chunk: list[dict] = []
+        current_chars = 0
+
+        for msg in evicted:
+            msg_chars = len(msg["content"])
+            if current_chunk and current_chars + msg_chars > char_limit:
+                chunks.append(current_chunk)
+                current_chunk = [msg]
+                current_chars = msg_chars
+            else:
+                current_chunk.append(msg)
+                current_chars += msg_chars
+
+        if current_chunk:
+            chunks.append(current_chunk)
+
+        # Fold sequentially: s = summarize_once(s, chunk) for each chunk.
+        running = previous
+        for chunk in chunks:
+            result = self._summarize_once(running, chunk)
+            if result is None:
+                return None
+            running = result
+
+        return running if running != previous else None
+
+    def _summarize_once(self, previous: str, chunk: list[dict]) -> str | None:
+        """Call the LLM to summarize one chunk of evicted turns.
+
+        Returns the summary string, or None on failure (LlmError, empty content,
+        or finish_reason == "length" — Oracle R8).
+        """
+        cfg = self.cfg
+
+        lines = []
+        if previous:
+            lines.append(f"Previous summary:\n{previous}\n")
+        lines.append("Conversation to summarize:")
+        for msg in chunk:
+            lines.append(f"{msg['role']}: {msg['content']}")
+        user_text = "\n".join(lines)
+
+        messages = [
+            {"role": "system", "content": cfg.compaction_system_prompt},
+            {"role": "user",   "content": user_text},
+        ]
+
+        try:
+            response = self.client.chat(
+                messages,
+                max_tokens=cfg.compaction_max_tokens,
+                timeout_s=cfg.compaction_timeout_s,
+            )
+        except LlmError:
+            logger.warning("Summarizer LLM call failed; falling back to block trim")
+            return None
+
+        if not response.content or response.finish_reason == "length":
+            logger.warning(
+                "Summarizer returned unusable reply (empty=%s, finish_reason=%r); "
+                "falling back to block trim",
+                not response.content,
+                response.finish_reason,
+            )
+            return None
+
+        return response.content

src/stt_daemon/assistant/runner.py

@@ -36,6 +36,7 @@ def main() -> int:
     executor = NullExecutor()
     chat_fn = None
     chat_reset = None
+    session = None
     if cfg.llm.enabled:
         client = LlmClient(cfg.llm)
         session = ChatSession(cfg.llm, client)
@@ -51,6 +52,8 @@ def main() -> int:
         chat_fn=chat_fn,
         chat_reset=chat_reset,
     )
+    if session is not None:
+        session.notify = service.chat_notify
     control = ControlServer(cfg, service)
     service.set_sinks(emit=control.broadcast, relay=control.relay)
 

src/stt_daemon/assistant/service.py

@@ -13,6 +13,7 @@ from typing import Callable
 from .. import events
 from ..config import Config
 from .executor import ExecutionResult
+from .inference import InferenceJob, InferenceQueue
 from .modes import Mode, ModeMachine
 
 logger = logging.getLogger(__name__)
@@ -23,6 +24,13 @@ ERROR_EVENT = "error"
 TOOL_CALL_EVENT = "tool_call"
 TOOL_RESULT_EVENT = "tool_result"
 CHAT_MESSAGE_EVENT = "chat_message"
+
+
+class _ExecuteBusy(str):
+    """Sentinel returned by `execute` when the inference queue rejects the job."""
+
+
+EXECUTE_BUSY = _ExecuteBusy("busy")
 HEARTBEAT_EVENT = "heartbeat"
 
 CAPTURING = "capturing"
@@ -99,14 +107,15 @@ class AssistantService:
         self._lock = threading.RLock()
         self._reader: threading.Thread | None = None
         self._ticker: threading.Thread | None = None
-        self._exec_thread: threading.Thread | None = None
+        self._inference = InferenceQueue(size=cfg.llm.inference_queue_size)
 
     # ------------------------------------------------------------------ lifecycle
 
     def start(self) -> None:
-        """Start the event reader and the timeout ticker."""
+        """Start the event reader, the timeout ticker, and the inference queue."""
         if self._reader is not None or self._ticker is not None:
             return
+        self._inference.start()
         self._reader = threading.Thread(
             target=self._read_loop, name="assistant-events", daemon=True
         )
@@ -117,12 +126,13 @@ class AssistantService:
         self._ticker.start()
 
     def stop(self) -> None:
-        """Stop the reader/ticker and close the subscription."""
+        """Stop the reader/ticker, inference queue, and close the subscription."""
         self._stop.set()
         try:
             self._subscriber.close()
         except Exception:  # noqa: BLE001 - best-effort shutdown
             pass
+        self._inference.stop()
         for thread in (self._reader, self._ticker):
             if thread is not None:
                 thread.join(timeout=1)
@@ -193,6 +203,26 @@ class AssistantService:
         self._emit = emit
         self._relay = relay
 
+    def chat_notify(self, kind: str, **fields) -> None:
+        """Route a `ChatSession` notification to the event stream.
+
+        `compacted` becomes a user-visible note in the conversation pane; `truncated`
+        becomes an error so a cut-off reply is never silent.
+        """
+        if kind == "compacted":
+            turns = int(fields.get("turns", 0) or 0)
+            self._emit(
+                CHAT_MESSAGE_EVENT,
+                role="note",
+                text=f"Compacted {turns} earlier turns into a summary.",
+            )
+        elif kind == "truncated":
+            self._emit(
+                ERROR_EVENT,
+                code="truncated",
+                message=str(fields.get("message", "response truncated")),
+            )
+
     def edit(self, text: str) -> bool:
         """Replace the command buffer text (refused unless command-armed)."""
         with self._lock:
@@ -228,8 +258,8 @@ class AssistantService:
     def execute(self) -> str | None:
         """Return the command text and enter `executing`, or None if not allowed.
 
-        When an executor is configured, execution runs on a worker thread so the
-        caller (control socket) is never blocked.
+        Execution runs through the inference queue so the caller (control socket)
+        is never blocked. Returns `EXECUTE_BUSY` when the queue rejects the job.
         """
         # The readiness check and the mode transition must be atomic: the
         # reader thread mutates speech_open/awaited_seg_id concurrently.
@@ -243,13 +273,14 @@ class AssistantService:
         self._emit_mode()
         self._emit_command_text(force=True)
         if self.executor is not None:
-            self._exec_thread = threading.Thread(
-                target=self._run_execution,
-                args=(text, edited),
-                name="assistant-exec",
-                daemon=True,
+            job = InferenceJob(
+                name="command",
+                run=lambda: self._run_execution(text, edited),
+                priority=-1,
             )
-            self._exec_thread.start()
+            if not self._inference.submit(job):
+                self.finish()
+                return EXECUTE_BUSY
         return text
 
     def _run_execution(self, text: str, edited: bool) -> None:
@@ -448,6 +479,7 @@ class AssistantService:
                 return False
             self.chat_history = []
             self._pending_chat = []
+        self._inference.drop(lambda j: j.name == "chat")
         if self.chat_reset is not None:
             try:
                 self.chat_reset()
@@ -465,10 +497,8 @@ class AssistantService:
         if not text:
             return
         self._emit(CHAT_MESSAGE_EVENT, role="user", text=text)
-        thread = threading.Thread(
-            target=self._run_chat, args=(text,), name="assistant-chat", daemon=True
-        )
-        thread.start()
+        job = InferenceJob(name="chat", run=lambda: self._run_chat(text))
+        self._inference.submit(job)
 
     def _run_chat(self, text: str) -> None:
         chat_fn = self.chat_fn

src/stt_daemon/config.py

@@ -119,7 +119,19 @@ class LlmConfig:
     max_tokens: int = 1024
     timeout_s: float = 120.0
     cache_prompt: bool = True
-    max_prompt_tokens: int = 64000
+    max_context_tokens: int = 131072
+    compaction_safety_margin: int = 256
+    compaction_enabled: bool = True
+    compaction_trigger_ratio: float = 0.8
+    compaction_keep_turns: int = 6
+    compaction_max_tokens: int = 512
+    compaction_timeout_s: float = 30.0
+    compaction_input_tokens: int = 8192
+    compaction_system_prompt: str = (
+        "You compress a conversation. Preserve names, numbers, decisions, user "
+        "preferences, and unresolved questions. Omit pleasantries. Output only the summary."
+    )
+    inference_queue_size: int = 8
     trim_block_turns: int = 4
     max_tool_iterations: int = 3
     system_prompt_chat: str = (
@@ -129,6 +141,11 @@ class LlmConfig:
         "You translate a spoken request into a single tool call when one applies."
     )
 
+    @property
+    def prompt_budget(self) -> int:
+        """Prompt tokens available: the context window minus generation and slack."""
+        return self.max_context_tokens - self.max_tokens - self.compaction_safety_margin
+
 
 @dataclass
 class SinkConfig:
@@ -331,6 +348,11 @@ def load_config(path: str | None = None) -> Config:
         # LLM section
         if "llm" in data:
             llm_data = data["llm"]
+            if "max_prompt_tokens" in llm_data:
+                logger.warning(
+                    "llm.max_prompt_tokens is deprecated and ignored; the prompt budget is "
+                    "derived as max_context_tokens - max_tokens - compaction_safety_margin"
+                )
             cfg.llm = LlmConfig(
                 enabled=llm_data.get("enabled", cfg.llm.enabled),
                 endpoint=llm_data.get("endpoint", cfg.llm.endpoint),
@@ -338,8 +360,35 @@ def load_config(path: str | None = None) -> Config:
                 max_tokens=llm_data.get("max_tokens", cfg.llm.max_tokens),
                 timeout_s=llm_data.get("timeout_s", cfg.llm.timeout_s),
                 cache_prompt=llm_data.get("cache_prompt", cfg.llm.cache_prompt),
-                max_prompt_tokens=llm_data.get(
-                    "max_prompt_tokens", cfg.llm.max_prompt_tokens
+                max_context_tokens=llm_data.get(
+                    "max_context_tokens", cfg.llm.max_context_tokens
+                ),
+                compaction_safety_margin=llm_data.get(
+                    "compaction_safety_margin", cfg.llm.compaction_safety_margin
+                ),
+                compaction_enabled=llm_data.get(
+                    "compaction_enabled", cfg.llm.compaction_enabled
+                ),
+                compaction_trigger_ratio=llm_data.get(
+                    "compaction_trigger_ratio", cfg.llm.compaction_trigger_ratio
+                ),
+                compaction_keep_turns=llm_data.get(
+                    "compaction_keep_turns", cfg.llm.compaction_keep_turns
+                ),
+                compaction_max_tokens=llm_data.get(
+                    "compaction_max_tokens", cfg.llm.compaction_max_tokens
+                ),
+                compaction_timeout_s=llm_data.get(
+                    "compaction_timeout_s", cfg.llm.compaction_timeout_s
+                ),
+                compaction_input_tokens=llm_data.get(
+                    "compaction_input_tokens", cfg.llm.compaction_input_tokens
+                ),
+                compaction_system_prompt=llm_data.get(
+                    "compaction_system_prompt", cfg.llm.compaction_system_prompt
+                ),
+                inference_queue_size=llm_data.get(
+                    "inference_queue_size", cfg.llm.inference_queue_size
                 ),
                 trim_block_turns=llm_data.get(
                     "trim_block_turns", cfg.llm.trim_block_turns

tests/test_assistant_service.py

@@ -5,7 +5,8 @@ import threading
 import time
 
 from stt_daemon import events
-from stt_daemon.assistant.service import AssistantService
+from stt_daemon.assistant.inference import InferenceQueue
+from stt_daemon.assistant.service import EXECUTE_BUSY, AssistantService
 from stt_daemon.config import Config
 
 _sequence = itertools.count(1)
@@ -284,6 +285,22 @@ def test_execute_requires_readiness():
     assert service.modes.mode.value == "idle"
 
 
+def test_execute_reports_busy_when_queue_rejects():
+    service, emitted, clock = _service()
+
+    # A zero-capacity lane rejects the priority command job on submit.
+    service._inference = InferenceQueue(size=0)
+    service.executor = object()
+    service.arm()
+    service.handle_event(_transcript(clock, "do it"))
+    service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
+    service.handle_event(_ev(events.SEGMENT_DONE, clock, seg_id=1, emitted=True))
+
+    assert service.execute() is EXECUTE_BUSY
+    assert service.modes.mode.value == "idle"
+    assert not any(e.get("code") == "busy" for e in emitted)
+
+
 def test_turn_ended_requires_quiet_gap():
     service, _, clock = _service()
     service.handle_event(_ev(events.SPEECH_STOP, clock, seg_id=1))
@@ -357,3 +374,23 @@ def test_pending_chat_is_not_lost_across_threads():
     producer.join(timeout=5)
 
     assert sum(len(text.split()) for text in dispatched) == total
+
+
+def test_chat_notify_emits_compaction_note():
+    service, emitted, _ = _service()
+
+    service.chat_notify("compacted", turns=4)
+
+    assert {
+        "type": "chat_message",
+        "role": "note",
+        "text": "Compacted 4 earlier turns into a summary.",
+    } in emitted
+
+
+def test_chat_notify_emits_truncated_error():
+    service, emitted, _ = _service()
+
+    service.chat_notify("truncated", message="cut off")
+
+    assert {"type": "error", "code": "truncated", "message": "cut off"} in emitted

tests/test_config.py

@@ -370,3 +370,75 @@ sticky_redetect_logprob = -0.5
         cfg = load_config(str(config_path))
 
     assert cfg.whisper.sticky_redetect_logprob == -0.5
+
+
+def test_llm_context_and_compaction_defaults():
+    with tempfile.TemporaryDirectory() as tmpdir:
+        import os
+        old_cwd = os.getcwd()
+        try:
+            os.chdir(tmpdir)
+            cfg = load_config()
+        finally:
+            os.chdir(old_cwd)
+
+    assert cfg.llm.max_context_tokens == 131072
+    assert cfg.llm.compaction_safety_margin == 256
+    assert cfg.llm.compaction_enabled is True
+    assert cfg.llm.compaction_trigger_ratio == 0.8
+    assert cfg.llm.compaction_keep_turns == 6
+    assert cfg.llm.compaction_max_tokens == 512
+    assert cfg.llm.compaction_timeout_s == 30.0
+    assert cfg.llm.compaction_input_tokens == 8192
+    assert cfg.llm.inference_queue_size == 8
+    assert cfg.llm.prompt_budget == 131072 - 1024 - 256
+    assert not hasattr(cfg.llm, "max_prompt_tokens")
+
+
+def test_llm_context_and_compaction_fields_parsed():
+    config_toml = """
+[llm]
+max_context_tokens = 8192
+max_tokens = 2048
+compaction_safety_margin = 128
+compaction_enabled = false
+compaction_trigger_ratio = 0.5
+compaction_keep_turns = 2
+compaction_max_tokens = 256
+compaction_timeout_s = 12.5
+compaction_input_tokens = 4096
+inference_queue_size = 3
+"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        config_path = Path(tmpdir) / "config.toml"
+        config_path.write_text(config_toml)
+
+        cfg = load_config(str(config_path))
+
+    assert cfg.llm.max_context_tokens == 8192
+    assert cfg.llm.compaction_safety_margin == 128
+    assert cfg.llm.compaction_enabled is False
+    assert cfg.llm.compaction_trigger_ratio == 0.5
+    assert cfg.llm.compaction_keep_turns == 2
+    assert cfg.llm.compaction_max_tokens == 256
+    assert cfg.llm.compaction_timeout_s == 12.5
+    assert cfg.llm.compaction_input_tokens == 4096
+    assert cfg.llm.inference_queue_size == 3
+    assert cfg.llm.prompt_budget == 8192 - 2048 - 128
+
+
+def test_max_prompt_tokens_is_deprecated_and_ignored(caplog):
+    config_toml = """
+[llm]
+max_prompt_tokens = 50
+"""
+    with tempfile.TemporaryDirectory() as tmpdir:
+        config_path = Path(tmpdir) / "config.toml"
+        config_path.write_text(config_toml)
+
+        with caplog.at_level("WARNING", logger="stt_daemon.config"):
+            cfg = load_config(str(config_path))
+
+    assert "max_prompt_tokens" in caplog.text
+    assert "deprecated" in caplog.text
+    assert cfg.llm.prompt_budget == 131072 - 1024 - 256

tests/test_control.py

@@ -105,6 +105,7 @@ def harness(sock_path, tmp_path):
     server = ControlServer(cfg, service, path=sock_path)
     service.set_sinks(emit=server.broadcast, relay=server.relay)
     server.start()
+    service.start()
 
     yield SimpleNamespace(
         cfg=cfg,

tests/test_executor.py

@@ -91,3 +91,23 @@ def test_tool_loop_is_capped():
     assert result.ok is False
     assert any(e["type"] == "error" and e["code"] == "tool_loop" for e in emitted)
     assert len(client.calls) == 3
+
+
+def test_truncated_empty_answer_is_an_error():
+    executor, _ = _executor([LlmResponse(content="", finish_reason="length")])
+    emitted, emit = _collector()
+    result = executor.run("do it", emit)
+
+    assert result.ok is False
+    assert result.result == "response truncated"
+    assert any(e["type"] == "error" and e["code"] == "truncated" for e in emitted)
+
+
+def test_truncated_partial_answer_reports_partial():
+    executor, _ = _executor([LlmResponse(content="par", finish_reason="length")])
+    emitted, emit = _collector()
+    result = executor.run("do it", emit)
+
+    assert result.ok is False
+    assert result.result == "par"
+    assert any(e["type"] == "error" and e["code"] == "truncated" for e in emitted)

tests/test_inference.py

@@ -0,0 +1,289 @@
+"""Tests for the inference queue (Phase B)."""
+
+import threading
+from stt_daemon.assistant.inference import InferenceJob, InferenceQueue
+
+
+class _JobRunner:
+    """Helper to track job execution timing and detect concurrency."""
+
+    def __init__(self):
+        self.events: list[tuple[str, str]] = []  # [(event, job_name), ...]
+        self.lock = threading.Lock()
+
+    def job(self, name: str, block_until: threading.Event | None = None):
+        """Return a callable that records start/end and optionally blocks."""
+
+        def run():
+            with self.lock:
+                self.events.append(("start", name))
+            if block_until is not None:
+                block_until.wait()
+            with self.lock:
+                self.events.append(("end", name))
+
+        return run
+
+    def in_flight(self) -> str | None:
+        """Return the name of the job currently running, or None."""
+        with self.lock:
+            for i in range(len(self.events) - 1, -1, -1):
+                event, name = self.events[i]
+                if event == "start":
+                    return name
+                if event == "end":
+                    return None
+        return None
+
+
+def test_jobs_run_one_at_a_time():
+    """Max concurrency is 1: job 2 does not start until job 1 ends."""
+    queue = InferenceQueue(size=4)
+    runner = _JobRunner()
+    unblock = threading.Event()
+
+    queue.start()
+
+    job1 = InferenceJob(name="job1", run=runner.job("job1", unblock))
+    job2 = InferenceJob(name="job2", run=runner.job("job2"))
+
+    queue.submit(job1)
+    queue.submit(job2)
+
+    # Give job1 time to start
+    import time
+    time.sleep(0.01)
+
+    # job2 must not have started yet
+    assert runner.in_flight() == "job1"
+
+    # Unblock job1
+    unblock.set()
+    time.sleep(0.05)
+
+    # Now job2 should be running or done
+    assert any(event[1] == "job2" for event in runner.events)
+
+    queue.stop()
+
+
+def test_priority_jobs_inserted_at_front():
+    """Priority job (priority < 0) submitted while normal job queued runs next."""
+    queue = InferenceQueue(size=4)
+    runner = _JobRunner()
+    unblock1 = threading.Event()
+    unblock_cmd = threading.Event()
+
+    queue.start()
+
+    job1 = InferenceJob(name="normal", run=runner.job("normal", unblock1), priority=0)
+    job_cmd = InferenceJob(name="command", run=runner.job("command", unblock_cmd), priority=-1)
+    job3 = InferenceJob(name="chat2", run=runner.job("chat2"), priority=0)
+
+    queue.submit(job1)
+    queue.submit(job3)
+    queue.submit(job_cmd)
+
+    import time
+    time.sleep(0.1)
+
+    # Release all blocks to complete all jobs
+    unblock1.set()
+    unblock_cmd.set()
+    time.sleep(0.1)
+
+    # Check that command ran before chat2 (priority ordering)
+    names = [name for _, name in runner.events if _ == "end"]
+    assert "command" in names
+    assert "chat2" in names
+    assert "normal" in names
+    # Priority guarantees: command should run before chat2
+    cmd_index = names.index("command")
+    chat_index = names.index("chat2")
+    assert cmd_index < chat_index
+
+    queue.stop()
+
+
+def test_priority_job_rejected_when_queue_full():
+    """Overflow: priority job is rejected (submit returns False)."""
+    queue = InferenceQueue(size=2)
+    runner = _JobRunner()
+    unblock = threading.Event()
+
+    queue.start()
+
+    job1 = InferenceJob(name="job1", run=runner.job("job1", unblock), priority=0)
+    job2 = InferenceJob(name="job2", run=runner.job("job2"), priority=0)
+    job_cmd = InferenceJob(name="command", run=runner.job("command"), priority=-1)
+
+    assert queue.submit(job1) is True
+    assert queue.submit(job2) is True
+    # Queue is now full (size=2: job1 in-flight, job2 queued)
+    assert queue.submit(job_cmd) is False
+
+    queue.stop()
+
+
+def test_normal_job_evicts_oldest_on_overflow():
+    """Overflow: normal job drops the oldest normal job."""
+    queue = InferenceQueue(size=2)
+    runner = _JobRunner()
+    unblock = threading.Event()
+
+    queue.start()
+
+    job1 = InferenceJob(name="job1", run=runner.job("job1", unblock), priority=0)
+    job2 = InferenceJob(name="job2", run=runner.job("job2"), priority=0)
+    job3 = InferenceJob(name="job3", run=runner.job("job3"), priority=0)
+
+    assert queue.submit(job1) is True
+    import time
+    time.sleep(0.05)
+
+    # Ensure job1 is in-flight
+    while not runner.in_flight():
+        time.sleep(0.001)
+
+    # Now queue is: job1 in-flight, job2 queued, room for 1 more
+    assert queue.submit(job2) is True
+    # Queue is now full (in-flight + 1 queued)
+    # job3 submits, evicts job2 (oldest queued normal job)
+    assert queue.submit(job3) is True
+
+    unblock.set()
+    time.sleep(0.1)
+
+    # Only job1 and job3 should have run; job2 was evicted
+    names = [name for _, name in runner.events if _ == "end"]
+    assert "job2" not in names
+    assert set(names) == {"job1", "job3"}
+
+    queue.stop()
+
+
+def test_drop_removes_matching_queued_jobs():
+    """drop(predicate) removes queued jobs matching predicate, returns count."""
+    queue = InferenceQueue(size=8)
+    runner = _JobRunner()
+    unblock = threading.Event()
+
+    queue.start()
+
+    job1 = InferenceJob(name="chat", run=runner.job("chat1", unblock), priority=0)
+    job2 = InferenceJob(name="chat", run=runner.job("chat2"), priority=0)
+    job3 = InferenceJob(name="cmd", run=runner.job("cmd"), priority=-1)
+
+    queue.submit(job1)
+    queue.submit(job2)
+    queue.submit(job3)
+
+    import time
+    time.sleep(0.05)
+
+    # Give time for job1 to start in-flight
+    while not runner.in_flight():
+        time.sleep(0.001)
+
+    # Drop all chat jobs; job1 is in-flight, job2 is queued
+    count = queue.drop(lambda j: j.name == "chat")
+    assert count == 1  # job2 is queued; job1 is in-flight and not dropped
+
+    unblock.set()
+    time.sleep(0.1)
+
+    # Only job1 (in-flight, not dropped) and job3 should run
+    names = [name for _, name in runner.events if _ == "end"]
+    assert set(names) == {"chat1", "cmd"}
+
+    queue.stop()
+
+
+def test_stop_discards_queued_and_joins():
+    """stop() discards queued jobs, lets in-flight finish, then joins."""
+    queue = InferenceQueue(size=4)
+    runner = _JobRunner()
+    unblock = threading.Event()
+
+    queue.start()
+
+    job1 = InferenceJob(name="job1", run=runner.job("job1", unblock))
+    job2 = InferenceJob(name="job2", run=runner.job("job2"))
+    job3 = InferenceJob(name="job3", run=runner.job("job3"))
+
+    queue.submit(job1)
+    queue.submit(job2)
+    queue.submit(job3)
+
+    import time
+    time.sleep(0.01)
+
+    queue.stop()
+
+    # job1 was in-flight, so let it finish
+    unblock.set()
+    time.sleep(0.05)
+
+    # Only job1 should have run; job2 and job3 were discarded
+    names = [name for _, name in runner.events if _ == "end"]
+    assert names == ["job1"]
+
+
+def test_depth_reflects_queued_count():
+    """depth() returns the number of queued (not in-flight) jobs."""
+    queue = InferenceQueue(size=8)
+    runner = _JobRunner()
+    unblock = threading.Event()
+
+    queue.start()
+
+    job1 = InferenceJob(name="job1", run=runner.job("job1", unblock))
+    job2 = InferenceJob(name="job2", run=runner.job("job2"))
+    job3 = InferenceJob(name="job3", run=runner.job("job3"))
+
+    queue.submit(job1)
+    import time
+    time.sleep(0.05)
+    # Give worker time to dequeue job1
+    while queue.depth() > 0 and not runner.in_flight():
+        time.sleep(0.001)
+
+    queue.submit(job2)
+    queue.submit(job3)
+    # job1 is in-flight, job2 and job3 are queued
+    assert queue.depth() >= 1
+
+    unblock.set()
+    time.sleep(0.1)
+
+    # All jobs should finish; no more queued
+    assert queue.depth() == 0
+
+    queue.stop()
+
+
+def test_no_submit_after_stop():
+    """submit() returns False after stop()."""
+    queue = InferenceQueue(size=4)
+    queue.start()
+    queue.stop()
+
+    job = InferenceJob(name="job", run=lambda: None)
+    assert queue.submit(job) is False
+
+
+def test_bound_holds_when_only_priority_jobs_are_present():
+    """A normal job is rejected when there is no normal job to evict."""
+    queue = InferenceQueue(size=1)
+    release = threading.Event()
+
+    def block():
+        release.wait()
+
+    queue.start()
+    try:
+        assert queue.submit(InferenceJob(name="cmd", run=block, priority=-1)) is True
+        assert queue.submit(InferenceJob(name="chat", run=lambda: None)) is False
+    finally:
+        release.set()
+        queue.stop()

tests/test_llm.py

@@ -102,7 +102,13 @@ def test_chat_session_appends_and_keeps_prefix():
 
 
 def test_chat_session_trims_in_blocks_and_keeps_system_prompt():
-    cfg = LlmConfig(max_prompt_tokens=50, trim_block_turns=1)
+    cfg = LlmConfig(
+        max_context_tokens=1330,
+        max_tokens=1024,
+        compaction_safety_margin=256,
+        trim_block_turns=1,
+        compaction_enabled=False,
+    )  # prompt_budget == 50
     client, calls = _make_client(lambda _p: _reply(content="ok"))
     session = ChatSession(cfg, client)
 
@@ -132,3 +138,449 @@ def test_chat_session_drops_reply_when_reset_mid_flight():
 
     assert holder[0].send("hi") == ""
     assert holder[0].history == []
+
+
+# ---------------------------------------------------------------------------
+# Phase C tests
+# ---------------------------------------------------------------------------
+
+def _compact_cfg(
+    *,
+    budget_chars=200,
+    trigger_ratio=0.8,
+    keep_turns=1,
+    compaction_max_tokens=64,
+    compaction_timeout_s=5.0,
+    compaction_input_tokens=8192,
+    compaction_enabled=True,
+    trim_block_turns=1,
+):
+    """Return a small LlmConfig whose prompt_budget == budget_chars // 4 tokens.
+
+    prompt_budget = max_context_tokens - max_tokens - compaction_safety_margin
+    We want budget_chars // 4 tokens, so:
+        max_context_tokens = (budget_chars // 4) + max_tokens + compaction_safety_margin
+    """
+    max_tokens              = 16
+    compaction_safety_margin = 0
+    max_context_tokens      = (budget_chars // 4) + max_tokens + compaction_safety_margin
+    return LlmConfig(
+        max_context_tokens=max_context_tokens,
+        max_tokens=max_tokens,
+        compaction_safety_margin=compaction_safety_margin,
+        compaction_enabled=compaction_enabled,
+        compaction_trigger_ratio=trigger_ratio,
+        compaction_keep_turns=keep_turns,
+        compaction_max_tokens=compaction_max_tokens,
+        compaction_timeout_s=compaction_timeout_s,
+        compaction_input_tokens=compaction_input_tokens,
+        trim_block_turns=trim_block_turns,
+        system_prompt_chat="SYS",
+        system_prompt_command="CMD",
+        compaction_system_prompt="COMPACT",
+    )
+
+
+def test_compaction_fires_and_window_shrinks():
+    """Crossing the trigger compacts: window shrinks, summary present in next prompt."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+    # prompt_budget = 200//4 = 50 tokens; trigger at 40 tokens (0.8 * 50)
+
+    summary_text = "summary of old turns"
+    call_index   = 0
+
+    def responder(payload):
+        nonlocal call_index
+        msgs = payload["body"]["messages"]
+        call_index += 1
+        # Summarizer calls have the compaction system prompt.
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content=summary_text)
+        return _reply(content="ok")
+
+    client, calls = _make_client(responder, cfg)
+    session = ChatSession(cfg, client)
+
+    # Send enough turns to exceed the trigger (each "x"*40 = 40 chars = 10 tokens).
+    # After 4 turns (user+assistant pairs = 8 messages * 10 tokens = 80 tokens > 40 trigger).
+    for _ in range(4):
+        session.send("x" * 40)
+
+    # After compaction the window must be smaller than the full history.
+    assert len(session.history) <= 4  # keep_turns=1 → keep=2 messages
+
+    # The summary must be set.
+    assert session.summary != ""
+
+    # The next chat call must include the summary as a second system message,
+    # with the chat system prompt FIRST.
+    session.send("after compaction")
+    chat_calls = [c for c in calls if c["body"]["messages"][0]["content"] == "SYS"]
+    last_chat  = chat_calls[-1]["body"]["messages"]
+    assert last_chat[0]["role"] == "system"
+    assert last_chat[0]["content"] == "SYS"
+    assert last_chat[1]["role"] == "system"
+    assert "Earlier conversation summary:" in last_chat[1]["content"]
+    assert summary_text in last_chat[1]["content"]
+
+
+def test_summary_never_appended_to_history():
+    """The summary text must not appear as a history entry."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+    summary_text = "SUMMARY_SENTINEL"
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content=summary_text)
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    for msg in session.history:
+        assert summary_text not in msg["content"]
+
+
+def test_summarizer_failure_falls_back_to_trim():
+    """On summarizer failure the session falls back to _trim() and stays within budget."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content="", finish="stop")  # empty → failure
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    # Summary must remain empty (fallback used trim, not compaction).
+    assert session.summary == ""
+    # History must be bounded within the budget.
+    assert session._estimate_tokens() <= cfg.prompt_budget
+
+
+def test_summarizer_length_finish_reason_falls_back_to_trim():
+    """finish_reason == 'length' on the summarizer is treated as failure."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content="partial summary", finish="length")
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    assert session.summary == ""
+    assert session._estimate_tokens() <= cfg.prompt_budget
+
+
+def test_summarizer_llmerror_falls_back_to_trim():
+    """LlmError from the summarizer falls back to _trim()."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            raise LlmError("network error")
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    assert session.summary == ""
+    assert session._estimate_tokens() <= cfg.prompt_budget
+
+
+def test_reset_clears_summary():
+    """reset() clears both history and summary."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+    summary_text = "old summary"
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content=summary_text)
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    assert session.summary != ""
+
+    session.reset()
+    assert session.history == []
+    assert session.summary == ""
+
+
+def test_compaction_disabled_never_calls_summarizer():
+    """compaction_enabled=False never calls the summarizer and still trims."""
+    cfg = _compact_cfg(
+        budget_chars=200,
+        trigger_ratio=0.8,
+        keep_turns=1,
+        compaction_enabled=False,
+    )
+
+    summarizer_called = False
+
+    def responder(payload):
+        nonlocal summarizer_called
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            summarizer_called = True
+            return _reply(content="should not happen")
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    for _ in range(6):
+        session.send("x" * 40)
+
+    assert not summarizer_called
+    assert session.summary == ""
+    # History must still be bounded (trim ran).
+    assert session._estimate_tokens() <= cfg.prompt_budget
+
+
+def test_nothing_to_evict_skips_summarizer():
+    """When len(history) <= keep, compaction is skipped and trim runs instead."""
+    # keep_turns=10 → keep=20 messages; we only send 2 turns (4 messages).
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=10)
+
+    summarizer_called = False
+
+    def responder(payload):
+        nonlocal summarizer_called
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            summarizer_called = True
+            return _reply(content="should not happen")
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    # Push over the trigger but keep history shorter than keep=20.
+    for _ in range(4):
+        session.send("x" * 40)
+
+    assert not summarizer_called
+
+
+def test_chunk_and_fold_large_evicted():
+    """Large evicted list triggers chunk-and-fold (more than one summarizer call).
+
+    budget_chars=160 → prompt_budget=40 tokens, trigger at 32 tokens (0.8 * 40).
+    Each turn: user='x'*40 (40 chars) + assistant='ok' (2 chars) = ~10 tokens.
+    Compaction fires at turn 4 with 6 evicted messages.
+    compaction_input_tokens=1 → char_limit=4 chars per chunk, so each 40-char
+    or 2-char message is its own chunk → 6 summarizer calls for 6 chunks.
+    """
+    cfg = _compact_cfg(
+        budget_chars=160,
+        trigger_ratio=0.8,
+        keep_turns=1,
+        compaction_input_tokens=1,  # 1 token * 4 = 4 chars per chunk → each msg is a chunk
+    )
+
+    summarizer_calls = []
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            summarizer_calls.append(payload)
+            return _reply(content="folded")
+        return _reply(content="ok")
+
+    client, _ = _make_client(responder, cfg)
+    session   = ChatSession(cfg, client)
+
+    # 4 turns → 8 messages → fires compaction with 6 evicted (keep=2).
+    for _ in range(4):
+        session.send("x" * 40)
+
+    # With compaction_input_tokens=1 (4 chars/chunk) and 40-char messages,
+    # each message is its own chunk → multiple summarizer calls.
+    assert len(summarizer_calls) > 1
+
+
+def test_chat_per_request_max_tokens_override():
+    """LlmClient.chat sends per-request max_tokens when given, cfg default otherwise."""
+    cfg = LlmConfig(max_tokens=1024, timeout_s=60.0)
+    client, calls = _make_client(lambda _p: _reply(content="ok"), cfg)
+
+    # Default: uses cfg.max_tokens.
+    client.chat([{"role": "user", "content": "hi"}])
+    assert calls[0]["body"]["max_tokens"] == 1024
+
+    # Override: uses the supplied value.
+    client.chat([{"role": "user", "content": "hi"}], max_tokens=128)
+    assert calls[1]["body"]["max_tokens"] == 128
+
+
+def test_chat_per_request_timeout_override():
+    """LlmClient.chat uses per-request timeout_s when given, cfg default otherwise."""
+    cfg = LlmConfig(max_tokens=1024, timeout_s=60.0)
+    recorded_timeouts: list[float] = []
+
+    def opener(payload, timeout):
+        recorded_timeouts.append(timeout)
+        return _reply(content="ok")
+
+    client = LlmClient(cfg, opener=opener)
+
+    # Default timeout.
+    client.chat([{"role": "user", "content": "hi"}])
+    assert recorded_timeouts[0] == 60.0
+
+    # Per-request override.
+    client.chat([{"role": "user", "content": "hi"}], timeout_s=5.0)
+    assert recorded_timeouts[1] == 5.0
+
+
+def test_summarizer_uses_compaction_max_tokens_and_timeout():
+    """The summarizer call uses compaction_max_tokens and compaction_timeout_s."""
+    cfg = _compact_cfg(
+        budget_chars=200,
+        trigger_ratio=0.8,
+        keep_turns=1,
+        compaction_max_tokens=42,
+        compaction_timeout_s=7.0,
+    )
+
+    recorded: list[dict] = []
+
+    def opener(payload, timeout):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            recorded.append({"max_tokens": payload["body"]["max_tokens"], "timeout": timeout})
+            return _reply(content="summary")
+        return _reply(content="ok")
+
+    client  = LlmClient(cfg, opener=opener)
+    session = ChatSession(cfg, client)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    assert len(recorded) >= 1
+    assert recorded[0]["max_tokens"] == 42
+    assert recorded[0]["timeout"] == 7.0
+
+
+def test_notify_called_on_compaction():
+    """notify('compacted', turns=N) is called after a successful compaction."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+
+    notifications: list[tuple] = []
+
+    def notify(kind, **fields):
+        notifications.append((kind, fields))
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content="summary")
+        return _reply(content="ok")
+
+    client  = LlmClient(cfg, opener=lambda p, t: responder(p))
+    session = ChatSession(cfg, client, notify=notify)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    assert any(kind == "compacted" for kind, _ in notifications)
+    compacted = [(k, f) for k, f in notifications if k == "compacted"]
+    assert compacted[0][1]["turns"] > 0
+
+
+def test_notify_not_called_when_none():
+    """ChatSession works without a notify callback (default None)."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content="summary")
+        return _reply(content="ok")
+
+    client  = LlmClient(cfg, opener=lambda p, t: responder(p))
+    session = ChatSession(cfg, client)  # no notify
+
+    # Must not raise.
+    for _ in range(4):
+        session.send("x" * 40)
+
+
+def test_system_prompt_is_first_message_after_compaction():
+    """After compaction the chat system prompt is always the first message."""
+    cfg = _compact_cfg(budget_chars=200, trigger_ratio=0.8, keep_turns=1)
+
+    def responder(payload):
+        msgs = payload["body"]["messages"]
+        if msgs[0]["content"] == "COMPACT":
+            return _reply(content="summary text")
+        return _reply(content="ok")
+
+    client, calls = _make_client(responder, cfg)
+    session       = ChatSession(cfg, client)
+
+    for _ in range(4):
+        session.send("x" * 40)
+
+    # Find the last chat call (system prompt == "SYS", not "COMPACT").
+    chat_calls = [c for c in calls if c["body"]["messages"][0]["content"] == "SYS"]
+    assert chat_calls, "no chat calls found after compaction"
+    last_msgs = chat_calls[-1]["body"]["messages"]
+    assert last_msgs[0]["role"] == "system"
+    assert last_msgs[0]["content"] == "SYS"
+
+
+def test_chat_session_truncated_empty_reply_notifies_and_is_not_appended():
+    """A length-limited empty reply is reported and never stored as an empty turn."""
+    notifications: list[tuple] = []
+
+    def notify(kind, **fields):
+        notifications.append((kind, fields))
+
+    client, _ = _make_client(lambda _p: _reply(content="", finish="length"))
+    session = ChatSession(LlmConfig(), client, notify=notify)
+
+    assert session.send("hi") == ""
+    assert session.history == [{"role": "user", "content": "hi"}]
+    assert notifications == [("truncated", {"message": "response truncated"})]
+
+
+def test_chat_session_truncated_partial_reply_is_appended_and_notified():
+    """A truncated but non-empty reply is kept and reported as truncated."""
+    notifications: list[tuple] = []
+
+    def notify(kind, **fields):
+        notifications.append((kind, fields))
+
+    client, _ = _make_client(lambda _p: _reply(content="partial answer", finish="length"))
+    session = ChatSession(LlmConfig(), client, notify=notify)
+
+    assert session.send("hi") == "partial answer"
+    assert session.history[-1] == {"role": "assistant", "content": "partial answer"}
+    assert any(kind == "truncated" for kind, _ in notifications)

tests/test_tui_state.py

@@ -173,6 +173,24 @@ def test_heartbeat_mode_change_clears_chat():
     assert state.lower_lines() == []
 
 
+def test_compaction_note_renders_and_clears_on_chat_exit():
+    state = ViewState()
+    state.ingest({"type": "mode", "mode": "chat"})
+    state.ingest(
+        {
+            "type": "chat_message",
+            "role": "note",
+            "text": "Compacted 4 earlier turns into a summary.",
+        }
+    )
+
+    assert "note: Compacted 4 earlier turns into a summary." in state.lower_lines()
+
+    state.ingest({"type": "mode", "mode": "idle"})
+
+    assert state.lower_lines() == []
+
+
 def test_entering_chat_does_not_clear_rendered_log():
     state = ViewState()
     state.ingest({"type": "tool_result", "tool": "get_datetime", "ok": True, "result": "x"})