"""
QThread worker that streams chat completions from a ChatProvider so
the UI stays responsive during long generations.
Emits:
stage_changed(str) — coarse progress: "connecting", "streaming"
chunk_ready(str) — a partial completion chunk
finished(bool, str) — (ok, full_text_or_error)
"""
from __future__ import annotations
import sys
import traceback
from typing import Dict, List, Optional
from PySide6.QtCore import QObject, QThread, Signal
from .providers import ChatProvider, ProviderFailed, question_cache_key
[docs]
class StreamWorker(QObject):
"""QObject that drives one provider stream on a worker QThread.
:ivar stage_changed: coarse progress signal ("connecting", "streaming").
:ivar chunk_ready: emitted with each partial completion chunk.
:ivar finished: emitted with ``(ok, full_text_or_error)`` on completion.
"""
stage_changed = Signal(str)
chunk_ready = Signal(str)
finished = Signal(bool, str)
def __init__(
self,
provider: ChatProvider,
messages: List[Dict],
system: str = "",
model: Optional[str] = None,
):
"""Prepare the worker; call :meth:`run` from a QThread's ``started`` signal.
:param provider: the ChatProvider to stream from.
:param messages: conversation history to send.
:param system: optional system prompt.
:param model: optional model override.
"""
super().__init__()
self._provider = provider
self._messages = messages
self._system = system
self._model = model
self._cancelled = False
self.saved = False
[docs]
def cancel(self) -> None:
"""Cancel: kill the subprocess so the reader unblocks.
Setting a Python flag alone isn't enough — the worker is
blocked in a `for line in proc.stdout` iteration until the
subprocess writes or closes. This method terminates the subprocess
directly via `provider.cancel_stream()`; the reader then
exits with an empty read and run() completes cleanly.
"""
self._cancelled = True
try:
self._provider.cancel_stream()
except Exception:
pass
[docs]
def run(self) -> None:
"""Consume the provider stream, emitting stage/chunk/finished signals.
``finished`` carries ``(True, the whole reply)`` only when the stream
ended on its own and the provider did not report a failure. Anything
else gives ``(False, <what to tell the user>)``, and whatever the
provider printed before failing is not an answer. A stream that was
cancelled gives ``(False, "Cancelled.")`` even when ending the child
made the provider raise, because the user asked for the stop.
The exception's class name is prefixed to its message, except for a
:class:`~spacr.qt.ai.providers.ProviderFailed` with a non-empty
message. The console writes this text after "[AI error] ", and a
ProviderFailed message is written to be read there, including the
sign-in command to run. For every other exception the class name is
kept because it often names the error: a bare ``[Errno 2] No such
file or directory`` does not say it is a FileNotFoundError.
A fresh, short question
(:func:`~spacr.qt.ai.providers.question_cache_key`) asked before is
answered from this computer's saved answers with no provider call,
and :attr:`saved` is then ``True``; a new complete answer to such a
question is saved, whatever the provider.
"""
from ... import crashreport
buf: List[str] = []
key = question_cache_key(self._messages)
if key:
saved = crashreport.saved_explanation(key)
if saved:
self.saved = True
self.stage_changed.emit("streaming")
self.chunk_ready.emit(saved["text"])
self.finished.emit(True, saved["text"])
return
try:
self.stage_changed.emit("connecting")
stream = self._provider.stream_chat(
self._messages, system=self._system, model=self._model
)
self.stage_changed.emit("streaming")
for chunk in stream:
if self._cancelled:
break
if chunk:
buf.append(chunk)
self.chunk_ready.emit(chunk)
if self._cancelled:
self.finished.emit(False, "Cancelled.")
else:
if key and buf:
crashreport.save_explanation(
key, "".join(buf),
model=str(getattr(self._provider, "name", "") or ""))
self.finished.emit(True, "".join(buf))
except BaseException as e:
if self._cancelled:
self.finished.emit(False, "Cancelled.")
return
tb = traceback.format_exc()
try:
print(f"[AI worker] error: {tb}", file=sys.__stderr__, flush=True)
except Exception:
pass
detail = str(e)
if not isinstance(e, ProviderFailed) or not detail:
detail = f"{type(e).__name__}: {e}"
self.finished.emit(False, detail)
[docs]
def make_stream_thread(
provider: ChatProvider,
messages: List[Dict],
system: str = "",
model: Optional[str] = None,
parent: Optional[QObject] = None,
) -> tuple[QThread, StreamWorker]:
"""Return (QThread, StreamWorker) — connect signals, then start().
IMPORTANT: pass a `parent` (typically the panel that owns this
stream). Without a Qt parent the QThread's C++ object gets tied
exclusively to Python's refcount — and dropping the ref while
QThread.isRunning() is still True (which happens in the tiny
window between worker.run returning and thread.finished firing)
triggers Qt's `QThread: Destroyed while thread is still running /
Aborted` crash. A parent keeps the C++ object alive until
deleteLater runs.
Callers must ALSO keep a Python reference to the worker until
the stream truly finishes (see ConsolePanel._retire).
Two wiring details are load-bearing; both are the same contract
:func:`spacr.qt.bridge.make_thread` documents, and this function used
to get them wrong:
* ``worker.finished -> thread.quit`` is a **DirectConnection**. The
QThread object is created here, on the GUI thread, so it is
GUI-affine — a queued ``quit()`` is posted to the *GUI* thread's
event queue, not to the worker's. Measured: with a queued
connection, a GUI thread that goes straight into ``thread.wait()``
(which is exactly what ``ConsolePanel.shutdown`` and every "drain
before closing" path does) waits out its whole timeout on a worker
that has already finished, because the event that would stop the
thread is sitting behind the wait. ``QThread::quit`` is explicitly
thread-safe, so calling it inline from the worker thread is correct.
* There is deliberately **no** ``worker.deleteLater``. The worker's
affinity is the worker thread, so a deferred delete is posted into a
loop that is stopping, while the panel drops the object's last
Python reference from the GUI thread — two owners, one object.
``bridge.make_thread``'s ownership essay records the gdb trace
(``QThread -> sendPostedEvents -> ~QObject -> Sbk_GetPyOverride``)
and the measurement: 3 crashes in 8 runs. A PySide6 object built in
Python is already owned by Python; ``ConsolePanel``/``AIChatPanel``
hold it in ``_retired`` until the thread has exited and free it
there, on the thread that holds it.
:param provider: the :class:`~spacr.qt.ai.providers.ChatProvider` whose
``stream_chat`` the worker consumes.
:param messages: conversation history as ``{role, content}`` dicts,
handed to the provider unchanged.
"""
from PySide6.QtCore import Qt
thread = QThread(parent)
worker = StreamWorker(provider, messages, system=system, model=model)
worker.setParent(None)
worker.moveToThread(thread)
thread.started.connect(worker.run)
worker.finished.connect(thread.quit, Qt.DirectConnection)
thread.finished.connect(thread.deleteLater, Qt.QueuedConnection)
return thread, worker