Jack1808/Claude_Code

1from unittest.mock import patch2 3from messaging.rendering.telegram_markdown import (4    escape_md_v2,5    escape_md_v2_code,6    mdv2_bold,7    mdv2_code_inline,8    render_markdown_to_mdv2,9)10from messaging.transcript import RenderCtx, TranscriptBuffer11 12 13def _ctx() -> RenderCtx:14    return RenderCtx(15        bold=mdv2_bold,16        code_inline=mdv2_code_inline,17        escape_code=escape_md_v2_code,18        escape_text=escape_md_v2,19        render_markdown=render_markdown_to_mdv2,20        thinking_tail_max=1000,21        tool_input_tail_max=1200,22        tool_output_tail_max=1600,23        text_tail_max=2000,24    )25 26 27def test_transcript_order_thinking_tool_text():28    t = TranscriptBuffer()29    t.apply({"type": "thinking_chunk", "text": "think1"})30    t.apply({"type": "tool_use", "id": "tool_1", "name": "ls", "input": {"path": "."}})31    t.apply({"type": "text_chunk", "text": "done"})32 33    out = t.render(_ctx(), limit_chars=3900, status=None)34    assert out.find("think1") < out.find("Tool call:") < out.find("done")35 36 37def test_transcript_subagent_suppresses_thinking_and_text_inside():38    t = TranscriptBuffer()39 40    # Enter subagent context (Task tool call).41    t.apply(42        {43            "type": "tool_use",44            "id": "task_1",45            "name": "Task",46            "input": {"description": "Fix bug"},47        }48    )49 50    # These should be suppressed while inside subagent context.51    t.apply({"type": "thinking_delta", "index": -1, "text": "secret"})52    t.apply({"type": "text_chunk", "text": "visible?"})53 54    # Tool activity should still show.55    t.apply({"type": "tool_use", "id": "tool_2", "name": "ls", "input": {"path": "."}})56    t.apply({"type": "tool_result", "tool_use_id": "tool_2", "content": "x"})57 58    # Close subagent context (Task tool result).59    t.apply({"type": "tool_result", "tool_use_id": "task_1", "content": "done"})60 61    # Now text should show again.62    t.apply({"type": "text_chunk", "text": "after"})63 64    out = t.render(_ctx(), limit_chars=3900, status=None)65    assert "Subagent:" in out66    assert "secret" not in out67    assert "visible?" not in out68    # Only the current tool call should be shown (not the full history).69    assert out.count("Tool call:") == 170    assert "\n  🛠" in out or out.startswith("  🛠") or "  🛠" in out71    assert "Tools used:" in out72    assert "Tool calls:" in out73    assert "after" in out74 75 76def test_transcript_subagent_closes_on_whitespace_tool_ids():77    t = TranscriptBuffer()78 79    # Provider emitted a Task tool_use id with leading whitespace.80    t.apply(81        {82            "type": "tool_use",83            "id": " functions.Task:0",84            "name": "Task",85            "input": {"description": "Outer"},86        }87    )88 89    # Task completes, but tool_result references a trimmed id (or vice versa).90    t.apply(91        {"type": "tool_result", "tool_use_id": "functions.Task:0", "content": "done"}92    )93 94    # Next Task should be top-level, not nested under the previous subagent.95    t.apply(96        {97            "type": "tool_use",98            "id": "functions.Task:1",99            "name": "Task",100            "input": {"description": "Next"},101        }102    )103 104    out = t.render(_ctx(), limit_chars=3900, status=None)105    assert out.count("Subagent:") == 2106    # If nesting is incorrect, the second subagent line will be indented under the first.107    assert "\n  🤖 *Subagent:* `Next`" not in out108 109 110def test_transcript_subagent_closes_on_task_result_id_suffix_match():111    t = TranscriptBuffer()112    t.apply(113        {114            "type": "tool_use",115            "id": "task_1",116            "name": "Task",117            "input": {"description": "Outer"},118        }119    )120    t.apply({"type": "tool_result", "tool_use_id": "task_1_result", "content": "done"})121    t.apply(122        {123            "type": "tool_use",124            "id": "task_2",125            "name": "Task",126            "input": {"description": "Next"},127        }128    )129 130    out = t.render(_ctx(), limit_chars=3900, status=None)131    assert out.count("Subagent:") == 2132    assert "\n  🤖 *Subagent:* `Next`" not in out133 134 135def test_transcript_unmatched_non_task_tool_result_does_not_pop_subagent():136    t = TranscriptBuffer()137    t.apply(138        {139            "type": "tool_use",140            "id": "task_1",141            "name": "Task",142            "input": {"description": "Outer"},143        }144    )145    t.apply({"type": "tool_result", "tool_use_id": "totally_unrelated", "content": "x"})146 147    assert t._subagent_stack == ["task_1"]148 149 150def test_transcript_sequential_tasks_mismatched_results_no_depth_drift():151    t = TranscriptBuffer()152    t.apply(153        {154            "type": "tool_use",155            "id": "task_1",156            "name": "Task",157            "input": {"description": "A"},158        }159    )160    t.apply({"type": "tool_result", "tool_use_id": "task_1_result", "content": "done"})161    t.apply(162        {163            "type": "tool_use",164            "id": "task_2",165            "name": "Task",166            "input": {"description": "B"},167        }168    )169    t.apply({"type": "tool_result", "tool_use_id": "task_2_result", "content": "done"})170    t.apply(171        {172            "type": "tool_use",173            "id": "task_3",174            "name": "Task",175            "input": {"description": "C"},176        }177    )178 179    out = t.render(_ctx(), limit_chars=3900, status=None)180    assert "🤖 *Subagent:* `A`\n  🤖 *Subagent:* `B`" not in out181    assert "\n  🤖 *Subagent:* `C`" not in out182    assert t._subagent_stack == ["task_3"]183 184 185def test_transcript_synthetic_task_start_closes_on_functions_task_result_id():186    t = TranscriptBuffer()187    t.apply(188        {189            "type": "tool_use_start",190            "index": 0,191            "id": "",192            "name": "Task",193            "input": {"description": "Outer"},194        }195    )196    t.apply({"type": "tool_result", "tool_use_id": "functions.Task:0", "content": "x"})197    t.apply(198        {199            "type": "tool_use_start",200            "index": 1,201            "id": "",202            "name": "Task",203            "input": {"description": "Next"},204        }205    )206 207    out = t.render(_ctx(), limit_chars=3900, status=None)208    assert out.count("Subagent:") == 2209    assert "\n  🤖 *Subagent:* `Next`" not in out210 211 212def test_transcript_synthetic_task_not_closed_by_unknown_non_task_result_id():213    t = TranscriptBuffer()214    t.apply(215        {216            "type": "tool_use_start",217            "index": 0,218            "id": "",219            "name": "Task",220            "input": {"description": "Outer"},221        }222    )223    t.apply({"type": "tool_result", "tool_use_id": "call_deadbeef", "content": "x"})224 225    assert t._subagent_stack == ["__task_1"]226 227 228def test_transcript_overlapping_tasks_are_flat_not_nested():229    t = TranscriptBuffer()230    t.apply(231        {232            "type": "tool_use",233            "id": "task_a",234            "name": "Task",235            "input": {"description": "A"},236        }237    )238    t.apply(239        {240            "type": "tool_use",241            "id": "task_b",242            "name": "Task",243            "input": {"description": "B"},244        }245    )246    t.apply({"type": "tool_result", "tool_use_id": "task_b", "content": "done"})247    t.apply({"type": "tool_result", "tool_use_id": "task_a", "content": "done"})248 249    out = t.render(_ctx(), limit_chars=3900, status=None)250    assert "🤖 *Subagent:* `A`" in out251    assert "🤖 *Subagent:* `B`" in out252    assert out.find("🤖 *Subagent:* `A`") < out.find("🤖 *Subagent:* `B`")253    assert "\n  🤖 *Subagent:* `B`" not in out254 255 256def test_transcript_truncates_by_dropping_oldest_segments():257    t = TranscriptBuffer()258 259    # Create many segments by opening/closing distinct text blocks.260    for i in range(60):261        t.apply({"type": "text_start", "index": i})262        t.apply(263            {"type": "text_delta", "index": i, "text": f"segment_{i} " + ("x" * 120)}264        )265        t.apply({"type": "block_stop", "index": i})266 267    out = t.render(_ctx(), limit_chars=600, status="status")268    assert escape_md_v2("... (truncated)") in out269    # We keep the tail and drop the oldest segments when truncating.270    assert escape_md_v2("segment_59") in out271    assert escape_md_v2("segment_0") not in out272 273 274def test_transcript_render_many_segments_completes_quickly():275    """Render with 200+ segments exercises O(n) truncation (deque popleft)."""276    t = TranscriptBuffer()277    for i in range(200):278        t.apply({"type": "text_start", "index": i})279        t.apply({"type": "text_delta", "index": i, "text": f"seg_{i} " + ("y" * 80)})280        t.apply({"type": "block_stop", "index": i})281 282    out = t.render(_ctx(), limit_chars=500, status="ok")283    assert escape_md_v2("... (truncated)") in out284    assert "199" in out  # last segment (MarkdownV2 escapes underscores)285    assert "seg_0 " not in out  # oldest segment dropped286 287 288def test_transcript_reused_index_closes_previous_open_block():289    t = TranscriptBuffer()290    # Open a text block at index 0, but never close it.291    t.apply({"type": "text_start", "index": 0})292    t.apply({"type": "text_delta", "index": 0, "text": "a"})293    # Provider reuses index 0 for a new tool block without a stop.294    t.apply(295        {"type": "tool_use_start", "index": 0, "id": "t1", "name": "ls", "input": {}}296    )297    # Old open text should have been closed.298    assert 0 not in t._open_text_by_index299    assert 0 in t._open_tools_by_index300 301 302def test_transcript_render_segment_exception_skipped():303    """When a segment's render() raises, that segment is skipped and rest is rendered."""304    t = TranscriptBuffer()305    t.apply({"type": "thinking_chunk", "text": "before"})306    t.apply({"type": "text_chunk", "text": "middle"})307    t.apply({"type": "text_chunk", "text": "after"})308 309    bad_segment = t._segments[1]310 311    def _raising_render(self, ctx):312        raise ValueError("render failed")313 314    with patch.object(bad_segment, "render", _raising_render):315        out = t.render(_ctx(), limit_chars=3900, status=None)316    assert "before" in out317    assert "after" in out318    assert "middle" not in out319 320 321def test_transcript_render_status_only_exceeds_limit():322    """When all segments dropped, status-only output; long status returned as-is."""323    t = TranscriptBuffer()324    t.apply({"type": "text_chunk", "text": "x" * 5000})325 326    long_status = "A" * 500327    msg = t.render(_ctx(), limit_chars=100, status=long_status)328    assert "... (truncated)" in msg or long_status in msg329 330 331def test_transcript_truncation_preserves_last_segment_tail():332    """When all segments exceed limit, preserve tail of last segment (not just marker+status)."""333    t = TranscriptBuffer()334    t.apply({"type": "thinking_chunk", "text": "Thinking..."})335    t.apply(336        {"type": "text_chunk", "text": "The actual output content here" + "x" * 500}337    )338 339    msg = t.render(_ctx(), limit_chars=100, status="✅ *Complete*")340    # Must include actual content (tail of last segment), not only "... (truncated)\n✅ *Complete*"341    assert escape_md_v2("... (truncated)") in msg342    assert "✅ *Complete*" in msg343    assert "actual output" in msg or "content" in msg or "x" in msg344