diff --git a/docs/Architecture.md b/docs/Architecture.md index a87329c..00cbd14 100644 --- a/docs/Architecture.md +++ b/docs/Architecture.md @@ -332,9 +332,15 @@ logs the outage; interrupted outbound effects are covered by intents (L8). ### L6. Context and memory **1. The working context of one model call** is rendered from the log at call time -for one role, fitted to that model's window, and never stored. Fitting never drops -the system prompt, the current plan or the most recent turns; a call that still -doesn't fit fails (L9). +for one role, fitted to that model's window, and never stored. +- **A long run is compacted:** well before the window fills, the model's own + notes replace the middle of its run in one step (a `compaction` event, L7), + keeping word for word what started the work and its last few turns. Rare, + large steps keep the start of the prompt unchanged between them, so the + model server reuses its cache. +- **Fitting is the last guard:** it never drops the system prompt, the current + plan or the most recent turns, and a call that still doesn't fit fails (L9). + Each cut it makes is logged; after compaction one is a sign of trouble. **2. The work-item transcript** (the log, L7) is durable and scoped to one item: nothing crosses items, and agent-level calls (switching, duty runs) see other items diff --git a/docs/Development-Plan.md b/docs/Development-Plan.md index b4af0b1..a63f20b 100644 --- a/docs/Development-Plan.md +++ b/docs/Development-Plan.md @@ -713,6 +713,10 @@ of earlier points, replies on threads, the round cap and its reset, `stop` and environment only. A stop ends a git command, and a cut-short first clone is removed. `review.clone` / `review.fetch` log the time and size. No size limits yet. Tests: `test_clones.py`, against local bare repositories. +- **Compaction** (#158; reviews of PRs touching about 20 files hit the turn cap, + re-reading files fitting had cut): shared by both agents' driver runs, as in + L6 above. Each cut fitting still makes is logged (`context.trimmed`). + Tests: `test_compact.py` in nisse-core. **M7.6 Deployment.** *Waits for the agent host (Phase 6).* Its image, container, Gitea account and secrets (R5). The image needs `git`, and `scratch/` a @@ -771,6 +775,15 @@ if the implementation differs. proposal comment older than the reply. - **L5 transport:** the `Transport` interface is send, ask and receive. Intake backoff is exponential, capped at a few minutes. +- **L6 compaction (#158):** before each driver call, over `compact_at` (0.6) of + the prompt budget, the same model is asked, over the same request plus one + message, for notes on everything between what started the work and the last + turns (at most `compact_keep_turns`, 3, in `compact_to`, 0.25). The notes, + with code's list of the calls they replace, become a `compaction` event + that renders in place of its range; a later one folds in the earlier. + Failed notes are logged (`context.compaction_failed`) and fitting takes + over; `context.compacted` logs each one. Not yet in nisse-dev's helper + loop. - **L6 fitting order:** first shorten old tool results to their first and last lines, then drop the oldest events with a note saying so; never the system prompt, the current plan or the most recent turns. View labels: diff --git a/docs/Tuning-Guide.md b/docs/Tuning-Guide.md index fdb25dc..1e11d70 100644 --- a/docs/Tuning-Guide.md +++ b/docs/Tuning-Guide.md @@ -283,9 +283,39 @@ what failed last; the start keeps the exit code. - **Too high:** the transcript outgrows the context window sooner. The context budget copes (see below), but by cutting older output harder. -### `tuning.context` — fitting each request into the window +### `tuning.context` — compaction, and fitting each request into the window -Before any request is sent, the agent checks it against the model's context +**Compaction comes first.** A driver run's request grows with every tool +result. Once it passes `compact_at` of the prompt budget, the model is asked +for notes on its run so far: findings with the code they rest on, what each +file told it, what is left to do. The notes replace everything between what +started the work (the brief, or the conversation) and the last few turns, +which stay word for word. The model server has the request cached already, +so the notes cost only their own output. Because it happens rarely and drops +the request well below the limit, the start of the prompt stays the same for +many calls, and the server keeps reusing its cache. + +Logged as `context.compacted` (tokens before and after, seconds). Notes that +fail, come back empty or leave the request still too big are logged as +`context.compaction_failed`; the run goes on and fitting (below) cuts +instead, and compaction isn't tried again for `compact_keep_turns` turns. + +| Knob | Default | | +|---|---|---| +| `compact_at` | 0.6 | fraction of the prompt budget that sets compaction off; 0 turns it off | +| `compact_to` | 0.25 | what the kept recent turns may take, at most | +| `compact_keep_turns` | 3 | recent turns kept word for word, at most | + +- **`compact_at` too high:** little room left for the notes call and a big + result that lands in one turn, so fitting cuts more often. **Too low:** + compactions come often, and each one loses detail. +- **`compact_to` / `compact_keep_turns` too low:** the model loses sight of + what it just did. **Too high:** the notes don't bring the request far + enough down, and compactions come sooner. +- **Deciding:** look for `context.trimmed` in the log. After compaction it + should be rare; each one means the model lost something it had read. + +**Fitting is the last guard.** Before any request is sent, the agent checks it against the model's context window and makes it fit. A server doesn't always refuse a prompt that's too long. Ollama silently drops its start, which is where the system prompt and the plan are, so an unfitted request can fail in a way nobody sees. @@ -304,6 +334,9 @@ order, stopping at the first that fits: 3. **If the system prompt and the recent turns still don't fit,** the request is not sent. The milestone stops and asks you. +Each cut is logged as `context.trimmed`, with how many results were +shortened and events dropped. + Tokens are estimated from characters (pessimistically, at 3 per token) and the estimate is corrected from the prompt sizes the server reports. The correction only ever makes the estimate larger. diff --git a/nisse-core/src/nisse_core/loop/driver.py b/nisse-core/src/nisse_core/loop/driver.py index 8d913d1..da54c83 100644 --- a/nisse-core/src/nisse_core/loop/driver.py +++ b/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 diff --git a/nisse-core/src/nisse_core/models/compact.py b/nisse-core/src/nisse_core/models/compact.py new file mode 100644 index 0000000..c623d2f --- /dev/null +++ b/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:] diff --git a/nisse-core/src/nisse_core/models/context.py b/nisse-core/src/nisse_core/models/context.py index 8cd3d71..98935fb 100644 --- a/nisse-core/src/nisse_core/models/context.py +++ b/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: diff --git a/nisse-core/src/nisse_core/models/interruptible.py b/nisse-core/src/nisse_core/models/interruptible.py index 0fe94ea..60559bd 100644 --- a/nisse-core/src/nisse_core/models/interruptible.py +++ b/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 diff --git a/nisse-core/src/nisse_core/models/render.py b/nisse-core/src/nisse_core/models/render.py index b3d5b23..2a87238 100644 --- a/nisse-core/src/nisse_core/models/render.py +++ b/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: diff --git a/nisse-core/src/nisse_core/state/store.py b/nisse-core/src/nisse_core/state/store.py index e9b3a72..9867898 100644 --- a/nisse-core/src/nisse_core/state/store.py +++ b/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 diff --git a/nisse-core/tests/test_compact.py b/nisse-core/tests/test_compact.py new file mode 100644 index 0000000..4c5ee5a --- /dev/null +++ b/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" + ] diff --git a/nisse-core/tests/test_context.py b/nisse-core/tests/test_context.py index 5c731f2..cdb87cb 100644 --- a/nisse-core/tests/test_context.py +++ b/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 diff --git a/nisse-core/tests/test_render.py b/nisse-core/tests/test_render.py index 06f5280..0985117 100644 --- a/nisse-core/tests/test_render.py +++ b/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: diff --git a/nisse-dev/src/nisse_dev/bench/runner.py b/nisse-dev/src/nisse_dev/bench/runner.py index 933fd26..89f7144 100644 --- a/nisse-dev/src/nisse_dev/bench/runner.py +++ b/nisse-dev/src/nisse_dev/bench/runner.py @@ -22,6 +22,7 @@ from nisse_core.log.canonical import ( ) from nisse_core.loop.driver import Driver, RunOutcome from nisse_core.models.catalog import ModelFacts +from nisse_core.models.compact import Compactor from nisse_core.oplog import OpLog from nisse_dev.bench.tasks import Task from nisse_dev.config import Tuning @@ -78,10 +79,13 @@ class MeteringClient: tools: list[dict[str, Any]] | None, model: str, system: str | None = None, + **options: Any, ) -> list[CanonicalEvent]: self.calls += 1 try: - produced = self._inner.complete(events, tools, model, system=system) + produced = self._inner.complete( + events, tools, model, system=system, **options + ) finally: # Counted even when the call fails: a rate limit that never # cleared still cost the run its wait. @@ -95,10 +99,16 @@ class MeteringClient: answered = getattr(self._inner, "last_model", None) if not self._facts.answered(answered): self.fallbacks.append(str(answered)) - if not any(isinstance(e, ToolCall) for e in produced): + # Compaction's notes are prose by request, not a broken turn. + asked_for_prose = options.get("require_tool_call") is False + if not asked_for_prose and not any(isinstance(e, ToolCall) for e in produced): self.prose_turns += 1 return produced + def __getattr__(self, name: str) -> Any: + # The budget (budget_for, estimate) compaction reads. + return getattr(self._inner, name) + @dataclass class RunRecord: @@ -222,6 +232,8 @@ def run_task( model, max_turns=tuning.driver.max_turns, failure_limit=tuning.driver.failure_limit, + # As the agent runs it, so a long task measures the same thing. + compactor=Compactor(meter, tuning.context.compaction()), ) system = SYSTEM.format( repo=f"bench/{task.name}", diff --git a/nisse-dev/src/nisse_dev/config.py b/nisse-dev/src/nisse_dev/config.py index 5acebd4..d89af65 100644 --- a/nisse-dev/src/nisse_dev/config.py +++ b/nisse-dev/src/nisse_dev/config.py @@ -34,6 +34,7 @@ from nisse_core.config import ( resolve, ) from nisse_core.conversation.matrix import MatrixConfig +from nisse_core.models.compact import CompactionTuning, check_tuning from nisse_core.models.llm import ( DEFAULT_MAX_OUTPUT_TOKENS, DEFAULT_REQUEST_TIMEOUT_S, @@ -134,6 +135,19 @@ class ToolsTuning: class ContextTuning: keep_recent_turns: int = 6 safety_margin: float = 0.1 + # Compaction (L6): over compact_at of the prompt budget, the model's + # notes replace the middle of the run, keeping at most compact_keep_turns + # recent turns, in compact_to of it. compact_at 0: never. + compact_at: float = field(default=0.6, metadata={"zero": True}) + compact_to: float = 0.25 + compact_keep_turns: int = 3 + + def compaction(self) -> CompactionTuning: + return CompactionTuning( + compact_at=self.compact_at, + compact_to=self.compact_to, + keep_turns=self.compact_keep_turns, + ) @dataclass(frozen=True) @@ -182,6 +196,7 @@ def _parse_tuning(raw: Any, errors: list[str]) -> Tuning: ) if tuning.context.safety_margin >= 1: errors.append("tuning.context.safety_margin: must be less than 1") + check_tuning(tuning.context.compaction(), "tuning.context", errors) return tuning diff --git a/nisse-dev/src/nisse_dev/loop/build.py b/nisse-dev/src/nisse_dev/loop/build.py index f19901c..d9b1dd0 100644 --- a/nisse-dev/src/nisse_dev/loop/build.py +++ b/nisse-dev/src/nisse_dev/loop/build.py @@ -15,6 +15,7 @@ from nisse_core.knowledge.memory import Memory from nisse_core.knowledge.vault import Vault from nisse_core.loop.driver import Driver from nisse_core.models.catalog import ModelCatalog, select_client +from nisse_core.models.compact import Compactor from nisse_core.models.context import BudgetedClient from nisse_core.models.llm import AnthropicPassthroughClient, LiteLLMClient from nisse_core.oplog import OpLog @@ -110,7 +111,7 @@ class RunningLoop: def make_llm_clients( - config: AgentConfig, catalog: ModelCatalog + config: AgentConfig, catalog: ModelCatalog, oplog: OpLog | None = None ) -> tuple[BudgetedClient, BudgetedClient]: """The two LLM routes, as every model call in the agent uses them. @@ -136,6 +137,7 @@ def make_llm_clients( enforces_tool_choice=lambda m: ( catalog.facts(m).supports_tool_choice is True ), + oplog=oplog, # a cut that fitting has to make is logged ) normalized = budgeted( @@ -276,7 +278,7 @@ def build_loop_agent( project_runtime(config, entry, store, gitea, account) for entry in config.projects ] - normalized, passthrough = make_llm_clients(config, catalog) + normalized, passthrough = make_llm_clients(config, catalog, oplog) cancel = Cancel() # The configured driver; each work item runs on the one it pinned (L9). driver = Driver( @@ -285,6 +287,9 @@ def build_loop_agent( max_turns=config.tuning.driver.max_turns, cancel=cancel, failure_limit=config.tuning.driver.failure_limit, + compactor=Compactor( + normalized, config.tuning.context.compaction(), oplog=oplog + ), ) tuning = config.tuning helpers = Helpers( diff --git a/nisse-dev/tests/test_config.py b/nisse-dev/tests/test_config.py index e7c03a4..90e9f88 100644 --- a/nisse-dev/tests/test_config.py +++ b/nisse-dev/tests/test_config.py @@ -354,6 +354,18 @@ def test_context_tuning_is_read_and_checked() -> None: build(tuning={"context": {"safety_margin": 1}}) +def test_compaction_tuning_is_read_and_checked() -> None: + compaction = build().tuning.context.compaction() + assert (compaction.compact_at, compaction.compact_to) == (0.6, 0.25) + assert compaction.keep_turns == 3 + off = build(tuning={"context": {"compact_at": 0}}).tuning.context + assert off.compaction().compact_at == 0 + with pytest.raises(ConfigError, match="compact_to must be less than compact_at"): + build(tuning={"context": {"compact_at": 0.3, "compact_to": 0.5}}) + with pytest.raises(ConfigError, match="compact_to must be less than compact_at"): + build(tuning={"context": {"compact_at": 1.5}}) + + def test_the_build_host_is_required_and_defaults_to_root_on_22() -> None: host = only(build()).build_host assert (host.address, host.user) == ("build.test", "root") diff --git a/nisse-review/src/nisse_review/build.py b/nisse-review/src/nisse_review/build.py index 7112765..d158f8f 100644 --- a/nisse-review/src/nisse_review/build.py +++ b/nisse-review/src/nisse_review/build.py @@ -10,6 +10,7 @@ from nisse_core.knowledge.vault import Vault from nisse_core.loop.driver import Driver from nisse_core.loop.expert import ExpertDesk from nisse_core.models.catalog import ModelCatalog +from nisse_core.models.compact import Compactor from nisse_core.models.context import BudgetedClient from nisse_core.models.interruptible import Interruptible from nisse_core.models.llm import LiteLLMClient @@ -65,7 +66,7 @@ def check_access( return account, repos -def llm_client(config: ReviewConfig) -> BudgetedClient: +def llm_client(config: ReviewConfig, oplog: OpLog | None = None) -> BudgetedClient: """The one route to the models, every request fitted to its window (L6), against what LiteLLM says the configured models are (L9).""" catalog = ModelCatalog.discover(config.llm_base_url, config.llm_token) @@ -92,6 +93,7 @@ def llm_client(config: ReviewConfig) -> BudgetedClient: keep_recent_turns=tuning.keep_recent_turns, safety_margin=tuning.safety_margin, enforces_tool_choice=lambda m: catalog.facts(m).supports_tool_choice is True, + oplog=oplog, # a cut that fitting has to make is logged ) @@ -106,7 +108,7 @@ def build( gitea = gitea or GiteaClient(config.gitea_url, config.gitea_token) account, repos = check_access(gitea, config.repositories, oplog) # Every model call, the reviewer's and ask_expert's, can be broken off. - model = Interruptible(client if client is not None else llm_client(config)) # type: ignore[arg-type] + model = Interruptible(client if client is not None else llm_client(config, oplog)) # type: ignore[arg-type] config.state_db.parent.mkdir(parents=True, exist_ok=True) store = StateStore(config.state_db, oplog=oplog) vault = ( @@ -129,7 +131,12 @@ def build( reviewer = Reviewer( store, gitea, - Driver(model, config.roles.resolve("driver"), cancel=cancel), # type: ignore[arg-type] + Driver( + model, # type: ignore[arg-type] + config.roles.resolve("driver"), + cancel=cancel, + compactor=Compactor(model, tuning.context.compaction(), oplog=oplog), + ), account=account, gitea_url=config.gitea_url, oplog=oplog, diff --git a/nisse-review/src/nisse_review/config.py b/nisse-review/src/nisse_review/config.py index bbf8b4a..b081676 100644 --- a/nisse-review/src/nisse_review/config.py +++ b/nisse-review/src/nisse_review/config.py @@ -22,6 +22,7 @@ from nisse_core.config import ( read_file, resolve, ) +from nisse_core.models.compact import CompactionTuning, check_tuning from nisse_core.models.llm import ( DEFAULT_MAX_OUTPUT_TOKENS, DEFAULT_REQUEST_TIMEOUT_S, @@ -72,6 +73,19 @@ class ExpertTuning: class ContextTuning: keep_recent_turns: int = 6 safety_margin: float = 0.1 + # Compaction (L6): over compact_at of the prompt budget, the model's + # notes replace the middle of the run, keeping at most compact_keep_turns + # recent turns, in compact_to of it. compact_at 0: never. + compact_at: float = field(default=0.6, metadata={"zero": True}) + compact_to: float = 0.25 + compact_keep_turns: int = 3 + + def compaction(self) -> CompactionTuning: + return CompactionTuning( + compact_at=self.compact_at, + compact_to=self.compact_to, + keep_turns=self.compact_keep_turns, + ) @dataclass(frozen=True) @@ -118,6 +132,7 @@ class ReviewConfig: tuning = parse_tuning(tree.get("tuning"), Tuning, errors) if tuning.context.safety_margin >= 1: errors.append("tuning.context.safety_margin: must be less than 1") + check_tuning(tuning.context.compaction(), "tuning.context", errors) vault = parse_vault(tree.get("vault"), errors) flat = flatten({k: v for k, v in tree.items() if k != "vault"}) flat.pop("tuning", None) diff --git a/nisse-review/tests/test_review.py b/nisse-review/tests/test_review.py index 16fa28b..48dfa66 100644 --- a/nisse-review/tests/test_review.py +++ b/nisse-review/tests/test_review.py @@ -494,6 +494,30 @@ def test_the_ci_waits_may_be_zero(tmp_path: Path) -> None: assert (review.ci_wait_s, review.ci_grace_s) == (0, 0) +def test_compaction_is_tuned_in_the_config(tmp_path: Path) -> None: + from nisse_core.config import ConfigError + from nisse_review.config import ReviewConfig + + tree: dict[str, Any] = { + "state_db": str(tmp_path / "s.db"), + "llm": {"base_url": "http://llm", "token": "k"}, + "roles": {"reviewer": "local", "expert": "frontier"}, + "gitea": {"url": "https://g", "token": "t", "operators": ["don"]}, + } + assert ( + ReviewConfig.from_tree(tree, {}).tuning.context.compaction().compact_at == 0.6 + ) + tuned = { + **tree, + "tuning": {"context": {"compact_at": 0.7, "compact_keep_turns": 2}}, + } + compaction = ReviewConfig.from_tree(tuned, {}).tuning.context.compaction() + assert (compaction.compact_at, compaction.keep_turns) == (0.7, 2) + wrong = {**tree, "tuning": {"context": {"compact_at": 0.2, "compact_to": 0.3}}} + with pytest.raises(ConfigError, match="compact_to must be less than compact_at"): + ReviewConfig.from_tree(wrong, {}) + + def test_diff_lines_are_the_heads_lines_a_diff_shows() -> None: diff = ( "--- a/x.py\n+++ b/x.py\n@@ -1,3 +1,4 @@\n a\n-b\n+c\n+d\n e\n" diff --git a/nisse-review/tests/test_review_stop.py b/nisse-review/tests/test_review_stop.py index b688d39..64e0678 100644 --- a/nisse-review/tests/test_review_stop.py +++ b/nisse-review/tests/test_review_stop.py @@ -163,6 +163,8 @@ def test_build_wires_a_stop_to_every_model_call(tmp_path: Path) -> None: try: driver = built.scheduler._reviewer._driver assert driver._client is built.model + # Compaction's notes are a model call a stop breaks off too. + assert driver._compactor._client is built.model assert built.scheduler._reviewer._expert._client is built.model built.stop() # no call in flight: nothing raised assert driver._cancel.requested and built.scheduler._stop.is_set()