1  """Tests for reply-to pointer injection in _prepare_inbound_message_text.
  2  
  3  The `[Replying to: "..."]` prefix is a *disambiguation pointer*, not
  4  deduplication. It must always be injected when the user explicitly replies
  5  to a prior message — even when the quoted text already exists somewhere
  6  in the conversation history. History can contain the same or similar text
  7  multiple times, and without an explicit pointer the agent has to guess
  8  which prior message the user is referencing.
  9  """
 10  import pytest
 11  
 12  from gateway.config import GatewayConfig, Platform, PlatformConfig
 13  from gateway.platforms.base import MessageEvent
 14  from gateway.run import GatewayRunner
 15  from gateway.session import SessionSource
 16  
 17  
 18  def _make_runner() -> GatewayRunner:
 19      runner = object.__new__(GatewayRunner)
 20      runner.config = GatewayConfig(
 21          platforms={Platform.TELEGRAM: PlatformConfig(enabled=True, token="fake")},
 22      )
 23      runner.adapters = {}
 24      runner._model = "openai/gpt-4.1-mini"
 25      runner._base_url = None
 26      return runner
 27  
 28  
 29  def _source() -> SessionSource:
 30      return SessionSource(
 31          platform=Platform.TELEGRAM,
 32          chat_id="123",
 33          chat_name="DM",
 34          chat_type="private",
 35          user_name="Alice",
 36      )
 37  
 38  
 39  @pytest.mark.asyncio
 40  async def test_reply_prefix_injected_when_text_absent_from_history():
 41      runner = _make_runner()
 42      source = _source()
 43      event = MessageEvent(
 44          text="What's the best time to go?",
 45          source=source,
 46          reply_to_message_id="42",
 47          reply_to_text="Japan is great for culture, food, and efficiency.",
 48      )
 49  
 50      result = await runner._prepare_inbound_message_text(
 51          event=event,
 52          source=source,
 53          history=[{"role": "user", "content": "unrelated"}],
 54      )
 55  
 56      assert result is not None
 57      assert result.startswith(
 58          '[Replying to: "Japan is great for culture, food, and efficiency."]'
 59      )
 60      assert result.endswith("What's the best time to go?")
 61  
 62  
 63  @pytest.mark.asyncio
 64  async def test_reply_prefix_still_injected_when_text_in_history():
 65      """Regression test: the pointer must survive even when the quoted text
 66      already appears in history. Previously a `found_in_history` guard
 67      silently dropped the prefix, leaving the agent to guess which prior
 68      message the user was referencing."""
 69      runner = _make_runner()
 70      source = _source()
 71      quoted = "Japan is great for culture, food, and efficiency."
 72      event = MessageEvent(
 73          text="What's the best time to go?",
 74          source=source,
 75          reply_to_message_id="42",
 76          reply_to_text=quoted,
 77      )
 78  
 79      history = [
 80          {"role": "user", "content": "I'm thinking of going to Japan or Italy."},
 81          {
 82              "role": "assistant",
 83              "content": (
 84                  f"{quoted} Italy is better if you prefer a relaxed pace."
 85              ),
 86          },
 87          {"role": "user", "content": "How long should I stay?"},
 88          {"role": "assistant", "content": "For Japan, 10-14 days is ideal."},
 89      ]
 90  
 91      result = await runner._prepare_inbound_message_text(
 92          event=event,
 93          source=source,
 94          history=history,
 95      )
 96  
 97      assert result is not None
 98      assert result.startswith(f'[Replying to: "{quoted}"]')
 99      assert result.endswith("What's the best time to go?")
100  
101  
102  @pytest.mark.asyncio
103  async def test_no_prefix_without_reply_context():
104      runner = _make_runner()
105      source = _source()
106      event = MessageEvent(text="hello", source=source)
107  
108      result = await runner._prepare_inbound_message_text(
109          event=event,
110          source=source,
111          history=[],
112      )
113  
114      assert result == "hello"
115  
116  
117  @pytest.mark.asyncio
118  async def test_no_prefix_when_reply_to_text_is_empty():
119      """reply_to_message_id alone without text (e.g. a reply to a media-only
120      message) should not produce an empty `[Replying to: ""]` prefix."""
121      runner = _make_runner()
122      source = _source()
123      event = MessageEvent(
124          text="hi",
125          source=source,
126          reply_to_message_id="42",
127          reply_to_text=None,
128      )
129  
130      result = await runner._prepare_inbound_message_text(
131          event=event,
132          source=source,
133          history=[],
134      )
135  
136      assert result == "hi"
137  
138  
139  @pytest.mark.asyncio
140  async def test_reply_snippet_truncated_to_500_chars():
141      runner = _make_runner()
142      source = _source()
143      long_text = "x" * 800
144      event = MessageEvent(
145          text="follow-up",
146          source=source,
147          reply_to_message_id="42",
148          reply_to_text=long_text,
149      )
150  
151      result = await runner._prepare_inbound_message_text(
152          event=event,
153          source=source,
154          history=[],
155      )
156  
157      assert result is not None
158      assert result.startswith('[Replying to: "' + "x" * 500 + '"]')
159      assert "x" * 501 not in result