fix(compression): ghost-skill defense — canonical marker constant, protected-skill prune guard, deterministic marker survival

Salvage rework of PR #44166 (@dolphin-creator) onto current main:

- ONE canonical prune marker: _skill_pruned_marker(name) builds
  '[SKILL_PRUNED: ... reload with skill_view(name='X')]'; both emit
  sites and the survival presence check use the same string, fixing the
  original PR's defect where the emitted marker was '[SKILL_PRUNED:'
  but the presence check looked for '[SKILL_PRUNED]' (re-injection
  duplicated markers that had survived).
- Phase-1 prune (_prune_old_tool_results) now threads a protected-skill
  set: skills whose skill_view call is within the last 10 messages, in
  the protected tail, or named in a tail user message keep their full
  bodies. Pass-4 pressure demotion deliberately overrides the guard so
  the #61932 dead-end shape cannot return.
- P2 deterministic marker survival: skill names are extracted from the
  summarizer INPUT (and the previous summary) before the aux LLM call
  and any dropped canonical markers are re-injected afterward under a
  '## Pruned Skills' section — routed through _redact_compaction_text,
  appended to the summary body only (never in front of SUMMARY_PREFIX
  or scaffolding start-of-content markers; classify_summary_content is
  unaffected). Same treatment on the static fallback path, re-applied
  after its size cap since truncation cuts exactly where markers land.
- Summarizer prompt gains a '## Pruned Skills' copy-verbatim section.

Fixes #32106.

agent/context_compressor.py

@@ -320,6 +320,172 @@ _SUMMARY_INPUT_MAX_CHARS = 160_000
 # 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:"
+# 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
+
+
+_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
@@ -894,7 +1060,13 @@ def _summarize_tool_result_unguarded(tool_name: str, tool_args: str, tool_conten
     if tool_name == "skill_view":
         name = args.get("name", "?")
         if content_len > 5000:
-            return f"[skill_view] name={name} ({content_len:,} chars) [SKILL_PRUNED: content lost in compression; reload with skill_view(name='{name}')]"
+            # Ghost-skill defense (#32106): a metadata-only summary makes the
+            # model believe the skill is still loaded. The canonical marker
+            # tells it the instructions are gone AND how to get them back.
+            return (
+                f"[skill_view] name={name} ({content_len:,} chars) "
+                + _skill_pruned_marker(str(name))
+            )
         return f"[skill_view] name={name} ({content_len:,} chars)"
 
     if tool_name in {"skills_list", "skill_manage"}:
@@ -2217,7 +2389,14 @@ class ContextCompressor(ContextEngine):
             else:
                 content_hashes[h] = (i, msg.get("tool_call_id", "?"))
 
-        def _demote_tool_result_at(idx: int) -> bool:
+        # Ghost-skill defense (#32106): skills just loaded (or actively
+        # referenced in the protected tail) keep their full skill_view
+        # bodies through the ordinary prune passes. Without this, a skill
+        # loaded moments before a compaction can be demoted to metadata
+        # while the model still believes its instructions are in context.
+        protected_skills = _collect_protected_skill_names(result, prune_boundary)
+
+        def _demote_tool_result_at(idx: int, *, spare_protected_skills: bool = True) -> bool:
             """Replace a bulky tool result at ``idx`` with a 1-line summary.
 
             Returns True when the message was modified.
@@ -2256,6 +2435,16 @@ class ContextCompressor(ContextEngine):
                 return False
             call_id = msg.get("tool_call_id", "")
             tool_name, tool_args = call_id_to_tool.get(call_id, ("unknown", ""))
+            if spare_protected_skills and tool_name == "skill_view" and protected_skills:
+                # Just-loaded / actively-referenced skills survive verbatim
+                # (#32106). Pass-4 pressure demotion overrides this.
+                try:
+                    _args = json.loads(tool_args) if tool_args else {}
+                except (json.JSONDecodeError, TypeError):
+                    _args = {}
+                _skill = _args.get("name", "") if isinstance(_args, dict) else ""
+                if isinstance(_skill, str) and _skill.lower() in protected_skills:
+                    return False
             summary = _summarize_tool_result(tool_name, tool_args, content)
             result[idx] = {**msg, "content": summary}
             pruned += 1
@@ -2320,7 +2509,10 @@ class ContextCompressor(ContextEngine):
             if demote_end > prune_boundary and _protected_region_tokens() > soft_ceiling:
                 pressure_hits = 0
                 for i in range(max(0, prune_boundary), demote_end):
-                    if _demote_tool_result_at(i):
+                    # Pressure passes override the just-loaded-skill guard:
+                    # when the protected region itself blows the soft budget,
+                    # sparing skill bodies would recreate the #61932 dead-end.
+                    if _demote_tool_result_at(i, spare_protected_skills=False):
                         pressure_hits += 1
                     if _truncate_tool_call_args_at(i):
                         pressure_hits += 1
@@ -2340,7 +2532,7 @@ class ContextCompressor(ContextEngine):
                         if last_tool_idx is not None and i == last_tool_idx:
                             continue
                         if result[i].get("role") == "tool":
-                            if _demote_tool_result_at(i):
+                            if _demote_tool_result_at(i, spare_protected_skills=False):
                                 pressure_hits += 1
                         elif result[i].get("role") == "assistant":
                             if _truncate_tool_call_args_at(i):
@@ -2354,7 +2546,9 @@ class ContextCompressor(ContextEngine):
                         and last_tool_idx >= prune_boundary
                         and _protected_region_tokens() > soft_ceiling
                     ):
-                        if _demote_tool_result_at(last_tool_idx):
+                        if _demote_tool_result_at(
+                            last_tool_idx, spare_protected_skills=False
+                        ):
                             pressure_hits += 1
                 if pressure_hits and not self.quiet_mode:
                     logger.info(
@@ -2754,9 +2948,25 @@ Continue from the most recent unfulfilled user ask and protected tail messages.
 
 ## Critical Context
 Summary generation was unavailable, so this is a best-effort deterministic fallback for {len(turns_to_summarize)} compacted message(s).{reason_text}"""
+        # Ghost-skill defense (#32106): the fallback's per-turn truncation
+        # (``_FALLBACK_TURN_MAX_CHARS``) routinely cuts [SKILL_PRUNED: ...]
+        # markers out of the compacted turns. Re-derive them from the raw
+        # turn contents and re-inject deterministically, exactly like the
+        # LLM-summary path.
+        _pruned_names: list[str] = []
+        for _turn in turns_to_summarize:
+            for _name in _extract_pruned_skill_names(
+                _content_text_for_contains(_turn.get("content"))
+            ):
+                if _name not in _pruned_names:
+                    _pruned_names.append(_name)
+        del _pruned_names[_MAX_PRUNED_SKILL_MARKERS:]
         summary = self._with_summary_prefix(_redact_compaction_text(body.strip()))
         if len(summary) > _FALLBACK_SUMMARY_MAX_CHARS:
             summary = summary[: _FALLBACK_SUMMARY_MAX_CHARS - 42].rstrip() + "\n...[fallback summary truncated]"
+        # Re-inject AFTER the size cap: the markers live at the end of the
+        # body, exactly where the truncation above cuts.
+        summary = _reinject_pruned_skill_markers(summary, _pruned_names)
         return summary
 
     @classmethod
@@ -2864,9 +3074,22 @@ Summary generation was unavailable, so this is a best-effort deterministic fallb
             self._previous_summary = _redact_compaction_text(self._previous_summary)
 
         summary_budget = self._compute_summary_budget(turns_to_summarize)
-        content_to_summarize = self._bound_summary_input(
-            self._serialize_for_summary(turns_to_summarize)
-        )
+        content_to_summarize = self._serialize_for_summary(turns_to_summarize)
+        # P2 ghost-skill defense (#32106): [SKILL_PRUNED: ...] markers entering
+        # the summarizer are prompt INPUT only — LLMs routinely paraphrase them
+        # into vague prose ("some skills were loaded"), which erases the reload
+        # instruction. Extract the referenced skill names deterministically
+        # BEFORE the call; ``_reinject_pruned_skill_markers`` restores any
+        # marker the model dropped AFTER the call. Markers already carried by
+        # the previous summary must survive iterative rewrites the same way.
+        # Extraction runs on the UNBOUNDED serialization so a marker that
+        # falls into the input bound's omitted middle is still re-injected.
+        _pruned_skill_names = _extract_pruned_skill_names(content_to_summarize)
+        for _name in _extract_pruned_skill_names(self._previous_summary or ""):
+            if _name not in _pruned_skill_names:
+                _pruned_skill_names.append(_name)
+        del _pruned_skill_names[_MAX_PRUNED_SKILL_MARKERS:]
+        content_to_summarize = self._bound_summary_input(content_to_summarize)
         _sanitized_memory_context = sanitize_memory_context(memory_context)
         _serialized_memory_context = json.dumps(
             _sanitized_memory_context,
@@ -3058,6 +3281,12 @@ Be specific with file paths, commands, line numbers, and results.]
 ## Critical Context
 [Any specific values, error messages, configuration details, or data that would be lost without explicit preservation. NEVER include API keys, tokens, passwords, or credentials — write [REDACTED] instead.]
 
+{_PRUNED_SKILLS_SECTION_HEADING}
+[If any [SKILL_PRUNED: ...reload with skill_view(...)] markers appear in the input,
+repeat each one verbatim here — copy the exact text, do NOT paraphrase, summarize,
+or describe them. These markers tell the agent which skills must be reloaded before
+use. If none appear, omit this section entirely.]
+
 Target ~{summary_budget} tokens. Be CONCRETE — include file paths, command outputs, error messages, line numbers, and specific values. Avoid vague descriptions like "made some changes" — say exactly what changed.
 {_temporal_anchoring_rule}
 Write only the summary body. Do not include any preamble or prefix."""
@@ -3209,6 +3438,9 @@ This compaction should PRIORITISE preserving all information related to the focu
             # Redact the summary output as well — the summarizer LLM may
             # ignore prompt instructions and echo back secrets verbatim.
             summary = _redact_compaction_text(content.strip())
+            # P2 ghost-skill defense (#32106): deterministically restore any
+            # [SKILL_PRUNED: ...] marker the summarizer paraphrased away.
+            summary = _reinject_pruned_skill_markers(summary, _pruned_skill_names)
             summary = self._ground_historical_task_snapshot(summary, turns_to_summarize)
             self._validate_summary_user_provenance(summary, has_user_turn)
             # Store for iterative updates on next compaction