diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 49907e2c3316..58829dbf4fb2 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -75,6 +75,44 @@ _IMAGE_CHAR_EQUIVALENT = _IMAGE_TOKEN_ESTIMATE * _CHARS_PER_TOKEN _SUMMARY_FAILURE_COOLDOWN_SECONDS = 600 +# Hard ceiling for the deterministic summary-failure handoff. The fallback is +# only meant to preserve continuity anchors from the dropped window, not to +# become another unbounded transcript copy after the LLM summarizer failed. +_FALLBACK_SUMMARY_MAX_CHARS = 8_000 +_FALLBACK_TURN_MAX_CHARS = 700 + + +_PATH_MENTION_RE = re.compile(r"(?:/|~/?|[A-Za-z]:\\)[^\s`'\")\]}<>]+") + + +def _dedupe_append(items: list[str], value: str, *, limit: int) -> None: + value = value.strip() + if value and value not in items and len(items) < limit: + items.append(value) + + +def _extract_tool_call_name_and_args(tool_call: Any) -> tuple[str, str]: + """Return a best-effort ``(name, arguments)`` pair for dict/object tool calls.""" + if isinstance(tool_call, dict): + fn = tool_call.get("function") or {} + return str(fn.get("name") or "unknown"), str(fn.get("arguments") or "") + + fn = getattr(tool_call, "function", None) + if fn is None: + return "unknown", "" + return str(getattr(fn, "name", None) or "unknown"), str(getattr(fn, "arguments", None) or "") + + +def _extract_tool_call_id(tool_call: Any) -> str: + if isinstance(tool_call, dict): + return str(tool_call.get("id") or "") + return str(getattr(tool_call, "id", "") or "") + + +def _collect_path_mentions(text: str, relevant_files: list[str], *, limit: int = 12) -> None: + for match in _PATH_MENTION_RE.findall(text): + _dedupe_append(relevant_files, match.rstrip(".,:;"), limit=limit) + def _content_length_for_budget(raw_content: Any) -> int: """Return the effective char-length of a message's content for token budgeting. @@ -537,8 +575,8 @@ def __init__( self.quiet_mode = quiet_mode # When True, summary-generation failure aborts compression entirely # (returns messages unchanged, sets _last_compress_aborted=True). - # When False (default = historical behavior), insert a static - # "summary unavailable" placeholder and drop the middle window. + # When False (default = historical behavior), insert a + # deterministic "summary unavailable" handoff and drop the middle window. self.abort_on_summary_failure = abort_on_summary_failure self.context_length = get_model_context_length( @@ -884,6 +922,195 @@ def _serialize_for_summary(self, turns: List[Dict[str, Any]]) -> str: return "\n\n".join(parts) + def _build_static_fallback_summary( + self, + turns_to_summarize: List[Dict[str, Any]], + reason: str | None = None, + ) -> str: + """Build a deterministic handoff when the LLM summarizer is unavailable. + + This is intentionally much less rich than an LLM-written summary, but it + is still better than a bare "N messages were removed" marker. It keeps + the most useful continuity anchors that can be extracted locally: + recent user asks, assistant/tool actions, files/commands mentioned in + tool calls, and any error text. The result uses the normal summary + structure so downstream prompts can recover gracefully after a provider + outage or summary-model failure. + """ + user_asks: list[str] = [] + assistant_actions: list[str] = [] + tool_actions: list[str] = [] + relevant_files: list[str] = [] + blockers: list[str] = [] + last_dropped_turns: list[str] = [] + + def _compact_fallback_turn(value: Any) -> str: + text = redact_sensitive_text(_content_text_for_contains(value)) + text = re.sub(r"\bgh[pousr]_[A-Za-z0-9_]{8,}\b", "[REDACTED]", text) + text = re.sub(r"\s+", " ", text).strip() + if len(text) > _FALLBACK_TURN_MAX_CHARS: + text = text[: _FALLBACK_TURN_MAX_CHARS - 15].rstrip() + " ...[truncated]" + return re.sub(r"\bgh[pousr]_[A-Za-z0-9_.-]+", "[REDACTED]", text) + + def _remember_dropped_turn(label: str, text: str, *, limit: int = 8) -> None: + text = text.strip() + if not text: + return + last_dropped_turns.append(f"{label}: {text}") + if len(last_dropped_turns) > limit: + del last_dropped_turns[0] + + def _collect_paths_from_jsonish(obj: Any) -> None: + if isinstance(obj, dict): + for key, val in obj.items(): + if key in {"path", "workdir", "file_path", "output_path"} and isinstance(val, str): + _dedupe_append(relevant_files, val, limit=12) + _collect_paths_from_jsonish(val) + elif isinstance(obj, list): + for val in obj: + _collect_paths_from_jsonish(val) + elif isinstance(obj, str): + _collect_path_mentions(obj, relevant_files) + + call_id_to_tool: dict[str, tuple[str, str]] = {} + for msg in turns_to_summarize: + if msg.get("role") == "assistant" and msg.get("tool_calls"): + for tc in msg.get("tool_calls") or []: + name, raw_args = _extract_tool_call_name_and_args(tc) + args = redact_sensitive_text(raw_args) + call_id = _extract_tool_call_id(tc) + if call_id: + call_id_to_tool[call_id] = (name, args) + if args: + try: + parsed = json.loads(args) + except Exception: + parsed = args + _collect_paths_from_jsonish(parsed) + + for msg in turns_to_summarize: + role = msg.get("role", "unknown") + text = _compact_fallback_turn(msg.get("content")) + _collect_path_mentions(text, relevant_files) + + turn_text = text + turn_tool_names: list[str] = [] + if role == "assistant" and msg.get("tool_calls"): + for tc in msg.get("tool_calls") or []: + name, _args = _extract_tool_call_name_and_args(tc) + turn_tool_names.append(name) + if turn_tool_names: + prefix = "tool calls: " + ", ".join(turn_tool_names[:6]) + turn_text = f"{prefix}; {turn_text}" if turn_text else prefix + _remember_dropped_turn(str(role).upper(), turn_text) + + if len(text) > 600: + text = text[:420].rstrip() + " ... " + text[-160:].lstrip() + + if role == "user" and text: + user_asks.append(text) + elif role == "assistant": + tool_names: list[str] = [] + for tc in msg.get("tool_calls") or []: + name, _args = _extract_tool_call_name_and_args(tc) + tool_names.append(name) + if tool_names: + assistant_actions.append( + "Called tool(s): " + ", ".join(tool_names[:6]) + ) + elif text: + assistant_actions.append(text) + elif role == "tool": + call_id = str(msg.get("tool_call_id") or "") + tool_name, tool_args = call_id_to_tool.get(call_id, ("unknown", "")) + tool_actions.append( + _summarize_tool_result(tool_name, tool_args, text or "") + ) + if re.search( + r"\b(error|failed|exception|traceback|timeout|timed out|fatal)\b", + text, + re.I, + ): + blockers.append(text[:500]) + + def _bullets(items: list[str], limit: int = 8) -> str: + unique: list[str] = [] + seen: set[str] = set() + for item in items: + item = item.strip() + if not item or item in seen: + continue + seen.add(item) + unique.append(item) + if len(unique) >= limit: + break + return "\n".join(f"- {item}" for item in unique) if unique else "None." + + completed: list[str] = [] + for idx, item in enumerate((assistant_actions + tool_actions)[:12], start=1): + completed.append(f"{idx}. {item}") + + active_task = ( + f"User asked: {user_asks[-1]!r}" + if user_asks + else "Unknown from deterministic fallback." + ) + previous_summary_note = "" + if self._previous_summary: + previous_summary_note = ( + "\n\nPrevious compaction summary was present and should still be treated as " + "background continuity context, but the latest LLM summary update failed." + ) + + reason_text = f" Summary failure reason: {reason}." if reason else "" + body = f"""## Active Task +{active_task} + +## Goal +Recovered from a deterministic fallback because the LLM context summarizer was unavailable. Continue from the protected recent messages after this summary and use current file/system state for exact details.{previous_summary_note} + +## Constraints & Preferences +- This fallback was generated locally without an LLM summary call. +- Secrets and credentials were redacted before preservation. +- The summary may be incomplete; prefer verifying current files, git state, processes, and test results instead of assuming omitted details. + +## Completed Actions +{chr(10).join(completed) if completed else "None recoverable from compacted turns."} + +## Active State +Unknown from deterministic fallback. Inspect current repository/session state if needed. + +## In Progress +{active_task} + +## Blocked +{_bullets(blockers, limit=5)} + +## Key Decisions +None recoverable from deterministic fallback. + +## Resolved Questions +None recoverable from deterministic fallback. + +## Pending User Asks +{active_task} + +## Relevant Files +{_bullets(relevant_files, limit=12)} + +## Remaining Work +Continue from the most recent unfulfilled user ask and protected tail messages. Verify state with tools before making claims. + +## Last Dropped Turns +{_bullets(last_dropped_turns, limit=8)} + +## Critical Context +Summary generation was unavailable, so this is a best-effort deterministic fallback for {len(turns_to_summarize)} compacted message(s).{reason_text}""" + summary = self._with_summary_prefix(redact_sensitive_text(body.strip())) + if len(summary) > _FALLBACK_SUMMARY_MAX_CHARS: + summary = summary[: _FALLBACK_SUMMARY_MAX_CHARS - 42].rstrip() + "\n...[fallback summary truncated]" + return summary + def _fallback_to_main_for_compression(self, e: Exception, reason: str) -> None: """Switch from a separate ``summary_model`` back to the main model. @@ -911,7 +1138,11 @@ def _fallback_to_main_for_compression(self, e: Exception, reason: str) -> None: self.summary_model = "" # empty = use main model self._summary_failure_cooldown_until = 0.0 # no cooldown — retry immediately - def _generate_summary(self, turns_to_summarize: List[Dict[str, Any]], focus_topic: str = None) -> Optional[str]: + def _generate_summary( + self, + turns_to_summarize: List[Dict[str, Any]], + focus_topic: Optional[str] = None, + ) -> Optional[str]: """Generate a structured summary of conversation turns. Uses a structured template (Goal, Progress, Decisions, Resolved/Pending @@ -1608,9 +1839,9 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f # True → ABORT compression entirely. Return messages unchanged # and set _last_compress_aborted=True so callers can warn # the user and stop the auto-compress retry loop. - # False → Fall through to the legacy fallback path below: insert - # a static "summary unavailable" placeholder and drop the - # middle window. Records _last_summary_fallback_used / + # False → Fall through to the default fallback path below: insert + # a deterministic "summary unavailable" handoff and drop + # the middle window. Records _last_summary_fallback_used / # _last_summary_dropped_count for gateway hygiene to # surface a warning. # Default is False (historical behavior). @@ -1643,21 +1874,18 @@ def compress(self, messages: List[Dict[str, Any]], current_tokens: int = None, f ) compressed.append(msg) - # Legacy fallback path: LLM summary failed and abort_on_summary_failure - # is False (the default). Insert a static placeholder so the model - # knows context was lost rather than silently dropping everything. + # If LLM summary failed, insert a deterministic fallback so the model + # gets at least locally recoverable continuity anchors instead of a + # content-free "N messages were removed" marker. if not summary: if not self.quiet_mode: - logger.warning("Summary generation failed — inserting static fallback context marker") + logger.warning("Summary generation failed — inserting deterministic fallback context summary") n_dropped = compress_end - compress_start self._last_summary_dropped_count = n_dropped self._last_summary_fallback_used = True - summary = ( - f"{SUMMARY_PREFIX}\n" - f"Summary generation was unavailable. {n_dropped} message(s) were " - f"removed to free context space but could not be summarized. The removed " - f"messages contained earlier work in this session. Continue based on the " - f"recent messages below and the current state of any files or resources." + summary = self._build_static_fallback_summary( + turns_to_summarize, + reason=self._last_summary_error, ) _merge_summary_into_tail = False diff --git a/tests/agent/test_context_compressor.py b/tests/agent/test_context_compressor.py index dca10bb44627..0d7aa81f41fd 100644 --- a/tests/agent/test_context_compressor.py +++ b/tests/agent/test_context_compressor.py @@ -67,6 +67,7 @@ def test_too_few_messages_returns_unchanged(self, compressor): def test_truncation_fallback_no_client(self, compressor): # Simulate "no summarizer available" explicitly. call_llm can otherwise # discover the developer's real auxiliary credentials from auth state. + # The failed summary should use the deterministic fallback path. msgs = [{"role": "system", "content": "System prompt"}] + self._make_messages(10) with patch("agent.context_compressor.call_llm", side_effect=RuntimeError("no provider")): result = compressor.compress(msgs) @@ -78,6 +79,64 @@ def test_truncation_fallback_no_client(self, compressor): assert compressor._last_compress_aborted is False assert compressor._last_summary_fallback_used is True + def test_summary_failure_uses_deterministic_fallback_with_recovered_context(self): + """Regression: failed LLM summaries should not emit a content-free marker. + + The fallback should preserve locally recoverable continuity details so a + future turn does not see only "messages were removed" after compaction. + """ + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor( + model="test/model", + protect_first_n=1, + protect_last_n=2, + quiet_mode=True, + ) + + msgs = [ + {"role": "system", "content": "System prompt"}, + {"role": "user", "content": "Please fix the compression summary failure"}, + { + "role": "assistant", + "content": None, + "tool_calls": [{ + "id": "call_1", + "type": "function", + "function": { + "name": "read_file", + "arguments": '{"path":"agent/context_compressor.py","offset":1}', + }, + }], + }, + { + "role": "tool", + "tool_call_id": "call_1", + "content": "read agent/context_compressor.py and found static fallback marker", + }, + {"role": "assistant", "content": "I found the issue."}, + {"role": "user", "content": "latest protected ask"}, + {"role": "assistant", "content": "ok"}, + ] + + with ( + patch.object(c, "_find_tail_cut_by_tokens", return_value=5), + patch( + "agent.context_compressor.call_llm", + side_effect=RuntimeError("provider down"), + ), + ): + result = c.compress(msgs) + + combined = "\n".join(str(m.get("content", "")) for m in result) + assert "## Active Task" in combined + assert "Please fix the compression summary failure" in combined + assert "read_file" in combined + assert "agent/context_compressor.py" in combined + assert "Summary generation was unavailable" in combined + assert "removed to free context space but could not be summarized" not in combined + assert c._last_summary_fallback_used is True + assert c._last_summary_dropped_count == 3 + def test_compression_increments_count(self, compressor): msgs = self._make_messages(10) # Default config (abort_on_summary_failure=False) — fallback path @@ -756,6 +815,123 @@ def test_compress_records_fallback_and_dropped_count_on_summary_failure(self): for m in result ) + def test_summary_failure_fallback_preserves_tool_paths_and_redacts_secret_context(self): + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=1, protect_last_n=1) + + secret = "ghp_" + ("a" * 36) + msgs = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": f"Fix /tmp/project/app.py and never leak {secret}"}, + { + "role": "assistant", + "content": "I will inspect it.", + "tool_calls": [ + { + "id": "call-1", + "function": { + "name": "read_file", + "arguments": '{"path":"/tmp/project/app.py"}', + }, + } + ], + }, + {"role": "tool", "tool_call_id": "call-1", "content": f"read /tmp/project/app.py with token {secret}"}, + {"role": "assistant", "content": "Found the bug in /tmp/project/app.py"}, + {"role": "user", "content": "Patch it after this"}, + {"role": "assistant", "content": "Ready to patch"}, + {"role": "user", "content": "current live request should stay in tail"}, + ] + + with patch("agent.context_compressor.call_llm", side_effect=Exception("timeout")): + result = c.compress(msgs) + + fallback = next(m["content"] for m in result if "Summary generation was unavailable" in m.get("content", "")) + assert "Called tool(s): read_file" in fallback + assert "/tmp/project/app.py" in fallback + assert secret not in fallback + assert "ghp_" not in fallback + + def test_summary_failure_fallback_supports_object_tool_calls_and_content_path_mentions(self): + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=1, protect_last_n=1) + + tool_call = MagicMock() + tool_call.id = "call-object" + tool_call.function.name = "terminal" + tool_call.function.arguments = '{"command":"python /repo/scripts/fix.py", "workdir":"/repo"}' + msgs = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "Review ~/src/pkg/module.py before editing"}, + {"role": "assistant", "content": "Running command", "tool_calls": [tool_call]}, + {"role": "tool", "tool_call_id": "call-object", "content": "Traceback in /repo/src/pkg/module.py: boom"}, + {"role": "assistant", "content": "Need to update C:\\work\\pkg\\module.py too"}, + {"role": "user", "content": "Patch ~/src/pkg/module.py after checking those files"}, + {"role": "assistant", "content": "Ready to patch"}, + {"role": "user", "content": "tail task"}, + ] + + with patch("agent.context_compressor.call_llm", side_effect=Exception("timeout")): + result = c.compress(msgs) + + fallback = next(m["content"] for m in result if "Summary generation was unavailable" in m.get("content", "")) + assert "Called tool(s): terminal" in fallback + assert "/repo/scripts/fix.py" in fallback + assert "/repo" in fallback + assert "/repo/src/pkg/module.py" in fallback + assert "C:\\work\\pkg\\module.py" in fallback + assert "Traceback" in fallback + assert "## Last Dropped Turns" in fallback + assert "TOOL: Traceback in /repo/src/pkg/module.py: boom" in fallback + + def test_summary_failure_fallback_preserves_last_dropped_turns_without_tail(self): + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=1, protect_last_n=1) + + msgs = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "Investigate dropped-window request in /tmp/active.py"}, + {"role": "assistant", "content": "I inspected /tmp/active.py and found the failing branch"}, + {"role": "tool", "tool_call_id": "call-old", "content": "ValueError: boom in /tmp/active.py"}, + {"role": "assistant", "content": "Next step is patching /tmp/active.py"}, + {"role": "user", "content": "Confirm regression coverage for /tmp/active.py"}, + {"role": "assistant", "content": "Regression note is ready"}, + {"role": "user", "content": "protected tail request must not be copied from dropped window"}, + ] + + with patch("agent.context_compressor.call_llm", side_effect=Exception("timeout")): + result = c.compress(msgs) + + fallback = next(m["content"] for m in result if "Summary generation was unavailable" in m.get("content", "")) + assert "## Last Dropped Turns" in fallback + assert "ASSISTANT: I inspected /tmp/active.py and found the failing branch" in fallback + assert "TOOL: ValueError: boom in /tmp/active.py" in fallback + assert "protected tail request must not be copied" not in fallback + + def test_summary_failure_fallback_is_bounded(self): + with patch("agent.context_compressor.get_model_context_length", return_value=100000): + c = ContextCompressor(model="test", quiet_mode=True, protect_first_n=1, protect_last_n=1) + + long_text = "important detail " * 2000 + msgs = [ + {"role": "system", "content": "sys"}, + {"role": "user", "content": "head user"}, + {"role": "assistant", "content": "head assistant"}, + {"role": "user", "content": long_text}, + {"role": "assistant", "content": long_text}, + {"role": "user", "content": long_text}, + {"role": "assistant", "content": long_text}, + {"role": "user", "content": "tail"}, + ] + + with patch("agent.context_compressor.call_llm", side_effect=Exception("timeout")): + result = c.compress(msgs) + + fallback = next(m["content"] for m in result if "Summary generation was unavailable" in m.get("content", "")) + assert len(fallback) <= 8300 + assert "deterministic fallback" in fallback + assert "important detail" in fallback + def test_compress_clears_fallback_flag_on_subsequent_success(self): mock_response = MagicMock() mock_response.choices = [MagicMock()]