242 lines
11 KiB
Python
242 lines
11 KiB
Python
"""Tests for reflect_distill.py — mechanical distillation of session logs."""
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
|
|
import reflect_distill
|
|
|
|
|
|
def _write_session(directory: Path, name: str, records: list[dict]) -> Path:
|
|
path = directory / f"{name}.jsonl"
|
|
path.write_text("\n".join(json.dumps(r, ensure_ascii=False) for r in records), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def _conversation(turns: int = 3) -> list[dict]:
|
|
"""A session long enough to clear MIN_MESSAGES."""
|
|
records: list[dict] = [{"_type": "metadata", "key": "websocket:x", "created_at": "2026-07-11T14:02:03"}]
|
|
for i in range(turns):
|
|
records.append({"role": "user", "content": f"dotaz {i}"})
|
|
records.append({"role": "assistant", "content": f"answer {i}"})
|
|
return records
|
|
|
|
|
|
class TestNoiseFilter:
|
|
@pytest.mark.parametrize(
|
|
"name",
|
|
[
|
|
"reflect_20260908-033000",
|
|
"compact-memory-auto_20260715-054147",
|
|
"detach_abc",
|
|
"dream_xyz",
|
|
"cron_e81cda77",
|
|
"cli_kimi-ollama-test",
|
|
"wiki-compile",
|
|
"note-compile",
|
|
],
|
|
)
|
|
def test_machinery_prefixes_are_noise(self, name):
|
|
assert reflect_distill.is_noise(name)
|
|
|
|
@pytest.mark.parametrize("name", ["wiki-capture-test2", "test-note-search-zzz"])
|
|
def test_prefix_variants_are_noise(self, name):
|
|
"""Exact matching used to let `…test2` through; the filter matches prefixes."""
|
|
assert reflect_distill.is_noise(name)
|
|
|
|
@pytest.mark.parametrize("name", ["websocket_e5a6aac3-2c0a", "telegram_8826147089"])
|
|
def test_real_conversations_are_kept(self, name):
|
|
assert not reflect_distill.is_noise(name)
|
|
|
|
def test_base64_session_names_are_decoded(self):
|
|
"""Older sessions are stored base64-encoded; `ZHJlYW06…` is `dream:…`."""
|
|
assert reflect_distill.is_noise("ZHJlYW06MjAyNjA4MzEtMTA0NjM2")
|
|
assert not reflect_distill.is_noise("d2Vic29ja2V0OjUyZDBmMzM4")
|
|
|
|
def test_reflect_excludes_its_own_sessions(self):
|
|
"""Without this the skill would analyse its own runs on the next pass."""
|
|
assert reflect_distill.is_noise("reflect_20260908-033000")
|
|
|
|
|
|
class TestDistillSession:
|
|
def test_short_sessions_are_skipped(self, tmp_path):
|
|
path = _write_session(tmp_path, "websocket_short", [{"role": "user", "content": "ahoj"}])
|
|
assert reflect_distill.distill_session(path) is None
|
|
|
|
def test_noise_sessions_are_skipped(self, tmp_path):
|
|
path = _write_session(tmp_path, "dream_nightly", _conversation())
|
|
assert reflect_distill.distill_session(path) is None
|
|
|
|
def test_header_carries_key_time_and_count(self, tmp_path):
|
|
path = _write_session(tmp_path, "websocket_abc", _conversation())
|
|
digest = reflect_distill.distill_session(path)
|
|
assert digest.text.startswith(reflect_distill.SESSION_MARKER)
|
|
assert "websocket_abc" in digest.text
|
|
assert "2026-07-11 14:02" in digest.text
|
|
assert digest.message_count == 6
|
|
|
|
def test_header_is_not_a_markdown_heading(self, tmp_path):
|
|
"""Assistant prose is full of `###`, so a heading would not read as a boundary."""
|
|
records = _conversation()
|
|
records.append({"role": "assistant", "content": "### Summary\ndone"})
|
|
path = _write_session(tmp_path, "websocket_abc", records)
|
|
digest = reflect_distill.distill_session(path)
|
|
assert not digest.text.startswith("#")
|
|
assert digest.text.count(reflect_distill.SESSION_MARKER) == 1
|
|
|
|
def test_user_and_assistant_prose_is_kept_whole(self, tmp_path):
|
|
prose = "this is a long answer " * 40
|
|
records = _conversation()
|
|
records.append({"role": "assistant", "content": prose})
|
|
path = _write_session(tmp_path, "websocket_abc", records)
|
|
assert prose.strip() in reflect_distill.distill_session(path).text
|
|
|
|
def test_slash_commands_are_dropped(self, tmp_path):
|
|
records = _conversation()
|
|
records.append({"role": "user", "content": "/model", "_command": True})
|
|
path = _write_session(tmp_path, "websocket_abc", records)
|
|
assert "/model" not in reflect_distill.distill_session(path).text
|
|
|
|
def test_decoded_key_is_used_as_name(self, tmp_path):
|
|
path = _write_session(tmp_path, "d2Vic29ja2V0OjUyZDBmMzM4", _conversation())
|
|
assert reflect_distill.distill_session(path).name == "websocket:52d0f338"
|
|
|
|
|
|
class TestToolRendering:
|
|
def _digest_with_tool(self, tmp_path, call, result):
|
|
records = _conversation()
|
|
records.append({"role": "assistant", "content": "", "tool_calls": [call]})
|
|
records.append(result)
|
|
path = _write_session(tmp_path, "websocket_abc", records)
|
|
return reflect_distill.distill_session(path).text
|
|
|
|
def test_tool_result_body_is_replaced_by_metadata(self, tmp_path):
|
|
"""Tool results are 89% of the corpus and carry almost no diagnostic value."""
|
|
body = "x" * 5000
|
|
text = self._digest_with_tool(
|
|
tmp_path,
|
|
{"id": "c1", "function": {"name": "read_file", "arguments": '{"path": "/tmp/a.md"}'}},
|
|
{"role": "tool", "tool_call_id": "c1", "name": "read_file", "content": body},
|
|
)
|
|
assert body not in text
|
|
assert "read_file: ok, 5.0 kB" in text
|
|
|
|
def test_errors_are_flagged_with_their_message(self, tmp_path):
|
|
text = self._digest_with_tool(
|
|
tmp_path,
|
|
{"id": "c1", "function": {"name": "web_fetch", "arguments": '{"url": "https://x.dev"}'}},
|
|
{"role": "tool", "tool_call_id": "c1", "name": "web_fetch", "content": "Error: 403 Forbidden"},
|
|
)
|
|
assert "ERROR Error: 403 Forbidden" in text
|
|
|
|
def test_workspace_path_prefix_is_stripped(self, tmp_path):
|
|
text = self._digest_with_tool(
|
|
tmp_path,
|
|
{
|
|
"id": "c1",
|
|
"function": {
|
|
"name": "read_file",
|
|
"arguments": '{"path": "/home/nanobot/.nanobot/workspace/skills/note/SKILL.md"}',
|
|
},
|
|
},
|
|
{"role": "tool", "tool_call_id": "c1", "name": "read_file", "content": "ok"},
|
|
)
|
|
assert "path=skills/note/SKILL.md" in text
|
|
|
|
def test_malformed_arguments_still_render(self, tmp_path):
|
|
"""A truncated argument blob is itself a signal worth seeing, not a crash."""
|
|
text = self._digest_with_tool(
|
|
tmp_path,
|
|
{"id": "c1", "function": {"name": "exec", "arguments": '{"command": "ls'}},
|
|
{"role": "tool", "tool_call_id": "c1", "name": "exec", "content": "ok"},
|
|
)
|
|
assert "exec(" in text
|
|
|
|
def test_long_urls_keep_their_distinguishing_tail(self, tmp_path):
|
|
"""Front-truncation made different fetches look identical and hid real retries."""
|
|
base = "https://raw.githubusercontent.com/some-owner/some-repo/main/packages/core/src/"
|
|
first = self._digest_with_tool(
|
|
tmp_path,
|
|
{"id": "c1", "function": {"name": "web_fetch", "arguments": json.dumps({"url": base + "alpha.ts"})}},
|
|
{"role": "tool", "tool_call_id": "c1", "name": "web_fetch", "content": "ok"},
|
|
)
|
|
second = self._digest_with_tool(
|
|
tmp_path,
|
|
{"id": "c1", "function": {"name": "web_fetch", "arguments": json.dumps({"url": base + "omega.ts"})}},
|
|
{"role": "tool", "tool_call_id": "c1", "name": "web_fetch", "content": "ok"},
|
|
)
|
|
assert "alpha.ts" in first
|
|
assert "omega.ts" in second
|
|
assert first != second
|
|
|
|
def test_repeated_identical_calls_render_identically(self, tmp_path):
|
|
"""The LLM detects retry loops by seeing the same line twice — so it must match."""
|
|
call = {"id": "c1", "function": {"name": "web_fetch", "arguments": '{"url": "https://x.dev/a"}'}}
|
|
records = _conversation()
|
|
for _ in range(2):
|
|
records.append({"role": "assistant", "content": "", "tool_calls": [call]})
|
|
records.append({"role": "tool", "tool_call_id": "c1", "name": "web_fetch", "content": "Error: 403"})
|
|
path = _write_session(tmp_path, "websocket_abc", records)
|
|
lines = [line for line in reflect_distill.distill_session(path).text.splitlines() if "→ web_fetch" in line]
|
|
assert len(lines) == 2
|
|
assert lines[0] == lines[1]
|
|
|
|
def test_reasoning_is_shortened_not_dropped(self, tmp_path):
|
|
records = _conversation()
|
|
records.append({"role": "assistant", "content": "ok", "reasoning_content": "reasoning " * 200})
|
|
path = _write_session(tmp_path, "websocket_abc", records)
|
|
text = reflect_distill.distill_session(path).text
|
|
assert "~ reasoning" in text
|
|
assert len(text) < 2000
|
|
|
|
|
|
class TestBatching:
|
|
def test_batches_respect_the_character_budget(self, tmp_path):
|
|
for i in range(6):
|
|
_write_session(tmp_path, f"websocket_{i}", _conversation(turns=8))
|
|
paths = reflect_distill.collect_sessions(tmp_path)
|
|
batches = list(reflect_distill.iter_batches(paths, budget_chars=400))
|
|
assert len(batches) > 1
|
|
assert all(batch for batch in batches)
|
|
|
|
def test_oversized_session_is_not_split_or_dropped(self, tmp_path):
|
|
"""Truncating a huge session would hide exactly the runaway loops worth finding."""
|
|
records = _conversation()
|
|
records.append({"role": "assistant", "content": "y" * 5000})
|
|
_write_session(tmp_path, "websocket_big", records)
|
|
batches = list(reflect_distill.iter_batches(reflect_distill.collect_sessions(tmp_path), budget_chars=100))
|
|
assert len(batches) == 1
|
|
assert len(batches[0][0]) > 5000
|
|
|
|
def test_collect_sessions_skips_noise(self, tmp_path):
|
|
_write_session(tmp_path, "websocket_keep", _conversation())
|
|
_write_session(tmp_path, "dream_drop", _conversation())
|
|
names = [p.stem for p in reflect_distill.collect_sessions(tmp_path)]
|
|
assert names == ["websocket_keep"]
|
|
|
|
def test_cursor_written_by_a_run_excludes_those_sessions_next_time(self, tmp_path):
|
|
"""The round trip reflect_auto actually makes: digest.started becomes the next `since`.
|
|
|
|
A display-formatted cursor (`2026-07-11 14:02`) compares wrong against the raw ISO in
|
|
the log, because `T` sorts above a space — every session of that day came back.
|
|
"""
|
|
_write_session(tmp_path, "websocket_done", _conversation())
|
|
digests = [
|
|
d
|
|
for batch in reflect_distill.iter_batches(reflect_distill.collect_sessions(tmp_path), 10**6)
|
|
for d in batch
|
|
]
|
|
cursor = max(digest.started for digest in digests)
|
|
assert reflect_distill.collect_sessions(tmp_path, since=cursor) == []
|
|
|
|
def test_since_excludes_already_processed_sessions(self, tmp_path):
|
|
_write_session(tmp_path, "websocket_old", _conversation())
|
|
new = _conversation()
|
|
new[0]["created_at"] = "2026-08-20T09:00:00"
|
|
_write_session(tmp_path, "websocket_new", new)
|
|
names = [p.stem for p in reflect_distill.collect_sessions(tmp_path, since="2026-08-01T00:00:00")]
|
|
assert names == ["websocket_new"]
|