mirror of
https://github.com/zvx-echo6/meshai.git
synced 2026-08-26 17:31:34 +00:00
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>
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
"""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
|