Compaction: the model's notes replace the middle of a long run (#158)

Before each driver call, over compact_at (0.6) of the prompt budget, the
same model is asked for notes over the very request it was about to get,
plus one message, so the server's cache covers all but the notes. The notes,
with code's list of the calls they replace, become a compaction event that
renders in place of its range, keeping word for word what started the work
and the last few turns. Rare, large steps keep the prompt's start unchanged
between them. A later compaction folds in the earlier; one from an earlier
run's view is hidden.

Failed notes are logged and fitting takes over. Fitting now applies
compactions before it measures, records what it cuts (Trim) and logs each
cut as context.trimmed. Wired into both agents' drivers and the benchmark;
nisse-dev's helper loop is a follow-up.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01J5Utyfzw7gF7UhjRDEtjTk

nisse-core/src/nisse_core/loop/driver.py

@@ -2,7 +2,8 @@
 tool, runs out of turns, or is stopped.
 
 Everything the run does is written to the item's transcript as the driver
-(L7), and every model call sees the driver's view of it (L6). Execution tools
+(L7), and every model call sees the driver's view of it (L6), compacted
+when it grows long (``compact``). Execution tools
 go to the item's executor, on its tool VM (L13); control tools are answered
 here, and a waiting one ends the run, so the scheduler can put it to the
 operator. Which tools wait is the agent's: each run is told.
@@ -14,6 +15,7 @@ from typing import Any, Literal, Protocol
 
 from nisse_core.log.canonical import AssistantText, Fact, ToolCall, ToolResult
 from nisse_core.loop.tools import names
+from nisse_core.models.compact import Compactor
 from nisse_core.models.llm import LLMClient, LLMError
 from nisse_core.state.cancel import Cancel, skip_calls
 from nisse_core.state.store import Transcript
@@ -82,8 +84,10 @@ class Driver:
         max_turns: int = DEFAULT_MAX_TURNS,
         cancel: Cancel | None = None,
         failure_limit: int = 0,  # 0: no limit
+        compactor: Compactor | None = None,
     ) -> None:
         self._client = client
+        self._compactor = compactor
         self._model = model
         self._max_turns = max_turns
         self._failure_limit = failure_limit
@@ -123,6 +127,12 @@ class Driver:
             # L2: stop takes effect at the next tool boundary.
             if self._cancel.requested:
                 return RunOutcome("stopped", note=f"before turn {turn + 1}")
+            if self._compactor is not None:
+                # Not a turn: the model's notes replace the middle of a long
+                # run before it outgrows the window (L6).
+                self._compactor.compact(
+                    transcript, system=system, tools=tools, model=model
+                )
             try:
                 produced = self._client.complete(
                     transcript.view(), tools=tools, model=model, system=system

nisse-core/src/nisse_core/models/compact.py

@@ -0,0 +1,262 @@
+"""Compaction (architecture L6, L7): the model's own notes replace the middle
+of a long run, so it stays inside the window without losing what it found.
+
+Before each model call of a driver run, ``Compactor.compact`` looks at the
+size of the request. Below ``compact_at`` (a fraction of the prompt budget)
+it does nothing. Above it, in one step:
+
+1. It keeps, word for word, everything before the first turn (the brief, or
+   the conversation that started the work) and the last few turns, as many
+   as fit in ``compact_to`` of the budget, at most ``keep_turns``.
+2. It asks the same model, over the very same request plus one more
+   message, for notes on everything in between. The model server has that
+   request cached from the last call, so the notes cost only their own
+   output. The model offered the run's tools is not made to call one.
+3. It records a ``Compaction`` event replacing that range, the notes with a
+   list of the calls made in it appended by code, so the model knows what it
+   can look at again. The log keeps everything it replaces.
+
+Doing this rarely, and well below the limit, keeps the start of the prompt
+the same for many calls, which is what lets the server reuse its cache.
+Cutting a little every call, as fitting does, changes it every time.
+
+A compaction that fails (the call fails, the notes are empty, or they leave
+the request still over ``compact_at``) is logged and the run goes on; the
+request is then fitted as before (``context``). It isn't tried again until
+``keep_turns`` more turns have passed.
+"""
+
+import json
+import time
+from collections.abc import Sequence
+from dataclasses import dataclass
+from typing import Any
+
+from nisse_core.log.canonical import (
+    AssistantText,
+    Compaction,
+    StoredEvent,
+    ToolCall,
+    UserMessage,
+)
+from nisse_core.models.context import _MODEL_SIDE, _turn_starts
+from nisse_core.models.llm import LLMError
+from nisse_core.models.render import effective_events
+from nisse_core.oplog import OpLog
+from nisse_core.state.store import Transcript
+
+# The list of calls appended to the notes: how many, and how long each line.
+MAX_LISTED_CALLS = 80
+LISTED_CALL_CHARS = 200
+
+NOTES_REQUEST = """\
+[orchestrator] Your context is nearly full. Write notes to replace the \
+conversation from your task up to your last few turns (the task itself and \
+those turns stay as they are). Reply with the notes as plain text, not a \
+tool call. Include:
+
+- What the task is, and what you have done so far.
+- Your findings and conclusions so far, each with its file and line and \
+the exact code or text it rests on, quoted.
+- Decisions made and anything the operator or a peer asked for, word for \
+word where the wording matters.
+- What each file you read or command you ran told you that you still need.
+- What is left to do, in order.
+
+Leave out what you no longer need. These notes are all you will have of that \
+part of the conversation; anything else you would have to look at again."""
+
+
+@dataclass(frozen=True)
+class CompactionTuning:
+    compact_at: float = 0.6  # of the prompt budget; 0: never compact
+    compact_to: float = 0.25  # what the kept recent turns may take
+    keep_turns: int = 3  # recent turns kept word for word, at most
+
+
+def check_tuning(tuning: CompactionTuning, section: str, errors: list[str]) -> None:
+    """The agents' config check: notes must leave the request smaller than
+    what set them off, or every call would compact again."""
+    if tuning.compact_at and not tuning.compact_to < tuning.compact_at <= 1:
+        errors.append(
+            f"{section}: compact_to must be less than compact_at, which is at "
+            "most 1 (or 0: never compact)"
+        )
+
+
+class Compactor:
+    def __init__(
+        self,
+        client: Any,
+        tuning: CompactionTuning | None = None,
+        *,
+        oplog: OpLog | None = None,
+    ) -> None:
+        """``client``: the run's client. Compaction needs its budget
+        (``budget_for``, ``estimate``); with a client that has none, as in
+        tests with a scripted model, it never compacts."""
+        self._client = client
+        self._tuning = tuning or CompactionTuning()
+        self._oplog = oplog
+        # Per run (work item, first event shown): the number of turns when
+        # compaction last failed.
+        self._failed_at: dict[tuple[str | None, int], int] = {}
+
+    def compact(
+        self,
+        transcript: Transcript,
+        *,
+        system: str | None,
+        tools: list[dict[str, Any]] | None,
+        model: str,
+    ) -> bool:
+        """Compact the transcript if its request is over ``compact_at``.
+        True if a compaction was recorded."""
+        tuning = self._tuning
+        budget_for = getattr(self._client, "budget_for", None)
+        estimate = getattr(self._client, "estimate", None)
+        if not tuning.compact_at or budget_for is None or estimate is None:
+            return False
+        limit = budget_for(model).for_prompt
+        events = effective_events(transcript.view())
+        before = estimate(events, system, tools, model)
+        if before <= tuning.compact_at * limit:
+            return False
+
+        starts = _turn_starts(events)
+        run = (transcript.work_item, events[0].id if events else 0)
+        failed_at = self._failed_at.get(run)
+        if failed_at is not None and len(starts) < failed_at + tuning.keep_turns:
+            return False
+        head = _head_end(events)
+        # The recent turns kept: at most keep_turns, as many as fit in
+        # compact_to, and at least one, so the model sees where it is.
+        later = [s for s in starts if s > head]
+        recent_from = len(events)
+        for start in reversed(later[-tuning.keep_turns :]):
+            if recent_from < len(events) and (
+                estimate(events[start:], None, None, model) > tuning.compact_to * limit
+            ):
+                break
+            recent_from = start
+        middle = events[head:recent_from]
+        turns = sum(1 for s in later if s < recent_from)
+        if turns < 2:
+            # Too little to be worth notes (or nothing at all): fitting
+            # covers it, and logs what it cuts.
+            return False
+
+        started = time.monotonic()
+        try:
+            notes = self._notes(events, system, tools, model)
+        except LLMError as exc:  # a failed compaction never fails the run
+            return self._failed(run, transcript, len(starts), f"the call failed: {exc}")
+        if not notes:
+            return self._failed(run, transcript, len(starts), "the notes were empty")
+
+        first = min(
+            e.event.replaces[0] if isinstance(e.event, Compaction) else e.id
+            for e in middle
+        )
+        last = max(e.id for e in middle if not isinstance(e.event, Compaction))
+        listed = _calls(middle)
+        summary = notes + (
+            "\n\nCalls made in the part these notes replace (make one again "
+            "to see its result):\n" + "\n".join(listed)
+            if listed
+            else ""
+        )
+        compaction = Compaction(summary=summary, model=model, replaces=[first, last])
+        placed = StoredEvent(id=first, ts="", event=compaction)
+        after = estimate(
+            [*events[:head], placed, *events[recent_from:]], system, tools, model
+        )
+        if after > tuning.compact_at * limit:
+            return self._failed(
+                run,
+                transcript,
+                len(starts),
+                f"the request would still be {after} tokens, over "
+                f"{int(tuning.compact_at * limit)}",
+            )
+        transcript.append(compaction)
+        self._failed_at.pop(run, None)
+        if self._oplog is not None:
+            self._oplog.emit(
+                "context.compacted",
+                cause=f"the request was {before} tokens of {limit}",
+                work_item=transcript.work_item,
+                model=model,
+                tokens_before=before,
+                tokens_after=after,
+                replaced=[first, last],
+                turns_replaced=turns,
+                seconds=round(time.monotonic() - started, 1),
+            )
+        return True
+
+    def _notes(
+        self,
+        events: list[StoredEvent],
+        system: str | None,
+        tools: list[dict[str, Any]] | None,
+        model: str,
+    ) -> str:
+        # The same system prompt and tools as the run's calls, so the server
+        # reuses the cached request: only the last message is new.
+        ask = StoredEvent(
+            id=0, ts="", event=UserMessage(text=NOTES_REQUEST, source="compaction")
+        )
+        produced = self._client.complete(
+            [*events, ask], tools, model, system=system, require_tool_call=False
+        )
+        return "\n".join(
+            e.text.strip() for e in produced if isinstance(e, AssistantText)
+        ).strip()
+
+    def _failed(
+        self,
+        run: tuple[str | None, int],
+        transcript: Transcript,
+        turns: int,
+        why: str,
+    ) -> bool:
+        self._failed_at[run] = turns
+        if self._oplog is not None:
+            self._oplog.emit(
+                "context.compaction_failed",
+                cause=why[:300],
+                work_item=transcript.work_item,
+            )
+        return False
+
+
+def _head_end(events: Sequence[StoredEvent]) -> int:
+    """Where the part before the first turn ends: the brief, or what started
+    the work. An earlier compaction is the start of the middle."""
+    for index, stored in enumerate(events):
+        if isinstance(stored.event, (*_MODEL_SIDE, Compaction)):
+            return index
+    return len(events)
+
+
+def _calls(events: Sequence[StoredEvent]) -> list[str]:
+    """The calls made in ``events``, once each, in order, as lines; earlier
+    compactions' lists carried over."""
+    lines: list[str] = []
+    for stored in events:
+        event = stored.event
+        if isinstance(event, Compaction):
+            _, marker, listed = event.summary.partition(
+                "(make one again to see its result):\n"
+            )
+            if marker:
+                lines.extend(listed.splitlines())
+        elif isinstance(event, ToolCall):
+            arguments = json.dumps(event.arguments, ensure_ascii=False)
+            line = f"- {event.tool} {arguments}"
+            if len(line) > LISTED_CALL_CHARS:
+                line = line[: LISTED_CALL_CHARS - 3] + "..."
+            lines.append(line)
+    unique = list(dict.fromkeys(lines))
+    return unique[-MAX_LISTED_CALLS:]

nisse-core/src/nisse_core/models/context.py

@@ -6,8 +6,15 @@ start of the prompt, which is where the system prompt and the plan are. So a
 request is made to fit the model's declared window *before* it is sent, and
 one that cannot be made to fit is never sent.
 
-The log itself is never changed. Fitting works on the events about to be
-rendered, in three steps, stopping at the first that fits:
+The log itself is never changed. Compaction (``compact``) is what keeps a
+long run inside the window: the model's own notes replace the middle of the
+conversation, now and then, in one large step. Fitting is the last guard
+under it, and normally sends everything as it is. When it has to cut, the
+cut is recorded (``Trim``) and logged as ``context.trimmed``: after
+compaction that should be rare, and each one is worth a look.
+
+Fitting works on the events about to be rendered, compactions applied, in
+three steps, stopping at the first that fits:
 
 1. Tool results older than the most recent turns are cut to their first and
    last lines, with a note saying how much was left out.
@@ -39,6 +46,8 @@ from nisse_core.log.canonical import (
     UserMessage,
 )
 from nisse_core.models.llm import DEFAULT_MAX_OUTPUT_TOKENS, LLMError
+from nisse_core.models.render import effective_events
+from nisse_core.oplog import OpLog
 
 # English and code run at about four characters a token. Three is deliberately
 # pessimistic, before any correction from what the server reports.
@@ -129,6 +138,18 @@ class Budget:
         return int(self.window * (1 - self.safety_margin)) - self.max_output
 
 
+@dataclass(frozen=True)
+class Trim:
+    """What fitting cut from one request."""
+
+    step: int  # 1: older results shortened; 2: the oldest events dropped too
+    tokens_before: int
+    tokens_after: int
+    limit: int
+    results_cut: int
+    events_dropped: int
+
+
 def fit(
     events: Sequence[StoredEvent],
     system: str | None,
@@ -138,14 +159,29 @@ def fit(
 ) -> list[StoredEvent]:
     """The events to render: all of them if they fit, otherwise the result of
     the three steps in the module docstring."""
+    return fit_report(events, system, tools, budget, factor)[0]
+
+
+def fit_report(
+    events: Sequence[StoredEvent],
+    system: str | None,
+    tools: list[dict[str, Any]] | None,
+    budget: Budget,
+    factor: float = 1.0,
+) -> tuple[list[StoredEvent], Trim | None]:
+    """``fit``, and what it cut: None when nothing was."""
     limit = budget.for_prompt
 
+    def tokens(candidate: Sequence[StoredEvent]) -> int:
+        return estimate_tokens(candidate, system, tools, factor)
+
     def fits(candidate: Sequence[StoredEvent]) -> bool:
-        return estimate_tokens(candidate, system, tools, factor) <= limit
+        return tokens(candidate) <= limit
 
     kept = list(events)
-    if fits(kept):
-        return kept
+    before = tokens(kept)
+    if before <= limit:
+        return kept, None
 
     starts = _turn_starts(kept)
     recent_from = (
@@ -155,6 +191,13 @@ def fit(
     )
 
     # Step 1: cut older tool results down.
+    results_cut = sum(
+        1
+        for index, stored in enumerate(kept)
+        if index < recent_from
+        and isinstance(stored.event, ToolResult)
+        and len(stored.event.result) > ELIDED_RESULT_CHARS
+    )
     kept = [
         replace(
             stored,
@@ -168,7 +211,7 @@ def fit(
         for index, stored in enumerate(kept)
     ]
     if fits(kept):
-        return kept
+        return kept, Trim(1, before, tokens(kept), limit, results_cut, 0)
 
     # Step 2: drop the oldest events, one turn at a time, never the recent
     # ones. The note keeps the transcript starting with a user message, which
@@ -189,7 +232,10 @@ def fit(
         )
         candidate = [note, *remaining]
         if fits(candidate):
-            return candidate
+            dropped = len(kept) - len(remaining)
+            return candidate, Trim(
+                2, before, tokens(candidate), limit, results_cut, dropped
+            )
 
     # Step 3: nothing more can go.
     needed = estimate_tokens(kept[recent_from:], system, tools, factor)
@@ -216,8 +262,10 @@ class BudgetedClient:
         keep_recent_turns: int = 6,
         safety_margin: float = 0.1,
         enforces_tool_choice: Callable[[str], bool] = lambda model: False,
+        oplog: OpLog | None = None,
     ) -> None:
         self._inner = inner
+        self._oplog = oplog
         self._limits_for = limits_for
         # Whether a model's server holds it to the offered tools when asked
         # (LiteLLM's supports_tool_choice). Asking one that doesn't can be
@@ -231,6 +279,7 @@ class BudgetedClient:
         # from reported usage. Starts at 1 and only grows.
         self._factors: dict[str, float] = {}
         self.trimmed_requests = 0
+        self.last_trim: Trim | None = None  # the last request's, if it was cut
 
     @property
     def last_usage(self) -> dict[str, int]:
@@ -260,20 +309,57 @@ class BudgetedClient:
             safety_margin=self._safety_margin,
         )
 
+    def estimate(
+        self,
+        events: Sequence[StoredEvent],
+        system: str | None,
+        tools: list[dict[str, Any]] | None,
+        model: str,
+    ) -> int:
+        """A request's size in tokens, compactions applied, as fitting
+        would count it."""
+        factor = self._factors.get(model, 1.0)
+        return estimate_tokens(effective_events(list(events)), system, tools, factor)
+
     def complete(
         self,
         events: list[StoredEvent],
         tools: list[dict[str, Any]] | None,
         model: str,
         system: str | None = None,
+        *,
+        require_tool_call: bool | None = None,
     ) -> list[CanonicalEvent]:
+        """``require_tool_call``: None asks for a tool call whenever tools
+        are offered (and the model's server enforces it); False lets the
+        model answer in prose, as compaction's notes do."""
         budget = self.budget_for(model)
         factor = self._factors.get(model, 1.0)
-        fitted = fit(events, system, tools, budget, factor)
-        # Counted so the benchmark can report how often a model's window was
-        # the constraint: any request that had to be cut down or shortened.
-        if fitted != list(events):
+        events = effective_events(list(events))
+        fitted, trim = fit_report(events, system, tools, budget, factor)
+        self.last_trim = trim
+        if trim is not None:
+            # Counted so the benchmark can report how often a model's window
+            # was the constraint, and logged: each cut lost the model
+            # something it had read.
             self.trimmed_requests += 1
+            if self._oplog is not None:
+                self._oplog.emit(
+                    "context.trimmed",
+                    cause=(
+                        "older tool results shortened"
+                        if trim.step == 1
+                        else "the oldest events dropped"
+                    ),
+                    model=model,
+                    tokens_before=trim.tokens_before,
+                    tokens_after=trim.tokens_after,
+                    limit=trim.limit,
+                    results_cut=trim.results_cut,
+                    events_dropped=trim.events_dropped,
+                )
+        if require_tool_call is None:
+            require_tool_call = bool(tools) and self._enforces_tool_choice(model)
         produced = self._inner.complete(
             fitted,
             tools,
@@ -283,7 +369,7 @@ class BudgetedClient:
             # Every caller that offers tools needs one called: the executor
             # finishes with finish, the planner with submit_plan or
             # ask_human, a delegation with hand_back.
-            require_tool_call=bool(tools) and self._enforces_tool_choice(model),
+            require_tool_call=require_tool_call,
         )
         reported = int(self.last_usage.get("input", 0))
         if reported > 0:

nisse-core/src/nisse_core/models/interruptible.py

@@ -35,10 +35,11 @@ class Interruptible:
         tools: list[dict[str, Any]] | None,
         model: str,
         system: str | None = None,
+        **options: Any,
     ) -> list[CanonicalEvent]:
         self._calling = True
         try:
-            return self._client.complete(events, tools, model, system)
+            return self._client.complete(events, tools, model, system, **options)
         finally:
             self._calling = False
 

nisse-core/src/nisse_core/models/render.py

@@ -4,8 +4,7 @@ call time, never stored provider-shaped.
 First adapter: the OpenAI-compatible chat format LiteLLM's normalized route
 speaks — the executor role's path (L9). The Anthropic passthrough adapter
 arrives with delegations (L9). Token budgeting and prompt-cache breakpoints
-are render-time concerns and are never stored (L7); neither exists yet in
-this skeleton.
+are render-time concerns and are never stored (L7).
 """
 
 import json
@@ -27,20 +26,54 @@ from nisse_core.log.canonical import (
 
 
 def effective_events(stored: list[StoredEvent]) -> list[StoredEvent]:
-    """Apply compactions: events whose ids fall in a compaction's replaced
-    range are dropped; the compaction event itself stays as their summary."""
-    replaced: list[tuple[int, int]] = [
-        (e.event.replaces[0], e.event.replaces[1])
-        for e in stored
-        if isinstance(e.event, Compaction)
+    """Apply compactions: the events in a compaction's replaced range give way
+    to the compaction, which takes the place of the first of them, so the
+    summary sits where the conversation it summarises was. A compaction
+    whose range a later one covers is folded into it and not shown. One
+    whose range is no longer in the list (it was applied already) stays
+    where it is, so applying this twice changes nothing."""
+    compactions = [e for e in stored if isinstance(e.event, Compaction)]
+    active = [
+        c
+        for c in compactions
+        if not any(
+            later.id > c.id
+            and later.event.replaces[0] <= c.event.replaces[0]
+            and later.event.replaces[1] >= c.event.replaces[1]
+            for later in compactions
+        )
     ]
 
-    def is_replaced(event_id: int) -> bool:
-        return any(first <= event_id <= last for first, last in replaced)
+    active_ids = {c.id for c in active}
 
-    return [
-        e for e in stored if isinstance(e.event, Compaction) or not is_replaced(e.id)
-    ]
+    def covering(event_id: int) -> StoredEvent | None:
+        for c in active:
+            first, last = c.event.replaces
+            if first <= event_id <= last:
+                return c
+        return None
+
+    present = {e.id for e in stored if not isinstance(e.event, Compaction)}
+    placed = {
+        c.id
+        for c in active
+        if any(c.event.replaces[0] <= i <= c.event.replaces[1] for i in present)
+    }
+    out: list[StoredEvent] = []
+    shown: set[int] = set()
+    for e in stored:
+        if isinstance(e.event, Compaction):
+            if e.id in active_ids and e.id not in placed and e.id not in shown:
+                out.append(e)
+                shown.add(e.id)
+            continue
+        c = covering(e.id)
+        if c is None:
+            out.append(e)
+        elif c.id not in shown:
+            out.append(c)
+            shown.add(c.id)
+    return out
 
 
 class OpenAIChatRenderer:

nisse-core/src/nisse_core/state/store.py

@@ -20,6 +20,7 @@ from typing import Any, Self
 
 from nisse_core.log.canonical import (
     CanonicalEvent,
+    Compaction,
     StoredEvent,
     UserMessage,
     event_type,
@@ -573,9 +574,15 @@ class Transcript:
     def view(self, after: int = 0) -> list[StoredEvent]:
         """The log as this actor is shown it: what a model's context is
         built from (L6). See ``views``."""
-        return project(
-            self.read(max(after, self._start)), self.actor, self._store.SHOWN
-        )
+        start = max(after, self._start)
+        # A compaction of events before the start belongs to an earlier
+        # run's view, not this one's.
+        events = [
+            e
+            for e in self.read(start)
+            if not (isinstance(e.event, Compaction) and e.event.replaces[0] <= start)
+        ]
+        return project(events, self.actor, self._store.SHOWN)
 
     def since(self, event_id: int) -> "Transcript":
         """The same transcript, its view starting after ``event_id``: a run

nisse-core/tests/test_compact.py

@@ -0,0 +1,310 @@
+"""Compaction (L6, L7): a long driver run's middle gives way to the model's
+own notes, in one step now and then, so the run stays inside the window and
+the start of its prompt stays the same between compactions."""
+
+import io
+import json
+from itertools import pairwise
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from nisse_core.log.canonical import (
+    AssistantText,
+    CanonicalEvent,
+    Compaction,
+    Fact,
+    StoredEvent,
+    ToolCall,
+    ToolResult,
+    UserMessage,
+)
+from nisse_core.loop.driver import Driver, ToolOutcome
+from nisse_core.loop.tools import tool
+from nisse_core.models.compact import CompactionTuning, Compactor
+from nisse_core.models.context import BudgetedClient
+from nisse_core.models.interruptible import Interruptible, Stopped
+from nisse_core.oplog import OpLog
+from nisse_core.state.store import StateStore
+
+READ = tool("read_file", "Read a file.", path="Its path.")
+POST = tool("post_review", "Post the review. Ends your turn.", body="The review.")
+TOOLS = [READ, POST]
+BRIEF = "BRIEF: review PR #1"
+FILE_CHARS = 6000  # about 2000 tokens a read
+
+
+class Model:
+    """Reads one file a turn, then posts; writes notes when asked for them."""
+
+    def __init__(self, reads: int, notes: Any = "NOTES: all fine so far") -> None:
+        self.reads = reads
+        self.notes = notes
+        self.requests: list[dict[str, Any]] = []
+        self.turn = 0
+        self.on_notes: Any = None
+
+    def complete(
+        self,
+        events: list[StoredEvent],
+        tools: Any,
+        model: str,
+        system: str | None = None,
+        *,
+        max_tokens: int | None = None,
+        require_tool_call: bool = False,
+    ) -> list[CanonicalEvent]:
+        asked = _asks_for_notes(events)
+        self.requests.append(
+            {
+                "events": list(events),
+                "tools": tools,
+                "system": system,
+                "require_tool_call": require_tool_call,
+                "notes": asked,
+            }
+        )
+        if asked:
+            if self.on_notes is not None:
+                self.on_notes()
+            notes = self.notes(events) if callable(self.notes) else self.notes
+            return [AssistantText(text=notes, model=model)]
+        self.turn += 1
+        if self.turn > self.reads:
+            return [ToolCall(call_id="c-post", tool="post_review", arguments={})]
+        return [
+            AssistantText(text=f"reading f{self.turn}", model=model),
+            ToolCall(
+                call_id=f"c-{self.turn}",
+                tool="read_file",
+                arguments={"path": f"f{self.turn}.py"},
+            ),
+        ]
+
+    def turns_sent(self) -> list[list[StoredEvent]]:
+        return [r["events"] for r in self.requests if not r["notes"]]
+
+
+def _asks_for_notes(events: list[StoredEvent]) -> bool:
+    last = events[-1].event if events else None
+    return isinstance(last, UserMessage) and last.source == "compaction"
+
+
+class Files:
+    def dispatch(self, name: str, arguments: dict[str, Any]) -> ToolOutcome:
+        path = arguments["path"]
+        return ToolOutcome(f"{path}\n" + "x" * FILE_CHARS, True)
+
+
+class World:
+    def __init__(
+        self,
+        tmp_path: Path,
+        model: Model,
+        *,
+        tuning: CompactionTuning | None = None,
+        interruptible: bool = False,
+    ) -> None:
+        self.stream = io.StringIO()
+        oplog = OpLog("t", self.stream)
+        store = StateStore(tmp_path / "s.db", oplog=OpLog("t", io.StringIO()))
+        self.log = store.transcript("x", "driver")
+        self.log.as_actor("orchestrator").append(Fact(text=BRIEF))
+        self.model = model
+        # A 20k window less the margin and output ceiling: 17,000 tokens for
+        # the prompt, so compaction starts at about 10,200.
+        budgeted = BudgetedClient(
+            model,
+            lambda m: (20_000, None),
+            max_output_tokens=1000,
+            oplog=oplog,
+            enforces_tool_choice=lambda m: True,
+        )
+        self.client: Any = Interruptible(budgeted) if interruptible else budgeted
+        self.driver = Driver(
+            self.client,
+            "m",
+            max_turns=60,
+            compactor=Compactor(self.client, tuning, oplog=oplog),
+        )
+
+    def run(self) -> Any:
+        return self.driver.run(
+            self.log,
+            system="you review",
+            tools=TOOLS,
+            waiting={"post_review"},
+            workspace=Files(),
+        )
+
+    def oplog(self, event: str) -> list[dict[str, Any]]:
+        records = [json.loads(line) for line in self.stream.getvalue().splitlines()]
+        return [r for r in records if r["event"] == event]
+
+    def compactions(self) -> list[Compaction]:
+        return [e.event for e in self.log.read() if isinstance(e.event, Compaction)]
+
+
+def test_a_long_run_is_compacted_keeping_its_brief_and_recent_turns(
+    tmp_path: Path,
+) -> None:
+    world = World(tmp_path, Model(reads=7))
+    assert world.run().kind == "wait"
+
+    [compaction] = world.compactions()
+    [logged] = world.oplog("context.compacted")
+    assert logged["tokens_after"] < logged["tokens_before"]
+    assert logged["work_item"] == "x"
+    assert world.oplog("context.trimmed") == []  # fitting never had to cut
+
+    last = world.model.turns_sent()[-1]
+    # The brief first, word for word; then the notes; then the recent turns.
+    assert last[0].event.text == BRIEF
+    assert isinstance(last[1].event, Compaction)
+    assert last[1].event.summary.startswith("NOTES: all fine so far")
+    assert any(
+        isinstance(e.event, ToolResult) and e.event.result.startswith("f7.py")
+        for e in last
+    )
+    assert not any(
+        isinstance(e.event, ToolResult) and e.event.result.startswith("f1.py")
+        for e in last
+    )
+    # The turn just before the compaction is still there, word for word.
+    sent = world.model.turns_sent()
+    after = next(
+        i for i, r in enumerate(sent) if any(isinstance(e.event, Compaction) for e in r)
+    )
+    latest = [e for e in sent[after - 1] if isinstance(e.event, ToolResult)][-1]
+    assert latest in sent[after]
+    # The calls it replaced are listed by code, so the model can make them again.
+    assert '- read_file {"path": "f1.py"}' in compaction.summary
+
+
+def test_the_notes_are_asked_for_over_the_runs_own_request(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=8))
+    world.run()
+    requests = world.model.requests
+    asked = next(i for i, r in enumerate(requests) if r["notes"])
+    before, notes = requests[asked - 1], requests[asked]
+    # Same system prompt and tools, and the last request plus the model's
+    # reply and its result before it: only the request for notes is new,
+    # so the server reuses its cache.
+    assert notes["system"] == before["system"]
+    assert notes["tools"] == before["tools"]
+    assert before["require_tool_call"] is True
+    assert notes["require_tool_call"] is False
+    ids = [e.id for e in notes["events"][:-1]]
+    assert ids[: len(before["events"])] == [e.id for e in before["events"]]
+
+
+def test_the_prompt_start_stays_the_same_between_compactions(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=20))
+    world.run()
+    assert len(world.compactions()) >= 2
+    sent = world.model.turns_sent()
+    unchanged = 0
+    for earlier, later in pairwise(sent):
+        compacted = any(isinstance(e.event, Compaction) for e in later) and not any(
+            isinstance(e.event, Compaction) and e.event == c.event
+            for e in later
+            for c in earlier
+            if isinstance(c.event, Compaction)
+        )
+        if compacted:
+            continue
+        # Between compactions, each request extends the one before it.
+        assert [e.id for e in later[: len(earlier)]] == [e.id for e in earlier]
+        unchanged += 1
+    assert unchanged > len(sent) // 2
+
+
+def test_a_second_compaction_folds_in_the_first(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=20))
+    world.run()
+    first, second, *_ = world.compactions()
+    assert second.replaces[0] == first.replaces[0]
+    assert second.replaces[1] > first.replaces[1]
+    # The first one's list of calls is carried into the second.
+    assert '- read_file {"path": "f1.py"}' in second.summary
+    last = world.model.turns_sent()[-1]
+    assert sum(isinstance(e.event, Compaction) for e in last) == 1
+    assert last[0].event.text == BRIEF
+
+
+def test_tool_calls_are_never_parted_from_their_results(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=20))
+    world.run()
+    for sent in world.model.turns_sent():
+        calls = {e.event.call_id for e in sent if isinstance(e.event, ToolCall)}
+        answered = {e.event.call_id for e in sent if isinstance(e.event, ToolResult)}
+        assert answered <= calls
+
+
+def test_a_short_run_is_never_compacted(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=2))
+    world.run()
+    assert world.compactions() == []
+    assert not any(r["notes"] for r in world.model.requests)
+
+
+def test_empty_notes_fail_without_failing_the_run(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=8, notes=""))
+    assert world.run().kind == "wait"
+    assert world.compactions() == []
+    failed = world.oplog("context.compaction_failed")
+    assert failed and failed[0]["cause"] == "the notes were empty"
+    # Not tried again every turn: only after keep_turns more.
+    asked = [i for i, r in enumerate(world.model.requests) if r["notes"]]
+    assert all(b - a > 3 for a, b in pairwise(asked))
+
+
+def test_with_no_compaction_fitting_cuts_and_says_so(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=12, notes=""))
+    world.run()
+    [first, *_] = world.oplog("context.trimmed")
+    assert first["results_cut"] > 0
+    assert first["tokens_after"] <= first["limit"] < first["tokens_before"]
+
+
+def test_notes_that_leave_it_too_long_are_not_kept(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=8, notes="n" * 40_000))
+    world.run()
+    assert world.compactions() == []
+    [failed, *_] = world.oplog("context.compaction_failed")
+    assert "would still be" in failed["cause"]
+
+
+def test_a_stop_while_writing_notes_breaks_off_cleanly(tmp_path: Path) -> None:
+    model = Model(reads=8)
+    world = World(tmp_path, model, interruptible=True)
+    model.on_notes = world.client.interrupt
+    with pytest.raises(Stopped):
+        world.run()
+    assert world.compactions() == []
+
+
+def test_compact_at_zero_turns_it_off(tmp_path: Path) -> None:
+    world = World(tmp_path, Model(reads=10), tuning=CompactionTuning(compact_at=0))
+    world.run()
+    assert world.compactions() == []
+    assert world.oplog("context.trimmed")
+
+
+def test_a_view_from_a_start_hides_compactions_of_what_came_before(
+    tmp_path: Path,
+) -> None:
+    world = World(tmp_path, Model(reads=0))
+    log = world.log
+    log.as_actor("orchestrator").append(Fact(text="old work"))
+    start = log.read()[-1].id
+    log.as_actor("orchestrator").append(Fact(text="BRIEF again"))
+    log.append(Compaction(summary="stale", model="m", replaces=[1, start]))
+    log.append(
+        Compaction(summary="this run's", model="m", replaces=[start + 1, start + 1])
+    )
+    shown = log.since(start).view()
+    assert [e.event.summary for e in shown if isinstance(e.event, Compaction)] == [
+        "this run's"
+    ]

nisse-core/tests/test_context.py

@@ -9,6 +9,7 @@ import pytest
 from nisse_core.log.canonical import (
     AssistantText,
     CanonicalEvent,
+    Compaction,
     StoredEvent,
     ToolCall,
     ToolResult,
@@ -189,6 +190,46 @@ def test_reported_usage_only_ever_raises_the_estimate() -> None:
     assert under_then_over._factors["m"] == 2.0  # never loosened
 
 
+def test_a_request_that_fits_is_sent_whole_and_nothing_is_logged() -> None:
+    stream = io.StringIO()
+    client = BudgetedClient(
+        RecordingClient(), lambda m: (100_000, None), oplog=OpLog("t", stream)
+    )
+    events = transcript(3, 100)
+    client.complete(events, None, "m")
+    assert client.last_trim is None
+    assert stream.getvalue() == ""
+
+
+def test_a_cut_is_recorded_and_logged() -> None:
+    stream = io.StringIO()
+    client = BudgetedClient(
+        RecordingClient(),
+        lambda m: (4000, 500),
+        keep_recent_turns=2,
+        oplog=OpLog("t", stream),
+    )
+    client.complete(transcript(20, 3000), None, "m", system="system")
+    assert client.last_trim is not None and client.last_trim.step == 2
+    assert '"event": "context.trimmed"' in stream.getvalue()
+
+
+def test_a_compaction_counts_for_what_it_replaced() -> None:
+    events = transcript(20, 3000)
+    last = events[-1].id
+    compaction = StoredEvent(
+        id=last + 1,
+        ts="t",
+        event=Compaction(summary="notes", model="m", replaces=[2, last - 4]),
+    )
+    inner = RecordingClient()
+    client = BudgetedClient(inner, lambda m: (8000, 500), keep_recent_turns=2)
+    client.complete([*events, compaction], None, "m")
+    assert client.last_trim is None  # fits once the summary stands in
+    sent = inner.calls[0]["events"]
+    assert isinstance(sent[1].event, Compaction)
+
+
 def test_last_usage_passes_through() -> None:
     client = BudgetedClient(RecordingClient(reported_input=7), lambda m: (100, None))
     assert client.last_usage["input"] == 7

nisse-core/tests/test_render.py

@@ -66,11 +66,25 @@ def test_compaction_replaces_its_range_only() -> None:
         )
     )
     remaining = effective_events(events)
-    assert [e.id for e in remaining] == [3, 4]
+    # The summary takes the place of what it replaces.
+    assert [e.id for e in remaining] == [4, 3]
+    assert effective_events(remaining) == remaining  # applying again: no change
     body = OpenAIChatRenderer().render(events, tools=None, model="q")
     contents = [m["content"] for m in body["messages"]]
     # Both render as user messages, so they share one: roles must alternate.
-    assert contents == ["kept\n\n[summary of earlier conversation]\nthe old stuff"]
+    assert contents == ["[summary of earlier conversation]\nthe old stuff\n\nkept"]
+
+
+def test_a_later_compaction_folds_in_an_earlier_one() -> None:
+    events = stored([UserMessage(text=f"e{i}", source="m") for i in range(1, 7)])
+    events.append(
+        StoredEvent(id=7, ts="t", event=Compaction("first", "q", replaces=[2, 3]))
+    )
+    events.append(
+        StoredEvent(id=8, ts="t", event=Compaction("second", "q", replaces=[2, 5]))
+    )
+    remaining = effective_events(events)
+    assert [e.id for e in remaining] == [1, 8, 6]
 
 
 def test_tool_call_without_a_result_gets_a_placeholder() -> None: