[Bug]: Triggered skills and path rules silently stop applying after condensation

"""Probe A — keyword-triggered skill loss through the PUBLIC Conversation.condense().

Differences from the reference repro:
- Condensation goes through the public `conversation.condense()` entry point
  (LocalConversation.condense -> CondensationRequest event -> Agent.step ->
  prepare_llm_messages -> condenser.condense -> on_event(Condensation)).
- keep_first=4, matching the production default used by
  `default_condenser()` (_DEFAULT_KEEP_FIRST = 4 in
  llm_summarizing_condenser.py). Note: the LLMSummarizingCondenser class field
  default itself is keep_first=2; the "default condenser" factory used by the
  default agent pins 4. We pass 4 explicitly.
- The skill activates on a LATER turn (outside the protected keep_first
  prefix), so the failure is not an artifact of keep_first=1.
"""

from openhands.sdk.agent import Agent
from openhands.sdk.context.agent_context import AgentContext
from openhands.sdk.context.condenser import LLMSummarizingCondenser
from openhands.sdk.context.condenser.llm_summarizing_condenser import (
    _DEFAULT_KEEP_FIRST,
    _DEFAULT_MAX_SIZE,
)
from openhands.sdk.conversation import Conversation
from openhands.sdk.event import MessageEvent
from openhands.sdk.llm import Message, TextContent
from openhands.sdk.skills import KeywordTrigger, Skill
from openhands.sdk.testing import TestLLM

sentinel = "PRESERVE_EXACT_SENTINEL_7F3A"
skill = Skill(
    name="python_tips",
    content=sentinel,
    source="python-tips.md",
    trigger=KeywordTrigger(keywords=["python"]),
)

class_keep_first_default = LLMSummarizingCondenser.model_fields["keep_first"].default
print(f"[setup] LLMSummarizingCondenser class-field default keep_first = {class_keep_first_default}")
print(f"[setup] default_condenser() factory uses keep_first={_DEFAULT_KEEP_FIRST}, max_size={_DEFAULT_MAX_SIZE}")
print("[setup] probe passes keep_first=4 explicitly (production default via default_condenser)")

agent_llm = TestLLM.from_messages([], model="agent-test-model")
summary_llm = TestLLM.from_messages(
    [
        Message(
            role="assistant",
            content=[TextContent(text="USER_CONTEXT: continuing work; skill omitted")],
        )
    ],
    model="summary-test-model",
)
condenser = LLMSummarizingCondenser(llm=summary_llm, max_size=12, keep_first=4)

agent = Agent(
    llm=agent_llm,
    tools=[],
    include_default_tools=[],
    agent_context=AgentContext(skills=[skill], current_datetime="2026-01-01T00:00:00Z"),
    condenser=condenser,
)
conversation = Conversation(agent=agent, visualizer=None)

# Filler BEFORE the skill turn so the skill-bearing MessageEvent lands outside
# the protected keep_first=4 prefix.
for i in range(4):
    conversation.send_message(f"prefix filler turn {i}")
conversation.send_message("help with python")  # skill activates here
for i in range(10):
    conversation.send_message(f"trailing filler turn {i}")

def view_str(conv):
    return "\n".join(map(str, conv.state.view.events))

print("\n[before] events in view (index: type : first line):")
for idx, ev in enumerate(conversation.state.view.events):
    first_line = str(ev).splitlines()[0] if str(ev) else ""
    print(f"  [{idx}] {type(ev).__name__}: {first_line[:80]}")

skill_idx = next(
    i
    for i, ev in enumerate(conversation.state.view.events)
    if isinstance(ev, MessageEvent) and sentinel in str(ev)
)
print(f"[before] skill-bearing MessageEvent index in view = {skill_idx} (keep_first=4 protects 0..3)")
print(f"[before] activated_knowledge_skills = {conversation.state.activated_knowledge_skills}")
print(f"[before] sentinel present in str(view.events): {sentinel in view_str(conversation)}")

# --- PUBLIC condensation path ---
conversation.condense()

print("\n[after] events in view (index: type : first line):")
for idx, ev in enumerate(conversation.state.view.events):
    first_line = str(ev).splitlines()[0] if str(ev) else ""
    print(f"  [{idx}] {type(ev).__name__}: {first_line[:80]}")

print(f"[after] activated_knowledge_skills = {conversation.state.activated_knowledge_skills}")
print(f"[after] sentinel present in str(view.events): {sentinel in view_str(conversation)}")

# --- reactivation check ---
conversation.send_message("python again")
last = conversation.state.events[-1]
assert isinstance(last, MessageEvent)
print(f"\n[next] last user MessageEvent.activated_skills = {last.activated_skills}")
print(f"[next] last user MessageEvent.extended_content = {[c.text for c in last.extended_content]}")

ok = (
    conversation.state.activated_knowledge_skills == ["python_tips"]
    and sentinel not in view_str(conversation)
    and last.activated_skills == []
    and last.extended_content == []
)
print(f"\n[verdict] skill content gone from view AND reactivation suppressed: {ok}")

"""Probe B — path-triggered rule: capture the exact summarizer input.

Wiring under test (all production code paths):
- LocalConversation._on_event is built as
  `_tree_stamping(_rules_injecting(base_callback))`
  (local_conversation.py:457). `_rules_injecting` ->
  `_maybe_inject_path_rules` (local_conversation.py:558) fires on every
  ObservationEvent the agent emits, correlates it to its ActionEvent, reads
  the action's `path`, matches PathTrigger skills via
  `AgentContext.get_tool_use_suffix`, appends the rendered rule to
  `ObservationEvent.extended_content`, and records the rule name in
  `state.activated_path_rules` (dedup).
- Agent.step / the run loop emit ObservationEvents through exactly this
  `conversation._on_event` callback, so calling it directly is the minimal
  real entry point — same code path, without scripting tool calls through
  TestLLM.
- Condensation goes through the PUBLIC `conversation.condense()` API.
- The summarizer LLM is a RecordingTestLLM (TestLLM subclass) that records
  the exact `messages` it receives, so we can inspect what the condenser's
  summarization prompt contained.
"""

from pydantic import PrivateAttr

from openhands.sdk.agent import Agent
from openhands.sdk.context.agent_context import AgentContext
from openhands.sdk.context.condenser import LLMSummarizingCondenser
from openhands.sdk.conversation import Conversation
from openhands.sdk.event import ActionEvent, MessageEvent, ObservationEvent
from openhands.sdk.llm import Message, MessageToolCall, TextContent
from openhands.sdk.skills import PathTrigger, Skill
from openhands.sdk.testing import TestLLM
from openhands.sdk.tool.schema import Action, Observation

sentinel = "PATH_RULE_SENTINEL_9C4E"
rule = Skill(
    name="style_rule",
    content=sentinel,
    source="rules/style.md",
    trigger=PathTrigger(paths=["src/**/*.py"]),
)
print(f"[setup] rule.disable_model_invocation (forced by Skill validator) = {rule.disable_model_invocation}")
print(f"[setup] match_path_trigger('src/pkg/app.py') = {rule.match_path_trigger('src/pkg/app.py')!r}")


class RecordingTestLLM(TestLLM):
    """TestLLM that records the messages of every completion call."""

    _received: list[list[Message]] = PrivateAttr(default_factory=list)

    def completion(self, messages: list[Message], **kwargs):
        self._received.append(messages)
        return super().completion(messages, **kwargs)


class FakeEditAction(Action):
    """Minimal file-touching action: only `path` matters to rule injection
    (_touched_rule_path reads `action.path` generically)."""

    path: str


class FakeObservation(Observation):
    pass


agent_llm = TestLLM.from_messages([], model="agent-test-model")
summary_llm = RecordingTestLLM.from_messages(
    [
        Message(
            role="assistant",
            content=[TextContent(text="USER_CONTEXT: continuing work; rule omitted")],
        )
    ],
    model="summary-test-model",
)
condenser = LLMSummarizingCondenser(llm=summary_llm, max_size=12, keep_first=4)

agent = Agent(
    llm=agent_llm,
    tools=[],
    include_default_tools=[],
    agent_context=AgentContext(skills=[rule], current_datetime="2026-01-01T00:00:00Z"),
    condenser=condenser,
)
conversation = Conversation(agent=agent, visualizer=None)


def emit_file_touch(conv, path: str, call_id: str, result_text: str) -> ObservationEvent:
    """Emit an ActionEvent/ObservationEvent pair through the conversation's
    production `_on_event` callback (the one wired with `_rules_injecting`)."""
    action_event = ActionEvent(
        thought=[TextContent(text=f"editing {path}")],
        action=FakeEditAction(path=path),
        tool_name="file_editor",
        tool_call_id=call_id,
        tool_call=MessageToolCall(
            id=call_id,
            name="file_editor",
            arguments=f'{{"path": "{path}"}}',
            origin="completion",
        ),
        llm_response_id=f"resp-{call_id}",
    )
    conv._on_event(action_event)
    obs = ObservationEvent(
        observation=FakeObservation.from_text(result_text),
        action_id=action_event.id,
        tool_name="file_editor",
        tool_call_id=call_id,
    )
    conv._on_event(obs)
    # _maybe_inject_path_rules returns a copied event; fetch the stored one.
    stored = conv.state.events[-1]
    assert isinstance(stored, ObservationEvent)
    return stored


# Filler BEFORE the rule touch so the ObservationEvent lands outside the
# protected keep_first=4 prefix.
for i in range(4):
    conversation.send_message(f"prefix filler turn {i}")
obs1 = emit_file_touch(conversation, "src/pkg/app.py", "call_1", "file contents here")
for i in range(8):
    conversation.send_message(f"trailing filler turn {i}")

obs_idx = next(
    i for i, ev in enumerate(conversation.state.view.events) if ev.id == obs1.id
)
print(f"\n[before] rule-carrying ObservationEvent index in view = {obs_idx} (keep_first=4 protects 0..3)")
print(f"[before] activated_path_rules = {conversation.state.activated_path_rules}")
print(f"[before] obs.extended_content = {[c.text for c in obs1.extended_content]}")
print(f"[before] sentinel in obs.extended_content: {any(sentinel in c.text for c in obs1.extended_content)}")
print(f"[before] sentinel in str(obs) (what the summarizer is fed): {sentinel in str(obs1)}")
print(f"[before] str(obs) = {str(obs1)!r}")

# --- PUBLIC condensation path ---
conversation.condense()

print(f"\n[after] activated_path_rules = {conversation.state.activated_path_rules}")
view_str = "\n".join(map(str, conversation.state.view.events))
print(f"[after] sentinel present anywhere in str(view.events): {sentinel in view_str}")

# --- KEY DELIVERABLE: exact summarizer input ---
print(f"\n[summarizer] completion calls recorded: {len(summary_llm._received)}")
for call_n, msgs in enumerate(summary_llm._received):
    full_text = "\n".join(
        c.text for m in msgs for c in m.content if isinstance(c, TextContent)
    )
    print(f"[summarizer] call {call_n}: {len(msgs)} message(s), {len(full_text)} chars")
    print(f"[summarizer] call {call_n}: sentinel present in input: {sentinel in full_text}")
    print(f"[summarizer] call {call_n}: 'file contents here' (obs result) present: {'file contents here' in full_text}")
    print(f"[summarizer] call {call_n}: 'EXTRA_INFO' (rule wrapper) present: {'EXTRA_INFO' in full_text}")

# --- re-injection check: touch the same path again AFTER condensation ---
obs2 = emit_file_touch(conversation, "src/pkg/app.py", "call_2", "file contents v2")
print(f"\n[reinject] second touch obs.extended_content = {[c.text for c in obs2.extended_content]}")
print(f"[reinject] rule re-injected after condensation: {any(sentinel in c.text for c in obs2.extended_content)}")
print(f"[reinject] activated_path_rules = {conversation.state.activated_path_rules}")

view_str2 = "\n".join(map(str, conversation.state.view.events))
ok = (
    conversation.state.activated_path_rules == ["style_rule"]
    and sentinel not in view_str2
    and not any(sentinel in c.text for c in obs2.extended_content)
    and all(
        sentinel
        not in "\n".join(
            c.text for m in msgs for c in m.content if isinstance(c, TextContent)
        )
        for msgs in summary_llm._received
    )
)
print(f"\n[verdict] rule body never reached summarizer AND gone from view AND re-injection suppressed: {ok}")