feat(ai): stream explanations live (Ollama NDJSON + Claude SSE) — 0.33.0

ai.explain_stream(findings_text, on_chunk) streams token deltas and returns
(ok, full_text). Ollama: stream=True NDJSON; Claude: stream=True SSE (parse
content_block_delta text deltas). The diagnostic dialog opens an explanation
window immediately and fills it token-by-token via a _chunk signal, then
re-renders the finished answer as Markdown — no more multi-second freeze on a
local model. Non-streaming explain() kept for the CLI. Tests for both parsers;
verified live against qwen2.5:7b.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-22 14:23:15 +02:00
parent 4bd51a40c3
commit bbc22fa288
6 changed files with 169 additions and 17 deletions
+46
View File
@@ -114,5 +114,51 @@ class ExplainTests(unittest.TestCase):
self.assertEqual(headers["x-api-key"], "sk-ant-x")
class _FakeResp:
"""A context-managed iterable of byte lines, like urlopen() returns."""
def __init__(self, lines):
self._lines = [l.encode("utf-8") for l in lines]
def __enter__(self):
return iter(self._lines)
def __exit__(self, *a):
return False
class StreamTests(unittest.TestCase):
def _cfg(self, **over):
base = {"ai_provider": "", "ai_model": "", "ai_endpoint": "http://localhost:11434"}
base.update(over)
return base
def test_ollama_stream_accumulates_and_callbacks(self):
lines = ['{"response": "It is ", "done": false}',
'{"response": "the PSU.", "done": false}',
'{"response": "", "done": true}']
chunks = []
with mock.patch.object(ai.config, "load_config",
return_value=self._cfg(ai_provider="ollama", ai_model="qwen2.5:7b")), \
mock.patch.object(ai, "_stream_request", return_value=_FakeResp(lines)):
ok, full = ai.explain_stream("Xid 79", on_chunk=chunks.append)
self.assertTrue(ok)
self.assertEqual(full, "It is the PSU.")
self.assertEqual(chunks, ["It is ", "the PSU."])
def test_claude_stream_parses_sse(self):
lines = [
'event: content_block_delta',
'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"Failing "}}',
'data: {"type":"content_block_delta","delta":{"type":"text_delta","text":"disk."}}',
'data: {"type":"message_stop"}',
]
chunks = []
with mock.patch.object(ai.config, "load_config", return_value=self._cfg(ai_provider="claude")), \
mock.patch.object(ai.config, "load_ai_key", return_value="sk-ant-x"), \
mock.patch.object(ai, "_stream_request", return_value=_FakeResp(lines)):
ok, full = ai.explain_stream("SMART 197", on_chunk=chunks.append)
self.assertTrue(ok)
self.assertEqual(full, "Failing disk.")
self.assertEqual(chunks, ["Failing ", "disk."])
if __name__ == "__main__":
unittest.main()