From f9f51c97c6b9a3e2fbbb8f99600ffbe1d22228df Mon Sep 17 00:00:00 2001 From: Rohit C Prasad Date: Thu, 30 Jul 2026 06:24:39 -0700 Subject: [PATCH] compaction: live progress signal + user-message cap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit COMPACTING event drives a 'Compacting context…' transient in the GUI. Cap the compacted block's user-message list at 40 with an honest omitted count. --- coworker/compaction.py | 37 ++++++++++++++++++++++-- coworker/engine.py | 45 +++++++++++++++++++---------- coworker/events.py | 1 + surfaces/gui/e2e/compaction.spec.ts | 5 ++++ surfaces/gui/e2e/fixtures.ts | 14 +++++---- surfaces/gui/src/App.tsx | 18 ++++++++++-- surfaces/gui/src/types.ts | 1 + tests/test_compaction.py | 31 ++++++++++++++++++++ tests/test_compaction_engine.py | 17 +++++++++++ 9 files changed, 144 insertions(+), 25 deletions(-) diff --git a/coworker/compaction.py b/coworker/compaction.py index ec0ec25a..678682a5 100644 --- a/coworker/compaction.py +++ b/coworker/compaction.py @@ -34,7 +34,11 @@ SUMMARY_MAX_TOKENS = 3_000 _SPAN_TOOL_RESULT_CLIP = 400 _SPAN_BUDGET_CHARS = 400_000 # User messages preserved mechanically in the compacted block ("trimmed of pasted bulk"). +# The list is capped to the newest N across repeated compactions — otherwise it appends +# forever and the block slowly reclaims the window it freed. Dropped ones stay counted +# (their intent lives in the summary, which is asked to list user messages too). _USER_MESSAGE_CLIP = 600 +_USER_MESSAGES_MAX = 40 _TRIM_FRACTION = 0.10 @@ -88,6 +92,9 @@ class CompactionState: summary_text: str working_state: str user_messages: list[str] = field(default_factory=list) + # How many older user messages were dropped by the _USER_MESSAGES_MAX cap, across + # all compactions of this session — keeps the block's "N earlier omitted" honest. + user_messages_dropped: int = 0 created_at: float = 0.0 model_used: str = "" trimmed: bool = False # True when this state came from the no-summary trim fallback @@ -98,6 +105,7 @@ class CompactionState: "summary_text": self.summary_text, "working_state": self.working_state, "user_messages": list(self.user_messages), + "user_messages_dropped": self.user_messages_dropped, "created_at": self.created_at, "model_used": self.model_used, "trimmed": self.trimmed, @@ -112,6 +120,7 @@ class CompactionState: 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 []], + user_messages_dropped=int(raw.get("user_messages_dropped", 0)), created_at=float(raw.get("created_at", 0.0)), model_used=str(raw.get("model_used", "")), trimmed=bool(raw.get("trimmed", False)), @@ -289,6 +298,15 @@ def extract_user_messages( return out +def _cap_user_messages( + messages: list[str], *, prior_dropped: int, limit: int = _USER_MESSAGES_MAX +) -> tuple[list[str], int]: + """Newest-`limit` slice plus the running total of everything ever dropped.""" + if len(messages) <= limit: + return messages, prior_dropped + return messages[-limit:], prior_dropped + (len(messages) - limit) + + # -- 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. @@ -419,11 +437,16 @@ def build_state( span, prior_summary=prior.summary_text if prior is not None else "", ) + users, dropped = _cap_user_messages( + prior_users + extract_user_messages(span), + prior_dropped=prior.user_messages_dropped if prior is not None else 0, + ) return CompactionState( boundary_index=boundary, summary_text=summary, working_state=extract_working_state(span), - user_messages=prior_users + extract_user_messages(span), + user_messages=users, + user_messages_dropped=dropped, created_at=time.time(), model_used=model, ) @@ -459,11 +482,16 @@ def trim_state( + "(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.)" ) + users, dropped = _cap_user_messages( + prior_users + extract_user_messages(span), + prior_dropped=prior.user_messages_dropped if prior is not None else 0, + ) return CompactionState( boundary_index=boundary, summary_text=summary, working_state=extract_working_state(span), - user_messages=prior_users + extract_user_messages(span), + user_messages=users, + user_messages_dropped=dropped, created_at=time.time(), model_used="", trimmed=True, @@ -483,6 +511,11 @@ def compacted_block(state: CompactionState) -> str: parts += ["", state.working_state] if state.user_messages: parts += ["", "## User messages in the compacted span (verbatim, chronological)"] + if state.user_messages_dropped: + parts += [ + f"({state.user_messages_dropped} earlier user messages omitted — " + "their intent is covered by the summary above)" + ] parts += [f"- {u}" for u in state.user_messages] parts += ["", CONTINUATION_CONTRACT, ""] return "\n".join(parts) diff --git a/coworker/engine.py b/coworker/engine.py index f3379d61..ae34d4a0 100644 --- a/coworker/engine.py +++ b/coworker/engine.py @@ -312,8 +312,13 @@ class TurnEngine: 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() + # turn's first call. Deliberately no "wrap up" warning to the model. The + # COMPACTING signal precedes the (multi-second) summarizer call so surfaces + # can show progress instead of a silent stall. + notice = None + if self._compaction_due(): + yield Event(EventType.COMPACTING, {}) + notice = await self._compact_now() if notice: self._append_notice("compacted", notice) yield Event(EventType.COMPACTED, {"text": notice}) @@ -350,6 +355,7 @@ class TurnEngine: # 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(): + yield Event(EventType.COMPACTING, {}) notice = await self._compact_now(force=True) if notice: self._append_notice("compacted", notice) @@ -427,25 +433,32 @@ class TurnEngine: cfg.setdefault("cap_tokens", _compaction.DEFAULT_CAP_TOKENS) return cfg + def _compaction_due(self) -> bool: + """The trigger check alone — cheap and side-effect free, so the loop can emit + the COMPACTING signal before committing to the (slow) summarizer call.""" + cfg = self._compaction_config() + if cfg.get("enabled") is False: + return False + signal = self._last_context_tokens or _compaction.estimate_tokens( + self._outbound_messages() + ) + return _compaction.should_compact( + signal, + cfg.get("context_window"), + threshold_pct=float(cfg["threshold_pct"]), + cap_tokens=int(cfg["cap_tokens"]), + ) + 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).""" + """Run the compaction policy. Callers gate on `_compaction_due()` (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) diff --git a/coworker/events.py b/coworker/events.py index 0b146485..cdb9fe00 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" + COMPACTING = "compacting" # compaction started — surfaces show a transient signal COMPACTED = "compacted" # outbound history was compacted (summary or trim) diff --git a/surfaces/gui/e2e/compaction.spec.ts b/surfaces/gui/e2e/compaction.spec.ts index 8a3bd0e6..c5fa0b61 100644 --- a/surfaces/gui/e2e/compaction.spec.ts +++ b/surfaces/gui/e2e/compaction.spec.ts @@ -64,9 +64,14 @@ test("the compacted divider renders mid-session and the transcript stays intact" await box.fill("compact the context"); await box.press("Enter"); + // The transient signal shows while the summarizer runs, then yields to the divider. + await expect(page.getByText("Compacting context…").first()).toBeVisible({ + timeout: 10_000, + }); await expect( page.getByText("Context compacted — earlier turns were summarized").first(), ).toBeVisible({ timeout: 10_000 }); + await expect(page.getByText("Compacting context…")).toHaveCount(0); await expect( page.getByText("Still on it — continuing where I left off.").first(), ).toBeVisible(); diff --git a/surfaces/gui/e2e/fixtures.ts b/surfaces/gui/e2e/fixtures.ts index 037fc617..73776136 100644 --- a/surfaces/gui/e2e/fixtures.ts +++ b/surfaces/gui/e2e/fixtures.ts @@ -681,12 +681,16 @@ 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. + // Auto-compaction (OPE-27): the server signals `compacting` (the transient + // spinner label), summarizes for a beat, then emits the marker and 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"); + send("compacting", {}); + setTimeout(() => { + send("compacted", { text: "Context compacted — earlier turns were summarized" }); + send("assistant_message", { text: "Still on it — continuing where I left off." }); + send("turn_done"); + }, 400); return; } // A turn that dies on a provider error; the follow-up {type:"retry"} recovers. diff --git a/surfaces/gui/src/App.tsx b/surfaces/gui/src/App.tsx index f814cdb8..dd78ad6d 100644 --- a/surfaces/gui/src/App.tsx +++ b/surfaces/gui/src/App.tsx @@ -172,6 +172,10 @@ export function App() { const [mode, setMode] = useState("interactive"); const [connected, setConnected] = useState(false); const [running, setRunning] = useState(false); + // Transient "Compacting context…" indicator (OPE-27): set by the `compacting` event, + // cleared by whatever the engine emits next — the summarizer call is otherwise a + // multi-second silent stall mid-turn. + const [compacting, setCompacting] = useState(false); const [items, setItems] = useState([]); const [streaming, setStreamingState] = useState(""); // Ref mirror of `streaming`: the WS handler closure is built once per socket and can't read @@ -576,6 +580,9 @@ export function App() { }, ]); }; + // Any engine event after `compacting` means the summarizer finished (compacted / + // silent no-op / failure prompt) — the transient must never outlive it. + if (ev.type !== "compacting") setCompacting(false); switch (ev.type) { case "ready": setConnected(true); @@ -709,6 +716,9 @@ export function App() { if (d.model) setModel(d.model); setItems((p) => [...p, { kind: "notice", tone: "info", text: d.text || "Model switched" }]); break; + case "compacting": + setCompacting(true); + 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. @@ -1519,7 +1529,11 @@ export function App() { )} + {/* Compaction runs between provider turns (nothing streams during it), so + the transient takes over the waiting slot with a specific label. */} + {running && compacting && } {running && + !compacting && !reasoningStream && (!streaming || streamMode(streaming, items, running) === "hold") && !lastItemIsAssistant(items) && } @@ -1685,12 +1699,12 @@ function lastItemIsAssistant(items: Item[]): boolean { return false; } -function WaitingForAgent() { +function WaitingForAgent({ label }: { label?: string }) { return (
- Waiting for agent... + {label || "Waiting for agent..."}
); diff --git a/surfaces/gui/src/types.ts b/surfaces/gui/src/types.ts index a05a37ab..28fca293 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" + | "compacting" | "compacted" | "turn_done"; diff --git a/tests/test_compaction.py b/tests/test_compaction.py index 1d25f18e..a480bfb6 100644 --- a/tests/test_compaction.py +++ b/tests/test_compaction.py @@ -313,3 +313,34 @@ def test_is_context_overflow(): 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")) + + +def test_user_messages_capped_across_repeated_compactions(): + # The mechanical user-message list must not grow forever — newest _USER_MESSAGES_MAX + # survive, the rest stay counted so the block's "omitted" note is honest. + from coworker.compaction import _USER_MESSAGES_MAX + + msgs = [{"role": "system", "content": "s"}] + for i in range(120): + msgs.append({"role": "user", "content": f"ask {i}"}) + msgs.append({"role": "assistant", "content": f"answer {i}"}) + + state = None + while True: + nxt = trim_state(msgs, prior=state, fraction=0.4) + if nxt is None: + break + state = nxt + + assert state is not None + assert len(state.user_messages) <= _USER_MESSAGES_MAX + assert state.user_messages_dropped > 0 + assert state.user_messages[-1].startswith("ask") # newest survive, oldest dropped + + block = compacted_block(state) + assert f"{state.user_messages_dropped} earlier user messages omitted" in block + + restored = CompactionState.from_dict(state.as_dict()) + assert restored is not None + assert restored.user_messages_dropped == state.user_messages_dropped + assert restored.user_messages == state.user_messages diff --git a/tests/test_compaction_engine.py b/tests/test_compaction_engine.py index 99e1270c..aabd0651 100644 --- a/tests/test_compaction_engine.py +++ b/tests/test_compaction_engine.py @@ -256,3 +256,20 @@ def test_compaction_state_survives_save_and_rebuild(tmp_path): rebuilt = mgr.get_engine(sid, agent="cowork", workspace=str(tmp_path)) assert rebuilt.compaction_state == engine.compaction_state + + +def test_compacting_signal_precedes_the_compacted_marker(tmp_path): + # The transient-progress contract: COMPACTING fires before the (slow) summarizer + # call, COMPACTED after — surfaces key the "Compacting context…" spinner on it. + provider = CompactingProvider([AssistantTurn(text="done", finish_reason="stop")]) + engine = make_engine(tmp_path, provider, messages=long_history(), cap=400) + events = collect(engine) + + types = [e.type for e in events] + assert EventType.COMPACTING in types + assert types.index(EventType.COMPACTING) < types.index(EventType.COMPACTED) + # The signal is not persisted — only the compacted marker lands in the transcript. + assert not any( + m.get("role") == "notice" and m.get("kind") == "compacting" + for m in engine.messages + )