diff --git a/coworker/compaction.py b/coworker/compaction.py
new file mode 100644
index 00000000..ec0ec25a
--- /dev/null
+++ b/coworker/compaction.py
@@ -0,0 +1,528 @@
+"""Auto-compaction of long session histories (OPE-27).
+
+When the outbound history approaches the model's context limit, the older portion of the
+*outbound* view is replaced with (a) an LLM-written structured summary and (b) mechanically
+extracted state — the recent turns and all user messages survive. The persisted transcript
+is never modified; only what is sent to the model. Full design: ocw-context
+docs/auto-compaction-spec.md (approved 2026-07-28).
+
+This module is pure functions + one dataclass; the engine owns *when* (its run loop) and
+*with what* (its provider/model), both injected here. That split keeps the engine.py
+footprint to a few lines and makes every policy testable without a provider.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import dataclass, field
+from typing import Any, Optional
+
+# Trigger: min(threshold_pct × context_window, cap_tokens). The cap exists so 1M-context
+# models compact early — quality and latency degrade well before the nominal limit.
+DEFAULT_THRESHOLD_PCT = 0.8
+DEFAULT_CAP_TOKENS = 250_000
+# Models without a verified context_window entry in the matrix.
+DEFAULT_CONTEXT_WINDOW = 128_000
+# The newest slice kept verbatim, as a fraction of the trigger (a token budget, not a
+# turn count — one huge tool loop shouldn't starve the working set).
+KEEP_RECENT_FRACTION = 0.25
+# The summarizer call itself: tools off, modest ceiling.
+SUMMARY_MAX_TOKENS = 3_000
+# Per-message clip when rendering the span for the summarizer; tool results are the
+# first casualty (huge and mostly stale — a file read 40 turns ago is better re-read).
+_SPAN_TOOL_RESULT_CLIP = 400
+_SPAN_BUDGET_CHARS = 400_000
+# User messages preserved mechanically in the compacted block ("trimmed of pasted bulk").
+_USER_MESSAGE_CLIP = 600
+_TRIM_FRACTION = 0.10
+
+
+# -- token math ---------------------------------------------------------------
+
+
+def estimate_tokens(messages: list[dict[str, Any]]) -> int:
+ """chars/4 over the serialized messages — the fallback signal for providers that
+ never report usage (documented in the metering code)."""
+ total = 0
+ for msg in messages:
+ try:
+ total += len(json.dumps(msg, default=str))
+ except (TypeError, ValueError):
+ total += len(str(msg))
+ return total // 4
+
+
+def trigger_tokens(
+ context_window: Optional[int],
+ *,
+ threshold_pct: float = DEFAULT_THRESHOLD_PCT,
+ cap_tokens: int = DEFAULT_CAP_TOKENS,
+) -> int:
+ window = context_window or DEFAULT_CONTEXT_WINDOW
+ return min(int(threshold_pct * window), int(cap_tokens))
+
+
+def should_compact(
+ signal: int,
+ context_window: Optional[int],
+ *,
+ threshold_pct: float = DEFAULT_THRESHOLD_PCT,
+ cap_tokens: int = DEFAULT_CAP_TOKENS,
+) -> bool:
+ return signal >= trigger_tokens(
+ context_window, threshold_pct=threshold_pct, cap_tokens=cap_tokens
+ )
+
+
+# -- state --------------------------------------------------------------------
+
+
+@dataclass
+class CompactionState:
+ """One compaction point. `boundary_index` is an index into the CANONICAL message list:
+ messages before it are represented by the compacted block in the outbound view; messages
+ from it on are sent verbatim. Persisted with the session so reloads keep the view."""
+
+ boundary_index: int
+ summary_text: str
+ working_state: str
+ user_messages: list[str] = field(default_factory=list)
+ created_at: float = 0.0
+ model_used: str = ""
+ trimmed: bool = False # True when this state came from the no-summary trim fallback
+
+ def as_dict(self) -> dict[str, Any]:
+ return {
+ "boundary_index": self.boundary_index,
+ "summary_text": self.summary_text,
+ "working_state": self.working_state,
+ "user_messages": list(self.user_messages),
+ "created_at": self.created_at,
+ "model_used": self.model_used,
+ "trimmed": self.trimmed,
+ }
+
+ @classmethod
+ def from_dict(cls, raw: Any) -> Optional["CompactionState"]:
+ if not isinstance(raw, dict) or "boundary_index" not in raw:
+ return None
+ return cls(
+ boundary_index=int(raw.get("boundary_index", 0)),
+ summary_text=str(raw.get("summary_text", "")),
+ working_state=str(raw.get("working_state", "")),
+ user_messages=[str(u) for u in raw.get("user_messages") or []],
+ created_at=float(raw.get("created_at", 0.0)),
+ model_used=str(raw.get("model_used", "")),
+ trimmed=bool(raw.get("trimmed", False)),
+ )
+
+
+# -- boundary -----------------------------------------------------------------
+
+
+def _turn_starts(messages: list[dict[str, Any]], *, start: int) -> tuple[list[int], list[int]]:
+ """Candidate boundary indexes past `start`: user-message indexes (turn starts,
+ preferred) and assistant indexes (iteration starts — legal suffix heads; a `tool`
+ message must never head the outbound view)."""
+ users, assistants = [], []
+ for i in range(start, len(messages)):
+ role = messages[i].get("role")
+ if role == "user":
+ users.append(i)
+ elif role == "assistant":
+ assistants.append(i)
+ return users, assistants
+
+
+def pick_boundary(messages: list[dict[str, Any]], *, keep_tokens: int) -> Optional[int]:
+ """The canonical index where the verbatim tail begins: the earliest turn start whose
+ suffix fits the keep budget. Prefers user-message boundaries; falls back to iteration
+ (assistant) boundaries when the newest turn alone exceeds the budget (a giant tool
+ loop). None when there is nothing meaningful to summarize."""
+ start = 1 if messages and messages[0].get("role") == "system" else 0
+ users, assistants = _turn_starts(messages, start=start)
+
+ def _fit(candidates: list[int]) -> Optional[int]:
+ for i in candidates: # earliest-first: keep as much verbatim as fits
+ if estimate_tokens(messages[i:]) <= keep_tokens:
+ return i
+ return None
+
+ boundary = _fit(users)
+ if boundary is None and users:
+ # The newest user turn alone blows the budget — cut inside it at an iteration
+ # boundary, keeping at least the most recent assistant step.
+ inside = [i for i in assistants if i > users[-1]]
+ boundary = _fit(inside)
+ if boundary is None:
+ boundary = inside[-1] if inside else users[-1]
+ if boundary is None:
+ boundary = _fit(assistants) or (assistants[-1] if assistants else None)
+ # A boundary at (or before) the first real message summarizes nothing — skip.
+ if boundary is None or boundary <= start:
+ return None
+ return boundary
+
+
+# -- mechanical extraction (no LLM — zero hallucination risk) -----------------
+
+_WRITE_HINTS = ("write", "edit", "append", "save", "create", "patch")
+_ARTIFACT_HINTS = ("artifact", "publish", "deploy")
+
+
+def _iter_tool_calls(span: list[dict[str, Any]]):
+ """(name, args, result_content) for every tool call in the span, in order."""
+ results = {
+ m.get("tool_call_id"): m.get("content")
+ for m in span
+ if m.get("role") == "tool"
+ }
+ for msg in span:
+ if msg.get("role") != "assistant":
+ continue
+ for tc in msg.get("tool_calls") or []:
+ fn = tc.get("function") or {}
+ try:
+ args = json.loads(fn.get("arguments") or "{}")
+ except (ValueError, TypeError):
+ args = {}
+ yield str(fn.get("name") or ""), args, results.get(tc.get("id"))
+
+
+def _result_status(result: Any) -> str:
+ if not isinstance(result, str):
+ return ""
+ try:
+ parsed = json.loads(result)
+ except (ValueError, TypeError):
+ return ""
+ if not isinstance(parsed, dict):
+ return ""
+ if parsed.get("error"):
+ return "error"
+ if "exit_code" in parsed:
+ code = parsed.get("exit_code")
+ return "ok" if code in (0, "0") else f"exit {code}"
+ return ""
+
+
+def extract_working_state(span: list[dict[str, Any]]) -> str:
+ """The mechanical block appended to the summary by CODE, from the span's tool-call
+ records: files written, recent commands (+ exit status), artifacts, tools used."""
+ files: list[str] = []
+ commands: list[str] = []
+ artifacts: list[str] = []
+ tools: list[str] = []
+ for name, args, result in _iter_tool_calls(span):
+ if name and name not in tools:
+ tools.append(name)
+ lowered = name.lower()
+ path = args.get("path") or args.get("file_path")
+ if path and any(h in lowered for h in _WRITE_HINTS):
+ files.append(str(path))
+ if lowered == "run_shell" and args.get("command"):
+ status = _result_status(result)
+ line = " ".join(str(args["command"]).split())[:160]
+ commands.append(f"{line}" + (f" [{status}]" if status else ""))
+ if any(h in lowered for h in _ARTIFACT_HINTS):
+ location = args.get("url") or args.get("path") or args.get("title")
+ if location:
+ artifacts.append(str(location))
+
+ def _dedupe_recent_first(items: list[str], limit: int) -> list[str]:
+ seen: list[str] = []
+ for item in reversed(items): # most recent first
+ if item not in seen:
+ seen.append(item)
+ if len(seen) >= limit:
+ break
+ return seen
+
+ lines = ["## Working state (extracted mechanically from tool records)"]
+ written = _dedupe_recent_first(files, 20)
+ if written:
+ lines.append("Files written/edited (most recent first):")
+ lines += [f"- {p}" for p in written]
+ recent_cmds = commands[-10:]
+ if recent_cmds:
+ lines.append("Recent shell commands:")
+ lines += [f"- {c}" for c in recent_cmds]
+ made = _dedupe_recent_first(artifacts, 10)
+ if made:
+ lines.append("Artifacts produced:")
+ lines += [f"- {a}" for a in made]
+ if tools:
+ lines.append("Tools used in the summarized span: " + ", ".join(sorted(tools)))
+ return "\n".join(lines) if len(lines) > 1 else ""
+
+
+def _text_of(content: Any) -> str:
+ """A message's text, whether plain or content-parts (images become a placeholder)."""
+ if isinstance(content, str):
+ return content
+ if isinstance(content, list):
+ parts = []
+ for p in content:
+ if isinstance(p, dict) and p.get("type") == "text":
+ parts.append(str(p.get("text", "")))
+ elif isinstance(p, dict) and p.get("type") == "image_url":
+ parts.append("[image]")
+ return "\n".join(parts)
+ return "" if content is None else str(content)
+
+
+def extract_user_messages(
+ span: list[dict[str, Any]], *, clip: int = _USER_MESSAGE_CLIP
+) -> list[str]:
+ """Every user message in the span, chronological, trimmed of pasted bulk. Preserved
+ mechanically — the summarizer is also asked to list them, but user words are the
+ ground truth of intent and must not depend on an LLM remembering to include them."""
+ out: list[str] = []
+ for msg in span:
+ if msg.get("role") != "user":
+ continue
+ text = " ".join(_text_of(msg.get("content")).split())
+ if not text:
+ continue
+ out.append(text[: clip - 1] + "…" if len(text) > clip else text)
+ return out
+
+
+# -- summarizer ---------------------------------------------------------------
+
+SUMMARY_SYSTEM_PROMPT = """You are compacting an AI coworker's session history so the coworker can continue working in a smaller context. Write a structured summary of the conversation below. It is the coworker's ONLY memory of these turns, so preserve everything load-bearing.
+
+Produce ALL of the following sections, in this order, each as a markdown heading:
+
+1. **Primary request and intent** — what the user is trying to get done, in their terms, including standing constraints stated at any point (e.g. "never send without my approval"). Constraints outlive the turns they were stated in.
+2. **Key concepts and decisions** — domain facts, technical choices, and rationale established so far. Include the WHY, not just the what — a decision without its reason gets relitigated.
+3. **Artifacts and files** — every file/deliverable created, modified, or read that still matters: path, its role, and a short excerpt of load-bearing content only.
+4. **Errors and fixes** — problems hit and how they were resolved, including user corrections ("no, do it this way") — those are feedback with lasting force.
+5. **All user messages** — a chronological list of every user message (trimmed of pasted bulk). This is the intent audit-trail.
+6. **Pending tasks** — explicitly incomplete items, promised follow-ups, things the user said "later" about.
+7. **Current work** — precisely what was in progress at this point: which step, which file, what state.
+8. **Next step** — the immediate next action, justified by the user's request.
+
+Rules:
+- Do NOT carry full file contents as truth. Note THAT a file was read/edited; the coworker re-reads if it needs the content again. Stale memory of a file is worse than no memory.
+- Be concrete: paths, names, commands, ids — not vague references.
+- Output only the summary sections, no preamble."""
+
+CONTINUATION_CONTRACT = (
+ "Continue where you left off: pick up the current work and next step exactly as "
+ "described. Do not re-ask answered questions, do not recap, do not mention that the "
+ "context was compacted. If you need the contents of a file noted above, re-read it."
+)
+
+
+def _render_span(span: list[dict[str, Any]], *, budget_chars: int = _SPAN_BUDGET_CHARS) -> str:
+ """The summarized span as compact text for the summarizer. Tool results are clipped
+ hard (first casualty); if the whole render still exceeds the budget, oldest lines are
+ dropped — the newest context is the most load-bearing."""
+ lines: list[str] = []
+ for msg in span:
+ role = msg.get("role")
+ if role == "system":
+ continue
+ if role == "notice":
+ continue
+ if role == "tool":
+ text = _text_of(msg.get("content"))
+ text = " ".join(text.split())
+ if len(text) > _SPAN_TOOL_RESULT_CLIP:
+ text = text[: _SPAN_TOOL_RESULT_CLIP - 1] + "…"
+ lines.append(f"[tool result] {text}")
+ continue
+ text = _text_of(msg.get("content"))
+ if role == "assistant":
+ for tc in msg.get("tool_calls") or []:
+ fn = tc.get("function") or {}
+ args = " ".join(str(fn.get("arguments", "")).split())
+ if len(args) > 200:
+ args = args[:199] + "…"
+ lines.append(f"[assistant → {fn.get('name')}] {args}")
+ if text:
+ lines.append(f"[assistant] {text}")
+ elif role == "user":
+ lines.append(f"[user] {text}")
+ rendered = "\n".join(lines)
+ if len(rendered) > budget_chars:
+ rendered = "(…oldest turns elided…)\n" + rendered[-budget_chars:]
+ return rendered
+
+
+def summarizer_messages(
+ span: list[dict[str, Any]], *, prior_summary: str = ""
+) -> list[dict[str, Any]]:
+ """The provider-ready messages for the summarizer call. On repeated compaction the
+ previous summary is message zero of the new span — summarized along with the turns
+ since."""
+ body = _render_span(span)
+ if prior_summary:
+ body = (
+ "[previous compaction summary — fold its still-relevant content into the new "
+ "summary]\n" + prior_summary + "\n\n[conversation since]\n" + body
+ )
+ return [
+ {"role": "system", "content": SUMMARY_SYSTEM_PROMPT},
+ {"role": "user", "content": body},
+ ]
+
+
+def summarize_span(
+ provider: Any,
+ model: str,
+ span: list[dict[str, Any]],
+ *,
+ prior_summary: str = "",
+ max_tokens: int = SUMMARY_MAX_TOKENS,
+) -> str:
+ """One summarizer round-trip (blocking — the engine runs it off-loop). Tools are
+ disabled; the Settings model override is just a different `model` id. Raises on
+ provider failure or an empty summary — the caller owns the retry/trim policy."""
+ turn = provider.complete(
+ model=model,
+ messages=summarizer_messages(span, prior_summary=prior_summary),
+ tools=None,
+ max_tokens=max_tokens,
+ )
+ text = (getattr(turn, "text", None) or "").strip()
+ if not text:
+ raise RuntimeError("summarizer returned an empty summary")
+ return text
+
+
+# -- building + applying a compaction -----------------------------------------
+
+
+def build_state(
+ messages: list[dict[str, Any]],
+ *,
+ provider: Any,
+ model: str,
+ keep_tokens: int,
+ prior: Optional[CompactionState] = None,
+) -> Optional[CompactionState]:
+ """Summarize everything older than the picked boundary into a new CompactionState.
+ On repeated compaction the prior summary heads the new span. Returns None when there
+ is nothing to compact; raises when the summarizer fails (caller applies policy)."""
+ boundary = pick_boundary(messages, keep_tokens=keep_tokens)
+ if boundary is None or (prior is not None and boundary <= prior.boundary_index):
+ return None
+ span_start = prior.boundary_index if prior is not None else 0
+ span = messages[span_start:boundary]
+ prior_users = list(prior.user_messages) if prior is not None else []
+ summary = summarize_span(
+ provider,
+ model,
+ span,
+ prior_summary=prior.summary_text if prior is not None else "",
+ )
+ return CompactionState(
+ boundary_index=boundary,
+ summary_text=summary,
+ working_state=extract_working_state(span),
+ user_messages=prior_users + extract_user_messages(span),
+ created_at=time.time(),
+ model_used=model,
+ )
+
+
+def trim_state(
+ messages: list[dict[str, Any]],
+ *,
+ prior: Optional[CompactionState] = None,
+ fraction: float = _TRIM_FRACTION,
+) -> Optional[CompactionState]:
+ """The no-LLM fallback: advance the boundary past ~`fraction` of the outbound
+ messages. No summary — but the mechanical block and the user-message list (never
+ trimmed away, per spec) are free, so the model still gets deterministic state."""
+ start = prior.boundary_index if prior is not None else 0
+ remaining = len(messages) - start
+ if remaining <= 2:
+ return None
+ step = max(1, int(remaining * fraction))
+ target = start + step
+ # Land on a legal suffix head at or after the target (never a tool message).
+ boundary = None
+ for i in range(target, len(messages)):
+ if messages[i].get("role") in ("user", "assistant"):
+ boundary = i
+ break
+ if boundary is None or boundary <= start or boundary >= len(messages):
+ return None
+ span = messages[start:boundary]
+ prior_users = list(prior.user_messages) if prior is not None else []
+ summary = (
+ (prior.summary_text + "\n\n" if prior is not None and prior.summary_text else "")
+ + "(Older turns were trimmed to fit the context window; no summary is available "
+ "for them. Re-read files and re-run commands if earlier results are needed.)"
+ )
+ return CompactionState(
+ boundary_index=boundary,
+ summary_text=summary,
+ working_state=extract_working_state(span),
+ user_messages=prior_users + extract_user_messages(span),
+ created_at=time.time(),
+ model_used="",
+ trimmed=True,
+ )
+
+
+def compacted_block(state: CompactionState) -> str:
+ """The single outbound message standing in for everything before the boundary."""
+ parts = [
+ "",
+ "Earlier turns of this session were compacted. The summary below is your memory "
+ "of them.",
+ "",
+ state.summary_text,
+ ]
+ if state.working_state:
+ parts += ["", state.working_state]
+ if state.user_messages:
+ parts += ["", "## User messages in the compacted span (verbatim, chronological)"]
+ parts += [f"- {u}" for u in state.user_messages]
+ parts += ["", CONTINUATION_CONTRACT, ""]
+ return "\n".join(parts)
+
+
+def apply_to_outbound(
+ messages: list[dict[str, Any]], state: Optional[CompactionState]
+) -> list[dict[str, Any]]:
+ """The outbound view: [system?] + the compacted block (as a user message) + the
+ verbatim tail. Canonical history is untouched; provider-private sidecars in the
+ summarized span vanish with it (replay chains legally restart after a compaction
+ point). No-op when state is absent or stale."""
+ if state is None:
+ return messages
+ boundary = state.boundary_index
+ if boundary <= 0 or boundary >= len(messages):
+ return messages
+ head: list[dict[str, Any]] = []
+ if messages and messages[0].get("role") == "system":
+ head.append(messages[0])
+ head.append({"role": "user", "content": compacted_block(state)})
+ return head + messages[boundary:]
+
+
+# -- overflow detection -------------------------------------------------------
+
+_OVERFLOW_MARKERS = (
+ "context_length_exceeded",
+ "maximum context length",
+ "context window",
+ "prompt is too long",
+ "input is too long",
+ "too many tokens",
+ "input length and `max_tokens` exceed",
+ "exceeds the maximum number of tokens",
+)
+
+
+def is_context_overflow(exc: BaseException) -> bool:
+ """A raw context-overflow 400 from the main model (compaction mispredicted, e.g. the
+ estimate path) — routed into the compaction policy instead of surfacing."""
+ text = str(exc).lower()
+ return any(marker in text for marker in _OVERFLOW_MARKERS)
diff --git a/coworker/conversations.py b/coworker/conversations.py
index fd2131bf..e67ef2fb 100644
--- a/coworker/conversations.py
+++ b/coworker/conversations.py
@@ -95,6 +95,7 @@ class ConversationStore:
"ALTER TABLE sessions ADD COLUMN auto_title TEXT",
"ALTER TABLE sessions ADD COLUMN renamed INTEGER DEFAULT 0",
"ALTER TABLE sessions ADD COLUMN grants TEXT",
+ "ALTER TABLE sessions ADD COLUMN compaction TEXT",
):
try:
self._conn.execute(ddl)
@@ -191,13 +192,14 @@ class ConversationStore:
title = record.title or title_from(record.messages)
self._conn.execute(
"""
- INSERT INTO sessions (session_id, workspace, model, mode, title, agent, n_msgs, messages, extra_roots, grants, updated_at)
- VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, CURRENT_TIMESTAMP)
+ INSERT INTO sessions (session_id, workspace, model, mode, title, agent, n_msgs, messages, extra_roots, grants, compaction, updated_at)
+ VALUES (?, ?, ?, ?, ?, ?, ?, NULL, ?, ?, ?, CURRENT_TIMESTAMP)
ON CONFLICT(session_id) DO UPDATE SET
workspace = excluded.workspace, model = excluded.model, mode = excluded.mode,
title = COALESCE(sessions.title, excluded.title), agent = excluded.agent,
n_msgs = excluded.n_msgs, messages = NULL, extra_roots = excluded.extra_roots,
- grants = excluded.grants, updated_at = CURRENT_TIMESTAMP
+ grants = excluded.grants, compaction = excluded.compaction,
+ updated_at = CURRENT_TIMESTAMP
""",
(
sid,
@@ -209,6 +211,7 @@ class ConversationStore:
len(record.messages),
json.dumps(record.extra_roots or []),
json.dumps(record.grants or {}),
+ json.dumps(record.compaction or {}),
),
)
self._conn.commit()
@@ -241,6 +244,10 @@ class ConversationStore:
row["extra_roots"] if "extra_roots" in row.keys() else None
),
grants=_load_grants(row["grants"] if "grants" in row.keys() else None),
+ # Auto-compaction state (OPE-27) — same defensive parse as grants.
+ compaction=_load_grants(
+ row["compaction"] if "compaction" in row.keys() else None
+ ),
pinned=bool(row["pinned"]),
archived=bool(row["archived"]),
origin=row["origin"],
diff --git a/coworker/engine.py b/coworker/engine.py
index 3ce77d1e..f3379d61 100644
--- a/coworker/engine.py
+++ b/coworker/engine.py
@@ -19,6 +19,7 @@ from dataclasses import dataclass
from enum import Enum
from typing import Any, AsyncIterator, Awaitable, Callable, Optional
+from . import compaction as _compaction
from .events import Event, EventType
from .permissions import Mode, PermissionEngine
from .providers import AssistantTurn, ProviderClient, ToolCall
@@ -103,6 +104,14 @@ class TurnEngine:
# (answerable inline in a live session or from the Inbox when unattended). None on surfaces
# that can't ask (the tool then no-ops).
self.question_asker = question_asker
+ # Auto-compaction (OPE-27) — set post-construction by the surface/manager so the
+ # constructor footprint stays put. `compaction_settings` is a live getter (Settings
+ # changes apply without a rebuild); `is_attended` gates the failure prompt (None →
+ # treat as unattended: never park a background run on internal bookkeeping).
+ self.compaction_state: Optional[_compaction.CompactionState] = None
+ self.compaction_settings: Optional[Callable[[], dict[str, Any]]] = None
+ self.is_attended: Optional[Callable[[], bool]] = None
+ self._last_context_tokens: Optional[int] = None
self.audit_context: dict[str, Any] = {}
if instructions and not (
self.messages and self.messages[0].get("role") == "system"
@@ -302,6 +311,13 @@ class TurnEngine:
return
iterations += 1
+ # Auto-compaction checkpoint (OPE-27): between tool turns and before a new
+ # turn's first call. Deliberately no "wrap up" warning to the model.
+ notice = await self._compact_now()
+ if notice:
+ self._append_notice("compacted", notice)
+ yield Event(EventType.COMPACTED, {"text": notice})
+
turn: Optional[AssistantTurn] = None
streamed: list[str] = []
streamed_reasoning: list[str] = []
@@ -329,6 +345,16 @@ class TurnEngine:
if chunk.turn is not None:
turn = chunk.turn
except Exception as exc: # provider failure
+ # A raw context-overflow 400 (compaction mispredicted, e.g. the estimate
+ # path) routes into the compaction policy instead of surfacing. The retry
+ # is progress-guarded: each pass moves the boundary forward or gives up,
+ # so a model that keeps overflowing still terminates in the error path.
+ if _compaction.is_context_overflow(exc) and not self._cancel.is_set():
+ notice = await self._compact_now(force=True)
+ if notice:
+ self._append_notice("compacted", notice)
+ yield Event(EventType.COMPACTED, {"text": notice})
+ continue
# Same contract as the stop path below: the partial the user watched
# arrive survives the failure.
if streamed or streamed_reasoning:
@@ -352,6 +378,10 @@ class TurnEngine:
return
if turn is None:
turn = AssistantTurn()
+ if turn.usage is not None:
+ # The trigger signal: the prompt-side total that actually occupied the
+ # window on this round-trip (estimate fallback when never reported).
+ self._last_context_tokens = turn.usage.context_tokens
self.messages.append(_assistant_message(turn, model=self.model))
payload: dict[str, Any] = {
@@ -386,6 +416,97 @@ class TurnEngine:
if self._steering:
self._inject_steering()
+ # -- auto-compaction (OPE-27) ------------------------------------------------
+ def _compaction_config(self) -> dict[str, Any]:
+ cfg = dict(self.compaction_settings() or {}) if self.compaction_settings else {}
+ if not cfg.get("context_window"):
+ from .providers.matrix import model_context_windows
+
+ cfg["context_window"] = model_context_windows().get(self.model)
+ cfg.setdefault("threshold_pct", _compaction.DEFAULT_THRESHOLD_PCT)
+ cfg.setdefault("cap_tokens", _compaction.DEFAULT_CAP_TOKENS)
+ return cfg
+
+ async def _compact_now(self, *, force: bool = False) -> Optional[str]:
+ """Run the compaction policy when the trigger fires (or `force`, the overflow
+ path). Returns the user-facing notice text when the outbound view changed, else
+ None. Failure policy per spec: retry once (both modes); attended → Retry / Trim
+ prompt; unattended → auto-trim and continue (never park a run on bookkeeping)."""
+ cfg = self._compaction_config()
+ if cfg.get("enabled") is False and not force:
+ return None
+ pct = float(cfg["threshold_pct"])
+ cap = int(cfg["cap_tokens"])
+ window = cfg.get("context_window")
+ if not force:
+ signal = self._last_context_tokens or _compaction.estimate_tokens(
+ self._outbound_messages()
+ )
+ if not _compaction.should_compact(
+ signal, window, threshold_pct=pct, cap_tokens=cap
+ ):
+ return None
+ keep = int(
+ _compaction.KEEP_RECENT_FRACTION
+ * _compaction.trigger_tokens(window, threshold_pct=pct, cap_tokens=cap)
+ )
+ model = str(cfg.get("model") or "") or self.model
+
+ def _build() -> Optional[_compaction.CompactionState]:
+ return _compaction.build_state(
+ self.messages,
+ provider=self.provider,
+ model=model,
+ keep_tokens=keep,
+ prior=self.compaction_state,
+ )
+
+ state: Optional[_compaction.CompactionState] = None
+ failed = False
+ for _attempt in range(2): # first try + the unconditional single retry
+ try:
+ state = await asyncio.to_thread(_build)
+ failed = False
+ break
+ except Exception:
+ failed = True
+ if failed and self.question_asker is not None and self.is_attended and self.is_attended():
+ while True:
+ answer = await self._interruptible(
+ self.question_asker(
+ {
+ "question": (
+ "Context compaction failed — the summarizer couldn't "
+ "condense this session's history. How should I proceed?"
+ ),
+ "options": ["Retry", "Trim oldest 10%"],
+ "allow_text": False,
+ "header": "Compaction",
+ },
+ None,
+ ),
+ interrupted=None,
+ )
+ if not answer or answer.get("answer") != "Retry":
+ break
+ try:
+ state = await asyncio.to_thread(_build)
+ failed = False
+ break
+ except Exception:
+ continue
+ if state is not None:
+ self.compaction_state = state
+ self._last_context_tokens = None # stale once the outbound view shrank
+ return "Context compacted — earlier turns were summarized"
+ if failed or force:
+ trimmed = _compaction.trim_state(self.messages, prior=self.compaction_state)
+ if trimmed is not None:
+ self.compaction_state = trimmed
+ self._last_context_tokens = None
+ return "Context trimmed — oldest turns dropped (summary unavailable)"
+ return None
+
# -- helpers ----------------------------------------------------------------
async def _astream(self):
"""Bridge the provider's blocking stream generator to the async loop via a
@@ -893,13 +1014,19 @@ class TurnEngine:
# one. Whole `notice` messages (error/interrupted/model-switch markers) are
# display-only too: dropped entirely.
_SIDECARS = ("source", "_display", "ts", "reasoning", "usage")
+ # Auto-compaction (OPE-27): everything before the boundary is represented by the
+ # compacted block. Outbound-only — the canonical history stays intact — and the
+ # block+tail are byte-stable between turns, so prompt caching keeps working.
+ source_messages = _compaction.apply_to_outbound(
+ self.messages, self.compaction_state
+ )
out = [
(
{k: v for k, v in msg.items() if k not in _SIDECARS}
if any(s in msg for s in _SIDECARS)
else msg
)
- for msg in self.messages
+ for msg in source_messages
if msg.get("role") != "notice"
]
# PDF attachments (stored as `file` parts) are adapted to the ACTIVE model right
diff --git a/coworker/events.py b/coworker/events.py
index 4cdfc192..0b146485 100644
--- a/coworker/events.py
+++ b/coworker/events.py
@@ -31,6 +31,7 @@ class EventType(str, Enum):
TURN_END = "turn_end"
ERROR = "error"
INTERRUPTED = "interrupted"
+ COMPACTED = "compacted" # outbound history was compacted (summary or trim)
@dataclass
diff --git a/coworker/server/app.py b/coworker/server/app.py
index 3b054945..55cd155f 100644
--- a/coworker/server/app.py
+++ b/coworker/server/app.py
@@ -1378,6 +1378,17 @@ def create_app(manager: SessionManager) -> FastAPI:
max_mb=b.get("pdf_max_mb"),
)
+ @app.post("/v1/settings/compaction")
+ def settings_set_compaction(body: dict) -> dict[str, Any]:
+ # Auto-compaction overrides (OPE-27): threshold % of the context window, the
+ # absolute token cap, and the summarizer-model pin ("" → session's own model).
+ b = body or {}
+ return manager.set_compaction_settings(
+ threshold_pct=b.get("compaction_threshold_pct"),
+ cap_tokens=b.get("compaction_cap_tokens"),
+ model=b.get("compaction_model"),
+ )
+
@app.post("/v1/attachments/inspect-pdf")
def attachments_inspect_pdf(body: dict) -> dict[str, Any]:
# Attach-time page/size probe for the composer's threshold check. Local only.
@@ -1677,6 +1688,9 @@ def create_app(manager: SessionManager) -> FastAPI:
)
await ws.close()
return
+ # Auto-compaction failure prompt (OPE-27): only an ATTENDED session may be asked
+ # Retry/Trim — unattended runs auto-trim (the policy in engine._compact_now).
+ engine.is_attended = lambda: _visibility() == VIS_INLINE
await ws.send_json(
{
"type": "ready",
diff --git a/coworker/server/manager.py b/coworker/server/manager.py
index 048604d7..9b79f345 100644
--- a/coworker/server/manager.py
+++ b/coworker/server/manager.py
@@ -460,6 +460,13 @@ class SessionManager:
)
if record is not None and record.grants:
self._apply_grants(engine, record.grants)
+ # Auto-compaction (OPE-27): restore the persisted view boundary and wire the live
+ # Settings getter — post-construction, so build_engine's signature stays put.
+ if record is not None and record.compaction:
+ from ..compaction import CompactionState
+
+ engine.compaction_state = CompactionState.from_dict(record.compaction)
+ engine.compaction_settings = self.compaction_settings
self._engines[session_id] = engine
if is_new_session:
self._emit_session_created(session_id, agent_name)
@@ -1784,6 +1791,7 @@ class SessionManager:
# hardcoded POSIX one (Windows -> %APPDATA%\coworker, macOS/Linux -> ~/.config).
"secrets_path": str(self.secrets.path),
**self.pdf_settings(),
+ **self.compaction_settings_payload(),
}
def _surfaces(self) -> dict[str, bool]:
@@ -1860,6 +1868,65 @@ class SessionManager:
"pdf_max_mb": max(1, min(mb, 10)),
}
+ def compaction_settings(self) -> dict[str, Any]:
+ """The live auto-compaction knobs (OPE-27) — read by every engine per check, so a
+ Settings change applies without a rebuild. Only the two spec'd overrides plus the
+ summarizer-model pin; absent keys fall back to compaction.py defaults."""
+ from ..compaction import DEFAULT_CAP_TOKENS, DEFAULT_THRESHOLD_PCT
+
+ return {
+ "threshold_pct": float(
+ self._prefs.get("compaction_threshold_pct") or DEFAULT_THRESHOLD_PCT
+ ),
+ "cap_tokens": int(
+ self._prefs.get("compaction_cap_tokens") or DEFAULT_CAP_TOKENS
+ ),
+ # "" → the session's own model (engine falls back to self.model).
+ "model": str(self._prefs.get("compaction_model") or ""),
+ }
+
+ def compaction_settings_payload(self) -> dict[str, Any]:
+ """The same knobs under REST-facing names (prefixed to keep /v1/settings flat)."""
+ settings = self.compaction_settings()
+ return {
+ "compaction_threshold_pct": settings["threshold_pct"],
+ "compaction_cap_tokens": settings["cap_tokens"],
+ "compaction_model": settings["model"],
+ }
+
+ def set_compaction_settings(
+ self,
+ threshold_pct: Any = None,
+ cap_tokens: Any = None,
+ model: Any = None,
+ ) -> dict[str, Any]:
+ """Persist the auto-compaction overrides (OPE-27). Threshold is a percentage of
+ the model's context window (10–95); the cap is an absolute token ceiling; model
+ pins the summarizer ('' → the session's own model). Engines read these live via
+ `compaction_settings()`, so changes apply to running sessions immediately."""
+ if threshold_pct is not None:
+ try:
+ pct = float(threshold_pct)
+ except (TypeError, ValueError):
+ return {"ok": False, "error": "compaction_threshold_pct must be a number"}
+ if not 0.10 <= pct <= 0.95:
+ return {
+ "ok": False,
+ "error": "compaction_threshold_pct must be between 0.10 and 0.95",
+ }
+ self._prefs["compaction_threshold_pct"] = pct
+ if cap_tokens is not None:
+ try:
+ self._prefs["compaction_cap_tokens"] = max(
+ 10_000, min(int(cap_tokens), 2_000_000)
+ )
+ except (TypeError, ValueError):
+ return {"ok": False, "error": "compaction_cap_tokens must be a number"}
+ if model is not None:
+ self._prefs["compaction_model"] = str(model)
+ self._save_prefs()
+ return {"ok": True, **self.compaction_settings()}
+
def set_pdf_settings(
self,
fallback: Any = None,
@@ -3249,6 +3316,11 @@ class SessionManager:
agent=getattr(engine, "agent_name", "code"),
extra_roots=self._extra_roots_of(engine),
grants=_grants_of(engine),
+ compaction=(
+ engine.compaction_state.as_dict()
+ if getattr(engine, "compaction_state", None)
+ else {}
+ ),
)
)
diff --git a/coworker/sessions.py b/coworker/sessions.py
index cc6c4cf5..4bd857c7 100644
--- a/coworker/sessions.py
+++ b/coworker/sessions.py
@@ -34,3 +34,6 @@ class SessionRecord:
# (e.g. origin="slack", origin_label="#general · T0ABCD"). Set once at spawn.
origin: Optional[str] = None
origin_label: Optional[str] = None
+ # Auto-compaction state (OPE-27): CompactionState.as_dict(), {} when never compacted.
+ # Persisted so a reloaded session keeps its compacted outbound view.
+ compaction: dict[str, Any] = field(default_factory=dict)
diff --git a/surfaces/gui/e2e/compaction.spec.ts b/surfaces/gui/e2e/compaction.spec.ts
new file mode 100644
index 00000000..8a3bd0e6
--- /dev/null
+++ b/surfaces/gui/e2e/compaction.spec.ts
@@ -0,0 +1,75 @@
+// OPE-27 — auto-compaction GUI: the Settings card's two overrides + summarizer-model
+// pin POST through, and the "context compacted" divider renders inline mid-session
+// (driven by the fixtures' scripted `compacted` event) without touching the transcript.
+import { expect } from "@playwright/test";
+import { test } from "./fixtures";
+
+test("Settings: Context compaction card edits threshold, cap, and summarizer model", async ({
+ page,
+}) => {
+ await page.goto("/");
+ await page.getByTestId("account-row").click();
+ await page.getByRole("button", { name: "Settings", exact: true }).click();
+ await page.getByRole("button", { name: "Models", exact: true }).click();
+
+ const card = page.getByTestId("compaction-card");
+ await expect(card).toBeVisible();
+ await expect(card.getByText("Context compaction")).toBeVisible();
+
+ // Defaults render when the backend doesn't send the fields (older-backend robustness).
+ await expect(card.getByTestId("compaction-threshold")).toHaveValue("80");
+ await expect(card.getByTestId("compaction-cap")).toHaveValue("250000");
+ await expect(card.getByTestId("compaction-model")).toHaveValue("");
+
+ // Threshold edits POST as a fraction, clamped to 10–95%.
+ const [req] = await Promise.all([
+ page.waitForRequest(
+ (r) => r.url().endsWith("/v1/settings/compaction") && r.method() === "POST",
+ ),
+ card.getByTestId("compaction-threshold").fill("70"),
+ ]);
+ expect(req.postDataJSON()).toEqual({ compaction_threshold_pct: 0.7 });
+
+ const [req2] = await Promise.all([
+ page.waitForRequest(
+ (r) => r.url().endsWith("/v1/settings/compaction") && r.method() === "POST",
+ ),
+ card.getByTestId("compaction-cap").fill("100000"),
+ ]);
+ expect(req2.postDataJSON()).toEqual({ compaction_cap_tokens: 100000 });
+
+ // Summarizer pin: the picker offers the session-default plus the configured models.
+ const [req3] = await Promise.all([
+ page.waitForRequest(
+ (r) => r.url().endsWith("/v1/settings/compaction") && r.method() === "POST",
+ ),
+ card.getByTestId("compaction-model").selectOption("gpt-4o-mini"),
+ ]);
+ expect(req3.postDataJSON()).toEqual({ compaction_model: "gpt-4o-mini" });
+});
+
+test("the compacted divider renders mid-session and the transcript stays intact", async ({
+ page,
+}) => {
+ await page.goto("/");
+ await page.getByText("Draft the launch note").first().click();
+ const box = page.getByPlaceholder(/Ask the coworker/);
+
+ // An earlier exchange that must survive the compaction marker (transcript intact).
+ await box.fill("remember the launch date");
+ await box.press("Enter");
+ await expect(page.getByText("Echo: remember the launch date").first()).toBeVisible({
+ timeout: 10_000,
+ });
+
+ await box.fill("compact the context");
+ await box.press("Enter");
+ await expect(
+ page.getByText("Context compacted — earlier turns were summarized").first(),
+ ).toBeVisible({ timeout: 10_000 });
+ await expect(
+ page.getByText("Still on it — continuing where I left off.").first(),
+ ).toBeVisible();
+ // Outbound-only: everything before the divider is still on screen.
+ await expect(page.getByText("Echo: remember the launch date").first()).toBeVisible();
+});
diff --git a/surfaces/gui/e2e/fixtures.ts b/surfaces/gui/e2e/fixtures.ts
index 4224bff7..037fc617 100644
--- a/surfaces/gui/e2e/fixtures.ts
+++ b/surfaces/gui/e2e/fixtures.ts
@@ -681,6 +681,14 @@ export async function mockApi(page: import("@playwright/test").Page) {
}, 120);
return;
}
+ // Auto-compaction (OPE-27): the server compacts mid-run and emits the marker,
+ // then the turn continues normally — the divider must render inline.
+ if (/compact the context/i.test(msg.text)) {
+ send("compacted", { text: "Context compacted — earlier turns were summarized" });
+ send("assistant_message", { text: "Still on it — continuing where I left off." });
+ send("turn_done");
+ return;
+ }
// A turn that dies on a provider error; the follow-up {type:"retry"} recovers.
if (/fail the turn/i.test(msg.text)) {
send("error", { error: "model unreachable" });
diff --git a/surfaces/gui/src/App.tsx b/surfaces/gui/src/App.tsx
index 6c261944..f814cdb8 100644
--- a/surfaces/gui/src/App.tsx
+++ b/surfaces/gui/src/App.tsx
@@ -709,6 +709,11 @@ export function App() {
if (d.model) setModel(d.model);
setItems((p) => [...p, { kind: "notice", tone: "info", text: d.text || "Model switched" }]);
break;
+ case "compacted":
+ // Auto-compaction marker (OPE-27): outbound-only — the transcript stays intact,
+ // this divider just shows where the model's memory was summarized.
+ setItems((p) => [...p, { kind: "notice", tone: "info", text: d.text || "Context compacted" }]);
+ break;
case "interrupted":
flushPartialStream();
setItems((p) => [...p, { kind: "notice", tone: "warn", text: "Interrupted." }]);
diff --git a/surfaces/gui/src/api.ts b/surfaces/gui/src/api.ts
index 80a5dcef..55d97f5e 100644
--- a/surfaces/gui/src/api.ts
+++ b/surfaces/gui/src/api.ts
@@ -702,6 +702,12 @@ export interface ModelSettings {
pdf_fallback?: "text" | "images";
pdf_max_pages?: number; // default 20, 1–100
pdf_max_mb?: number; // default 10, 1–10
+ // Auto-compaction of long histories (OPE-27): trigger = min(threshold% × context
+ // window, cap tokens); model pins the summarizer ("" → the session's own model).
+ // Optional so the GUI is robust to an older backend.
+ compaction_threshold_pct?: number; // default 0.8, 0.10–0.95
+ compaction_cap_tokens?: number; // default 250000
+ compaction_model?: string;
}
export interface PdfSettings {
@@ -722,6 +728,24 @@ export async function setPdfSettings(
return res.json();
}
+export interface CompactionSettings {
+ compaction_threshold_pct: number;
+ compaction_cap_tokens: number;
+ compaction_model: string;
+}
+
+/** Persist the auto-compaction overrides (threshold %, token cap, summarizer model). */
+export async function setCompactionSettings(
+ patch: Partial,
+): Promise<{ ok: boolean; error?: string }> {
+ const res = await fetch(`${httpBase()}/v1/settings/compaction`, {
+ method: "POST",
+ headers: { "Content-Type": "application/json" },
+ body: JSON.stringify(patch),
+ });
+ return res.json();
+}
+
/** Local page/size probe for a PDF data URL — the composer's attach-time threshold check. */
export async function inspectPdf(
dataUrl: string,
diff --git a/surfaces/gui/src/components/SettingsView.tsx b/surfaces/gui/src/components/SettingsView.tsx
index 5b88cc84..fcc15505 100644
--- a/surfaces/gui/src/components/SettingsView.tsx
+++ b/surfaces/gui/src/components/SettingsView.tsx
@@ -2,11 +2,13 @@ import { useEffect, useState } from "react";
import {
getSettings,
getTrustedWorkspaces,
+ setCompactionSettings,
setOnboarded,
setPdfSettings,
setScratchBase,
setSessionsPeek,
setWorkspaceTrusted,
+ type CompactionSettings,
type ModelSettings,
type PdfSettings,
type WorkspaceCommandTrust,
@@ -118,6 +120,7 @@ export function SettingsView({
not under General. */}
+
) : tab === "voice" ? (
@@ -584,9 +587,9 @@ function UpdateInline() {
// -- Sidebar density -------------------------------------------------------------
// -- Token savings (PDF attachments; owner ask, 2026-07-17) ---------------------
// Attachments replay with EVERY turn, so a big PDF quietly multiplies token spend.
-// Auto-compaction of long histories is a planned follow-up (punchlist §7) — until
-// then this card is the user's dial: attach thresholds + the fallback for models
-// without native PDF support.
+// This card is the attachment dial: attach thresholds + the fallback for models
+// without native PDF support. (Long-history spend is handled by auto-compaction —
+// the CompactionCard below, OPE-27.)
function TokenSavingsCard() {
const [pdf, setPdf] = useState(null);
@@ -672,6 +675,121 @@ function TokenSavingsCard() {
);
}
+// -- Context compaction (OPE-27) ------------------------------------------------
+// Long sessions are summarized automatically when they approach the model's context
+// limit, so work continues instead of hitting a raw provider error. Two spec'd
+// overrides (trigger % + token cap) and the summarizer-model pin — nothing more.
+function CompactionCard() {
+ const [cfg, setCfg] = useState(null);
+ const [models, setModels] = useState([]);
+ const [labels, setLabels] = useState>({});
+
+ useEffect(() => {
+ getSettings()
+ .then((s) => {
+ setCfg({
+ compaction_threshold_pct: s.compaction_threshold_pct ?? 0.8,
+ compaction_cap_tokens: s.compaction_cap_tokens ?? 250_000,
+ compaction_model: s.compaction_model ?? "",
+ });
+ setModels(s.models || []);
+ setLabels(s.model_labels || {});
+ })
+ .catch(() =>
+ setCfg({
+ compaction_threshold_pct: 0.8,
+ compaction_cap_tokens: 250_000,
+ compaction_model: "",
+ }),
+ );
+ }, []);
+
+ const save = async (patch: Partial) => {
+ setCfg((p) => (p ? { ...p, ...patch } : p));
+ await setCompactionSettings(patch);
+ };
+
+ if (!cfg) return null;
+ const modelLabel = (id: string) => labels[id]?.split(" · ")[0] || id;
+ return (
+
+
Context compaction
+
+ Long sessions are compacted automatically: older turns are summarized so the
+ coworker keeps working instead of running out of context. Your visible transcript
+ is never changed — a small marker shows where compaction happened.
+
+
+
+
+
+
+
+ The cap makes very-large-context models compact early — quality and speed degrade
+ well before their nominal limit.
+
+
+
+ Summarizer model
+
+
+
+ The summary is written by this model. The default follows whatever model the
+ session is using.
+
+
+ );
+}
+
function SidebarCard() {
const [peek, setPeek] = useState(null);
diff --git a/surfaces/gui/src/itemsFromMessages.test.ts b/surfaces/gui/src/itemsFromMessages.test.ts
index ad025475..b87b8717 100644
--- a/surfaces/gui/src/itemsFromMessages.test.ts
+++ b/surfaces/gui/src/itemsFromMessages.test.ts
@@ -83,6 +83,20 @@ describe("itemsFromMessages model switch", () => {
});
});
+describe("itemsFromMessages compaction", () => {
+ it("replays the persisted compacted marker as an info notice (the divider)", () => {
+ const items = itemsFromMessages([
+ { role: "user", content: "hi" },
+ { role: "notice", kind: "compacted", text: "Context compacted — earlier turns were summarized" },
+ ] as any);
+ expect(items[1]).toEqual({
+ kind: "notice",
+ tone: "info",
+ text: "Context compacted — earlier turns were summarized",
+ });
+ });
+});
+
describe("itemsFromMessages reasoning", () => {
it("attaches the reasoning sidecar to assistant items; thinking-only messages still render", () => {
const items = itemsFromMessages([
diff --git a/surfaces/gui/src/itemsFromMessages.ts b/surfaces/gui/src/itemsFromMessages.ts
index b13b81e2..62e9a931 100644
--- a/surfaces/gui/src/itemsFromMessages.ts
+++ b/surfaces/gui/src/itemsFromMessages.ts
@@ -72,7 +72,10 @@ export function itemsFromMessages(messages: ConversationMessage[]): Item[] {
? { kind: "notice", tone: "warn", text: "Interrupted." }
: m.kind === "model_switch"
? { kind: "notice", tone: "info", text: m.text || "Model switched" }
- : { kind: "notice", tone: "warn", text: "Error: " + (m.text || "unknown"), retriable: true },
+ : m.kind === "compacted"
+ ? // The subtle "compacted here" divider (OPE-27) — the transcript itself is intact.
+ { kind: "notice", tone: "info", text: m.text || "Context compacted" }
+ : { kind: "notice", tone: "warn", text: "Error: " + (m.text || "unknown"), retriable: true },
);
}
// system messages are omitted; tool-result messages are folded into the tool row above
diff --git a/surfaces/gui/src/types.ts b/surfaces/gui/src/types.ts
index f86b3ce8..a05a37ab 100644
--- a/surfaces/gui/src/types.ts
+++ b/surfaces/gui/src/types.ts
@@ -18,6 +18,7 @@ export type EventType =
| "input_rejected"
| "interrupted"
| "model_changed"
+ | "compacted"
| "turn_done";
export interface WsEvent {
diff --git a/tests/test_compaction.py b/tests/test_compaction.py
new file mode 100644
index 00000000..1d25f18e
--- /dev/null
+++ b/tests/test_compaction.py
@@ -0,0 +1,315 @@
+"""OPE-27 — auto-compaction pure functions: trigger math, boundary picking, mechanical
+extraction, summarizer seam, trim fallback, outbound view. No engine involved."""
+
+import json
+
+import pytest
+
+from coworker.compaction import (
+ CompactionState,
+ DEFAULT_CAP_TOKENS,
+ DEFAULT_CONTEXT_WINDOW,
+ apply_to_outbound,
+ build_state,
+ compacted_block,
+ estimate_tokens,
+ extract_user_messages,
+ extract_working_state,
+ is_context_overflow,
+ pick_boundary,
+ should_compact,
+ summarize_span,
+ summarizer_messages,
+ trigger_tokens,
+ trim_state,
+)
+
+
+# -- message builders ---------------------------------------------------------
+
+
+def user(text):
+ return {"role": "user", "content": text, "ts": 1.0}
+
+
+_call_seq = 0
+
+
+def assistant(text="", tool_calls=None):
+ global _call_seq
+ msg = {"role": "assistant", "content": text, "ts": 1.0}
+ if tool_calls:
+ calls = []
+ for name, args in tool_calls:
+ calls.append(
+ {
+ "id": f"c{_call_seq}",
+ "type": "function",
+ "function": {"name": name, "arguments": json.dumps(args)},
+ }
+ )
+ _call_seq += 1
+ msg["tool_calls"] = calls
+ return msg
+
+
+def tool(call_id, content):
+ return {
+ "role": "tool",
+ "tool_call_id": call_id,
+ "content": content if isinstance(content, str) else json.dumps(content),
+ "ts": 1.0,
+ }
+
+
+def tool_turn(name, args, result):
+ """[assistant tool-call, matching tool result] with a properly paired call id."""
+ a = assistant(tool_calls=[(name, args)])
+ return [a, tool(a["tool_calls"][0]["id"], result)]
+
+
+def convo(turns=6, bulk=2000):
+ """system + N user/assistant turns with bulky assistant text."""
+ msgs = [{"role": "system", "content": "You are a coworker."}]
+ for i in range(turns):
+ msgs.append(user(f"request {i}"))
+ msgs.append(assistant(f"answer {i} " + "x" * bulk))
+ return msgs
+
+
+class FakeSummarizer:
+ def __init__(self, text="## Summary\nall good", fail_times=0):
+ self.text = text
+ self.fail_times = fail_times
+ self.calls = []
+
+ def complete(self, *, model, messages, tools=None, **settings):
+ self.calls.append({"model": model, "messages": messages, "tools": tools, **settings})
+ if self.fail_times > 0:
+ self.fail_times -= 1
+ raise RuntimeError("summarizer down")
+
+ class Turn:
+ pass
+
+ t = Turn()
+ t.text = self.text
+ return t
+
+
+# -- trigger math -------------------------------------------------------------
+
+
+def test_trigger_is_min_of_pct_and_cap():
+ assert trigger_tokens(100_000) == 80_000
+ assert trigger_tokens(1_000_000) == DEFAULT_CAP_TOKENS # the 250k cap wins
+ assert trigger_tokens(None) == int(0.8 * DEFAULT_CONTEXT_WINDOW)
+ # both knobs are user-overridable
+ assert trigger_tokens(100_000, threshold_pct=0.5, cap_tokens=40_000) == 40_000
+ assert trigger_tokens(100_000, threshold_pct=0.5, cap_tokens=999_999) == 50_000
+
+
+def test_should_compact_crosses_threshold():
+ assert not should_compact(79_999, 100_000)
+ assert should_compact(80_000, 100_000)
+
+
+def test_estimate_tokens_is_chars_over_four():
+ msgs = [user("a" * 400)]
+ est = estimate_tokens(msgs)
+ assert 100 <= est <= 120 # 400 chars of content + json overhead, /4
+
+
+# -- boundary -----------------------------------------------------------------
+
+
+def test_boundary_prefers_earliest_user_turn_that_fits():
+ msgs = convo(turns=6)
+ per_turn = estimate_tokens(msgs[1:3])
+ boundary = pick_boundary(msgs, keep_tokens=per_turn * 2 + 10)
+ assert msgs[boundary]["role"] == "user"
+ assert msgs[boundary]["content"] == "request 4" # newest two turns survive
+
+
+def test_boundary_falls_inside_a_giant_final_turn():
+ # One user turn followed by a huge tool loop: the turn alone exceeds the budget,
+ # so the cut lands on an assistant (iteration) boundary inside it — never a tool row.
+ msgs = [{"role": "system", "content": "s"}, user("go")]
+ for i in range(8):
+ a = assistant("step " + "y" * 3000, tool_calls=[("run_shell", {"command": f"cmd{i}"})])
+ msgs += [a, tool(a["tool_calls"][0]["id"], {"exit_code": 0, "out": "z" * 3000})]
+ boundary = pick_boundary(msgs, keep_tokens=estimate_tokens(msgs[-3:]))
+ assert msgs[boundary]["role"] == "assistant"
+
+
+def test_boundary_none_when_nothing_to_summarize():
+ msgs = [{"role": "system", "content": "s"}, user("hi"), assistant("hello")]
+ assert pick_boundary(msgs, keep_tokens=10_000_000) is None
+
+
+# -- mechanical extraction ----------------------------------------------------
+
+
+def test_working_state_files_commands_tools():
+ span = [
+ user("write it"),
+ *tool_turn("write_file", {"path": "a.py", "content": "x"}, {"ok": True}),
+ *tool_turn("run_shell", {"command": "pytest -q"}, {"exit_code": 1}),
+ *tool_turn("write_file", {"path": "b.py", "content": "y"}, {"ok": True}),
+ *tool_turn("write_file", {"path": "a.py", "content": "x2"}, {"ok": True}),
+ ]
+ block = extract_working_state(span)
+ # deduped, most recent first
+ assert block.index("- a.py") < block.index("- b.py")
+ assert block.count("a.py") == 1
+ assert "pytest -q" in block and "[exit 1]" in block
+ assert "run_shell" in block and "write_file" in block
+
+
+def test_working_state_empty_span():
+ assert extract_working_state([user("hi"), assistant("yo")]) == ""
+
+
+def test_user_messages_extracted_verbatim_and_clipped():
+ span = [
+ user("first ask"),
+ assistant("a"),
+ user([{"type": "text", "text": "second"}, {"type": "image_url", "image_url": {}}]),
+ assistant("b"),
+ user("bulk " + "z" * 2000),
+ ]
+ out = extract_user_messages(span)
+ assert out[0] == "first ask"
+ assert out[1] == "second [image]"
+ assert out[2].endswith("…") and len(out[2]) <= 600
+
+
+# -- summarizer seam ----------------------------------------------------------
+
+
+def test_summarizer_messages_clip_tool_results_and_fold_prior():
+ span = [user("go"), *tool_turn("read_file", {"path": "big.txt"}, "huge " * 500)]
+ msgs = summarizer_messages(span, prior_summary="OLD SUMMARY")
+ body = msgs[1]["content"]
+ assert "OLD SUMMARY" in body
+ assert len(body) < 3000 # the 2500-char tool result got clipped hard
+ assert msgs[0]["role"] == "system" and "Primary request and intent" in msgs[0]["content"]
+
+
+def test_summarize_span_passes_model_and_raises_on_empty():
+ fake = FakeSummarizer(text="## ok")
+ out = summarize_span(fake, "prov:model-x", [user("hi")])
+ assert out == "## ok"
+ assert fake.calls[0]["model"] == "prov:model-x"
+ assert fake.calls[0]["tools"] is None
+
+ with pytest.raises(RuntimeError):
+ summarize_span(FakeSummarizer(text=" "), "m", [user("hi")])
+
+
+# -- build + repeated compaction ----------------------------------------------
+
+
+def test_build_state_and_outbound_view():
+ msgs = convo(turns=6)
+ fake = FakeSummarizer(text="## Summary\nthe gist")
+ state = build_state(
+ msgs, provider=fake, model="m", keep_tokens=estimate_tokens(msgs[-4:]) + 10
+ )
+ assert state is not None and not state.trimmed
+ assert state.user_messages[0] == "request 0"
+
+ out = apply_to_outbound(msgs, state)
+ assert out[0]["role"] == "system" # instructions survive
+ assert "" in out[1]["content"]
+ assert "the gist" in out[1]["content"]
+ assert "request 0" in out[1]["content"] # mechanical user-message list
+ assert out[2] is msgs[state.boundary_index] # verbatim tail, canonical untouched
+ assert len(msgs) == 13 # canonical history unchanged
+
+
+def test_repeated_compaction_summarizes_prior_plus_new_turns():
+ msgs = convo(turns=4)
+ fake = FakeSummarizer()
+ first = build_state(msgs, provider=fake, model="m", keep_tokens=estimate_tokens(msgs[-4:]) + 10)
+ # session grows
+ for i in range(4, 8):
+ msgs.append(user(f"request {i}"))
+ msgs.append(assistant(f"answer {i} " + "x" * 2000))
+ second = build_state(
+ msgs, provider=fake, model="m",
+ keep_tokens=estimate_tokens(msgs[-4:]) + 10, prior=first,
+ )
+ assert second is not None and second.boundary_index > first.boundary_index
+ # the second summarizer call folds the prior summary in
+ assert "previous compaction summary" in fake.calls[1]["messages"][1]["content"]
+ # user messages accumulate across compactions
+ assert "request 0" in second.user_messages[0]
+ assert any("request 5" in u for u in second.user_messages)
+
+
+def test_build_state_none_when_boundary_stale():
+ msgs = convo(turns=3)
+ fake = FakeSummarizer()
+ state = build_state(msgs, provider=fake, model="m", keep_tokens=estimate_tokens(msgs[-2:]) + 10)
+ again = build_state(
+ msgs, provider=fake, model="m",
+ keep_tokens=10_000_000, prior=state,
+ )
+ assert again is None # nothing new fits below the prior boundary
+
+
+# -- trim fallback ------------------------------------------------------------
+
+
+def test_trim_advances_boundary_and_keeps_user_messages():
+ msgs = convo(turns=10)
+ state = trim_state(msgs)
+ assert state is not None and state.trimmed
+ assert msgs[state.boundary_index]["role"] in ("user", "assistant")
+ assert state.user_messages # preserved mechanically even without a summary
+ assert "trimmed" in state.summary_text
+ out = apply_to_outbound(msgs, state)
+ assert len(out) < len(msgs) + 1
+
+
+def test_trim_from_prior_state_never_lands_on_tool_row():
+ msgs = [{"role": "system", "content": "s"}, user("go")]
+ for i in range(10):
+ msgs += tool_turn("run_shell", {"command": f"c{i}"}, {"exit_code": 0})
+ prior = trim_state(msgs)
+ later = trim_state(msgs, prior=prior)
+ assert later.boundary_index > prior.boundary_index
+ assert msgs[later.boundary_index]["role"] != "tool"
+
+
+def test_trim_none_when_too_small():
+ assert trim_state([user("hi"), assistant("yo")]) is None
+
+
+# -- state round-trip + overflow detection ------------------------------------
+
+
+def test_state_dict_round_trip():
+ state = CompactionState(
+ boundary_index=7, summary_text="s", working_state="w",
+ user_messages=["u1"], created_at=1.5, model_used="m", trimmed=True,
+ )
+ assert CompactionState.from_dict(state.as_dict()) == state
+ assert CompactionState.from_dict(None) is None
+ assert CompactionState.from_dict({}) is None
+
+
+def test_apply_to_outbound_noop_on_stale_or_missing_state():
+ msgs = convo(turns=2)
+ assert apply_to_outbound(msgs, None) is msgs
+ stale = CompactionState(boundary_index=999, summary_text="s", working_state="")
+ assert apply_to_outbound(msgs, stale) is msgs
+
+
+def test_is_context_overflow():
+ assert is_context_overflow(Exception("Error 400: maximum context length is 128000 tokens"))
+ assert is_context_overflow(Exception("context_length_exceeded"))
+ assert is_context_overflow(Exception("Prompt is too long: 210000 tokens > limit"))
+ assert not is_context_overflow(Exception("rate limit exceeded"))
+ assert not is_context_overflow(Exception("connection reset"))
diff --git a/tests/test_compaction_engine.py b/tests/test_compaction_engine.py
new file mode 100644
index 00000000..99e1270c
--- /dev/null
+++ b/tests/test_compaction_engine.py
@@ -0,0 +1,258 @@
+"""OPE-27 engine hook: the mid-run trigger, the outbound view, the usage signal, the
+failure policy (attended prompt / unattended auto-trim), raw-overflow routing, and the
+session persistence round-trip. Scripted providers, tiny forced windows, no network."""
+
+import asyncio
+
+from coworker.engine import TurnEngine
+from coworker.events import EventType
+from coworker.permissions import PermissionEngine
+from coworker.providers import (
+ AssistantTurn,
+ ModelCapabilities,
+ ProviderClient,
+ ToolCall,
+)
+from coworker.providers.base import TokenUsage
+from coworker.tools import ToolRegistry
+
+SUMMARY = "## Primary request and intent\nkeep building the report"
+
+
+class CompactingProvider(ProviderClient):
+ """Scripted main turns; summarizer calls (recognized by the compaction system prompt)
+ are answered out-of-band so they never consume the main script."""
+
+ def __init__(self, turns, *, summary=SUMMARY, summary_fails=0, main_overflows=0):
+ self._turns = list(turns)
+ self.summary = summary
+ self.summary_fails = summary_fails
+ self.main_overflows = main_overflows
+ self.summary_calls = []
+ self.main_calls = 0
+
+ def complete(self, *, model, messages, tools=None, **settings):
+ if messages and "compacting an AI coworker" in str(
+ messages[0].get("content", "")
+ ):
+ self.summary_calls.append({"model": model, "messages": messages})
+ if self.summary_fails > 0:
+ self.summary_fails -= 1
+ raise RuntimeError("summarizer down")
+ return AssistantTurn(text=self.summary, finish_reason="stop")
+ self.main_calls += 1
+ if self.main_overflows > 0:
+ self.main_overflows -= 1
+ raise RuntimeError(
+ "Error 400: maximum context length is 100000 tokens, request used more"
+ )
+ return self._turns.pop(0)
+
+ def capabilities(self, model):
+ return ModelCapabilities()
+
+
+def long_history(turns=8, bulk=1500):
+ msgs = [{"role": "system", "content": "be helpful"}]
+ for i in range(turns):
+ msgs.append({"role": "user", "content": f"request {i}", "ts": 1.0})
+ msgs.append(
+ {"role": "assistant", "content": f"answer {i} " + "x" * bulk, "ts": 1.0}
+ )
+ return msgs
+
+
+def make_engine(tmp_path, provider, *, messages=None, cap=400):
+ engine = TurnEngine(
+ provider=provider,
+ registry=ToolRegistry(),
+ permissions=PermissionEngine(workspace_root=tmp_path),
+ model="gpt-5.5",
+ messages=messages,
+ )
+ engine.compaction_settings = lambda: {
+ "cap_tokens": cap,
+ "threshold_pct": 0.8,
+ "context_window": 100_000,
+ }
+ return engine
+
+
+def collect(engine, text="continue"):
+ async def _run():
+ return [e async for e in engine.run(text)]
+
+ return asyncio.run(_run())
+
+
+def test_compacts_before_the_turn_when_estimate_crosses(tmp_path):
+ provider = CompactingProvider([AssistantTurn(text="done", finish_reason="stop")])
+ engine = make_engine(tmp_path,provider, messages=long_history(), cap=400)
+ events = collect(engine)
+
+ assert any(e.type == EventType.COMPACTED for e in events)
+ assert not any(e.type == EventType.ERROR for e in events)
+ state = engine.compaction_state
+ assert state is not None and not state.trimmed
+ assert provider.summary_calls[0]["model"] == "gpt-5.5" # session's own model
+
+ # Outbound view: system survives, the block stands in for the old turns, the
+ # canonical transcript is untouched, and the persisted notice marks the spot.
+ out = engine._outbound_messages()
+ assert out[0]["role"] == "system"
+ assert "" in out[1]["content"]
+ assert SUMMARY.splitlines()[-1] in out[1]["content"]
+ assert "request 0" in out[1]["content"] # mechanical user-message list
+ assert any("answer 0" in str(m.get("content")) for m in engine.messages)
+ assert any(
+ m.get("role") == "notice" and m.get("kind") == "compacted"
+ for m in engine.messages
+ )
+
+
+def test_usage_signal_triggers_between_tool_turns(tmp_path):
+ # History too small for the estimate path — only the reported usage crosses the
+ # trigger, after iteration 1's round-trip. The compaction runs before iteration 2.
+ provider = CompactingProvider(
+ [
+ AssistantTurn(
+ tool_calls=[ToolCall(id="c1", name="nonexistent_tool", arguments={})],
+ finish_reason="tool_calls",
+ usage=TokenUsage(input=90_000, output=10),
+ ),
+ AssistantTurn(text="done", finish_reason="stop"),
+ ]
+ )
+ engine = make_engine(tmp_path,provider, messages=long_history(turns=2, bulk=10), cap=400)
+ events = collect(engine)
+ assert any(e.type == EventType.COMPACTED for e in events)
+ assert provider.summary_calls # driven by usage, not the (tiny) estimate
+ assert engine._last_context_tokens is None # reset once the view shrank
+
+
+def test_summarizer_failure_unattended_auto_trims(tmp_path):
+ provider = CompactingProvider(
+ [AssistantTurn(text="done", finish_reason="stop")], summary_fails=99
+ )
+ engine = make_engine(tmp_path,provider, messages=long_history(), cap=400)
+ events = collect(engine) # is_attended is None → unattended policy
+
+ compacted = [e for e in events if e.type == EventType.COMPACTED]
+ assert compacted and "trimmed" in compacted[0].data["text"].lower()
+ assert engine.compaction_state is not None and engine.compaction_state.trimmed
+ assert len(provider.summary_calls) == 2 # the one unconditional retry, then trim
+
+
+def test_summarizer_failure_attended_prompts_retry_then_succeeds(tmp_path):
+ provider = CompactingProvider(
+ [AssistantTurn(text="done", finish_reason="stop")], summary_fails=2
+ )
+ engine = make_engine(tmp_path,provider, messages=long_history(), cap=400)
+ engine.is_attended = lambda: True
+ asked = []
+
+ async def asker(args, tool_call_id=None):
+ asked.append(args)
+ return {"answer": "Retry"}
+
+ engine.question_asker = asker
+ collect(engine)
+
+ assert asked and asked[0]["options"] == ["Retry", "Trim oldest 10%"]
+ assert engine.compaction_state is not None and not engine.compaction_state.trimmed
+
+
+def test_summarizer_failure_attended_choose_trim(tmp_path):
+ provider = CompactingProvider(
+ [AssistantTurn(text="done", finish_reason="stop")], summary_fails=99
+ )
+ engine = make_engine(tmp_path,provider, messages=long_history(), cap=400)
+ engine.is_attended = lambda: True
+
+ async def asker(args, tool_call_id=None):
+ return {"answer": "Trim oldest 10%"}
+
+ engine.question_asker = asker
+ collect(engine)
+ assert engine.compaction_state is not None and engine.compaction_state.trimmed
+
+
+def test_raw_overflow_routes_into_compaction_and_retries(tmp_path):
+ # Trigger never fires (huge cap) — the provider 400 is the only signal. The engine
+ # must compact (force) and retry the call instead of surfacing the error.
+ provider = CompactingProvider(
+ [AssistantTurn(text="recovered", finish_reason="stop")], main_overflows=1
+ )
+ engine = make_engine(tmp_path,provider, messages=long_history(), cap=1_000_000)
+ events = collect(engine)
+
+ assert any(e.type == EventType.COMPACTED for e in events)
+ assert not any(e.type == EventType.ERROR for e in events)
+ finals = [e for e in events if e.type == EventType.ASSISTANT_MESSAGE]
+ assert finals and finals[-1].data["text"] == "recovered"
+ assert provider.main_calls == 2
+
+
+def test_non_overflow_provider_errors_still_surface(tmp_path):
+ class FailingProvider(CompactingProvider):
+ def complete(self, *, model, messages, tools=None, **settings):
+ raise RuntimeError("rate limit exceeded")
+
+ engine = make_engine(tmp_path,FailingProvider([]), messages=long_history(turns=1), cap=1_000_000)
+ events = collect(engine)
+ assert any(e.type == EventType.ERROR for e in events)
+ assert not any(e.type == EventType.COMPACTED for e in events)
+
+
+def test_set_compaction_settings_validates_and_round_trips(tmp_path):
+ from coworker.server.manager import SessionManager
+
+ class Provider(ProviderClient):
+ def complete(self, *, model, messages, tools=None, **settings):
+ return AssistantTurn(text="hi")
+
+ def capabilities(self, model):
+ return ModelCapabilities()
+
+ mgr = SessionManager(workspace=tmp_path, provider=Provider())
+ out = mgr.set_compaction_settings(
+ threshold_pct=0.5, cap_tokens=100_000, model="gpt-4o-mini"
+ )
+ assert out["ok"] and out["threshold_pct"] == 0.5 and out["cap_tokens"] == 100_000
+ assert mgr.compaction_settings()["model"] == "gpt-4o-mini"
+ # validation: out-of-range % and non-numeric cap are rejected, tiny caps clamp up
+ assert mgr.set_compaction_settings(threshold_pct=0.05)["ok"] is False
+ assert mgr.set_compaction_settings(cap_tokens="lots")["ok"] is False
+ assert mgr.set_compaction_settings(cap_tokens=1)["cap_tokens"] == 10_000
+ # the flat /v1/settings names
+ payload = mgr.compaction_settings_payload()
+ assert payload["compaction_threshold_pct"] == 0.5
+ assert payload["compaction_model"] == "gpt-4o-mini"
+
+
+def test_compaction_state_survives_save_and_rebuild(tmp_path):
+ from coworker.compaction import CompactionState
+ from coworker.server.manager import SessionManager
+
+ class Provider(ProviderClient):
+ def complete(self, *, model, messages, tools=None, **settings):
+ return AssistantTurn(text="hi", finish_reason="stop")
+
+ def capabilities(self, model):
+ return ModelCapabilities()
+
+ mgr = SessionManager(workspace=tmp_path, provider=Provider())
+ sid = "compact-persist"
+ engine = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path))
+ assert callable(engine.compaction_settings) # live Settings getter is wired
+ assert engine.compaction_settings()["threshold_pct"] == 0.8
+
+ engine.messages += long_history(turns=3)[1:]
+ engine.compaction_state = CompactionState(
+ boundary_index=3, summary_text="the gist", working_state="", user_messages=["u"]
+ )
+ mgr.save(sid, engine)
+ mgr._engines.pop(sid)
+
+ rebuilt = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path))
+ assert rebuilt.compaction_state == engine.compaction_state
diff --git a/tests/test_compaction_smoke.py b/tests/test_compaction_smoke.py
new file mode 100644
index 00000000..dd5f2b48
--- /dev/null
+++ b/tests/test_compaction_smoke.py
@@ -0,0 +1,114 @@
+"""OPE-27 smoke (4/4) — a long multi-turn session driven through the real SessionManager
+across REPEATED forced compactions: the provider must actually receive the compacted
+view (summary block + verbatim tail), user intent must survive every compaction, and the
+state must survive a save/rebuild mid-conversation. This is the scripted stand-in for
+the live-model smoke (which needs a configured provider key)."""
+
+import json
+
+import asyncio
+
+from coworker.providers import AssistantTurn, ModelCapabilities, ProviderClient
+from coworker.providers.base import TokenUsage
+from coworker.server.manager import SessionManager
+
+BULK = "analysis paragraph " * 400 # ~7.6k chars (~1.9k tokens) per turn → triggers by turn 2
+
+
+class LongSessionProvider(ProviderClient):
+ """Main turns: bulky text answers with realistic (growing) usage reporting.
+ Summarizer turns: a structured summary echoing the required sections."""
+
+ def __init__(self):
+ self.main_messages_seen: list[list[dict]] = []
+ self.summary_prompts: list[str] = []
+
+ def complete(self, *, model, messages, tools=None, **settings):
+ if messages and "compacting an AI coworker" in str(
+ messages[0].get("content", "")
+ ):
+ self.summary_prompts.append(str(messages[1]["content"]))
+ return AssistantTurn(
+ text=(
+ "## Primary request and intent\nBuild the Q3 report; never email "
+ "it without approval.\n## Current work\nDrafting section "
+ f"{len(self.summary_prompts)}.\n## Next step\nContinue drafting."
+ ),
+ finish_reason="stop",
+ )
+ self.main_messages_seen.append([dict(m) for m in messages])
+ # Usage mirrors the outbound size (chars/4), like a real provider would bill it.
+ prompt_tokens = sum(len(json.dumps(m, default=str)) for m in messages) // 4
+ return AssistantTurn(
+ text=f"turn {len(self.main_messages_seen)}: {BULK}",
+ finish_reason="stop",
+ usage=TokenUsage(input=prompt_tokens, output=500),
+ )
+
+ def capabilities(self, model):
+ return ModelCapabilities()
+
+
+def test_long_session_survives_repeated_compaction(tmp_path):
+ provider = LongSessionProvider()
+ mgr = SessionManager(workspace=tmp_path, provider=provider)
+ # Force tiny windows straight through the real Settings plumbing.
+ mgr._prefs["compaction_cap_tokens"] = 3_000
+ sid = "smoke-long"
+
+ boundaries = []
+
+ # ONE event loop for the whole session, like the real server — the engine's asyncio
+ # primitives bind to the loop they first run on, so a per-turn asyncio.run() would
+ # silently drop every stream after the first (found the hard way in the live smoke).
+ async def scenario():
+ engine = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path))
+ for i in range(8):
+ async for _ in engine.run(f"user step {i}: keep drafting the Q3 report"):
+ pass
+ # Every turn must produce a real reply — an empty assistant message means
+ # the stream got dropped, not answered.
+ last = next(
+ m for m in reversed(engine.messages) if m.get("role") == "assistant"
+ )
+ assert f"turn {i + 1}:" in str(last.get("content", ""))
+ if engine.compaction_state is not None:
+ if (
+ not boundaries
+ or engine.compaction_state.boundary_index != boundaries[-1]
+ ):
+ boundaries.append(engine.compaction_state.boundary_index)
+ mgr.save(sid, engine)
+ if i == 4: # mid-conversation restart: state must survive the rebuild
+ mgr._engines.pop(sid)
+ engine = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path))
+ assert engine.compaction_state is not None
+ return engine
+
+ engine = asyncio.run(scenario())
+
+ # Repeated compaction actually happened, moving forward each time.
+ assert len(boundaries) >= 2
+ assert boundaries == sorted(boundaries)
+ # Later summarizer calls fold the previous summary in (summary is message zero).
+ assert any("previous compaction summary" in p for p in provider.summary_prompts)
+
+ # What the MODEL actually received after the last compaction: the block + the tail,
+ # bounded — not the whole ever-growing canonical history.
+ final_view = provider.main_messages_seen[-1]
+ assert final_view[0]["role"] == "system"
+ block = final_view[1]["content"]
+ assert "" in block
+ assert "Q3 report" in block # the summary carries the intent
+ assert "user step 0" in block # mechanical user-message preservation, from turn 0
+ assert "do not recap" in block # the continuation contract
+ assert len(final_view) < len(engine.messages)
+
+ # Canonical transcript: untouched (every turn still present) + the divider notices.
+ texts = [str(m.get("content", "")) for m in engine.messages]
+ assert all(any(f"user step {i}" in t for t in texts) for i in range(8))
+ assert sum(1 for m in engine.messages if m.get("kind") == "compacted") >= 2
+
+ # The persisted record round-trips the final state.
+ record = mgr.session_store.load(sid)
+ assert record.compaction["boundary_index"] == engine.compaction_state.boundary_index