diff --git a/agent/context_compressor.py b/agent/context_compressor.py index 7dd1e8967316..7a41fbca6ca4 100644 --- a/agent/context_compressor.py +++ b/agent/context_compressor.py @@ -40,6 +40,24 @@ estimate_tokens_rough, ) from agent.redact import redact_sensitive_text +from agent.context_compressor_text_utils import ( + _content_text_for_contains, + _redact_compaction_text, +) +from agent.context_compressor_skill_prune import ( # noqa: E402 + SKILL_PRUNED_MARKER_PREFIX, + _MAX_PRUNED_SKILL_MARKERS, + _PRUNED_SKILLS_SECTION_HEADING, + _SKILL_PRUNE_RECENT_WINDOW, + _SKILL_PRUNED_MARKER_RE, + _SKILL_VIEW_PRUNE_MIN_CHARS, + _collect_ghosted_skill_names, + _collect_protected_skill_names, + _extract_pruned_skill_names, + _reinject_pruned_skill_markers, + _skill_pruned_marker, + _skill_view_call_sites, +) from agent.turn_context import drop_stale_api_content from tools.todo_tool import TODO_INJECTION_HEADER @@ -398,221 +416,6 @@ def _strip_persistence_markers(messages: List[Dict[str, Any]]) -> None: # Placeholder used when pruning old tool results _PRUNED_TOOL_PLACEHOLDER = "[Old tool output cleared to save context space]" -# Ghost-skill defense (#32106): when compaction reduces an old ``skill_view`` -# result to a 1-line metadata summary, the model still believes the skill is -# loaded even though its instructions are gone. The marker below is the ONE -# canonical prune signal — ``_skill_pruned_marker()`` builds it and every -# presence check matches against the same string, so the emit side and the -# check side can never drift apart (the original PR #44166 emitted -# ``[SKILL_PRUNED:`` but presence-checked ``[SKILL_PRUNED]``, making -# re-injection fire even when the marker had survived). -SKILL_PRUNED_MARKER_PREFIX = "[SKILL_PRUNED:" -# skill_view results at or below this size stay verbatim in pruned -# summaries — small skills are cheap to keep and their loss is unlikely to -# ghost the model. Shared by the emit site and the summarizer-input scan. -_SKILL_VIEW_PRUNE_MIN_CHARS = 5000 -# Cap for the deterministic marker re-injection list — keeps a very long -# session from growing an unbounded "## Pruned Skills" block in every -# iterative summary update. Newest-referenced skills win. -_MAX_PRUNED_SKILL_MARKERS = 20 - - -def _skill_pruned_marker(skill_name: str) -> str: - """Return the canonical prune marker for *skill_name*. - - Used verbatim by BOTH the emit sites (tool-result summarization, - summary re-injection) and the survival check in - ``_reinject_pruned_skill_markers`` — one string, no drift. - """ - return ( - f"{SKILL_PRUNED_MARKER_PREFIX} content lost in compression; " - f"reload with skill_view(name='{skill_name}')]" - ) - - -# Matches the canonical marker and captures the skill name. Anchored on the -# shared prefix constant so a wording change to the marker body updates the -# emit helper and this extractor together. -_SKILL_PRUNED_MARKER_RE = re.compile( - re.escape(SKILL_PRUNED_MARKER_PREFIX) - + r"[^\]]*?reload with skill_view\(name='([^']+)'\)" -) - - -def _extract_pruned_skill_names(text: str) -> list[str]: - """Return skill names referenced by prune markers in *text*, in order.""" - names: list[str] = [] - for match in _SKILL_PRUNED_MARKER_RE.finditer(text or ""): - name = match.group(1) - if name not in names: - names.append(name) - return names - - -def _collect_ghosted_skill_names(turns: List[Dict[str, Any]]) -> list[str]: - """Skill names whose instructions are about to be lost in compaction. - - Covers BOTH shapes a compacted middle window can carry: - - - a ``skill_view`` result already demoted by Phase-1 pruning — the - canonical ``[SKILL_PRUNED: ...]`` marker is in the row content; - - a RAW ``skill_view`` body that was never demoted (it sat inside the - protected tail of an earlier prune, then aged into the compression - window). The summarizer will paraphrase the instructions away, which - is exactly the ghost-skill failure — so it needs a marker too. - """ - names: list[str] = [] - - def _add(name: str) -> None: - if name and name not in names: - names.append(name) - - call_id_to_skill: dict[str, str] = {} - for idx, skill in _skill_view_call_sites(turns): - msg = turns[idx] - for tc in msg.get("tool_calls") or []: - tc_fn = tc.get("function", {}) if isinstance(tc, dict) else getattr(tc, "function", None) - tc_name = tc_fn.get("name", "") if isinstance(tc_fn, dict) else getattr(tc_fn, "name", "") - if tc_name != "skill_view": - continue - cid = tc.get("id", "") if isinstance(tc, dict) else (getattr(tc, "id", "") or "") - if cid: - call_id_to_skill[cid] = skill - for msg in turns: - content = msg.get("content") - text = content if isinstance(content, str) else _content_text_for_contains(content) - for name in _extract_pruned_skill_names(text): - _add(name) - if ( - msg.get("role") == "tool" - and isinstance(content, str) - and len(content) > _SKILL_VIEW_PRUNE_MIN_CHARS - ): - skill = call_id_to_skill.get(str(msg.get("tool_call_id") or "")) - if skill: - _add(skill) - return names - - -_PRUNED_SKILLS_SECTION_HEADING = "## Pruned Skills" - - -def _reinject_pruned_skill_markers(summary: str, skill_names: list[str]) -> str: - """Deterministically restore prune markers the summarizer dropped. - - ``skill_names`` was extracted from the summarizer INPUT before the LLM - call. For every skill whose canonical marker (``_skill_pruned_marker``) - is absent from the model's output, append it under a ``## Pruned - Skills`` section. Presence is checked against the SAME canonical string - the emit sites produce — a paraphrased or renamed marker counts as - dropped and is restored (the original PR checked the literal - ``[SKILL_PRUNED]``, which never matches the emitted ``[SKILL_PRUNED:`` - form, so it duplicated markers that HAD survived). - - The appended block is plain body text: it never carries a handoff - prefix, the merged-summary delimiter, or a start-of-content scaffolding - marker, so ``classify_summary_content`` / todo-snapshot flag handling - are unaffected. The block is routed through ``_redact_compaction_text`` - like every other compaction-boundary text. - """ - if not skill_names: - return summary - missing = [ - name for name in skill_names - if _skill_pruned_marker(name) not in summary - ] - if not missing: - return summary - lines = [_skill_pruned_marker(name) for name in missing] - block = ( - "\n\n" + _PRUNED_SKILLS_SECTION_HEADING + "\n" - + "\n".join(lines) - + "\n(The listed skills' instructions were pruned during context " - "compression. Reload with the skill_view call in each marker before " - "relying on that skill; one reload per skill is enough — ignore any " - "older markers for the same skill.)" - ) - return summary + _redact_compaction_text(block) - - -# A skill_view call within this many trailing messages counts as "just -# loaded": its full instruction body must survive the Phase-1 prune even when -# the token-budget boundary would otherwise demote it (#32106). Distinct from -# the protected-tail boundary, which is token-based and can land immediately -# after a bulky just-loaded skill body. -_SKILL_PRUNE_RECENT_WINDOW = 10 - - -def _skill_view_call_sites( - messages: List[Dict[str, Any]], -) -> list[tuple[int, str]]: - """Yield ``(message_index, skill_name)`` for every skill_view tool call.""" - sites: list[tuple[int, str]] = [] - for i, msg in enumerate(messages): - if msg.get("role") != "assistant": - continue - for tc in msg.get("tool_calls") or []: - if isinstance(tc, dict): - fn = tc.get("function", {}) - name = fn.get("name", "") if isinstance(fn, dict) else "" - args_str = fn.get("arguments", "") if isinstance(fn, dict) else "" - else: - fn = getattr(tc, "function", None) - name = getattr(fn, "name", "") if fn else "" - args_str = getattr(fn, "arguments", "") if fn else "" - if name != "skill_view" or not isinstance(args_str, str) or not args_str: - continue - try: - args = json.loads(args_str) - except (json.JSONDecodeError, TypeError): - continue - if isinstance(args, dict): - skill = args.get("name", "") - if isinstance(skill, str) and skill: - sites.append((i, skill)) - return sites - - -def _collect_protected_skill_names( - messages: List[Dict[str, Any]], prune_boundary: int, -) -> set[str]: - """Skill names whose skill_view bodies must survive Phase-1 demotion. - - A skill is protected (lower-cased set) when any of these hold: - - - its most recent ``skill_view`` call sits within the last - ``_SKILL_PRUNE_RECENT_WINDOW`` messages (just loaded / just reloaded); - - its most recent ``skill_view`` call sits inside the protected tail - (at or after *prune_boundary*); - - its name is mentioned in a user message inside the protected tail - (the user is actively steering work that depends on it). - - Protection applies to the ordinary Phase-1/2 prune only. The Pass-4 - pressure demotion deliberately ignores it: when the protected region - itself exceeds the soft budget, exempting skill bodies would recreate - the #61932 dead-end shape. - """ - total = len(messages) - if not total: - return set() - recent_start = max(0, total - _SKILL_PRUNE_RECENT_WINDOW) - tail_start = max(0, prune_boundary) - tail_user_texts: list[str] = [] - for msg in messages[tail_start:]: - if msg.get("role") != "user": - continue - content = msg.get("content") - if isinstance(content, str) and content: - tail_user_texts.append(content.lower()) - protected: set[str] = set() - for idx, skill in _skill_view_call_sites(messages): - key = skill.lower() - if idx >= recent_start or idx >= tail_start: - protected.add(key) - elif any(key in text for text in tail_user_texts): - protected.add(key) - return protected - # Chars per token rough estimate _CHARS_PER_TOKEN = 4 # Flat token cost per attached image part. Real cost varies by provider and @@ -676,27 +479,6 @@ def _collect_protected_skill_names( ) -def _redact_compaction_text(text: Any) -> str: - """Redact text that crosses a compaction summary boundary. - - Compaction summaries persist across sessions and are re-injected into - every subsequent summarizer prompt, so this boundary uses strict mode: - - - ``force=True`` — deliberately overrides ``security.redact_secrets: - false``. That opt-out targets *live tool output* (e.g. working on the - redactor itself); a summary is a persistence boundary where a leaked - credential keeps re-entering prompts indefinitely. - - ``redact_url_credentials=True`` — OAuth callback codes, magic-link - tokens, and URL userinfo never need to survive summarization the way - they must survive live navigation flows. - """ - return redact_sensitive_text( - text or "", - force=True, - redact_url_credentials=True, - ) - - 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: @@ -867,29 +649,6 @@ def _estimate_msg_budget_tokens(msg: dict) -> int: return tokens -def _content_text_for_contains(content: Any) -> str: - """Return a best-effort text view of message content. - - Used only for substring checks when we need to know whether we've already - appended a note to a message. Keeps multimodal lists intact elsewhere. - """ - if content is None: - return "" - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for item in content: - if isinstance(item, str): - parts.append(item) - elif isinstance(item, dict): - text = item.get("text") - if isinstance(text, str): - parts.append(text) - return "\n".join(part for part in parts if part) - return str(content) - - def _append_text_to_content(content: Any, text: str, *, prepend: bool = False) -> Any: """Append or prepend plain text to message content safely. diff --git a/agent/context_compressor_skill_prune.py b/agent/context_compressor_skill_prune.py new file mode 100644 index 000000000000..ccca302bbb8c --- /dev/null +++ b/agent/context_compressor_skill_prune.py @@ -0,0 +1,236 @@ +"""Skill-prune / ghost-skill defense helpers extracted from context_compressor. + +Ghost-skill defense (#32106): when compaction reduces an old ``skill_view`` +result to a 1-line metadata summary, the model still believes the skill is +loaded even though its instructions are gone. This module owns the canonical +prune marker, extraction, re-injection, and protected-name collection. + +Part of #78645 + #78647. +""" +from __future__ import annotations + +import json +import re +from typing import Any, Dict, List + +from agent.context_compressor_text_utils import ( + _content_text_for_contains, + _redact_compaction_text, +) + + + +# Ghost-skill defense (#32106): when compaction reduces an old ``skill_view`` +# result to a 1-line metadata summary, the model still believes the skill is +# loaded even though its instructions are gone. The marker below is the ONE +# canonical prune signal — ``_skill_pruned_marker()`` builds it and every +# presence check matches against the same string, so the emit side and the +# check side can never drift apart (the original PR #44166 emitted +# ``[SKILL_PRUNED:`` but presence-checked ``[SKILL_PRUNED]``, making +# re-injection fire even when the marker had survived). +SKILL_PRUNED_MARKER_PREFIX = "[SKILL_PRUNED:" +# skill_view results at or below this size stay verbatim in pruned +# summaries — small skills are cheap to keep and their loss is unlikely to +# ghost the model. Shared by the emit site and the summarizer-input scan. +_SKILL_VIEW_PRUNE_MIN_CHARS = 5000 +# Cap for the deterministic marker re-injection list — keeps a very long +# session from growing an unbounded "## Pruned Skills" block in every +# iterative summary update. Newest-referenced skills win. +_MAX_PRUNED_SKILL_MARKERS = 20 + + +def _skill_pruned_marker(skill_name: str) -> str: + """Return the canonical prune marker for *skill_name*. + + Used verbatim by BOTH the emit sites (tool-result summarization, + summary re-injection) and the survival check in + ``_reinject_pruned_skill_markers`` — one string, no drift. + """ + return ( + f"{SKILL_PRUNED_MARKER_PREFIX} content lost in compression; " + f"reload with skill_view(name='{skill_name}')]" + ) + + +# Matches the canonical marker and captures the skill name. Anchored on the +# shared prefix constant so a wording change to the marker body updates the +# emit helper and this extractor together. +_SKILL_PRUNED_MARKER_RE = re.compile( + re.escape(SKILL_PRUNED_MARKER_PREFIX) + + r"[^\]]*?reload with skill_view\(name='([^']+)'\)" +) + + +def _extract_pruned_skill_names(text: str) -> list[str]: + """Return skill names referenced by prune markers in *text*, in order.""" + names: list[str] = [] + for match in _SKILL_PRUNED_MARKER_RE.finditer(text or ""): + name = match.group(1) + if name not in names: + names.append(name) + return names + + +def _collect_ghosted_skill_names(turns: List[Dict[str, Any]]) -> list[str]: + """Skill names whose instructions are about to be lost in compaction. + + Covers BOTH shapes a compacted middle window can carry: + + - a ``skill_view`` result already demoted by Phase-1 pruning — the + canonical ``[SKILL_PRUNED: ...]`` marker is in the row content; + - a RAW ``skill_view`` body that was never demoted (it sat inside the + protected tail of an earlier prune, then aged into the compression + window). The summarizer will paraphrase the instructions away, which + is exactly the ghost-skill failure — so it needs a marker too. + """ + names: list[str] = [] + + def _add(name: str) -> None: + if name and name not in names: + names.append(name) + + call_id_to_skill: dict[str, str] = {} + for idx, skill in _skill_view_call_sites(turns): + msg = turns[idx] + for tc in msg.get("tool_calls") or []: + tc_fn = tc.get("function", {}) if isinstance(tc, dict) else getattr(tc, "function", None) + tc_name = tc_fn.get("name", "") if isinstance(tc_fn, dict) else getattr(tc_fn, "name", "") + if tc_name != "skill_view": + continue + cid = tc.get("id", "") if isinstance(tc, dict) else (getattr(tc, "id", "") or "") + if cid: + call_id_to_skill[cid] = skill + for msg in turns: + content = msg.get("content") + text = content if isinstance(content, str) else _content_text_for_contains(content) + for name in _extract_pruned_skill_names(text): + _add(name) + if ( + msg.get("role") == "tool" + and isinstance(content, str) + and len(content) > _SKILL_VIEW_PRUNE_MIN_CHARS + ): + skill = call_id_to_skill.get(str(msg.get("tool_call_id") or "")) + if skill: + _add(skill) + return names + + +_PRUNED_SKILLS_SECTION_HEADING = "## Pruned Skills" + + +def _reinject_pruned_skill_markers(summary: str, skill_names: list[str]) -> str: + """Deterministically restore prune markers the summarizer dropped. + + ``skill_names`` was extracted from the summarizer INPUT before the LLM + call. For every skill whose canonical marker (``_skill_pruned_marker``) + is absent from the model's output, append it under a ``## Pruned + Skills`` section. Presence is checked against the SAME canonical string + the emit sites produce — a paraphrased or renamed marker counts as + dropped and is restored (the original PR checked the literal + ``[SKILL_PRUNED]``, which never matches the emitted ``[SKILL_PRUNED:`` + form, so it duplicated markers that HAD survived). + + The appended block is plain body text: it never carries a handoff + prefix, the merged-summary delimiter, or a start-of-content scaffolding + marker, so ``classify_summary_content`` / todo-snapshot flag handling + are unaffected. The block is routed through ``_redact_compaction_text`` + like every other compaction-boundary text. + """ + if not skill_names: + return summary + missing = [ + name for name in skill_names + if _skill_pruned_marker(name) not in summary + ] + if not missing: + return summary + lines = [_skill_pruned_marker(name) for name in missing] + block = ( + "\n\n" + _PRUNED_SKILLS_SECTION_HEADING + "\n" + + "\n".join(lines) + + "\n(The listed skills' instructions were pruned during context " + "compression. Reload with the skill_view call in each marker before " + "relying on that skill; one reload per skill is enough — ignore any " + "older markers for the same skill.)" + ) + return summary + _redact_compaction_text(block) + + +# A skill_view call within this many trailing messages counts as "just +# loaded": its full instruction body must survive the Phase-1 prune even when +# the token-budget boundary would otherwise demote it (#32106). Distinct from +# the protected-tail boundary, which is token-based and can land immediately +# after a bulky just-loaded skill body. +_SKILL_PRUNE_RECENT_WINDOW = 10 + + +def _skill_view_call_sites( + messages: List[Dict[str, Any]], +) -> list[tuple[int, str]]: + """Yield ``(message_index, skill_name)`` for every skill_view tool call.""" + sites: list[tuple[int, str]] = [] + for i, msg in enumerate(messages): + if msg.get("role") != "assistant": + continue + for tc in msg.get("tool_calls") or []: + if isinstance(tc, dict): + fn = tc.get("function", {}) + name = fn.get("name", "") if isinstance(fn, dict) else "" + args_str = fn.get("arguments", "") if isinstance(fn, dict) else "" + else: + fn = getattr(tc, "function", None) + name = getattr(fn, "name", "") if fn else "" + args_str = getattr(fn, "arguments", "") if fn else "" + if name != "skill_view" or not isinstance(args_str, str) or not args_str: + continue + try: + args = json.loads(args_str) + except (json.JSONDecodeError, TypeError): + continue + if isinstance(args, dict): + skill = args.get("name", "") + if isinstance(skill, str) and skill: + sites.append((i, skill)) + return sites + + +def _collect_protected_skill_names( + messages: List[Dict[str, Any]], prune_boundary: int, +) -> set[str]: + """Skill names whose skill_view bodies must survive Phase-1 demotion. + + A skill is protected (lower-cased set) when any of these hold: + + - its most recent ``skill_view`` call sits within the last + ``_SKILL_PRUNE_RECENT_WINDOW`` messages (just loaded / just reloaded); + - its most recent ``skill_view`` call sits inside the protected tail + (at or after *prune_boundary*); + - its name is mentioned in a user message inside the protected tail + (the user is actively steering work that depends on it). + + Protection applies to the ordinary Phase-1/2 prune only. The Pass-4 + pressure demotion deliberately ignores it: when the protected region + itself exceeds the soft budget, exempting skill bodies would recreate + the #61932 dead-end shape. + """ + total = len(messages) + if not total: + return set() + recent_start = max(0, total - _SKILL_PRUNE_RECENT_WINDOW) + tail_start = max(0, prune_boundary) + tail_user_texts: list[str] = [] + for msg in messages[tail_start:]: + if msg.get("role") != "user": + continue + content = msg.get("content") + if isinstance(content, str) and content: + tail_user_texts.append(content.lower()) + protected: set[str] = set() + for idx, skill in _skill_view_call_sites(messages): + key = skill.lower() + if idx >= recent_start or idx >= tail_start: + protected.add(key) + elif any(key in text for text in tail_user_texts): + protected.add(key) + return protected diff --git a/agent/context_compressor_text_utils.py b/agent/context_compressor_text_utils.py new file mode 100644 index 000000000000..7c8f3fcbea48 --- /dev/null +++ b/agent/context_compressor_text_utils.py @@ -0,0 +1,56 @@ +"""Text helpers extracted from context_compressor (leaf util, epic #78647). + +Pure module-level helpers used across skill-prune, summary serialize, and +identity paths. Extracted first so skill-prune can import without a cycle +back into the god file. + +Part of #78645 + #78647. +""" +from __future__ import annotations + +from typing import Any + +from agent.redact import redact_sensitive_text + + +def _redact_compaction_text(text: Any) -> str: + """Redact text that crosses a compaction summary boundary. + + Compaction summaries persist across sessions and are re-injected into + every subsequent summarizer prompt, so this boundary uses strict mode: + + - ``force=True`` — deliberately overrides ``security.redact_secrets: + false``. That opt-out targets *live tool output* (e.g. working on the + redactor itself); a summary is a persistence boundary where a leaked + credential keeps re-entering prompts indefinitely. + - ``redact_url_credentials=True`` — OAuth callback codes, magic-link + tokens, and URL userinfo never need to survive summarization the way + they must survive live navigation flows. + """ + return redact_sensitive_text( + text or "", + force=True, + redact_url_credentials=True, + ) + +def _content_text_for_contains(content: Any) -> str: + """Return a best-effort text view of message content. + + Used only for substring checks when we need to know whether we've already + appended a note to a message. Keeps multimodal lists intact elsewhere. + """ + if content is None: + return "" + if isinstance(content, str): + return content + if isinstance(content, list): + parts: list[str] = [] + for item in content: + if isinstance(item, str): + parts.append(item) + elif isinstance(item, dict): + text = item.get("text") + if isinstance(text, str): + parts.append(text) + return "\n".join(part for part in parts if part) + return str(content) diff --git a/tests/agent/test_context_compressor_skill_prune_seam.py b/tests/agent/test_context_compressor_skill_prune_seam.py new file mode 100644 index 000000000000..5563d7a0c6e5 --- /dev/null +++ b/tests/agent/test_context_compressor_skill_prune_seam.py @@ -0,0 +1,69 @@ +"""Seam identity for context_compressor_skill_prune extract (LB2). + +Part of #78645 + #78647. +""" + +from agent import context_compressor as cc +from agent import context_compressor_skill_prune as sp + + +def test_all_members_resolve_is_identical_through_godfile(): + members = [ + "SKILL_PRUNED_MARKER_PREFIX", + "_SKILL_VIEW_PRUNE_MIN_CHARS", + "_MAX_PRUNED_SKILL_MARKERS", + "_SKILL_PRUNED_MARKER_RE", + "_PRUNED_SKILLS_SECTION_HEADING", + "_SKILL_PRUNE_RECENT_WINDOW", + "_skill_pruned_marker", + "_extract_pruned_skill_names", + "_collect_ghosted_skill_names", + "_reinject_pruned_skill_markers", + "_skill_view_call_sites", + "_collect_protected_skill_names", + ] + for m in members: + assert getattr(cc, m) is getattr(sp, m), f"{m} not is-identical" + + +def test_no_duplicate_defs_in_godfile(): + from pathlib import Path + + src = Path(cc.__file__).read_text(encoding="utf-8") + for name in [ + "_skill_pruned_marker", + "_extract_pruned_skill_names", + "_collect_ghosted_skill_names", + "_reinject_pruned_skill_markers", + "_skill_view_call_sites", + "_collect_protected_skill_names", + ]: + assert src.count(f"def {name}") == 0, f"duplicate def {name} left in godfile" + assert "context_compressor_skill_prune" in src + + +def test_behavior_smoke(): + # marker build + extract round trip + marker = cc._skill_pruned_marker("my-skill") + assert marker.startswith("[SKILL_PRUNED:") + assert "my-skill" in marker + names = cc._extract_pruned_skill_names(marker) + assert "my-skill" in names + # ghosted collection on a turn list + turns = [{"role": "user", "content": f"loaded {marker}"}] + ghosted = cc._collect_ghosted_skill_names(turns) + assert "my-skill" in ghosted + # reinject restores marker + out = cc._reinject_pruned_skill_markers("summary text", ["my-skill"]) + assert "[SKILL_PRUNED:" in out + + +def test_import_orders_no_cycle(): + import importlib + + import agent.context_compressor_skill_prune as a + import agent.context_compressor as b + + importlib.reload(a) + importlib.reload(b) + assert b._skill_pruned_marker is a._skill_pruned_marker diff --git a/tests/agent/test_context_compressor_text_utils_seam.py b/tests/agent/test_context_compressor_text_utils_seam.py new file mode 100644 index 000000000000..a2bdaee7bb78 --- /dev/null +++ b/tests/agent/test_context_compressor_text_utils_seam.py @@ -0,0 +1,41 @@ +"""Seam identity for context_compressor_text_utils leaf extract (LB-textutil). + +Part of #78645 + #78647. +""" + +from agent import context_compressor as cc +from agent import context_compressor_text_utils as tu + + +def test_redact_and_content_text_resolve_is_identical_through_godfile(): + assert cc._redact_compaction_text is tu._redact_compaction_text + assert cc._content_text_for_contains is tu._content_text_for_contains + + +def test_no_duplicate_defs_in_godfile(): + import inspect + from pathlib import Path + + src = Path(cc.__file__).read_text(encoding="utf-8") + assert src.count("def _redact_compaction_text") == 0 + assert src.count("def _content_text_for_contains") == 0 + assert "context_compressor_text_utils" in src + + +def test_redact_behavior_smoke(): + # None-safety preserved + assert cc._redact_compaction_text(None) == "" + # content text view + assert cc._content_text_for_contains(None) == "" + assert cc._content_text_for_contains("hi") == "hi" + assert cc._content_text_for_contains([{"type": "text", "text": "a"}, "b"]) == "a\nb" + + +def test_import_orders_no_cycle(): + import importlib + import agent.context_compressor_text_utils as a + import agent.context_compressor as b + + importlib.reload(a) + importlib.reload(b) + assert b._redact_compaction_text is a._redact_compaction_text