feat(reply): cap interactive LLM replies to 3 mesh packets (#7)

Add terse-answer guidance to the interactive system prompt and a hard
ceiling of 3 packets (3 x connector.max_chars) on LLM replies, with an
"ask for more" indicator when truncated. Protects LoRa airtime from
runaway replies. Broadcast chunking unchanged.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
malice 2026-07-02 16:49:10 -06:00 committed by GitHub
commit 7c15fa3f09
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 160 additions and 1 deletions

View file

@ -0,0 +1,117 @@
"""Tests for the interactive LLM reply cap (MAX_REPLY_PACKETS).
All tests are hermetic no real LLM, no real socket.
"""
import pytest
from meshai.chunker import cap_reply_chunks, chunk_response, MAX_REPLY_PACKETS, _TRUNCATION_INDICATOR
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_chunks(n: int, chars_each: int = 50) -> list[str]:
"""Build n chunks of exactly chars_each ASCII characters."""
return [f"chunk{i}" + "x" * (chars_each - len(f"chunk{i}")) for i in range(n)]
# ---------------------------------------------------------------------------
# Test 1 — long reply produces exactly 3 chunks with truncation indicator
# ---------------------------------------------------------------------------
def test_long_reply_capped_to_three_chunks():
"""A reply long enough to produce >3 chunks results in exactly 3 delivered
chunks, and the truncation indicator is present in the last chunk."""
max_chars = 140
# Build a response that will produce >3 chunks at 140 chars each.
# Each sentence is ~130 chars so 5 sentences → 5 chunks.
sentence = "A" * 130 + "."
text = " ".join([sentence] * 5)
# chunk_response with a high max_messages so it happily returns 5 chunks
raw_chunks, _ = chunk_response(text, max_chars=max_chars, max_messages=10)
assert len(raw_chunks) > MAX_REPLY_PACKETS, (
f"Precondition: expected >3 raw chunks, got {len(raw_chunks)}"
)
capped = cap_reply_chunks(raw_chunks, MAX_REPLY_PACKETS, max_chars)
assert len(capped) == MAX_REPLY_PACKETS
assert _TRUNCATION_INDICATOR in capped[-1], (
f"Expected truncation indicator in last chunk, got: {capped[-1]!r}"
)
# ---------------------------------------------------------------------------
# Test 2 — short reply is delivered unchanged, no indicator
# ---------------------------------------------------------------------------
def test_short_reply_unchanged():
"""A short reply (<=3 chunks) is delivered unchanged with no indicator."""
max_chars = 140
# One chunk well under the limit
chunks_1 = ["Hello, this is a short reply."]
result_1 = cap_reply_chunks(chunks_1, MAX_REPLY_PACKETS, max_chars)
assert result_1 == chunks_1
assert _TRUNCATION_INDICATOR not in result_1[-1]
# Exactly 3 chunks — boundary condition
chunks_3 = _make_chunks(3, chars_each=50)
result_3 = cap_reply_chunks(chunks_3, MAX_REPLY_PACKETS, max_chars)
assert result_3 == chunks_3
assert _TRUNCATION_INDICATOR not in result_3[-1]
# ---------------------------------------------------------------------------
# Test 3 — cap respects connector.max_chars=140
# ---------------------------------------------------------------------------
def test_cap_respects_max_chars_140():
"""cap_reply_chunks uses the max_chars budget (140) from connector.max_chars."""
max_chars = 140
# 5 chunks of 80 chars each (well under max_chars individually)
chunks = _make_chunks(5, chars_each=80)
capped = cap_reply_chunks(chunks, MAX_REPLY_PACKETS, max_chars)
assert len(capped) == 3
# Last chunk must not exceed max_chars in total length
assert len(capped[-1]) <= max_chars
assert _TRUNCATION_INDICATOR in capped[-1]
# ---------------------------------------------------------------------------
# Test 4 — 3rd chunk trimmed correctly when at max_chars limit
# ---------------------------------------------------------------------------
def test_third_chunk_trimmed_to_fit_indicator():
"""When the 3rd chunk fills max_chars exactly, it is trimmed so that
trimmed_chunk + indicator == max_chars exactly."""
max_chars = 140
indicator = _TRUNCATION_INDICATOR # len 16 Python chars (" …(ask for more)")
# 3rd chunk that is exactly max_chars long
full_chunk = "B" * max_chars
chunks = [
"First chunk.",
"Second chunk.",
full_chunk,
"Fourth chunk (should be dropped).",
]
capped = cap_reply_chunks(chunks, MAX_REPLY_PACKETS, max_chars)
assert len(capped) == MAX_REPLY_PACKETS
last = capped[-1]
assert last.endswith(indicator), f"Expected indicator at end, got: {last!r}"
assert len(last) == max_chars, (
f"Expected trimmed chunk + indicator == {max_chars}, got len={len(last)}"
)
# The trimmed body is max_chars - len(indicator) Bs
expected_body = "B" * (max_chars - len(indicator))
assert last == expected_body + indicator