{escaped_code}")
+ def _replace_inline_code(code: str) -> str:
+ return _make_placeholder(f"{escape_html(code)}")
- text = re.sub(r"`([^`\n]+)`", _replace_inline_code, text)
+ text = _extract_code_spans(text, _replace_inline_code)
# --- 3. HTML-escape remaining text ---
text = escape_html(text)
diff --git a/src/claude/sdk_integration.py b/src/claude/sdk_integration.py
index d740e135e..c67b16f85 100644
--- a/src/claude/sdk_integration.py
+++ b/src/claude/sdk_integration.py
@@ -54,6 +54,60 @@
# Fallback message when Claude produces no text but did use tools.
TASK_COMPLETED_MSG = "✅ Task completed. Tools used: {tools_summary}"
+# Fallback message when a run stopped early without producing any text. The
+# stop-reason footer carries the warning and the reason, so this only reports
+# what the run got done -- otherwise the two stack up as "stopped" twice.
+TASK_STOPPED_MSG = "No final response. Tools used: {tools_summary}"
+
+# ResultMessage.subtype reported by the CLI for a run that ran to completion.
+RESULT_SUBTYPE_SUCCESS = "success"
+
+
+def _as_error_list(value: Any) -> List[str]:
+ """Normalise ResultMessage.errors into a list of strings."""
+ if not value:
+ return []
+ if isinstance(value, str):
+ return [value]
+ if isinstance(value, list):
+ return [str(item) for item in value if item]
+ return [str(value)]
+
+
+def _as_denial_list(value: Any) -> List[Dict[str, Any]]:
+ """Normalise ResultMessage.permission_denials into a list of dicts.
+
+ The SDK types this ``list[Any]`` and passes the CLI payload through
+ untouched, so accept both snake_case and camelCase keys and tolerate
+ entries that are not dicts at all.
+ """
+ if not value or not isinstance(value, list):
+ return []
+
+ denials: List[Dict[str, Any]] = []
+ for item in value:
+ if isinstance(item, dict):
+ tool_name = item.get("tool_name") or item.get("toolName")
+ tool_input = item.get("tool_input")
+ if tool_input is None:
+ tool_input = item.get("toolInput")
+ denials.append(
+ {
+ "tool_name": str(tool_name) if tool_name else "unknown",
+ "tool_input": tool_input if isinstance(tool_input, dict) else {},
+ }
+ )
+ else:
+ name = getattr(item, "tool_name", None)
+ tool_input = getattr(item, "tool_input", None)
+ denials.append(
+ {
+ "tool_name": str(name) if name else "unknown",
+ "tool_input": tool_input if isinstance(tool_input, dict) else {},
+ }
+ )
+ return denials
+
@dataclass
class ClaudeResponse:
@@ -68,6 +122,23 @@ class ClaudeResponse:
error_type: Optional[str] = None
tools_used: List[Dict[str, Any]] = field(default_factory=list)
interrupted: bool = False
+ # Why the run ended. All optional so existing construction sites keep
+ # working; populated from ResultMessage when the SDK reports them.
+ result_subtype: Optional[str] = None
+ stop_reason: Optional[str] = None
+ terminal_reason: Optional[str] = None
+ errors: List[str] = field(default_factory=list)
+ permission_denials: List[Dict[str, Any]] = field(default_factory=list)
+
+ @property
+ def completed_normally(self) -> bool:
+ """Whether the run reached its own end rather than being cut short.
+
+ ``None`` means the CLI reported no subtype at all (older versions, or a
+ result that bypassed the query loop), which we treat as normal so that
+ nothing regresses into a spurious warning.
+ """
+ return self.result_subtype in (None, RESULT_SUBTYPE_SUCCESS)
@dataclass
@@ -640,16 +711,32 @@ async def _cancel_on_interrupt() -> None:
if last_exc is not None:
raise last_exc
- # Extract cost, tools, and session_id from result message
+ # Extract cost, tools, session_id and stop reason from result message
cost = 0.0
tools_used: List[Dict[str, Any]] = []
claude_session_id = None
result_content = None
+ result_subtype: Optional[str] = None
+ result_num_turns: Optional[int] = None
+ stop_reason: Optional[str] = None
+ terminal_reason: Optional[str] = None
+ result_errors: List[str] = []
+ permission_denials: List[Dict[str, Any]] = []
for message in messages:
if isinstance(message, ResultMessage):
cost = getattr(message, "total_cost_usd", 0.0) or 0.0
claude_session_id = getattr(message, "session_id", None)
result_content = getattr(message, "result", None)
+ # getattr (not attribute access) throughout: older CLI
+ # versions and the test doubles omit these fields.
+ result_subtype = getattr(message, "subtype", None)
+ result_num_turns = getattr(message, "num_turns", None)
+ stop_reason = getattr(message, "stop_reason", None)
+ terminal_reason = getattr(message, "terminal_reason", None)
+ result_errors = _as_error_list(getattr(message, "errors", None))
+ permission_denials = _as_denial_list(
+ getattr(message, "permission_denials", None)
+ )
current_time = asyncio.get_event_loop().time()
for msg in messages:
if isinstance(msg, AssistantMessage):
@@ -710,6 +797,8 @@ async def _cancel_on_interrupt() -> None:
content_parts.append(str(msg_content))
content = "\n".join(content_parts).strip()
+ ran_to_completion = result_subtype in (None, RESULT_SUBTYPE_SUCCESS)
+
if not content and tools_used:
tool_names = [
tool.get("name", "")
@@ -718,22 +807,54 @@ async def _cancel_on_interrupt() -> None:
]
unique_tool_names = list(dict.fromkeys(tool_names))
tools_summary = ", ".join(unique_tool_names) or "unknown"
- content = TASK_COMPLETED_MSG.format(tools_summary=tools_summary)
+ # Only claim completion when the CLI says the run completed.
+ # A run killed at the turn limit takes this same path (tools
+ # ran, no final text) and must not report success (#172).
+ template = TASK_COMPLETED_MSG if ran_to_completion else TASK_STOPPED_MSG
+ content = template.format(tools_summary=tools_summary)
+
+ # The CLI reports the authoritative turn count. Counting messages
+ # over-reports it -- every tool result arrives as another
+ # UserMessage -- and that number is now shown to the user in the
+ # stop-reason footer, so the approximation is only a fallback for
+ # a result that did not carry one.
+ if isinstance(result_num_turns, int) and result_num_turns >= 0:
+ num_turns = result_num_turns
+ else:
+ num_turns = len(
+ [
+ m
+ for m in messages
+ if isinstance(m, (UserMessage, AssistantMessage))
+ ]
+ )
+
+ if not ran_to_completion or permission_denials or result_errors:
+ logger.info(
+ "Claude run did not end cleanly",
+ result_subtype=result_subtype,
+ stop_reason=stop_reason,
+ terminal_reason=terminal_reason,
+ permission_denials=len(permission_denials),
+ denied_tools=[d["tool_name"] for d in permission_denials],
+ errors=result_errors,
+ num_turns=num_turns,
+ session_id=final_session_id,
+ )
return ClaudeResponse(
content=content,
session_id=final_session_id,
cost=cost,
duration_ms=duration_ms,
- num_turns=len(
- [
- m
- for m in messages
- if isinstance(m, (UserMessage, AssistantMessage))
- ]
- ),
+ num_turns=num_turns,
tools_used=tools_used,
interrupted=interrupted,
+ result_subtype=result_subtype,
+ stop_reason=stop_reason,
+ terminal_reason=terminal_reason,
+ errors=result_errors,
+ permission_denials=permission_denials,
)
except asyncio.TimeoutError:
diff --git a/src/events/handlers.py b/src/events/handlers.py
index d2dcc334a..74dbc6d55 100644
--- a/src/events/handlers.py
+++ b/src/events/handlers.py
@@ -10,6 +10,7 @@
import structlog
+from ..bot.utils.formatting import with_stop_reason
from ..claude.facade import ClaudeIntegration
from .bus import Event, EventBus
from .types import AgentResponseEvent, ScheduledEvent, WebhookEvent
@@ -65,7 +66,11 @@ async def handle_webhook(self, event: Event) -> None:
user_id=self.default_user_id,
)
- if response.content:
+ # Nobody is watching a webhook run, so the reason it stopped is
+ # the whole of what the notification can say about it (#172).
+ text = with_stop_reason(response)
+
+ if text:
# We don't know which chat to send to from a webhook alone.
# The notification service needs configured target chats.
# Publish with chat_id=0 — the NotificationService
@@ -73,7 +78,7 @@ async def handle_webhook(self, event: Event) -> None:
await self.event_bus.publish(
AgentResponseEvent(
chat_id=0,
- text=response.content,
+ text=text,
originating_event_id=event.id,
)
)
@@ -122,12 +127,14 @@ async def _run_scheduled(self, event: ScheduledEvent) -> None:
user_id=self.default_user_id,
)
- if response.content:
+ text = with_stop_reason(response)
+
+ if text:
for chat_id in event.target_chat_ids:
await self.event_bus.publish(
AgentResponseEvent(
chat_id=chat_id,
- text=response.content,
+ text=text,
originating_event_id=event.id,
)
)
@@ -137,7 +144,7 @@ async def _run_scheduled(self, event: ScheduledEvent) -> None:
await self.event_bus.publish(
AgentResponseEvent(
chat_id=0,
- text=response.content,
+ text=text,
originating_event_id=event.id,
)
)
diff --git a/tests/unit/test_bot/test_callback_stop_reason.py b/tests/unit/test_bot/test_callback_stop_reason.py
new file mode 100644
index 000000000..de8f9498e
--- /dev/null
+++ b/tests/unit/test_bot/test_callback_stop_reason.py
@@ -0,0 +1,304 @@
+"""The classic-mode callbacks build HTML by hand, so they carry the footer too."""
+
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock
+
+from src.bot.handlers.callback import (
+ TRUNCATION_NOTE,
+ _clip_escaped,
+ _compose_reply,
+ _handle_continue_action,
+ _stop_reason_html,
+ handle_quick_action_callback,
+)
+from src.claude.sdk_integration import ClaudeResponse
+
+
+def _response(**kwargs) -> ClaudeResponse:
+ defaults = {
+ "content": "Ran the checks.",
+ "session_id": "s1",
+ "cost": 0.01,
+ "duration_ms": 100,
+ "num_turns": 10,
+ }
+ defaults.update(kwargs)
+ return ClaudeResponse(**defaults)
+
+
+class TestStopReasonHtml:
+ """format_stop_reason output converted for a hand-built HTML message."""
+
+ def test_empty_for_a_clean_run(self):
+ assert _stop_reason_html(_response(result_subtype="success")) == ""
+
+ def test_turn_limit_renders_as_html(self):
+ html = _stop_reason_html(_response(result_subtype="error_max_turns"))
+
+ assert "turn limit reached after 10 turns" in html
+ assert "<" not in html.replace("", "").replace("", "")
+
+ def test_blocked_call_is_escaped_and_monospaced(self):
+ html = _stop_reason_html(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Bash", "tool_input": {"command": "cd / && ls"}}
+ ],
+ )
+ )
+
+ assert "cd / && ls" in html
+
+
+async def _run_quick_action(claude_response, tmp_path):
+ """Drive handle_quick_action_callback and return the text it replied with."""
+ action = MagicMock()
+ action.icon = "🔍"
+ action.name = "Run tests"
+ action.prompt = "run the tests"
+
+ quick_actions = MagicMock()
+ quick_actions.get_action = MagicMock(return_value=action)
+
+ claude_integration = AsyncMock()
+ claude_integration.run_command = AsyncMock(return_value=claude_response)
+
+ settings = MagicMock()
+ settings.approved_directory = tmp_path
+
+ query = MagicMock()
+ query.from_user.id = 123
+ query.edit_message_text = AsyncMock()
+ query.message.reply_text = AsyncMock()
+
+ context = MagicMock()
+ context.user_data = {"current_directory": tmp_path}
+ context.bot_data = {
+ "quick_actions": quick_actions,
+ "claude_integration": claude_integration,
+ "settings": settings,
+ }
+
+ await handle_quick_action_callback(query, "test", context)
+
+ assert query.message.reply_text.call_args is not None, "no reply was sent"
+ return query.message.reply_text.call_args.args[0]
+
+
+class TestQuickActionHeading:
+ """The heading said "Complete" whatever the run did — #172 in a header."""
+
+ async def test_clean_run_still_says_complete(self, tmp_path):
+ text = await _run_quick_action(
+ _response(result_subtype="success"), Path(tmp_path)
+ )
+
+ assert "Complete" in text
+ assert "Stopped" not in text
+
+ async def test_truncated_run_says_stopped_and_explains(self, tmp_path):
+ text = await _run_quick_action(
+ _response(result_subtype="error_max_turns", terminal_reason="max_turns"),
+ Path(tmp_path),
+ )
+
+ assert "Complete" not in text
+ assert "Stopped" in text
+ assert "turn limit reached after 10 turns" in text
+
+
+class TestQuickActionMessageLength:
+ """Telegram caps a message at 4096 characters and reply_text will not split.
+
+ Appending the footer after a fixed-size clip pushed a long reply past the
+ cap; the send raised, the handler's except reported "Action Error", and
+ the run that most needed its stop reason was the one that lost it.
+ """
+
+ TELEGRAM_LIMIT = 4096
+
+ async def test_long_stopped_reply_with_denials_still_fits(self, tmp_path):
+ response = _response(
+ content="x" * 10000,
+ result_subtype="error_during_execution",
+ errors=["y" * 500],
+ permission_denials=[
+ {"tool_name": f"Tool{i}", "tool_input": {"file_path": "z" * 100}}
+ for i in range(8)
+ ],
+ )
+
+ text = await _run_quick_action(response, Path(tmp_path))
+
+ assert len(text) <= self.TELEGRAM_LIMIT
+
+ async def test_the_footer_is_kept_and_the_body_is_what_gives(self, tmp_path):
+ response = _response(
+ content="x" * 10000,
+ result_subtype="error_max_turns",
+ terminal_reason="max_turns",
+ permission_denials=[
+ {"tool_name": "Write", "tool_input": {"file_path": "/etc/hosts"}}
+ ],
+ )
+
+ text = await _run_quick_action(response, Path(tmp_path))
+
+ assert len(text) <= self.TELEGRAM_LIMIT
+ assert "Stopped" in text
+ assert "turn limit reached after 10 turns" in text
+ assert "1 tool call was blocked" in text
+ assert "(Response truncated)" in text
+
+ async def test_a_short_reply_is_not_truncated(self, tmp_path):
+ text = await _run_quick_action(
+ _response(content="Short.", result_subtype="success"), Path(tmp_path)
+ )
+
+ assert "Short." in text
+ assert "truncated" not in text
+
+
+class TestComposeReply:
+ """Both hand-built HTML messages compose through here."""
+
+ TELEGRAM_LIMIT = 4096
+
+ def test_short_reply_is_untouched(self):
+ text = _compose_reply(
+ "Heading",
+ _response(content="Short.", result_subtype="success"),
+ body_limit=500,
+ )
+
+ assert text == "Heading\n\nShort."
+
+ def test_body_limit_is_respected_below_the_cap(self):
+ text = _compose_reply(
+ "Heading",
+ _response(content="x" * 2000, result_subtype="success"),
+ body_limit=500,
+ )
+
+ assert len(text) < 600
+ assert "(Response truncated)" in text
+
+ def test_html_escaping_cannot_push_it_past_the_cap(self):
+ """500 raw characters of & escape to 2500, and the footer adds more."""
+ response = _response(
+ content="&" * 500,
+ result_subtype="error_during_execution",
+ errors=["&" * 500],
+ permission_denials=[
+ {"tool_name": "Bash", "tool_input": {"command": "&" * 100}}
+ for _ in range(8)
+ ],
+ )
+
+ text = _compose_reply("⚠️ Session Continued", response, body_limit=500)
+
+ assert len(text) <= self.TELEGRAM_LIMIT
+ assert text.startswith("⚠️ Session Continued")
+ assert "the run hit an error" in text
+ assert "8 tool calls were blocked" in text
+
+
+async def _run_continue_action(claude_response, tmp_path):
+ """Drive _handle_continue_action and return the text it replied with."""
+ claude_integration = AsyncMock()
+ claude_integration.continue_session = AsyncMock(return_value=claude_response)
+
+ settings = MagicMock()
+ settings.approved_directory = tmp_path
+
+ query = MagicMock()
+ query.from_user.id = 123
+ query.edit_message_text = AsyncMock()
+ query.message.reply_text = AsyncMock()
+
+ context = MagicMock()
+ context.user_data = {"current_directory": tmp_path}
+ context.bot_data = {
+ "claude_integration": claude_integration,
+ "settings": settings,
+ }
+
+ await _handle_continue_action(query, context)
+
+ assert query.message.reply_text.call_args is not None, "no reply was sent"
+ return query.message.reply_text.call_args.args[0]
+
+
+class TestContinueSessionHeading:
+ """A green tick above a "Stopped" footer is the #172 contradiction again."""
+
+ async def test_clean_run_keeps_the_tick(self, tmp_path):
+ text = await _run_continue_action(
+ _response(result_subtype="success"), Path(tmp_path)
+ )
+
+ assert text.startswith("✅ Session Continued")
+
+ async def test_stopped_run_loses_the_tick_but_keeps_the_words(self, tmp_path):
+ """The session did continue; it is the tick that would be false."""
+ text = await _run_continue_action(
+ _response(result_subtype="error_max_turns", terminal_reason="max_turns"),
+ Path(tmp_path),
+ )
+
+ assert text.startswith("⚠️ Session Continued")
+ assert "✅" not in text
+ assert "turn limit reached after 10 turns" in text
+
+
+class TestEntitySafeClipping:
+ """A cut inside & leaves &am, which Telegram rejects outright.
+
+ The handler's except would then report a generic failure, losing the
+ reply for exactly the stopped run the footer exists to explain.
+ """
+
+ @staticmethod
+ def _dangling(text: str) -> bool:
+ """True when the text ends in a half-written HTML entity."""
+ opener = text.rfind("&")
+ return opener != -1 and ";" not in text[opener:]
+
+ def test_cut_inside_an_entity_drops_it(self):
+ assert _clip_escaped("ab&", 5) == "ab"
+
+ def test_cut_on_the_entity_boundary_keeps_it(self):
+ assert _clip_escaped("ab&cd", 7) == "ab&"
+
+ def test_a_complete_entity_earlier_is_not_disturbed(self):
+ assert _clip_escaped("&xyz", 6) == "&x"
+
+ def test_nothing_to_clip(self):
+ assert _clip_escaped("abc", 10) == "abc"
+
+ def test_no_cut_point_leaves_it_empty(self):
+ assert _clip_escaped("&", 2) == ""
+
+ def test_compose_reply_never_emits_a_half_entity(self):
+ """Walk the cut across every offset in a run of escaping characters."""
+ for pad in range(64):
+ response = _response(
+ content="x" * pad + "&<>" * 200, result_subtype="success"
+ )
+
+ text = _compose_reply("H", response, body_limit=200)
+
+ body = text[len("H\n\n") :]
+ if body.endswith(TRUNCATION_NOTE):
+ body = body[: -len(TRUNCATION_NOTE)]
+ assert not self._dangling(body), f"pad={pad}: {body[-10:]!r}"
+
+ async def test_quick_action_reply_never_emits_a_half_entity(self, tmp_path):
+ text = await _run_quick_action(
+ _response(content="&" * 5000, result_subtype="error_max_turns"),
+ Path(tmp_path),
+ )
+
+ head, _, tail = text.rpartition("...")
+ assert not self._dangling(head or text)
diff --git a/tests/unit/test_bot/test_code_span_scanner.py b/tests/unit/test_bot/test_code_span_scanner.py
new file mode 100644
index 000000000..56f246881
--- /dev/null
+++ b/tests/unit/test_bot/test_code_span_scanner.py
@@ -0,0 +1,148 @@
+r"""The inline-code pass runs on the event loop, so it has to stay linear.
+
+Pairing equal-length backtick runs with a regex needs a backreference inside a
+lazy middle -- ``(`+)([^\n]*?)\1`` -- which re-scans the rest of the line for
+every opener that never finds a partner. On a reply whose backtick runs are
+all of different lengths that is superlinear: the regex this replaced took
+2.5 seconds on 256KB, blocking every other user's message for that long.
+"""
+
+import re
+import time
+
+from src.bot.utils.html_format import _extract_code_spans, markdown_to_telegram_html
+
+# The pattern this scanner replaced, kept so the two can be compared directly.
+BACKREF_PATTERN = re.compile(r"(? str:
+ return f"[{code}]"
+
+
+def _distinct_runs(size: int) -> str:
+ """Backtick runs of every length, so no opener ever finds its closer."""
+ parts, length = [], 1
+ while sum(len(p) for p in parts) < size:
+ parts.append("`" * length + "a")
+ length += 1
+ return "".join(parts)
+
+
+def _mirrored_across_a_newline(distinct_lengths: int) -> str:
+ """Runs 1..K, a newline, then K..1.
+
+ Every opener's only same-length partner is at the far end of the text and
+ on the other side of the line break, so every one of them is looked up and
+ rejected. Deciding that by scanning the gap costs the whole distance and
+ puts the pass back where the regex was; the distinct-length corpus above
+ does not reach the check at all, because those openers have no partner to
+ look at.
+ """
+ head = "".join("`" * k + "a" for k in range(1, distinct_lengths + 1))
+ tail = "".join("`" * k + "a" for k in range(distinct_lengths, 0, -1))
+ return head + "\n" + tail
+
+
+class TestMatchesTheRegexItReplaced:
+ """Same output, or the linearity fix would be a behaviour change."""
+
+ CASES = [
+ "Run `ls -la` then `cd /tmp`.",
+ "Use ``a`b`` for a literal backtick.",
+ "Blocked: Bash(`` `whoami` ``)",
+ "unclosed ` backtick and `another` one",
+ "multi\nline ` spanning ` attempt",
+ "` spaced `",
+ "` `",
+ "`` ``",
+ "``",
+ "```",
+ "a`b``c```d",
+ "`x`\n`y`",
+ "",
+ ]
+
+ @staticmethod
+ def _via_regex(text: str) -> str:
+ def replace(m: "re.Match[str]") -> str:
+ code = m.group(2)
+ if (
+ len(code) >= 2
+ and code[0] == " "
+ and code[-1] == " "
+ and code.strip(" ")
+ ):
+ code = code[1:-1]
+ return _render(code)
+
+ return BACKREF_PATTERN.sub(replace, text)
+
+ def test_every_case_renders_identically(self):
+ for case in self.CASES:
+ assert _extract_code_spans(case, _render) == self._via_regex(
+ case
+ ), f"diverged on {case!r}"
+
+
+class TestSpanSemantics:
+ def test_a_run_closes_only_on_the_same_length(self):
+ assert _extract_code_spans("``a`b``", _render) == "[a`b]"
+
+ def test_an_unpaired_run_is_left_alone(self):
+ assert _extract_code_spans("``a`b", _render) == "``a`b"
+
+ def test_a_span_does_not_cross_a_newline(self):
+ assert _extract_code_spans("`a\nb`", _render) == "`a\nb`"
+
+ def test_one_space_is_stripped_from_each_end(self):
+ assert _extract_code_spans("`` `x` ``", _render) == "[`x`]"
+
+ def test_a_span_of_only_spaces_keeps_them(self):
+ assert _extract_code_spans("` `", _render) == "[ ]"
+
+ def test_spans_after_an_unpaired_run_are_still_found(self):
+ assert _extract_code_spans("``` then `a`", _render) == "``` then [a]"
+
+
+class TestStaysLinear:
+ """A ceiling with three orders of magnitude of headroom.
+
+ The scanner does this in under a millisecond; the regex it replaced took
+ ~2.5s. The bound is loose enough that only a return to superlinear
+ behaviour can trip it, however slow the runner.
+ """
+
+ BUDGET_SECONDS = 1.0
+
+ def test_pathological_input_converts_promptly(self):
+ text = _distinct_runs(256_000)
+
+ started = time.perf_counter()
+ markdown_to_telegram_html(text)
+ elapsed = time.perf_counter() - started
+
+ assert elapsed < self.BUDGET_SECONDS, f"took {elapsed:.2f}s"
+
+ def test_the_pathological_input_really_has_nothing_to_match(self):
+ """Otherwise the budget above would be measuring an early exit."""
+ text = _distinct_runs(64_000)
+
+ assert _extract_code_spans(text, _render) == text
+
+ # Scanning the gap instead of searching it takes 1.07s on the input below,
+ # so this budget fails on that implementation and passes on this one with
+ # roughly fifty times to spare.
+ NEWLINE_BUDGET_SECONDS = 0.5
+
+ def test_partners_rejected_across_a_line_break_are_cheap_too(self):
+ """Every opener here is looked up, and every one is rejected."""
+ text = _mirrored_across_a_newline(2560)
+ assert len(text) > 6_000_000
+
+ started = time.perf_counter()
+ result = _extract_code_spans(text, _render)
+ elapsed = time.perf_counter() - started
+
+ assert result == text, "nothing should pair across the newline"
+ assert elapsed < self.NEWLINE_BUDGET_SECONDS, f"took {elapsed:.2f}s"
diff --git a/tests/unit/test_bot/test_formatting.py b/tests/unit/test_bot/test_formatting.py
index a2e97d44a..6f38b872c 100644
--- a/tests/unit/test_bot/test_formatting.py
+++ b/tests/unit/test_bot/test_formatting.py
@@ -299,6 +299,42 @@ def test_inline_code(self):
result = markdown_to_telegram_html("`code here`")
assert "code here" in result
+ def test_multi_backtick_span_carries_backticks(self):
+ """A run of N backticks closes only on a run of N."""
+ result = markdown_to_telegram_html("``echo `whoami` in a shell``")
+ assert "echo `whoami` in a shell" in result
+
+ def test_triple_backtick_span_on_one_line(self):
+ result = markdown_to_telegram_html("``` ``x`` ```")
+ assert "``x``" in result
+
+ def test_one_space_is_stripped_from_each_end(self):
+ result = markdown_to_telegram_html("`` ` ``")
+ assert "`" in result
+
+ def test_space_is_not_stripped_from_one_end_alone(self):
+ result = markdown_to_telegram_html("` x`")
+ assert " x" in result
+
+ def test_all_space_content_is_left_alone(self):
+ result = markdown_to_telegram_html("` `")
+ assert " " in result
+
+ def test_unpaired_backtick_is_literal(self):
+ result = markdown_to_telegram_html("a ` b")
+ assert "" not in result
+ assert "a ` b" in result
+
+ def test_two_spans_on_one_line_stay_separate(self):
+ result = markdown_to_telegram_html("`a` and `b`")
+ assert "a and b" in result
+
+ def test_span_content_is_not_markdown(self):
+ result = markdown_to_telegram_html("`_x_ **y**`")
+ assert "_x_ **y**" in result
+ assert "" not in result
+ assert "" not in result
+
def test_fenced_code_block(self):
result = markdown_to_telegram_html("```python\nprint('hi')\n```")
assert "" in result
diff --git a/tests/unit/test_bot/test_stop_reason.py b/tests/unit/test_bot/test_stop_reason.py
new file mode 100644
index 000000000..624b62afb
--- /dev/null
+++ b/tests/unit/test_bot/test_stop_reason.py
@@ -0,0 +1,429 @@
+"""Tests for the footer that says why a Claude run stopped (#230, #172)."""
+
+from src.bot.utils.formatting import (
+ format_permission_denials,
+ format_stop_reason,
+ with_stop_reason,
+)
+from src.bot.utils.html_format import markdown_to_telegram_html
+from src.claude.sdk_integration import ClaudeResponse
+
+
+def _response(**kwargs) -> ClaudeResponse:
+ defaults = {
+ "content": "Some output",
+ "session_id": "s1",
+ "cost": 0.01,
+ "duration_ms": 100,
+ "num_turns": 10,
+ }
+ defaults.update(kwargs)
+ return ClaudeResponse(**defaults)
+
+
+class TestFormatStopReason:
+ """The footer appended to Claude's reply."""
+
+ def test_no_footer_for_a_clean_run(self):
+ assert format_stop_reason(_response(result_subtype="success")) is None
+
+ def test_no_footer_when_the_cli_reported_no_subtype(self):
+ """Older CLI versions omit subtype; that must not read as a failure."""
+ assert format_stop_reason(_response()) is None
+
+ def test_turn_limit(self):
+ footer = format_stop_reason(
+ _response(result_subtype="error_max_turns", terminal_reason="max_turns")
+ )
+
+ assert footer is not None
+ assert "turn limit reached after 10 turns" in footer
+ assert "Send a message to continue" in footer
+ assert footer.startswith("\n\n")
+
+ def test_turn_limit_without_a_terminal_reason(self):
+ """subtype alone is enough; terminal_reason is extra detail."""
+ footer = format_stop_reason(_response(result_subtype="error_max_turns"))
+
+ assert footer is not None
+ assert "turn limit reached" in footer
+
+ def test_cost_budget(self):
+ """CLAUDE_MAX_COST_PER_REQUEST is passed to the SDK as max_budget_usd."""
+ footer = format_stop_reason(_response(result_subtype="error_max_budget_usd"))
+
+ assert footer is not None
+ assert "cost budget reached" in footer
+
+ def test_error_during_execution_shows_the_cli_prose(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="error_during_execution",
+ errors=["Tool ran out of memory"],
+ )
+ )
+
+ assert footer is not None
+ assert "the run hit an error" in footer
+ assert "Tool ran out of memory" in footer
+
+ def test_unknown_subtype_names_the_raw_value(self):
+ footer = format_stop_reason(_response(result_subtype="error_brand_new"))
+
+ assert footer is not None
+ assert "error_brand_new" in footer
+
+ def test_terminal_reason_wins_over_an_unknown_subtype(self):
+ footer = format_stop_reason(
+ _response(result_subtype="error_something", terminal_reason="api_error")
+ )
+
+ assert footer is not None
+ assert "the API returned an error" in footer
+
+ def test_a_single_turn_is_singular(self):
+ footer = format_stop_reason(
+ _response(result_subtype="error_max_turns", num_turns=1)
+ )
+
+ assert footer is not None
+ assert "after 1 turn." in footer
+
+ def test_no_turn_count_when_none_were_recorded(self):
+ footer = format_stop_reason(
+ _response(result_subtype="error_max_turns", num_turns=0)
+ )
+
+ assert footer is not None
+ assert "after" not in footer
+
+ def test_long_error_detail_is_clipped(self):
+ footer = format_stop_reason(
+ _response(result_subtype="error_during_execution", errors=["x" * 500])
+ )
+
+ assert footer is not None
+ assert "…" in footer
+ assert len(footer) < 400
+
+ def test_denials_are_reported_even_on_a_successful_run(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Write", "tool_input": {"file_path": "/etc/hosts"}}
+ ],
+ )
+ )
+
+ assert footer is not None
+ assert "1 tool call was blocked: Write(`/etc/hosts`)" in footer
+ assert "Stopped" not in footer
+
+ def test_stop_line_and_denials_together(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="error_max_turns",
+ permission_denials=[{"tool_name": "Bash", "tool_input": {}}],
+ )
+ )
+
+ assert footer is not None
+ assert "turn limit reached" in footer
+ assert "1 tool call was blocked: Bash" in footer
+
+
+class TestFormatPermissionDenials:
+ """The blocked-calls line."""
+
+ def test_none_when_nothing_was_denied(self):
+ assert format_permission_denials([]) is None
+
+ def test_plural_wording_and_argument_extraction(self):
+ line = format_permission_denials(
+ [
+ {"tool_name": "Write", "tool_input": {"file_path": "/etc/hosts"}},
+ {"tool_name": "Bash", "tool_input": {"command": "cd /"}},
+ ]
+ )
+
+ assert line == "🚫 2 tool calls were blocked: Write(`/etc/hosts`), Bash(`cd /`)"
+
+ def test_tool_without_a_recognised_argument(self):
+ line = format_permission_denials([{"tool_name": "WebSearch", "tool_input": {}}])
+
+ assert line == "🚫 1 tool call was blocked: WebSearch"
+
+ def test_long_arguments_are_clipped(self):
+ line = format_permission_denials(
+ [{"tool_name": "Bash", "tool_input": {"command": "echo " + "a" * 200}}]
+ )
+
+ assert line is not None
+ assert "…" in line
+ assert len(line) < 100
+
+ def test_whitespace_in_arguments_is_collapsed(self):
+ line = format_permission_denials(
+ [{"tool_name": "Bash", "tool_input": {"command": "ls\n -la"}}]
+ )
+
+ assert line == "🚫 1 tool call was blocked: Bash(`ls -la`)"
+
+ def test_long_lists_are_summarised(self):
+ denials = [{"tool_name": f"Tool{i}", "tool_input": {}} for i in range(8)]
+
+ line = format_permission_denials(denials)
+
+ assert line is not None
+ assert line.startswith("🚫 8 tool calls were blocked:")
+ assert "and 3 more" in line
+ assert "Tool5" not in line
+
+ def test_non_dict_entries_are_ignored(self):
+ """permission_denials is typed list[Any]; the CLI payload is opaque."""
+ assert format_permission_denials(["not a dict", None]) is None
+
+ def test_non_list_input_is_ignored(self):
+ assert format_permission_denials(None) is None
+
+ def test_malformed_entries_do_not_hide_the_ones_behind_them(self):
+ """Filtering after the slice let five bad entries swallow the list."""
+ denials = ["junk"] * 5 + [
+ {"tool_name": "Write", "tool_input": {"file_path": "/etc/hosts"}}
+ ]
+
+ line = format_permission_denials(denials)
+
+ assert line is not None
+ assert "Write(`/etc/hosts`)" in line
+ assert "6 tool calls were blocked" in line
+
+ def test_missing_tool_name_falls_back(self):
+ line = format_permission_denials([{"tool_input": {"file_path": "/x"}}])
+
+ assert line == "🚫 1 tool call was blocked: unknown(`/x`)"
+
+
+class TestInterruptedRuns:
+ """The user pressed Stop; they do not need to be told why it ended."""
+
+ def test_note_replaces_the_stop_reason(self):
+ footer = format_stop_reason(
+ _response(
+ interrupted=True,
+ result_subtype="error_during_execution",
+ terminal_reason="aborted_streaming",
+ )
+ )
+
+ assert footer == "\n\n_(Interrupted by user)_"
+
+ def test_wording_is_unchanged_from_before_the_footer_existed(self):
+ response = _response(content="Partial output.", interrupted=True)
+
+ assert (
+ with_stop_reason(response) == "Partial output.\n\n_(Interrupted by user)_"
+ )
+
+ def test_blocked_calls_are_still_listed(self):
+ footer = format_stop_reason(
+ _response(
+ interrupted=True,
+ permission_denials=[{"tool_name": "Bash", "tool_input": {}}],
+ )
+ )
+
+ assert footer is not None
+ assert "_(Interrupted by user)_" in footer
+ assert "1 tool call was blocked: Bash" in footer
+ assert "Stopped" not in footer
+
+
+class TestFooterSurvivesTheMarkdownPass:
+ """The footer is rendered with Claude's reply, so Markdown runs over it."""
+
+ def test_underscores_in_a_path_are_not_italicised(self):
+ """Without inline code, /tmp/_a_b_ renders as /tmp/a_b."""
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Write", "tool_input": {"file_path": "/tmp/_a_b_"}}
+ ],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "/tmp/_a_b_" in html
+ assert "" not in html
+
+ def test_italics_do_not_bleed_between_two_denials(self):
+ """One underscore per entry used to open an italic span in the next."""
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Read", "tool_input": {"file_path": "/x/_p_"}},
+ {"tool_name": "Write", "tool_input": {"file_path": "/y/_q_"}},
+ ],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "/x/_p_" in html
+ assert "/y/_q_" in html
+ assert "" not in html
+
+ def test_shell_metacharacters_are_html_escaped(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Bash", "tool_input": {"command": "cd / && ls"}}
+ ],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "cd / && ls" in html
+
+ def test_backticks_in_the_argument_are_preserved(self):
+ """`whoami` runs a command; whoami prints a word. Dropping the
+ backticks would report a different call than the one blocked."""
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Bash", "tool_input": {"command": "echo `whoami`"}}
+ ],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "echo `whoami`" in html
+
+ def test_consecutive_backticks_are_preserved(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Bash", "tool_input": {"command": "echo ``x``"}}
+ ],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "echo ``x``" in html
+
+ def test_an_argument_that_is_only_backticks_survives(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Bash", "tool_input": {"command": "`"}}
+ ],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "`" in html
+
+ def test_backticks_with_html_metacharacters(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="success",
+ permission_denials=[
+ {
+ "tool_name": "Bash",
+ "tool_input": {"command": "echo `a && b` > /x"},
+ }
+ ],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "echo `a && b` > /x" in html
+
+ def test_error_prose_is_not_italicised(self):
+ footer = format_stop_reason(
+ _response(
+ result_subtype="error_during_execution",
+ errors=["cannot read _config_ from *here*"],
+ )
+ )
+
+ assert footer is not None
+ html = markdown_to_telegram_html(footer)
+ assert "" not in html
+ assert "" not in html
+
+
+class TestWithStopReason:
+ """The helper every render site uses."""
+
+ def test_clean_run_is_the_content_unchanged(self):
+ response = _response(content="All done.", result_subtype="success")
+
+ assert with_stop_reason(response) == "All done."
+
+ def test_stopped_run_gets_the_footer(self):
+ response = _response(content="Partial.", result_subtype="error_max_turns")
+
+ text = with_stop_reason(response)
+
+ assert text.startswith("Partial.")
+ assert "turn limit reached" in text
+
+ def test_empty_content_still_carries_the_footer(self):
+ response = _response(content="", result_subtype="error_max_turns")
+
+ assert "turn limit reached" in with_stop_reason(response)
+
+
+class TestEveryRenderSiteCarriesTheFooter:
+ """A new entry point that renders a reply must not skip the footer.
+
+ The first cut of this fix covered `agentic_text` and the four sites in
+ `handlers/message.py` and missed five others, so a run truncated at the
+ turn limit still reported success through document upload, voice, photo,
+ `/continue` and the quick-action buttons. This walks the source rather
+ than trusting a list.
+ """
+
+ def test_format_claude_response_is_always_given_the_footer(self):
+ import ast
+ from pathlib import Path
+
+ import src.bot as bot_package
+
+ offenders = []
+ for path in sorted(Path(bot_package.__file__).parent.rglob("*.py")):
+ tree = ast.parse(path.read_text(), filename=str(path))
+ for node in ast.walk(tree):
+ if not isinstance(node, ast.Call):
+ continue
+ func = node.func
+ name = func.attr if isinstance(func, ast.Attribute) else None
+ if name != "format_claude_response" or not node.args:
+ continue
+ argument = node.args[0]
+ wrapped = (
+ isinstance(argument, ast.Call)
+ and isinstance(argument.func, ast.Name)
+ and argument.func.id == "with_stop_reason"
+ )
+ if not wrapped:
+ offenders.append(f"{path.name}:{node.lineno}")
+
+ assert offenders == [], (
+ "these calls render a Claude reply without the stop-reason "
+ f"footer: {offenders}"
+ )
diff --git a/tests/unit/test_claude/test_sdk_integration.py b/tests/unit/test_claude/test_sdk_integration.py
index 7183f3340..5c81dadc2 100644
--- a/tests/unit/test_claude/test_sdk_integration.py
+++ b/tests/unit/test_claude/test_sdk_integration.py
@@ -13,11 +13,14 @@
ResultMessage,
TextBlock,
ToolPermissionContext,
+ ToolUseBlock,
)
from claude_agent_sdk.types import StreamEvent
from src.claude.sdk_integration import (
GUARDED_TOOLS,
+ TASK_COMPLETED_MSG,
+ TASK_STOPPED_MSG,
ClaudeResponse,
ClaudeSDKManager,
StreamUpdate,
@@ -1698,3 +1701,211 @@ async def test_setting_sources_includes_project(self, sdk_manager, tmp_path):
opts = captured[0]
assert opts.setting_sources == ["project"]
+
+
+class TestStopReasonCapture:
+ """ResultMessage stop-reason fields reach ClaudeResponse (#230, #172)."""
+
+ @pytest.fixture
+ def config(self, tmp_path):
+ return Settings(
+ telegram_bot_token="test:token",
+ telegram_bot_username="testbot",
+ approved_directory=tmp_path,
+ claude_timeout_seconds=2,
+ )
+
+ @pytest.fixture
+ def sdk_manager(self, config):
+ return ClaudeSDKManager(config)
+
+ @staticmethod
+ def _tool_use_message(name="Bash"):
+ """An assistant turn that used a tool but produced no text."""
+ return AssistantMessage(
+ content=[ToolUseBlock(id="tool-1", name=name, input={"command": "ls"})],
+ model="claude-sonnet-4-20250514",
+ )
+
+ async def test_success_subtype_still_reports_completion(self, sdk_manager):
+ mock_factory = _mock_client_factory(
+ self._tool_use_message(),
+ _make_result_message(subtype="success", result=None),
+ )
+
+ with patch(
+ "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory
+ ):
+ response = await sdk_manager.execute_command(
+ prompt="Test prompt",
+ working_directory=Path("/test"),
+ )
+
+ assert response.content == TASK_COMPLETED_MSG.format(tools_summary="Bash")
+ assert response.result_subtype == "success"
+ assert response.completed_normally is True
+
+ async def test_max_turns_does_not_claim_completion(self, sdk_manager):
+ """A run killed at the turn limit used to report success (#172)."""
+ mock_factory = _mock_client_factory(
+ self._tool_use_message(),
+ _make_result_message(
+ subtype="error_max_turns",
+ is_error=True,
+ result=None,
+ terminal_reason="max_turns",
+ errors=["Reached maximum number of turns"],
+ ),
+ )
+
+ with patch(
+ "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory
+ ):
+ response = await sdk_manager.execute_command(
+ prompt="Test prompt",
+ working_directory=Path("/test"),
+ )
+
+ assert TASK_COMPLETED_MSG.format(tools_summary="Bash") not in response.content
+ assert response.content == TASK_STOPPED_MSG.format(tools_summary="Bash")
+ assert response.result_subtype == "error_max_turns"
+ assert response.terminal_reason == "max_turns"
+ assert response.errors == ["Reached maximum number of turns"]
+ assert response.completed_normally is False
+
+ async def test_permission_denials_reach_response(self, sdk_manager):
+ mock_factory = _mock_client_factory(
+ _make_assistant_message("Done what I could"),
+ _make_result_message(
+ result="Done what I could",
+ permission_denials=[
+ {
+ "tool_name": "Write",
+ "tool_use_id": "t1",
+ "tool_input": {"file_path": "/etc/hosts"},
+ },
+ {"toolName": "Bash", "toolInput": {"command": "cd /"}},
+ ],
+ ),
+ )
+
+ with patch(
+ "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory
+ ):
+ response = await sdk_manager.execute_command(
+ prompt="Test prompt",
+ working_directory=Path("/test"),
+ )
+
+ assert response.permission_denials == [
+ {"tool_name": "Write", "tool_input": {"file_path": "/etc/hosts"}},
+ {"tool_name": "Bash", "tool_input": {"command": "cd /"}},
+ ]
+ # A denial on its own does not make the run a failure.
+ assert response.completed_normally is True
+
+ async def test_missing_fields_are_tolerated(self, sdk_manager):
+ """Older CLI versions omit the 0.2 fields entirely."""
+ mock_factory = _mock_client_factory(
+ _make_assistant_message("Test response"),
+ _make_result_message(),
+ )
+
+ with patch(
+ "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory
+ ):
+ response = await sdk_manager.execute_command(
+ prompt="Test prompt",
+ working_directory=Path("/test"),
+ )
+
+ assert response.stop_reason is None
+ assert response.terminal_reason is None
+ assert response.errors == []
+ assert response.permission_denials == []
+
+ def test_response_defaults_keep_existing_callers_working(self):
+ response = ClaudeResponse(
+ content="hi", session_id="s", cost=0.0, duration_ms=1, num_turns=1
+ )
+ assert response.result_subtype is None
+ assert response.errors == []
+ assert response.permission_denials == []
+ assert response.completed_normally is True
+
+
+class TestNumTurns:
+ """num_turns comes from the CLI, not from counting messages."""
+
+ @pytest.fixture
+ def config(self, tmp_path):
+ return Settings(
+ telegram_bot_token="test:token",
+ telegram_bot_username="testbot",
+ approved_directory=tmp_path,
+ claude_timeout_seconds=2,
+ )
+
+ @pytest.fixture
+ def sdk_manager(self, config):
+ return ClaudeSDKManager(config)
+
+ async def test_result_message_wins_over_the_message_count(self, sdk_manager):
+ """Every tool result arrives as another UserMessage, so counting
+ messages over-reports the turns -- and the stop-reason footer shows
+ that number to the user."""
+ mock_factory = _mock_client_factory(
+ _make_assistant_message("one"),
+ _make_assistant_message("two"),
+ _make_assistant_message("three"),
+ _make_result_message(num_turns=10, subtype="error_max_turns"),
+ )
+
+ with patch(
+ "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory
+ ):
+ response = await sdk_manager.execute_command(
+ prompt="Test prompt",
+ working_directory=Path("/test"),
+ )
+
+ assert response.num_turns == 10
+
+ async def test_zero_turns_is_taken_at_face_value(self, sdk_manager):
+ mock_factory = _mock_client_factory(
+ _make_assistant_message("one"),
+ _make_result_message(num_turns=0),
+ )
+
+ with patch(
+ "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory
+ ):
+ response = await sdk_manager.execute_command(
+ prompt="Test prompt",
+ working_directory=Path("/test"),
+ )
+
+ assert response.num_turns == 0
+
+ async def test_falls_back_to_counting_when_the_cli_reports_nothing(
+ self, sdk_manager
+ ):
+ """An older CLI, or a result that never reached the query loop."""
+ result = _make_result_message()
+ del result.num_turns
+
+ mock_factory = _mock_client_factory(
+ _make_assistant_message("one"),
+ _make_assistant_message("two"),
+ result,
+ )
+
+ with patch(
+ "src.claude.sdk_integration.ClaudeSDKClient", side_effect=mock_factory
+ ):
+ response = await sdk_manager.execute_command(
+ prompt="Test prompt",
+ working_directory=Path("/test"),
+ )
+
+ assert response.num_turns == 2
diff --git a/tests/unit/test_events/test_handlers.py b/tests/unit/test_events/test_handlers.py
index b9a6e9d03..ba8ae14b0 100644
--- a/tests/unit/test_events/test_handlers.py
+++ b/tests/unit/test_events/test_handlers.py
@@ -1,15 +1,29 @@
"""Tests for event handlers."""
from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock
import pytest
+from src.claude.sdk_integration import ClaudeResponse
from src.events.bus import EventBus
from src.events.handlers import AgentHandler
from src.events.types import AgentResponseEvent, ScheduledEvent, WebhookEvent
+def _response(**kwargs: object) -> ClaudeResponse:
+ """A real ClaudeResponse -- these paths now read stop-reason fields."""
+ defaults: dict = {
+ "content": "Analysis complete",
+ "session_id": "s1",
+ "cost": 0.0,
+ "duration_ms": 1,
+ "num_turns": 1,
+ }
+ defaults.update(kwargs)
+ return ClaudeResponse(**defaults) # type: ignore[arg-type]
+
+
@pytest.fixture
def event_bus() -> EventBus:
return EventBus()
@@ -41,9 +55,7 @@ async def test_webhook_event_triggers_claude(
self, event_bus: EventBus, mock_claude: AsyncMock, agent_handler: AgentHandler
) -> None:
"""Webhook events are processed through Claude."""
- mock_response = MagicMock()
- mock_response.content = "Analysis complete"
- mock_claude.run_command.return_value = mock_response
+ mock_claude.run_command.return_value = _response(content="Analysis complete")
published: list = []
original_publish = event_bus.publish
@@ -76,9 +88,7 @@ async def test_scheduled_event_triggers_claude(
self, event_bus: EventBus, mock_claude: AsyncMock, agent_handler: AgentHandler
) -> None:
"""Scheduled events invoke Claude with the job's prompt."""
- mock_response = MagicMock()
- mock_response.content = "Standup summary"
- mock_claude.run_command.return_value = mock_response
+ mock_claude.run_command.return_value = _response(content="Standup summary")
published: list = []
original_publish = event_bus.publish
@@ -108,9 +118,7 @@ async def test_scheduled_event_with_skill(
self, event_bus: EventBus, mock_claude: AsyncMock, agent_handler: AgentHandler
) -> None:
"""Scheduled events with skill_name prepend the skill invocation."""
- mock_response = MagicMock()
- mock_response.content = "Done"
- mock_claude.run_command.return_value = mock_response
+ mock_claude.run_command.return_value = _response(content="Done")
event = ScheduledEvent(
job_name="standup",
diff --git a/tests/unit/test_events/test_stop_reason_notifications.py b/tests/unit/test_events/test_stop_reason_notifications.py
new file mode 100644
index 000000000..168353615
--- /dev/null
+++ b/tests/unit/test_events/test_stop_reason_notifications.py
@@ -0,0 +1,185 @@
+"""Webhook and scheduled runs carry the stop reason too.
+
+Nobody is watching these. A scheduled job that dies at the turn limit sends
+"No final response. Tools used: ..." and, without the footer, nothing at all
+about why -- the #172 ambiguity, on the one path where no user was there to
+notice the run was short.
+"""
+
+from pathlib import Path
+from typing import Any, Dict, List
+from unittest.mock import AsyncMock
+
+import pytest
+
+from src.claude.sdk_integration import ClaudeResponse
+from src.events.bus import EventBus
+from src.events.handlers import AgentHandler
+from src.events.types import AgentResponseEvent, ScheduledEvent, WebhookEvent
+
+
+def _stopped(**kwargs: Any) -> ClaudeResponse:
+ defaults: Dict[str, Any] = {
+ "content": "No final response. Tools used: Bash, Read",
+ "session_id": "s1",
+ "cost": 0.02,
+ "duration_ms": 900,
+ "num_turns": 10,
+ "result_subtype": "error_max_turns",
+ "terminal_reason": "max_turns",
+ }
+ defaults.update(kwargs)
+ return ClaudeResponse(**defaults)
+
+
+@pytest.fixture
+def event_bus() -> EventBus:
+ return EventBus()
+
+
+@pytest.fixture
+def mock_claude() -> AsyncMock:
+ mock = AsyncMock()
+ mock.run_command = AsyncMock()
+ return mock
+
+
+@pytest.fixture
+def handler(event_bus: EventBus, mock_claude: AsyncMock) -> AgentHandler:
+ return AgentHandler(
+ event_bus=event_bus,
+ claude_integration=mock_claude,
+ default_working_directory=Path("/tmp/test"),
+ default_user_id=42,
+ )
+
+
+def _capture(event_bus: EventBus) -> List[AgentResponseEvent]:
+ published: List[AgentResponseEvent] = []
+ original = event_bus.publish
+
+ async def capture(event: Any) -> None:
+ if isinstance(event, AgentResponseEvent):
+ published.append(event)
+ await original(event)
+
+ event_bus.publish = capture # type: ignore[assignment]
+ return published
+
+
+class TestWebhookNotifications:
+ async def test_a_truncated_webhook_run_says_so(
+ self, event_bus: EventBus, mock_claude: AsyncMock, handler: AgentHandler
+ ) -> None:
+ mock_claude.run_command.return_value = _stopped()
+ published = _capture(event_bus)
+
+ await handler.handle_webhook(
+ WebhookEvent(
+ provider="github",
+ event_type_name="push",
+ payload={},
+ delivery_id="d1",
+ )
+ )
+
+ assert len(published) == 1
+ assert "turn limit reached after 10 turns" in published[0].text
+
+ async def test_a_clean_webhook_run_reads_as_before(
+ self, event_bus: EventBus, mock_claude: AsyncMock, handler: AgentHandler
+ ) -> None:
+ mock_claude.run_command.return_value = _stopped(
+ content="All good.", result_subtype="success", terminal_reason="completed"
+ )
+ published = _capture(event_bus)
+
+ await handler.handle_webhook(
+ WebhookEvent(
+ provider="github",
+ event_type_name="push",
+ payload={},
+ delivery_id="d2",
+ )
+ )
+
+ assert [e.text for e in published] == ["All good."]
+
+ async def test_blocked_calls_reach_the_notification(
+ self, event_bus: EventBus, mock_claude: AsyncMock, handler: AgentHandler
+ ) -> None:
+ mock_claude.run_command.return_value = _stopped(
+ content="Done.",
+ result_subtype="success",
+ terminal_reason="completed",
+ permission_denials=[
+ {"tool_name": "Write", "tool_input": {"file_path": "/etc/hosts"}}
+ ],
+ )
+ published = _capture(event_bus)
+
+ await handler.handle_webhook(
+ WebhookEvent(
+ provider="github",
+ event_type_name="push",
+ payload={},
+ delivery_id="d3",
+ )
+ )
+
+ assert "1 tool call was blocked" in published[0].text
+ assert "/etc/hosts" in published[0].text
+
+
+class TestScheduledNotifications:
+ async def _drain(self, handler: AgentHandler) -> None:
+ for task in list(handler._background_tasks):
+ await task
+
+ async def test_a_truncated_scheduled_run_says_so(
+ self, event_bus: EventBus, mock_claude: AsyncMock, handler: AgentHandler
+ ) -> None:
+ mock_claude.run_command.return_value = _stopped()
+ published = _capture(event_bus)
+
+ await handler.handle_scheduled(
+ ScheduledEvent(
+ job_name="standup", prompt="summarise", target_chat_ids=[100, 200]
+ )
+ )
+ await self._drain(handler)
+
+ assert [e.chat_id for e in published] == [100, 200]
+ for event in published:
+ assert "turn limit reached after 10 turns" in event.text
+
+ async def test_the_default_broadcast_carries_it_too(
+ self, event_bus: EventBus, mock_claude: AsyncMock, handler: AgentHandler
+ ) -> None:
+ mock_claude.run_command.return_value = _stopped()
+ published = _capture(event_bus)
+
+ await handler.handle_scheduled(
+ ScheduledEvent(job_name="nightly", prompt="run", target_chat_ids=[])
+ )
+ await self._drain(handler)
+
+ assert len(published) == 1
+ assert published[0].chat_id == 0
+ assert "turn limit reached" in published[0].text
+
+ async def test_nothing_is_published_when_there_is_nothing_to_say(
+ self, event_bus: EventBus, mock_claude: AsyncMock, handler: AgentHandler
+ ) -> None:
+ """An empty reply from a clean run still publishes nothing."""
+ mock_claude.run_command.return_value = _stopped(
+ content="", result_subtype="success", terminal_reason="completed"
+ )
+ published = _capture(event_bus)
+
+ await handler.handle_scheduled(
+ ScheduledEvent(job_name="quiet", prompt="run", target_chat_ids=[100])
+ )
+ await self._drain(handler)
+
+ assert published == []
diff --git a/tests/unit/test_orchestrator.py b/tests/unit/test_orchestrator.py
index 108e0f808..1902f1898 100644
--- a/tests/unit/test_orchestrator.py
+++ b/tests/unit/test_orchestrator.py
@@ -1040,3 +1040,129 @@ async def test_bot_suffixed_command_not_forwarded(agentic_settings, deps):
) as mock_claude:
await orchestrator._handle_unknown_command(update, context)
mock_claude.assert_not_called()
+
+
+async def _run_agentic_text_with(claude_response, agentic_settings, deps):
+ """Drive agentic_text with a given ClaudeResponse and return the texts sent."""
+ orchestrator = MessageOrchestrator(agentic_settings, deps)
+
+ claude_integration = AsyncMock()
+ claude_integration.run_command = AsyncMock(return_value=claude_response)
+
+ update = MagicMock()
+ update.effective_user.id = 123
+ update.message.text = "Do something big"
+ update.message.message_id = 1
+ update.message.chat.send_action = AsyncMock()
+ update.message.reply_text = AsyncMock()
+
+ progress_msg = AsyncMock()
+ progress_msg.delete = AsyncMock()
+ update.message.reply_text.return_value = progress_msg
+
+ context = MagicMock()
+ context.user_data = {}
+ context.bot_data = {
+ "settings": agentic_settings,
+ "claude_integration": claude_integration,
+ "storage": None,
+ "rate_limiter": None,
+ "audit_logger": None,
+ }
+
+ await orchestrator.agentic_text(update, context)
+
+ # The first reply_text call is the progress message, not a response.
+ return [
+ call.args[0]
+ for call in update.message.reply_text.call_args_list[1:]
+ if call.args
+ ]
+
+
+async def test_agentic_text_reports_turn_limit(agentic_settings, deps):
+ """A run killed at the turn limit says so instead of claiming success (#172)."""
+ from src.claude.sdk_integration import ClaudeResponse
+
+ response = ClaudeResponse(
+ content="Started refactoring...",
+ session_id="session-abc",
+ cost=0.1,
+ duration_ms=100,
+ num_turns=10,
+ result_subtype="error_max_turns",
+ terminal_reason="max_turns",
+ )
+
+ sent = await _run_agentic_text_with(response, agentic_settings, deps)
+
+ assert sent, "expected a response message"
+ body = "\n".join(sent)
+ assert "turn limit reached after 10 turns" in body
+ assert "Send a message to continue" in body
+
+
+async def test_agentic_text_lists_blocked_tool_calls(agentic_settings, deps):
+ """Denials the bot itself generated are reported to the user."""
+ from src.claude.sdk_integration import ClaudeResponse
+
+ response = ClaudeResponse(
+ content="I could not write that file.",
+ session_id="session-abc",
+ cost=0.1,
+ duration_ms=100,
+ num_turns=2,
+ result_subtype="success",
+ permission_denials=[
+ {"tool_name": "Write", "tool_input": {"file_path": "/etc/hosts"}}
+ ],
+ )
+
+ sent = await _run_agentic_text_with(response, agentic_settings, deps)
+
+ body = "\n".join(sent)
+ assert "1 tool call was blocked" in body
+ assert "/etc/hosts" in body
+
+
+async def test_agentic_text_adds_no_footer_to_a_clean_run(agentic_settings, deps):
+ """A run that finished normally reads exactly as it did before."""
+ from src.claude.sdk_integration import ClaudeResponse
+
+ response = ClaudeResponse(
+ content="All done.",
+ session_id="session-abc",
+ cost=0.1,
+ duration_ms=100,
+ num_turns=2,
+ result_subtype="success",
+ )
+
+ sent = await _run_agentic_text_with(response, agentic_settings, deps)
+
+ body = "\n".join(sent)
+ assert "Stopped" not in body
+ assert "blocked" not in body
+
+
+async def test_agentic_text_does_not_say_stopped_twice(agentic_settings, deps):
+ """The turn-limit-mid-tool-use case: no final text, so the placeholder
+ stands in for the reply and the footer explains. One warning, not two."""
+ from src.claude.sdk_integration import TASK_STOPPED_MSG, ClaudeResponse
+
+ response = ClaudeResponse(
+ content=TASK_STOPPED_MSG.format(tools_summary="Bash, Read"),
+ session_id="session-abc",
+ cost=0.1,
+ duration_ms=100,
+ num_turns=10,
+ result_subtype="error_max_turns",
+ terminal_reason="max_turns",
+ )
+
+ sent = await _run_agentic_text_with(response, agentic_settings, deps)
+
+ body = "\n".join(sent)
+ assert "No final response. Tools used: Bash, Read" in body
+ assert "turn limit reached after 10 turns" in body
+ assert body.count("⚠️") == 1