"""Combine pipeline output, errors, and AI chat in one console panel.
Topic bars separate pipeline stages and AI conversations in a shared scrolling
view. The input area and transcript form a vertical ``QSplitter``; when a
``persist_key`` is supplied, its state is stored under ``console/split/<key>``
and restored per screen.
:class:`ConsolePanel` can start a topic, append ordinary or error output,
launch the AI error-explanation flow, and clear the transcript. It owns its AI
worker so streamed state remains coherent while the active pipeline screen
changes.
Widget creation always occurs on the GUI thread. Calls to
:meth:`ConsolePanel.append_stdout` and :meth:`ConsolePanel.append_error` from
logging or pipeline workers are relayed through queued Qt signals before they
modify the transcript.
"""
from __future__ import annotations
import re
from typing import Dict, List, Optional
from PySide6.QtCore import (QByteArray, QSize, Qt, QThread, QTimer, Slot,
Signal)
from PySide6.QtGui import (
QColor,
QFont,
QFontDatabase,
QKeyEvent,
QTextBlockFormat,
QTextCursor,
QTextLayout,
)
from PySide6.QtWidgets import (
QAbstractButton,
QComboBox,
QFrame,
QHBoxLayout,
QLabel,
QMessageBox,
QPlainTextEdit,
QScrollArea,
QSizePolicy,
QSpinBox,
QSplitter,
QTextEdit,
QVBoxLayout,
QWidget,
)
from .. import ai as ai_module
from ..ai import settings as ai_settings
from ..ai.providers import ChatProvider
from ..ai.worker import StreamWorker, make_stream_thread
from ..i18n import retranslate_widget_tree, tr
from ..theme import FONT_SIZE, QSS_FONT_FAMILY, SPACING, _painting_palette, active_palette
from ..verbose_logger import console_write, console_write_in_progress
from .flash import Flash
from .. import _after
#: Soft budget for pipeline scrollback attached to one AI turn. A complete
#: traceback is never cut to satisfy it; ordinary stdout yields first.
AI_CONSOLE_CONTEXT_CHARS = 24_000
#: A deliberately small, visible heuristic for the Auto context mode. The
#: adjacent dropdown is the override when wording falls outside this set.
_CONSOLE_QUESTION_TERMS = (
"console", "traceback", "error", "exception", "failed", "failure",
"went wrong", "what happened", "why did", "debug", "log output",
)
#: Palette role behind each of the three legacy ``COLOR_*`` names.
_TEXT_ROLES = {
"COLOR_OUTPUT": "accent",
"COLOR_USER": "success",
"COLOR_ERROR": "error",
}
[docs]
def color_output() -> str:
"""Pipeline stdout colour for the theme on screen right now."""
return active_palette()["accent"]
[docs]
def color_user() -> str:
"""User-input colour for the theme on screen right now."""
return active_palette()["success"]
[docs]
def color_error() -> str:
"""Error colour for the theme on screen right now."""
return active_palette()["error"]
[docs]
def color_warning() -> str:
"""Warning colour for the theme on screen right now.
Distinct from :func:`color_error` on purpose -- see
:meth:`ConsolePanel.append_warning`. Every theme already defines the
``warning`` role; nothing here needed inventing.
"""
return active_palette()["warning"]
[docs]
def __getattr__(name: str) -> str:
"""Serve ``COLOR_OUTPUT`` / ``COLOR_USER`` / ``COLOR_ERROR`` live.
PEP 562. Reading one of the three resolves it against the current
theme, so ``from ...console_panel import COLOR_USER`` can no longer
freeze a dark-theme hex into a caller at import time.
"""
role = _TEXT_ROLES.get(name)
if role is None:
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
return active_palette()[role]
[docs]
def color_spacr_ai() -> str:
"""spaCR AI's own console colour: the theme's teal.
The palette's ``chip_class`` role is the spaCR teal on every theme, so
a "spaCR AI" entry reads apart from the blue output and the red error.
"""
palette = active_palette()
return palette.get("chip_class") or palette["accent"]
_MD_HEADING_RE = re.compile(
r"^\s*(?:#{1,6}\s*)?\*\*(?P<head>[^*]+?)\*\*\s*(?:[-–—:]\s*)?(?P<rest>.*)$")
_MD_ATX_RE = re.compile(r"^\s*#{1,6}\s+(?P<head>.+?)\s*#*\s*$")
_MD_BREAK_RE = re.compile(
r"\s+(?=\*\*[^*\n]{2,60}\*\*)|\s+(?=\d{1,2}\.\s)|(?<!\*\*)\s+(?=[-•]\s+\S)")
def _is_a_heading(match: "re.Match", line: str) -> bool:
"""Whether a line's opening ``**bold**`` is a section heading.
It is when nothing follows it, a dash or colon follows it, or it reads
like a title (two words or more, or ending in ``?`` or ``:``) -- not a
single bold word opening a sentence.
"""
if "rest" not in match.groupdict():
return True
head = match.group("head").strip()
rest = match.group("rest")
if not rest.strip() or re.match(r"^\s*\*\*[^*]+\*\*\s*[-–—:]", line):
return True
return len(head.split()) >= 2 or head.endswith(("?", ":"))
[docs]
def markdown_lines(text: str) -> List[tuple]:
"""Lay an AI answer's Markdown out as lines a console block can show.
A ``**Heading**`` (or ``# Heading``) that opens a line becomes a bold
line of its own and its text the next line; numbered and bulleted steps
keep a line each. An answer that arrived as one run-on line is broken
before its headings and steps first.
:param text: the answer.
:returns: ``[(line, bold), ...]``; ``line`` keeps any inline ``**bold**``
markers, which :meth:`_StdoutBlock.show_markdown` draws.
"""
raw = str(text or "").replace("\r\n", "\n").strip()
if "\n" not in raw:
raw = _MD_BREAK_RE.sub("\n", raw)
out: List[tuple] = []
for line in raw.split("\n"):
line = line.rstrip()
heading = _MD_ATX_RE.match(line) or _MD_HEADING_RE.match(line)
if heading is not None and not _is_a_heading(heading, line):
heading = None
if heading is not None:
if out and out[-1][0]:
out.append(("", False))
out.append((heading.group("head").strip(), True))
rest = heading.groupdict().get("rest") or ""
if rest.strip():
out.append((rest.strip(), False))
continue
bullet = re.match(r"^\s*[-*]\s+(.*)$", line)
if bullet is not None:
line = "• " + bullet.group(1)
if line.strip() or (out and out[-1][0]):
out.append((line.strip(), False))
while out and not out[-1][0]:
out.pop()
return out
AI_COLOR_CLAUDE = "#DE7356"
AI_COLOR_OPENAI = "#74AA9C"
AI_COLOR_GEMINI = "#74AA9C"
AI_COLOR_DEFAULT = "#74AA9C"
[docs]
def ai_color_for_provider(provider_name: Optional[str]) -> str:
"""Return the spaCR-AI text colour for a provider name.
:param provider_name: the provider's name, matched case-insensitively by
substring (``claude``, ``gpt``, ``gemini`` and the like); None or
unknown gives the default colour.
"""
p = (provider_name or "").lower()
if "claude" in p or "anthropic" in p:
return AI_COLOR_CLAUDE
if "gpt" in p or "openai" in p or "chatgpt" in p:
return AI_COLOR_OPENAI
if "gemini" in p or "google" in p:
return AI_COLOR_GEMINI
return AI_COLOR_DEFAULT
#: Default height, in pixels, of the AI chat box — the height the old
#: hard-coded ``setMaximumHeight(120)`` pinned it at.
DEFAULT_CHAT_HEIGHT = 120
#: Floor for the chat box. Also the splitter's stop, since
#: ``setChildrenCollapsible(False)`` honours a child's minimum.
CHAT_MIN_HEIGHT = 48
#: Floor for the console box. Without one, a QScrollArea's minimum size hint
#: is a couple of pixels and "not collapsible" would still let the console be
#: dragged down to a sliver.
CONSOLE_MIN_HEIGHT = 80
#: QSettings key prefix for the persisted splitter state, one entry per
#: screen (``console/split/mask``, ``console/split/measure``, …). Screens are
#: used for different things and a user who wants a tall chat box on one does
#: not necessarily want it on all of them.
_SPLIT_KEY_PREFIX = "console/split"
def _settings():
"""The app's ``QSettings``, borrowed from :mod:`spacr.qt.preferences`.
Going through preferences rather than constructing ``QSettings`` here
keeps the org/app pair single-sourced — and, just as important, keeps this
module inside the sandbox the test suite installs, which redirects by
*path* and so only catches settings opened the same way everything else
opens them.
"""
from ..preferences import _settings as _prefs_settings
return _prefs_settings()
[docs]
def get_split_state(screen_key: str):
"""Return the saved ``QSplitter.saveState()`` blob for ``screen_key``.
:param screen_key: the screen's app key, e.g. ``"mask"``.
:returns: the stored ``QByteArray``, or ``None`` when the user has never
dragged this screen's handle (or the stored value is unusable).
"""
key = str(screen_key or "").strip()
if not key:
return None
try:
raw = _settings().value(f"{_SPLIT_KEY_PREFIX}/{key}")
except Exception:
return None
if isinstance(raw, (bytes, bytearray, QByteArray)) and len(raw):
return QByteArray(raw)
return None
[docs]
def set_split_state(screen_key: str, state) -> None:
"""Persist a ``QSplitter.saveState()`` blob against ``screen_key``.
Stored as the splitter's own state rather than as a pixel pair on
purpose: a saved ``[572, 120]`` means something different on a laptop
panel than on the 4K display the same user docks into, whereas
``restoreState`` is the mechanism Qt itself defines for this.
:param screen_key: the screen the state belongs to; stripped, and a blank
key stores nothing.
:param state: the ``QSplitter.saveState()`` bytes, stored as a
``QByteArray``.
"""
key = str(screen_key or "").strip()
if not key:
return
try:
_settings().setValue(f"{_SPLIT_KEY_PREFIX}/{key}", QByteArray(state))
except Exception:
pass
class _CopyGlyphButton(QAbstractButton):
"""The two-offset-squares copy mark, drawn rather than shipped.
An icon file would need a light and a dark variant and would have to be
kept in step with the theme; two rounded rectangles in the current
foreground colour follow it for free, at any DPI.
:param parent: parent widget; ownership only.
"""
#: Side of the front square, in px. The back one is drawn behind it,
#: offset by :data:`_OFFSET`, which is what reads as "a copy".
_SIDE = 9
_OFFSET = 3
def __init__(self, parent=None):
"""Build the copy mark, drawn rather than shipped as an icon."""
super().__init__(parent)
self.setObjectName("ConsoleCopyGlyph")
self.setFocusPolicy(Qt.NoFocus)
edge = self._SIDE + self._OFFSET + 5
self.setFixedSize(edge, edge)
self._flash = Flash(self)
def flash_copied(self) -> None:
"""Briefly mark the glyph, so a silent clipboard write is visible."""
self._flash.trigger()
def paintEvent(self, _event) -> None: # noqa: N802 (Qt naming)
"""Draw the two offset squares that read as one sheet over another.
The accent colour while the copy flash is active, the dim foreground
otherwise, and lighter under the pointer -- so the button says it is
hoverable before it is pressed and says it worked after.
:param _event: the paint event; unused.
"""
from PySide6.QtGui import QPainter, QPen
painter = QPainter(self)
painter.setRenderHint(QPainter.Antialiasing, True)
try:
palette = _painting_palette()
colour = QColor(palette["button_accent"] if self._flash.active
else palette["fg_dim"])
except Exception:
colour = QColor("#888888")
if self.underMouse() and not self._flash.active:
colour = colour.lighter(150)
pen = QPen(colour)
pen.setWidth(1)
painter.setPen(pen)
painter.setBrush(Qt.NoBrush)
side, off = self._SIDE, self._OFFSET
painter.drawRoundedRect(off + 1, 1, side, side, 2, 2)
painter.drawRoundedRect(1, off + 1, side, side, 2, 2)
painter.end()
class _TopicBar(QFrame):
"""Dark-gray divider bar with a topic label ("spaCR output — …", …).
An optional ``accent`` colour tints the label so each banner reads in
the same colour as the text that follows it. A trailing ``widget`` (e.g.
an animated working indicator) can be pinned to the right.
"""
def __init__(self, label: str, parent=None, accent: Optional[str] = None,
trailing: Optional[QWidget] = None):
"""Build one console section heading.
:param label: the heading text.
:param parent: parent widget.
:param accent: colour for the heading, or ``None`` for the theme's.
:param trailing: an optional widget pinned to the right of the
heading, for a count or a control belonging to the section.
The heading is a CONTROL, not a caption: clicking it brings its
section to the top and expands it, so it takes strong focus while
preserving the operating system's cursor. A control only a mouse can
reach is one some users cannot reach at all.
"""
super().__init__(parent)
self.setObjectName("ConsoleTopicBar")
self.setFocusPolicy(Qt.StrongFocus)
self.setAccessibleName(label)
self._expanded = True
lay = QHBoxLayout(self)
lay.setContentsMargins(SPACING["md"], SPACING["xs"],
SPACING["md"], SPACING["xs"])
self._chevron = QLabel("▾")
self._chevron.setObjectName("ConsoleTopicChevron")
self._chevron.setProperty("i18nSkipText", True)
lay.addWidget(self._chevron)
self._label = QLabel(label)
self._label.setObjectName("ConsoleTopicLabel")
self._label.setProperty("i18nSkipText", True)
if accent:
self._label.setStyleSheet(
f"QLabel#ConsoleTopicLabel {{ color: {accent}; "
"background: transparent; }")
lay.addWidget(self._label)
if trailing is not None:
lay.addWidget(trailing)
self._copy_btn = _CopyGlyphButton(self)
copy_tip = "Copy this section, header and all"
self._copy_btn.setProperty("_spacr_i18n_tooltip", copy_tip)
self._copy_btn.setToolTip(tr(copy_tip))
self._copy_btn.clicked.connect(self._copy_section)
lay.addWidget(self._copy_btn)
lay.addStretch(1)
def text(self) -> str:
"""The header text, for the plain-text export."""
return self._label.text()
def is_expanded(self) -> bool:
"""Whether this section's body is showing."""
return self._expanded
def set_expanded(self, expanded: bool) -> None:
"""Record the state and turn the chevron to match."""
self._expanded = bool(expanded)
self._chevron.setText("▾" if self._expanded else "▸")
def _panel(self):
"""The owning :class:`ConsolePanel`, or None."""
node = self.parentWidget()
while node is not None and not hasattr(node, "toggle_section"):
node = node.parentWidget()
return node
def _activate(self) -> None:
"""Fold or unfold this section, if the panel is still there."""
panel = self._panel()
if panel is not None:
panel.toggle_section(self)
def mouseReleaseEvent(self, event): # noqa: N802 (Qt naming)
"""Raise this section on a click inside the bar.
On RELEASE rather than press, so dragging off cancels. The copy button
and any trailing widget are children with their own handlers, so a click
on them never reaches here -- which is what keeps "copy this section"
from also moving the viewport.
:param event: the mouse event.
"""
if (event.button() == Qt.LeftButton
and self.rect().contains(event.pos())):
self._activate()
super().mouseReleaseEvent(event)
def keyPressEvent(self, event): # noqa: N802 (Qt naming)
"""Raise this section on Return, Enter or Space.
:param event: the key event.
"""
if event.key() in (Qt.Key_Return, Qt.Key_Enter, Qt.Key_Space):
self._activate()
return
super().keyPressEvent(event)
def _copy_section(self) -> None:
"""Put this section on the clipboard, asking the panel for its span."""
panel = self.parent()
for _ in range(6):
if panel is None:
return
if hasattr(panel, "section_text"):
break
panel = panel.parent()
else:
return
text = panel.section_text(self)
if not text.strip():
return
try:
from PySide6.QtWidgets import QApplication
QApplication.clipboard().setText(text)
except Exception:
return
self._copy_btn.flash_copied()
class _WorkingDots(QLabel):
"""Three dots that cycle (. → .. → ...) to show work is in progress.
:param color: the dots' colour, as a CSS string. Baked into a stylesheet
at construction, so a theme change needs a new widget rather than a
setter.
:param parent: parent widget; ownership only.
"""
def __init__(self, color: str = AI_COLOR_DEFAULT, parent=None):
"""Build the dots at the AI colour, baked into a stylesheet."""
super().__init__(parent)
self.setObjectName("ConsoleWorkingDots")
self.setProperty("i18nSkipText", True)
self._color = color
self._n = 0
self.setStyleSheet(
f"QLabel#ConsoleWorkingDots {{ color: {color}; "
f"font-size: {max(6, FONT_SIZE['xs'] - 2)}px; font-weight: 400; "
"background: transparent; }")
from PySide6.QtCore import QTimer
self._timer = QTimer(self)
self._timer.setInterval(350)
self._timer.timeout.connect(self._tick)
self._render()
def set_color(self, color: str) -> None:
"""Re-ink the working dots.
:param color: the provider's colour, so the indicator matches the reply
it belongs to.
"""
self._color = color
self.setStyleSheet(
f"QLabel#ConsoleWorkingDots {{ color: {color}; "
f"font-size: {max(6, FONT_SIZE['xs'] - 2)}px; font-weight: 400; "
"background: transparent; }")
def _render(self) -> None:
"""Draw the current dot count, padded to a fixed width.
Padded so the row does not JITTER as the count cycles: three glyph-slots
either way, and the text beside it stays where it is.
"""
dots = "●" * (self._n + 1)
pad = " " * (2 - self._n)
self.setText(dots + pad)
def _tick(self) -> None:
"""Advance one step through . / .. / ... and redraw."""
self._n = (self._n + 1) % 3
self._render()
@Slot()
def _noop_slot(self) -> None: # pragma: no cover - anchor
"""Present so the Slot import is used even if the others change."""
def _on_gui_thread(self) -> bool:
"""Whether the caller is the thread this widget lives on."""
from PySide6.QtCore import QThread
return QThread.currentThread() is self.thread()
@Slot()
def start(self) -> None:
"""Start the activity animation on the widget's Qt thread.
This method may be called from any thread. Calls from a worker are
queued onto the widget's owning thread before the timer is started.
"""
if not self._on_gui_thread():
from PySide6.QtCore import QMetaObject, Qt
QMetaObject.invokeMethod(self, "start", Qt.QueuedConnection)
return
self._n = 0
self._render()
self._timer.start()
self.show()
@Slot()
def stop(self) -> None:
"""Stop the animation and hide the dots.
Marshalled onto the GUI thread when called from another: a stream
finishes on a worker, and touching a widget from there is undefined.
"""
if not self._on_gui_thread():
from PySide6.QtCore import QMetaObject, Qt
QMetaObject.invokeMethod(self, "stop", Qt.QueuedConnection)
return
self._timer.stop()
self.hide()
class _StdoutBlock(QPlainTextEdit):
"""Readable text block that grows in place as pipeline output arrives.
A single block is reused for a whole stdout run so line breaks do
not fragment the console into one widget per line. A read-only
``QPlainTextEdit`` gives us selectable plain text while also exposing
``QTextBlockFormat``—the reliable Qt API for real line spacing. QSS
does not implement CSS ``line-height`` for a ``QLabel``.
The viewport never fills its own background (item 515). A scroll area's
viewport fills with ``QPalette.Base`` by default and only the stylesheet
turns that off; a polish nested in another stylesheet style's call lets
the unpolish through (item 408), the fill comes back in the window
colour, and the console text sat on black until the module was left and
reopened. Switched off before any sheet sees it, there is nothing for an
unpolish to restore.
"""
LINE_HEIGHT_PERCENT = 145
#: Characters kept in one block. Older text is dropped from the head.
MAX_CHARS = 200_000
def __init__(self, text: str = "", error: bool = False, parent=None,
text_color: Optional[str] = None):
"""Build the block one stdout run grows into.
:param text: the initial contents.
:param error: whether this is the error stream. Chooses the object
name, so the theme colours it without this class deciding what
"error" looks like.
:param parent: parent widget.
:param text_color: an explicit colour, or ``None`` for the theme's.
The block is reused for a whole run rather than made per line: a
widget per line fragments the console, and :data:`MAX_CHARS` drops
text from the HEAD so a long run cannot grow without bound.
Its accessible name, "Log" or "Error", is what VoiceOver says when the
block takes focus; without one it was the only unnamed control the
macOS accessibility audit found (2026-10-19).
"""
super().__init__(parent)
self.viewport().setAutoFillBackground(False)
self.setObjectName("ConsoleStdoutBlockError"
if error else "ConsoleStdoutBlock")
self.setAccessibleName(tr("Error") if error else tr("Log"))
self.setReadOnly(True)
self.setFrameShape(QFrame.NoFrame)
self.setLineWrapMode(QPlainTextEdit.WidgetWidth)
self.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self.setVerticalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self.setTextInteractionFlags(
Qt.TextSelectableByMouse | Qt.TextSelectableByKeyboard
)
self.setSizePolicy(QSizePolicy.Expanding, QSizePolicy.Minimum)
self.viewport().setMouseTracking(True)
self._font_pt = 10
self._font = QFont("Open Sans", self._font_pt, QFont.Light)
self._font.setStyleName("Light")
self.setFont(self._font)
self.document().setDefaultFont(self._font)
self.document().setDocumentMargin(0)
if text_color is None:
text_color = color_error() if error else color_output()
self._text_color = text_color
self._refresh_style()
#: Characters currently in the document. Tracked rather than derived
#: from ``toPlainText()``, which copies the whole document.
self._chars = 0
#: ``sizeHint`` cache — see :meth:`sizeHint`.
self._size_key: tuple = ()
self._size_value = 32
#: Summed paragraph heights, kept current line by line while the
#: wrap width and font stay what :attr:`_height_key` records.
self._height_sum = 0.0
self._height_key: tuple = ()
self._user_height: Optional[int] = None
self._height_handle = _BlockHeightHandle(self)
self._height_handle.show()
if text:
self.append(text)
def _refresh_style(self) -> None:
"""Keep inline, theme-aware ink and the user-selected point size."""
self.setStyleSheet(
"QPlainTextEdit#%s { color: %s; background-color: transparent; "
"border: none; "
"font-family: %s; "
"font-weight: 300; font-size: %dpt; "
"padding: %dpx %dpx; }" % (
self.objectName(), self._text_color, QSS_FONT_FAMILY, self._font_pt,
SPACING["sm"], SPACING["md"]))
def _block_format(self) -> QTextBlockFormat:
"""The 145% leading every paragraph in this block carries."""
block_format = QTextBlockFormat()
block_format.setLineHeight(
float(self.LINE_HEIGHT_PERCENT),
QTextBlockFormat.ProportionalHeight.value,
)
return block_format
def _apply_line_spacing(self) -> None:
"""Apply the leading to every paragraph in the document.
Whole-document, therefore O(document): only for the rare events that
genuinely change every paragraph, such as a font-size change.
:meth:`append` formats the paragraphs it creates and nothing else.
"""
cursor = QTextCursor(self.document())
cursor.select(QTextCursor.Document)
cursor.mergeBlockFormat(self._block_format())
def set_console_font_pt(self, pt: int) -> None:
"""Apply the console size while retaining Open Sans Light."""
self._font_pt = int(pt)
self._font.setPointSize(self._font_pt)
self._font.setWeight(QFont.Light)
self._font.setStyleName("Light")
self.setFont(self._font)
self.document().setDefaultFont(self._font)
self._refresh_style()
self._apply_line_spacing()
self._height_key = ()
self.updateGeometry()
def text(self) -> str:
"""Compatibility with the former QLabel-backed output block."""
return self.toPlainText()
def append(self, text: str) -> None:
"""Append ``text`` in place, trimming the head past :attr:`MAX_CHARS`.
Costs what the new text costs, not what the console already holds.
This used to rebuild the entire document on every line —
``setPlainText("".join(buf))`` plus a document-wide
``mergeBlockFormat`` — which made a run's own output quadratic in
its length. Measured on this tree: 0.56 ms per line for the first
500, 6.64 ms per line by line 3000, and level there only because
the 200k cap had been reached.
That is not merely slow. With Verbose logging on,
``spacr.logging_util``'s profile hook emits a record on entry to
every spaCR function — including the ones inside Qt event delivery
— and each record lands here. At 7 ms a line the GUI thread cannot
drain its own queue: the process sits at 100% CPU making no forward
progress, which is exactly how the Qt shard "live-lock" presented.
"""
if not text:
return
doc = self.document()
cursor = QTextCursor(doc)
cursor.movePosition(QTextCursor.End)
first_touched = cursor.blockNumber()
tracking = self._height_key == self._layout_key()
if tracking:
self._height_sum -= self._block_height(
doc.findBlockByNumber(first_touched))
cursor.insertText(text)
self._chars += len(text)
fmt_cursor = QTextCursor(doc.findBlockByNumber(first_touched))
fmt_cursor.setPosition(cursor.position(), QTextCursor.KeepAnchor)
fmt_cursor.mergeBlockFormat(self._block_format())
if tracking:
block = doc.findBlockByNumber(first_touched)
while block.isValid():
self._height_sum += self._block_height(block)
block = block.next()
self._trim_to_cap()
self.updateGeometry()
def show_markdown(self, text: str) -> None:
"""Replace the contents with an AI answer, its Markdown drawn.
Headings are bold lines of their own and every step keeps its line
(:func:`markdown_lines`); ``**bold**`` inside a line is bold too.
:param text: the answer, in Markdown.
"""
from PySide6.QtGui import QTextCharFormat
doc = self.document()
doc.clear()
cursor = QTextCursor(doc)
plain = QTextCharFormat()
bold = QTextCharFormat()
bold.setFontWeight(QFont.Bold)
for index, (line, strong) in enumerate(markdown_lines(text)):
if index:
cursor.insertBlock(self._block_format())
else:
cursor.mergeBlockFormat(self._block_format())
if strong:
cursor.insertText(line, bold)
continue
for part, piece in enumerate(re.split(r"\*\*(.+?)\*\*", line)):
if piece:
cursor.insertText(piece, bold if part % 2 else plain)
self._chars = doc.characterCount()
self._height_key = ()
self.updateGeometry()
def _layout_key(self) -> tuple:
"""What paragraph heights depend on: the wrap width and the font."""
return (self.viewport().width(), self._font_pt)
def _block_height(self, block) -> float:
"""One paragraph's laid-out height in pixels."""
return self.document().documentLayout().blockBoundingRect(
block).height()
def _trim_to_cap(self) -> None:
"""Drop whole paragraphs off the head until back under the cap.
Removing from the front costs what is removed, in one edit however
many paragraphs go. Re-setting the document to its own tail costs
what is kept, on every single line once the cap is reached.
"""
if self._chars <= self.MAX_CHARS:
return
doc = self.document()
tracking = self._height_key == self._layout_key()
last = doc.lastBlock().blockNumber()
block = doc.begin()
removed = 0
while (self._chars - removed > self.MAX_CHARS
and block.isValid() and block.blockNumber() < last):
removed += block.length()
if tracking:
self._height_sum -= self._block_height(block)
block = block.next()
if not removed:
return
cursor = QTextCursor(doc)
cursor.setPosition(0)
cursor.setPosition(block.position(), QTextCursor.KeepAnchor)
cursor.removeSelectedText()
self._chars = max(0, self._chars - removed)
_LINK = re.compile(r"https?://[^\s<>\"')\]]+")
def link_at(self, pos) -> str:
"""The web address under a point of the block, or an empty string.
:param pos: a point in the viewport's coordinates.
"""
cursor = self.cursorForPosition(pos)
line = cursor.block().text()
column = cursor.positionInBlock()
for match in self._LINK.finditer(line):
if match.start() <= column <= match.end():
return match.group(0).rstrip(".,;:")
return ""
def mouseMoveEvent(self, event) -> None: # noqa: N802
"""Show the pointing hand over an address."""
try:
over = bool(self.link_at(event.position().toPoint()))
except Exception: # noqa: BLE001
over = False
self.viewport().setCursor(
Qt.PointingHandCursor if over else Qt.IBeamCursor)
super().mouseMoveEvent(event)
def mouseReleaseEvent(self, event) -> None: # noqa: N802
"""Open the address that was clicked, unless text was selected."""
super().mouseReleaseEvent(event)
if event.button() != Qt.LeftButton or self.textCursor().hasSelection():
return
url = self.link_at(event.position().toPoint())
if url:
from PySide6.QtCore import QUrl
from PySide6.QtGui import QDesktopServices
QDesktopServices.openUrl(QUrl(url))
def sizeHint(self) -> QSize:
"""Report the full document height; the outer console owns scrolling.
Qt asks for a size hint several times per layout pass, and
:meth:`append` asks for a new layout on every line. Walking every
paragraph here made each line cost what the whole block holds: 35 ms
a line once the block was full, so output arriving faster than about
thirty lines a second -- verbose runs, DEBUG in the console --
queued faster than the GUI thread could draw it and the window
froze while the queue grew. The total is therefore kept as a running
sum that :meth:`append` and :meth:`_trim_to_cap` adjust by the
paragraphs they touch, and the full walk happens only when the wrap
width or the font changes.
"""
if self._user_height is not None:
return QSize(
max(120, super().sizeHint().width()), self._user_height)
layout_key = self._layout_key()
if layout_key != self._height_key:
height = 0.0
block = self.document().begin()
while block.isValid():
height += self._block_height(block)
block = block.next()
self._height_sum = height
self._height_key = layout_key
key = (self._chars,) + layout_key + (round(self._height_sum),)
if key != self._size_key:
chrome = (2 * SPACING["sm"]) + 2
self._size_key = key
self._size_value = max(
32, int(round(self._height_sum)) + chrome)
return QSize(max(120, super().sizeHint().width()), self._size_value)
def resizeEvent(self, event) -> None:
"""Re-wrap the text and keep the drag handle pinned to the bottom edge.
:param event: the resize event.
"""
super().resizeEvent(event)
self.document().setTextWidth(max(1, self.viewport().width()))
handle_height = self._height_handle.sizeHint().height()
self._height_handle.setGeometry(
0, max(0, self.height() - handle_height),
self.width(), handle_height,
)
self._height_handle.raise_()
self.updateGeometry()
def set_user_height(self, height: int) -> None:
"""Pin this section to a user-selected height."""
self._user_height = max(48, min(4000, int(height)))
self.setFixedHeight(self._user_height)
self.updateGeometry()
def reset_user_height(self) -> None:
"""Return to automatic document-height sizing."""
self._user_height = None
self.setMinimumHeight(0)
self.setMaximumHeight(16_777_215)
self.updateGeometry()
class _BlockHeightHandle(QFrame):
"""Thin drag handle along a console section's lower edge.
:param block: the section this handle resizes. ALSO ITS QWIDGET PARENT,
so the handle is laid out inside the block it drags and cannot
outlive it; there is no separate ``parent``.
"""
HEIGHT = 7
def __init__(self, block: _StdoutBlock):
"""Build the handle with a vertical-resize cursor and its tooltip."""
super().__init__(block)
self._block = block
self._press_y: Optional[float] = None
self._start_height = 0
self.setObjectName("ConsoleSectionResizeHandle")
self.setCursor(Qt.SizeVerCursor)
self.setFixedHeight(self.HEIGHT)
source = (
"Drag to resize this console section. Double-click for auto height."
)
self.setProperty("_spacr_i18n_tooltip", source)
self.setToolTip(tr(source))
def sizeHint(self) -> QSize:
"""Return the handle's preferred size.
:returns: a strip as tall as the handle and nominally 80 wide -- the
width comes from the block it spans, not from this hint.
"""
return QSize(80, self.HEIGHT)
def mousePressEvent(self, event) -> None:
"""Begin a drag, recording where it started and how tall the block was.
Both are needed: the new height is the starting height plus the total
movement, so a drag that reverses returns to where it began rather than
accumulating.
:param event: the mouse event.
"""
if event.button() == Qt.LeftButton:
self._press_y = event.globalPosition().y()
self._start_height = self._block.height()
event.accept()
return
super().mousePressEvent(event)
def mouseMoveEvent(self, event) -> None:
"""Resize the block to follow the drag.
:param event: the mouse event.
"""
if self._press_y is not None and event.buttons() & Qt.LeftButton:
delta = event.globalPosition().y() - self._press_y
self._block.set_user_height(self._start_height + int(delta))
event.accept()
return
super().mouseMoveEvent(event)
def mouseReleaseEvent(self, event) -> None:
"""End the drag.
:param event: the mouse event.
"""
self._press_y = None
super().mouseReleaseEvent(event)
def mouseDoubleClickEvent(self, event) -> None:
"""Return the block to its automatic height.
A double click is the undo for the drag: without it a block dragged
short can only be restored by guessing its original size.
:param event: the mouse event.
"""
if event.button() == Qt.LeftButton:
self._block.reset_user_height()
event.accept()
return
super().mouseDoubleClickEvent(event)
class _Bubble(QFrame):
"""Chat bubble — a coloured QFrame that renders wrapped rich text.
Manual sizing: on every resizeEvent we clamp the inner label's
width to our own width minus padding, then set the label's fixed
height from QFontMetrics.boundingRect for that wrap width. The
frame's height is set to match. Simple, works reliably even
inside a QScrollArea.
:param role: ``"user"`` or anything else, which is read as the AI. It
picks the object name and so the whole appearance -- there is no
third style, and an unrecognised role is drawn as the AI rather
than refused.
:param text: the initial message; may be set later instead.
:param parent: parent widget; ownership only.
"""
_H_PAD = 24
_V_PAD = 12
def __init__(self, role: str, text: str = "", parent=None):
"""Build the bubble, styled by role, sized to its text."""
super().__init__(parent)
self.role = role
self.setObjectName(
"ConsoleBubbleUser" if role == "user" else "ConsoleBubbleAI"
)
self._recalc_guard = False
self._label = QLabel(self)
self._label.setObjectName("ConsoleBubbleText")
self._label.setProperty("i18nSkipText", True)
self._label.setTextFormat(Qt.RichText)
self._label.setTextInteractionFlags(
Qt.TextSelectableByMouse | Qt.LinksAccessibleByMouse
)
self._label.setOpenExternalLinks(True)
self._label.setWordWrap(True)
self._label.setAlignment(Qt.AlignVCenter | Qt.AlignLeft)
self._label.setStyleSheet(
"QLabel#ConsoleBubbleText {"
f" color: {active_palette()['fg']};"
f" font-size: {FONT_SIZE['body']}px;"
" background: transparent;"
" border: none;"
"}"
)
lay = QVBoxLayout(self)
lay.setContentsMargins(SPACING["md"], SPACING["sm"],
SPACING["md"], SPACING["sm"])
lay.setSpacing(0)
lay.addWidget(self._label)
self._raw_text = ""
self._prefix_source = "spaCR user" if role == "user" else "spaCR AI"
if text:
self.set_text(text)
def set_text(self, text: str) -> None:
"""Replace the bubble's body with ``text`` (HTML-escaped, wrapped)."""
self._raw_text = text or ""
safe = self._raw_text.replace("<", "<").replace(">", ">")
safe = safe.replace("\n", "<br>")
prefix = tr(self._prefix_source)
html = f'<span style="opacity:0.7;">{prefix}: </span>{safe}'
self._label.setText(html)
self._recalc()
def _recalc(self) -> None:
"""Fit the label + frame to the wrapped text at our current
width. Uses QLabel.heightForWidth which — for a word-wrap
enabled label — returns the correct line-broken height."""
if self._recalc_guard:
return
w = self.width()
if w <= 0:
return
text_width = max(120, w - self._H_PAD)
self._recalc_guard = True
try:
self._label.setMaximumWidth(text_width)
self._label.setMinimumWidth(text_width)
h = self._label.heightForWidth(text_width)
if h <= 0:
h = self._label.sizeHint().height()
self._label.setFixedHeight(h)
self.setFixedHeight(h + self._V_PAD)
finally:
self._recalc_guard = False
def resizeEvent(self, event):
"""Re-fit label height to the new wrap width."""
super().resizeEvent(event)
self._recalc()
def showEvent(self, event):
"""Re-fit label height once the bubble becomes visible."""
super().showEvent(event)
self._recalc()
class _ChatInput(QTextEdit):
"""Multi-line chat input: Enter sends, Shift+Enter inserts a newline.
:param parent: parent widget; ownership only.
"""
submitted = Signal()
def __init__(self, parent=None):
"""Build the input, floored at one line and capped so it cannot take the pane."""
super().__init__(parent)
self.setMinimumHeight(CHAT_MIN_HEIGHT)
self.setAcceptRichText(False)
def keyPressEvent(self, event: QKeyEvent):
"""Emit ``submitted`` on plain Enter; forward Shift+Enter as newline."""
if event.key() in (Qt.Key_Return, Qt.Key_Enter):
if event.modifiers() & Qt.ShiftModifier:
super().keyPressEvent(event)
return
self.submitted.emit()
return
super().keyPressEvent(event)
def canInsertFromMimeData(self, source) -> bool:
"""Reject file/URL drops. A plain QTextEdit answers yes to a dropped file and
then tries to read it into the text buffer — which freezes the whole app when
the file (or folder) is large. Datasets belong on the app's dropzone, not the
chat box, so only real text is insertable here."""
if source.hasUrls():
return False
return super().canInsertFromMimeData(source)
def insertFromMimeData(self, source) -> None:
"""Paste text, ignoring dropped files.
A file dropped on the chat box is never read here: the console is a
place to type a question, and silently pasting a path -- or worse, a
file's contents -- is not what the gesture meant.
:param source: the mime data being inserted.
"""
if source.hasUrls():
return
super().insertFromMimeData(source)
_LOG_FLUSH_MS = 50
[docs]
class ConsolePanel(QWidget):
"""Merged pipeline stdout + AI chat panel.
Owns the AI stream thread so provider switches and app changes
do not orphan a running subprocess. See the module docstring for
the full public surface.
The console box and the AI chat box are the two halves of a vertical
:class:`~PySide6.QtWidgets.QSplitter`, so the user can drag the handle
between them to trade height — a taller chat box is a shorter console.
Pass ``persist_key`` and the position is remembered per screen.
:ivar ai_stream_finished: emitted when an AI stream ends (ok or
error) so the parent screen can flip its Cancel button back.
"""
ai_stream_finished = Signal()
#: Internal relays used by :meth:`append_stdout` / :meth:`append_error`
#: to hop a call made on a worker thread onto the thread that owns
#: this widget. Both are connected to the very method that emits
#: them — a bound method of this QObject, so Qt queues the delivery
#: rather than running it inline — and the second entry finds itself
#: on the GUI thread and falls through to the real body.
_relay_stdout = Signal(str)
_relay_error = Signal(str)
_relay_notice = Signal(str, object)
def __init__(self, active_app_label: str = "", parent=None,
persist_key: str = "", *, follow_log: bool = True,
chat: bool = True):
"""
:param active_app_label: the app name shown in the output banner.
:param parent: parent widget.
:param persist_key: screen key the console/chat split is remembered
against (usually the screen's ``app_key``). Empty means the split
is not persisted, which is what a bare panel in a test wants.
:param follow_log: False keeps the application-wide log out of this
panel, for a screen whose console carries only its own messages
(Make Masks); the log still reaches the file and the shell's
console.
:param chat: False hides the chat row, for a console that only
reports.
"""
super().__init__(parent)
self.setObjectName("ConsolePanel")
self._log_pending: List[str] = []
self._log_flush = QTimer(self)
self._log_flush.setSingleShot(True)
self._log_flush.setInterval(_LOG_FLUSH_MS)
self._log_flush.timeout.connect(self._flush_log_records)
self._persist_key = str(persist_key or "").strip()
self.setAttribute(Qt.WA_StyledBackground, True)
self._active_app_label = active_app_label or ""
self._run_module: str = ""
self._run_function: str = ""
self._last_entry_kind: str = ""
self._current_stdout: Optional[_StdoutBlock] = None
#: Label of the topic bar currently showing, so an
#: identical one is not drawn again. See begin_topic.
self._current_topic_label: Optional[str] = None
self._working_dots: Optional[_WorkingDots] = None
#: The "spaCR AI" heading opened when a reply was asked for, and the
#: empty block under it, until the first chunk arrives. See
#: :meth:`_take_down_the_waiting_heading`.
self._pending_ai_topic: Optional[_TopicBar] = None
self._pending_ai_block: Optional[_StdoutBlock] = None
#: The traceback the AI is currently explaining, and its answer once
#: the stream finishes. Read by the bug reporter -- see
#: :meth:`ai_explanation_of`.
self._ai_error_traceback: str = ""
self._ai_error_explanation: str = ""
self._ai_messages: List[Dict] = []
self._ai_buf: List[str] = []
self._ai_thread: Optional[QThread] = None
self._ai_worker: Optional[StreamWorker] = None
self._console_sent_lengths: Dict[int, int] = {}
self._retired: List = []
self._relay_stdout.connect(self.append_stdout)
self._relay_error.connect(self.append_error)
self._relay_notice.connect(self._append_notice_on_gui_thread)
self._build_ui()
if not chat:
self._chat_row.setVisible(False)
if follow_log:
try:
from ..logging_util import get_signal_handler
get_signal_handler().records_ready.connect(
self._on_log_records)
except Exception:
pass
retranslate_widget_tree(self)
def _build_ui(self):
"""Lay out the console box and the chat row as the two halves of a splitter.
The handle between them trades height: drag it up and the chat grows
while the console shrinks by the same amount. Only the console carries a
stretch factor, so a taller window grows the scrollback and leaves the
chat box at whatever height the user gave it.
"""
outer = QVBoxLayout(self)
outer.setContentsMargins(0, 0, 0, 0)
outer.setSpacing(SPACING["sm"])
self._split = QSplitter(Qt.Vertical)
self._split.setObjectName("ConsoleSplit")
self._split.setChildrenCollapsible(False)
self._split.setHandleWidth(SPACING["sm"])
try:
from ..theme import make_transparent
make_transparent(self._split)
except Exception:
pass
outer.addWidget(self._split, 1)
self._console_box = QFrame()
self._console_box.setObjectName("ConsoleBox")
self._console_box.setMinimumHeight(CONSOLE_MIN_HEIGHT)
box_lay = QVBoxLayout(self._console_box)
inset = SPACING["sm"]
box_lay.setContentsMargins(inset, inset, inset, inset)
box_lay.setSpacing(0)
self._scroll = QScrollArea()
self._scroll.setObjectName("ConsoleScroll")
self._scroll.setWidgetResizable(True)
self._scroll.setFrameShape(QScrollArea.NoFrame)
self._scroll.viewport().setAutoFillBackground(False)
self._scroll.viewport().setStyleSheet("background: transparent;")
self._scroll.setStyleSheet("background: transparent;")
self._scroll.setHorizontalScrollBarPolicy(Qt.ScrollBarAlwaysOff)
self._holder = QWidget()
self._holder.setObjectName("ConsoleHolder")
self._holder.setStyleSheet("background: transparent;")
self._entries = QVBoxLayout(self._holder)
self._entries.setContentsMargins(0, 0, 0, 0)
self._entries.setSpacing(SPACING["xs"])
self._entries.addStretch(1)
self._scroll.setWidget(self._holder)
#: Whether the view follows new output. Cleared by raising a section
#: and restored by scrolling back to the bottom.
self._follow_output = True
self._scroll.verticalScrollBar().valueChanged.connect(
self._on_console_scrolled)
self._scroll.verticalScrollBar().valueChanged.connect(
self._refresh_jump_button)
box_lay.addWidget(self._scroll, 1)
from PySide6.QtGui import QKeySequence, QShortcut
from PySide6.QtWidgets import QPushButton
self._jump = QPushButton("↓ jump to the end", self._console_box)
self._jump.setToolTip(
"Go to the newest line. Ctrl+End does the same from anywhere in "
"the scrollback.")
jump_policy = self._jump.sizePolicy()
jump_policy.setRetainSizeWhenHidden(True)
self._jump.setSizePolicy(jump_policy)
self._jump.clicked.connect(self.jump_to_the_end)
self._jump.setVisible(False)
box_lay.addWidget(self._jump)
self._end_shortcut = QShortcut(QKeySequence("Ctrl+End"), self)
self._end_shortcut.activated.connect(self.jump_to_the_end)
self._split.addWidget(self._console_box)
input_row = QWidget()
self._chat_row = input_row
chat_column = QVBoxLayout(input_row)
chat_column.setContentsMargins(0, 0, 0, 0)
chat_column.setSpacing(SPACING["xs"])
row = QHBoxLayout()
row.setContentsMargins(0, 0, 0, 0)
row.setSpacing(SPACING["sm"])
chat_column.addLayout(row, 1)
self._input = _ChatInput()
self._input.setObjectName("ConsoleChatInput")
self._input.setPlaceholderText(
"Type here and hit Enter… (toggle AI at the bottom-right "
"to route through your chat subscription)"
)
self._input.submitted.connect(self._on_submit)
row.addWidget(self._input, 1)
from PySide6.QtWidgets import QProgressBar
self._usage_row = QWidget()
self._usage_row.setObjectName("AiUsageRow")
usage = QHBoxLayout(self._usage_row)
usage.setContentsMargins(0, 0, 0, 0)
usage.setSpacing(SPACING["sm"])
self._usage_bar = QProgressBar()
self._usage_bar.setObjectName("AiUsageBar")
self._usage_bar.setRange(0, 100)
self._usage_bar.setTextVisible(False)
self._usage_bar.setFixedHeight(6)
self._usage_bar.setFixedWidth(96)
self._usage_text = QLabel()
self._usage_text.setObjectName("AiUsageText")
usage.addWidget(self._usage_bar)
usage.addWidget(self._usage_text, 1)
self._usage_row.setVisible(False)
chat_column.addWidget(self._usage_row)
self._split.addWidget(input_row)
self._split.setStretchFactor(0, 1)
self._split.setStretchFactor(1, 0)
self._apply_default_split()
self._restore_split()
self._split.splitterMoved.connect(self._on_split_moved)
#: The right-hand column's text size (item 529); see
#: :meth:`apply_column_text_scale`.
self._column_text_scale = 1.0
self._font_pt = self._zoomed_font_pt()
self._ai_active: bool = False
self._current_provider_name: Optional[str] = None
def _apply_default_split(self) -> None:
"""Seat the handle where the panel used to draw it with no splitter.
``setSizes`` totals are advisory — Qt rescales them to the height the
splitter actually has, distributing the difference by stretch factor.
With the chat box on stretch 0 the whole difference lands on the
console, so the chat box comes out at exactly
:data:`DEFAULT_CHAT_HEIGHT` at any window size, which is what the old
``setMaximumHeight(120)`` produced.
"""
self._split.setSizes([max(CONSOLE_MIN_HEIGHT, 400), DEFAULT_CHAT_HEIGHT])
def _restore_split(self) -> None:
"""Re-seat the handle where this screen's user last dragged it."""
if not self._persist_key:
return
state = get_split_state(self._persist_key)
if state is None:
return
try:
self._split.restoreState(state)
except Exception:
self._apply_default_split()
self._split.setChildrenCollapsible(False)
def _on_split_moved(self, _pos: int = 0, _index: int = 0) -> None:
"""Persist the split as the user drags the handle."""
if not self._persist_key:
return
set_split_state(self._persist_key, self._split.saveState())
[docs]
def split_sizes(self) -> List[int]:
"""Current ``[console_height, chat_height]`` in pixels.
Public so a test, or a caller arranging the screen, can read it rather
than reaching into ``_split``.
"""
return list(self._split.sizes())
[docs]
def set_split_sizes(self, console_px: int, chat_px: int) -> None:
"""Move the handle programmatically and persist the result.
Same end state as a user drag, so a caller restoring a layout and a
user dragging leave the panel in the same place.
:param console_px: height for the console box.
:param chat_px: height for the AI chat box.
"""
self._split.setSizes([int(console_px), int(chat_px)])
self._on_split_moved()
@staticmethod
def _zoomed_font_pt() -> int:
"""The console point size, from the platform's fixed font x Zoom.
The base is the system's monospace size so the console still looks
native, and the Zoom preference multiplies it so the console tracks
the rest of the interface. Falls back to the unscaled base if
preferences cannot be read at all, which is what a first run
mid-generation gets.
"""
base = int(QFontDatabase.systemFont(
QFontDatabase.FixedFont).pointSize()) or 10
try:
from ..preferences import get_font_scale
return max(1, int(round(base * get_font_scale())))
except Exception:
return base
#: Glyphs the run notices use that Open Sans does not have, so Qt finds
#: them in a fallback font the first time each size draws them.
NOTICE_GLYPHS = "→✓■…⚠"
[docs]
def warm_notice_glyphs(self) -> bool:
"""Lay out :data:`NOTICE_GLYPHS` once in the size new entries get.
The first "→ Starting ..." notice of a session cost 33 ms on the GUI
thread, almost all of it Qt looking up a fallback font for the arrow
at the console's size; the same text costs 0.5 ms after that. The
lookup is cached per thread and per size, so it can only be done here,
on the GUI thread, ahead of the notice: a Run waiting on its
pipeline's import calls this while the busy bar is already showing.
The font is taken from a stand-in entry, because the one an entry
draws with carries the family list its sheet gives it, and a font
that differs only in that list is a different cache entry.
:returns: whether a layout was made (``False`` when this size was
warmed before).
"""
warmed = self.__dict__.setdefault("_warmed_glyph_sizes", set())
pt = int(getattr(self, "_font_pt", 10))
if pt in warmed:
return False
warmed.add(pt)
probe = _StdoutBlock()
probe.set_console_font_pt(pt)
layout = QTextLayout(self.NOTICE_GLYPHS,
probe.document().defaultFont())
layout.beginLayout()
layout.createLine()
layout.endLayout()
probe.deleteLater()
return True
[docs]
def apply_zoom(self) -> None:
"""Re-read Zoom and restyle every entry. Called on a preferences save."""
self.set_console_font_pt(max(1, int(round(
self._zoomed_font_pt()
* getattr(self, "_column_text_scale", 1.0)))))
[docs]
def apply_column_text_scale(self, scale: float) -> None:
"""Size the console's text with the column it sits in.
The entries carry their own point size (a per-widget sheet and an
explicit font), which a sheet on the column cannot reach, so the
column's Ctrl + wheel hands its size here and it multiplies Zoom.
:param scale: 1.0 for the size Zoom alone gives.
"""
scale = float(scale or 1.0)
if scale == getattr(self, "_column_text_scale", 1.0):
return
self._column_text_scale = scale
self.apply_zoom()
[docs]
def set_console_font_pt(self, pt: int) -> None:
"""Set the console font size and apply it to every existing entry.
:param pt: the font size in points, converted to int.
"""
self._font_pt = int(pt)
for block in self._holder.findChildren(_StdoutBlock):
block.set_console_font_pt(self._font_pt)
for lbl in self._holder.findChildren(QLabel):
f = lbl.font()
f.setPointSize(self._font_pt)
lbl.setFont(f)
def _apply_font(self, w: QWidget) -> None:
"""Apply the current console font size to a newly-created entry."""
if isinstance(w, _StdoutBlock):
w.set_console_font_pt(getattr(self, "_font_pt", 10))
for lbl in ([w] if isinstance(w, QLabel) else w.findChildren(QLabel)):
f = lbl.font()
f.setPointSize(getattr(self, "_font_pt", 10))
lbl.setFont(f)
def _insert_entry(self, w: QWidget) -> None:
"""Every entry — topic bar, stdout block, chat bubble — spans
the full width of the console. Bubbles no longer get a
horizontal offset row."""
self._apply_font(w)
self._entries.insertWidget(self._entries.count() - 1, w)
self._scroll_to_bottom()
def _scroll_to_bottom(self) -> None:
"""Follow the newest line, unless the user has scrolled away.
Raising a section is a statement that the user is reading there, and a
log that scrolls away from what is being read cannot be read at all.
"""
if not getattr(self, "_follow_output", True):
return
sb = self._scroll.verticalScrollBar()
sb.setValue(sb.maximum())
def _on_console_scrolled(self, value: int) -> None:
"""Follow the tail while the view is at the bottom, and only then.
The convention every log viewer uses: scrolling up means "let me
read", scrolling back down means "keep going". A few pixels of
tolerance because a scrollbar dragged to the end does not always land
exactly on maximum().
BOTH DIRECTIONS, because only one of them was ever wired. Raising a
section cleared the follow, so a reader who got to the middle of the
log by clicking a heading stayed there; a reader who got to the same
place by dragging the scrollbar was thrown back to the bottom by the
next line written, which is the thing that makes a live log
unreadable. Where the viewport is answers that question, and it
answers it the same way whichever gesture put it there.
"""
scrollbar = self._scroll.verticalScrollBar()
self._follow_output = value >= scrollbar.maximum() - 4
[docs]
def jump_to_the_end(self) -> None:
"""Show the newest line, and follow the tail again.
Both are done together: a console that jumped without resuming the
follow would slide back off the end on the very next line written, and
the user would press it again.
"""
bar = self._scroll.verticalScrollBar()
bar.setValue(bar.maximum())
self._follow_output = True
self._refresh_jump_button()
[docs]
def at_the_end(self) -> bool:
"""Whether the view is showing the newest line.
A few pixels of tolerance, because a scrollbar dragged to the end
does not always land exactly on maximum() -- the same tolerance
`_on_console_scrolled` uses, and for the same reason.
"""
bar = self._scroll.verticalScrollBar()
return bar.value() >= bar.maximum() - 4
def _refresh_jump_button(self, *_args) -> None:
"""Show the control only when it would do something."""
button = getattr(self, "_jump", None)
if button is not None:
button.setVisible(not self.at_the_end())
def _needs_topic(self, kind: str) -> bool:
"""Report whether a new banner is needed before writing this kind of entry.
:param kind: the entry kind about to be written.
:returns: ``True`` when it differs from the last one written, so
consecutive entries of one kind share a single banner.
"""
return self._last_entry_kind != kind
def _on_gui_thread(self) -> bool:
"""True when the caller is on the thread that owns this widget.
Everything this panel appends is a QWidget, and Qt only allows a
QWidget to be built on the GUI thread. Python's logging module
does not care: it runs handlers inline on whatever thread logged
the record, so a pipeline worker's ``LOG.warning`` used to land
in :meth:`append_stdout` on the worker thread and construct a
``_TopicBar`` there.
"""
return QThread.currentThread() is self.thread()
[docs]
def set_active_app(self, label: str) -> None:
"""Set the label used in the next auto-inserted topic divider.
:param label: the text shown in the next automatic topic divider.
"""
self._active_app_label = label
[docs]
def set_run_context(self, module: str = "", function: str = "") -> None:
"""Record the module/function the pipeline output comes from.
Shown in the "spaCR output — <module> — <function>" banner so users
can see the source of the output at a glance.
"""
self._run_module = module or ""
self._run_function = function or ""
def _output_banner(self, head: str) -> str:
"""Build a banner like 'spaCR output — mask — preprocess_generate_masks'."""
parts = [tr(head)]
mod = self._run_module or self._active_app_label
if mod:
parts.append(tr(str(mod)))
if self._run_function:
parts.append(str(self._run_function))
return " — ".join(parts)
[docs]
def begin_topic(self, label: str, accent: Optional[str] = None,
trailing: Optional[QWidget] = None
) -> Optional["_TopicBar"]:
"""Insert a divider bar labeled `label` (e.g. 'spaCR output — …').
A bar is not redrawn when its label would repeat. Three bands now write
under the "spaCR output" heading -- stdout, warnings, and the notice
path -- and each opens its topic. A run that alternates between them
therefore drew the identical banner before every line:
=== spaCR output — Mask Generation ===
Source directory (src): ...
=== spaCR output — Mask Generation ===
12:09:37 [WARNING] cellpose.vit: Could not import CPDINO...
=== spaCR output — Mask Generation ===
12:09:47 [INFO] spacr.qt.resource_cleanup: memory budget...
which is what the console looked like when this was reported. The
divider marks a change of subject; repeating it adds three lines to the
panel and no information.
The accent is deliberately not part of the comparison. It is applied to
the text below the bar -- amber for a warning, blue for output -- so a
warning still reads differently without a second identical heading
above it.
:param label: the heading text.
:param accent: colour for the heading, or ``None`` for the theme's.
:param trailing: a widget pinned to the right of the heading, such as
a working indicator.
:returns: the bar that was drawn, or ``None`` when the one already
showing says the same thing and was kept. The caller needs the
widget to be able to take an empty heading down again -- see
:meth:`_take_down_the_waiting_heading`.
"""
if label and label == getattr(self, "_current_topic_label", None):
self._last_entry_kind = ""
self._current_stdout = None
return None
bar = _TopicBar(label, accent=accent, trailing=trailing)
self._insert_entry(bar)
self._current_topic_label = label
self._last_entry_kind = ""
self._current_stdout = None
return bar
[docs]
def append_stdout(self, text: str) -> None:
"""Append pipeline output as blue text under a 'spaCR output' banner.
Safe to call from any thread: an off-thread call is re-posted to
the GUI thread through :attr:`_relay_stdout` and returns without
touching a widget. See :meth:`_on_gui_thread`.
Re-entrant calls on the same thread are refused. Drawing a line
runs Python inside a QWidget, and with verbose logging on the
function-trace profile hook logs on entry to every spaCR function
it passes through — including this one. Both console log sinks feed
that record straight back here, and ``_StdoutBlock.append`` answers
it with a nested ``setPlainText`` whose first act is to destroy the
QTextDocument's frames — the ones the outer call is still inside.
gdb: ``QTextFrame::~QTextFrame -> QTextDocumentPrivate::clear``,
``#0`` in freed memory. Reproduced as ``pytest
tests/qt/test_all_module_smoke.py
tests/qt/test_batch_f_diagnostics.py`` (exit 139).
:param text: the pipeline output to append; empty does nothing.
"""
if not text:
return
if not self._on_gui_thread():
self._relay_stdout.emit(text)
return
if console_write_in_progress():
return
with console_write():
if (self._current_stdout is None
or self._needs_topic("stdout")
or self._current_stdout.property(
"consoleContextKind") != "stdout"):
accent = color_output()
self.begin_topic(self._output_banner("spaCR output"),
accent=accent)
self._current_stdout = _StdoutBlock(text_color=accent)
self._current_stdout.setProperty(
"consoleContextKind", "stdout")
self._insert_entry(self._current_stdout)
self._last_entry_kind = "stdout"
self._current_stdout.append(text)
self._scroll_to_bottom()
[docs]
def append_notice(self, source: str, **values: object) -> None:
"""Append one localized spaCR-authored UI notice.
This is intentionally separate from :meth:`append_stdout`: arbitrary
worker stdout, logs, tracebacks, paths and AI responses must remain
byte-for-byte English/canonical. Off-thread notices carry their stable
English template to the GUI thread and are translated only there.
:param source: the notice's untranslated English template; empty does
nothing. It is translated on the GUI thread and filled with the
keyword values.
"""
if not source:
return
if not self._on_gui_thread():
self._relay_notice.emit(str(source), dict(values))
return
self._append_notice_on_gui_thread(str(source), dict(values))
def _append_notice_on_gui_thread(
self, source: str, values: object = None,
) -> None:
"""Translate a notice and append it, keeping its surrounding whitespace.
Call sites add line breaks for console layout, but translation keys omit
incidental leading and trailing whitespace -- so the framing is stripped
off, the core translated, and the framing put back.
:param source: the untranslated notice, with whatever framing it
carries; a blank one is dropped.
:param values: substitutions for the translated template; anything that
is not a mapping is treated as none.
"""
mapping = values if isinstance(values, dict) else {}
core = source.strip()
if not core:
return
leading = source[:len(source) - len(source.lstrip())]
trailing = source[len(source.rstrip()):]
self.append_stdout(leading + tr(core, **mapping) + trailing)
def _on_log_record(self, text: str, level: int) -> None:
"""Slot for QtLogHandler.record_ready, routed by level.
A WARNING IS NOT AN ERROR. This used to send everything at or above
WARNING through :meth:`append_error`, which draws the red "spaCR
ERROR" banner -- so a routine Qt warning ("libpyside: addMetaMethod
...") was presented to the user as a failure, a dozen times, on
merely opening a module. The cost is not the wrong colour:
it is that a pane which cries error over routine noise is a pane
people stop reading, and the next line in it might be the one that
matters.
Three bands now, not two.
"""
import logging as _logging
if level < _logging.WARNING:
self._log_pending.append(text)
if not self._log_flush.isActive():
self._log_flush.start()
return
self._flush_log_records()
if level >= _logging.ERROR:
self.append_error(text)
else:
self.append_warning(text)
def _on_log_records(self, records: list) -> None:
"""Slot for ``QtLogHandler.records_ready``: a batch, in order.
:param records: ``(text, level)`` pairs as :meth:`_on_log_record`
takes them.
"""
import logging as _logging
for text, level in records:
if level < _logging.WARNING:
self._log_pending.append(text)
else:
self._on_log_record(text, level)
if self._log_pending and not self._log_flush.isActive():
self._log_flush.start()
def _flush_log_records(self) -> None:
"""Append the informational records waiting, as one write.
A run logging thousands of records a second costs one append per
50 ms instead of one per record; a warning or an
error flushes what came before it, so the order is kept.
"""
self._log_flush.stop()
if not self._log_pending:
return
text = "".join(self._log_pending)
self._log_pending.clear()
self.append_stdout(text)
[docs]
def append_warning(self, text: str) -> None:
"""Append warning text in the theme's amber, under the output banner.
Warnings have no banner of their own: a "spaCR warning" heading would
need a new row in ``spacr.qt.i18n._ROWS`` with nine translations, which
moves the COMPACT caption ratchet. Amber under the existing translated
"spaCR output" heading keeps warnings out of the error pane and errors
in it without that change.
:param text: the formatted record; empty strings are ignored.
"""
if not text:
return
with console_write():
amber = color_warning()
self.begin_topic(self._output_banner("spaCR output"), accent=amber)
block = _StdoutBlock(text, text_color=amber)
block.setProperty("consoleContextKind", "warning")
self._insert_entry(block)
self._last_entry_kind = "stdout"
[docs]
def append_error(self, tb: str) -> None:
"""Append red error text under a 'spaCR ERROR — <module> — <function>'
banner.
:param tb: traceback text; empty strings are ignored.
Thread-safe in the same way as :meth:`append_stdout`.
"""
if not tb:
return
if not self._on_gui_thread():
self._relay_error.emit(tb)
return
if console_write_in_progress():
return
with console_write():
red = color_error()
self.begin_topic(self._output_banner("spaCR ERROR"), accent=red)
block = _StdoutBlock(tb, error=True, text_color=red)
block.setProperty("consoleContextKind", "traceback")
self._insert_entry(block)
self._last_entry_kind = "stdout"
[docs]
def as_text(self, start: int = 0, stop: Optional[int] = None) -> str:
"""The console as plain text, section headers included.
:param start: first entry index to include.
:param stop: one past the last, or ``None`` for the rest.
:returns: the text a person would have selected by hand.
"""
parts = []
last = self._entries.count() - 1
stop = last if stop is None else min(stop, last)
for index in range(max(0, start), stop):
item = self._entries.itemAt(index)
widget = item.widget() if item is not None else None
if widget is None:
continue
if isinstance(widget, _TopicBar):
parts.append(f"\n=== {widget.text()} ===")
elif hasattr(widget, "toPlainText"):
text = widget.toPlainText().rstrip()
if text:
parts.append(text)
elif hasattr(widget, "text"):
text = (widget.text() or "").strip()
if text:
parts.append(text)
return "\n".join(parts).strip() + "\n"
def _section_span(self, bar: "_TopicBar"):
"""Entry indices ``(start, stop)`` spanning ``bar`` and its content.
``stop`` is exclusive and may be ``None``, meaning "to the end".
ONE definition of where a section ends, so copying, raising and
collapsing cannot disagree about it.
The span runs to the next header that actually has something under
it. ``append_stdout`` inserts its own "spaCR output" bar, so a module
banner is followed immediately by another banner: stopping at the
first boundary copies a title and nothing else, and folds a section
that hides nothing.
"""
start = None
last = self._entries.count() - 1
boundaries = []
for index in range(last):
item = self._entries.itemAt(index)
widget = item.widget() if item is not None else None
if widget is bar:
start = index
continue
if start is not None and isinstance(widget, _TopicBar):
boundaries.append(index)
if start is None:
return None, None
for boundary in boundaries:
text = self.as_text(start, boundary)
if len(text.strip().splitlines()) > 1:
return start, boundary
return start, None
[docs]
def section_text(self, bar: "_TopicBar") -> str:
"""Return a topic bar and its content up to the next topic bar.
:param bar: the topic bar (section heading) whose section is meant. A
bar not in the console gives ``""``.
"""
start, stop = self._section_span(bar)
if start is None:
return ""
return self.as_text(start, stop)
[docs]
def section_body(self, bar: "_TopicBar"):
"""The widgets under ``bar``, up to the next topic bar.
The same span :meth:`section_text` copies, as widgets rather than as
text, so raising, collapsing and copying a section cannot disagree
about where it ends. A nested heading inside the span is part of the
body: folding a module banner folds the "spaCR output" banner under
it too, because that banner is the section's own content.
:param bar: the topic bar (section heading) whose section is meant. A
bar not in the console gives an empty list.
"""
start, stop = self._section_span(bar)
if start is None:
return []
end = self._entries.count() - 1 if stop is None else stop
body = []
for index in range(start + 1, end):
item = self._entries.itemAt(index)
widget = item.widget() if item is not None else None
if widget is not None:
body.append(widget)
return body
[docs]
def raise_section(self, bar: "_TopicBar") -> None:
"""Bring ``bar``'s section to the top of the view and expand it.
The console is a transcript, so the order of its sections is the one
property a log has: this scrolls, it does not reorder.
Raising a section also stops the view following new output. A user who
clicked a heading is reading that section, and new output would
otherwise move the viewport away from it. Following resumes when they
scroll back to the bottom, which is the convention every log viewer
uses.
:param bar: the topic bar (section heading) whose section is meant.
"""
bar.set_expanded(True)
folded = False
for widget in self.section_body(bar):
if isinstance(widget, _TopicBar):
folded = not widget.is_expanded()
widget.setVisible(True)
continue
widget.setVisible(not folded)
self._follow_output = False
_after(0, self, lambda: self._scroll_widget_to_top(bar))
def _scroll_widget_to_top(self, bar) -> None:
"""Scroll the console so a widget sits at the top of the viewport.
:param bar: the widget to bring to the top -- typically a section
heading that was just clicked. A section torn down between the click
and the layout is ignored rather than raising.
"""
try:
top = bar.mapTo(self._holder, bar.rect().topLeft()).y()
except RuntimeError:
return
scrollbar = self._scroll.verticalScrollBar()
scrollbar.setValue(min(top, scrollbar.maximum()))
[docs]
def collapse_section(self, bar: "_TopicBar") -> None:
"""Hide ``bar``'s body, leaving its heading in place.
:param bar: the topic bar (section heading) whose section is meant.
"""
bar.set_expanded(False)
for widget in self.section_body(bar):
widget.setVisible(False)
def _is_raised(self, bar: "_TopicBar") -> bool:
"""Whether ``bar`` is already sitting at the top of the viewport."""
try:
top = bar.mapTo(self._holder, bar.rect().topLeft()).y()
except RuntimeError:
return False
scrollbar = self._scroll.verticalScrollBar()
return abs(scrollbar.value() - min(top, scrollbar.maximum())) <= 4
[docs]
def toggle_section(self, bar: "_TopicBar") -> None:
"""Scroll to the section first; fold it on a second click.
Sections are created expanded, so a plain expanded/collapsed toggle
would hide the section on the first click without scrolling to it.
A heading that is not already at the top of the viewport is therefore a
request to scroll there, whatever its state. Only a heading already at
the top is collapsed by the click.
:param bar: the topic bar (section heading) whose section is meant.
"""
if bar.is_expanded() and self._is_raised(bar):
self.collapse_section(bar)
else:
self.raise_section(bar)
[docs]
def copy_all(self) -> str:
"""Put the whole console on the clipboard; return what was copied."""
text = self.as_text()
try:
from PySide6.QtWidgets import QApplication
QApplication.clipboard().setText(text)
except Exception:
pass
return text
[docs]
def clear(self) -> None:
"""Wipe every entry (topic bars, stdout blocks, chat bubbles)."""
while self._entries.count() > 1:
item = self._entries.takeAt(0)
w = item.widget() if item else None
if w is not None:
w.setParent(None)
w.deleteLater()
self._last_entry_kind = ""
self._current_stdout = None
self._current_topic_label = None
self._pending_ai_topic = None
self._pending_ai_block = None
self._ai_messages.clear()
self._console_sent_lengths.clear()
[docs]
def set_ai_active(self, on: bool) -> None:
"""Enable/disable AI routing for Enter-submits from the input.
:param on: whether Enter in the input goes to the AI; converted to
bool.
"""
self._ai_active = bool(on)
[docs]
def set_ai_provider(self, provider_name: Optional[str]) -> None:
"""Select the provider used for AI submissions, or None to unset.
:param provider_name: the provider's name, or None to unset it.
"""
self._current_provider_name = provider_name
self.refresh_ai_usage()
def _hosted_is_current(self) -> bool:
"""Whether the selected provider is spaCR's hosted one."""
from ..ai.providers import HOSTED_PROVIDER_NAME
return self._current_provider_name == HOSTED_PROVIDER_NAME
[docs]
def refresh_ai_usage(self) -> None:
"""Refresh the usage bar under the chat box for the selected provider.
For spaCR AI (free) the relay is asked, off the GUI thread, how much
of today's share is left; the request runs on a plain thread that
touches no Qt object, and a single-shot timer tied to this panel
collects the answer on the GUI thread, so a panel closed meanwhile
never hears back. A vendor CLI has no usage source, so its line is
shown at once and the bar is hidden
(:func:`spacr.qt.ai.providers.usage_bar`). No provider hides the row.
"""
name = self._current_provider_name
if not self._hosted_is_current():
self._show_ai_usage(None)
return
import threading
from ... import crashreport
from .. import _after
box: dict = {}
def _fetch() -> None:
"""Fetch the allowance into ``box``; touches no Qt object."""
try:
box["allowance"] = crashreport.fetch_allowance()
except Exception:
box["allowance"] = None
fetcher = threading.Thread(target=_fetch,
name="spacr-hosted-allowance", daemon=True)
def _collect() -> None:
"""On the GUI thread: show the allowance once it has arrived."""
if "allowance" not in box and fetcher.is_alive():
_after(100, self, _collect)
return
if self._current_provider_name == name:
self._show_ai_usage(box.get("allowance"))
fetcher.start()
_after(100, self, _collect)
def _show_ai_usage(self, allowance: object) -> None:
"""Draw the usage bar and its text for the selected provider.
:param allowance: for spaCR AI, what
:func:`spacr.crashreport.fetch_allowance` returned, or ``None``.
"""
from ..ai.providers import hosted_cli_hint, usage_bar
percent, text = usage_bar(
self._current_provider_name,
allowance if isinstance(allowance, dict) else None)
self._usage_bar.setVisible(percent is not None)
if percent is not None:
self._usage_bar.setValue(int(percent))
self._usage_text.setText(text)
self._usage_row.setToolTip(
hosted_cli_hint() if self._hosted_is_current() else "")
self._usage_row.setVisible(bool(text))
def _current_provider(self) -> Optional[ChatProvider]:
"""Resolve the selected AI provider.
:returns: the provider, or ``None`` when none is selected.
"""
if not self._current_provider_name:
return None
return ai_module.get_provider(self._current_provider_name)
def _on_submit(self) -> None:
"""Send the chat box's contents, to the AI or to the console.
An empty box does nothing. With AI off the text is written as a local
note under its own banner rather than dropped, so a typed thought stays
in the transcript beside the run it was about.
"""
text = self._input.toPlainText().strip()
if not text:
return
self._input.clear()
if self._ai_active:
self._send_to_ai(text)
else:
self._append_user(text)
def _append_user(self, text: str) -> None:
"""Insert a 'spaCR user' banner + green user text."""
green = color_user()
self.begin_topic(tr("spaCR user"), accent=green)
block = _StdoutBlock(text, text_color=green)
self._insert_entry(block)
self._current_stdout = None
self._last_entry_kind = "user"
def _send_to_ai(self, text: str) -> None:
"""Send one question to the provider and open a reply block for it.
Console context is attached according to the AI preference and reported
on the message it went with, rather than as furniture that goes stale
between asks. Asking a question of the user's own also ends the error
pairing: whatever comes back answers this, not the crash, so it must not
later be filed as an analysis of the crash.
:param text: the user's question. With no provider configured this says
so; with a stream already running it is silently dropped, since the
Cancel button lives on the actions row rather than here.
"""
provider = self._current_provider()
if provider is None:
self.append_notice(
"[AI] No provider configured. Open Providers…\n"
)
return
if self._ai_thread is not None:
return
context, status = self._console_context_for_question(text)
prompt = text
if context:
prompt += (
"\n\n<spacr_console_context>\n" + context
+ "\n</spacr_console_context>")
self._ai_messages.append({"role": "user", "content": prompt})
self._ai_error_traceback = ""
self._append_user(text + f"\n\n[{status}]")
self._begin_ai_reply()
def _begin_ai_reply(self) -> None:
"""Open the "spaCR AI" heading and reply block, then start the stream."""
ai_color = ai_color_for_provider(self._current_provider_name)
self._working_dots = _WorkingDots(color=ai_color)
self._pending_ai_topic = self.begin_topic(
tr("spaCR AI"), accent=ai_color, trailing=self._working_dots)
self._working_dots.start()
self._current_stdout = _StdoutBlock(text_color=ai_color)
self._insert_entry(self._current_stdout)
self._pending_ai_block = self._current_stdout
self._last_entry_kind = "ai"
self._start_stream(system=ai_settings.get_system_prompt())
def _pipeline_console_blocks(self):
"""Yield rendered pipeline blocks that are eligible as AI context."""
last = self._entries.count() - 1
for index in range(last):
item = self._entries.itemAt(index)
widget = item.widget() if item is not None else None
kind = widget.property("consoleContextKind") if widget else None
if kind in {"stdout", "traceback"} and hasattr(
widget, "toPlainText"):
yield widget, str(kind), widget.toPlainText()
def _console_context_for_question(self, question: str):
"""Package unsent console context when an AI question is submitted.
Return ``(context, visible_status)``. Complete traceback blocks take
priority over ordinary output and may exceed the soft context budget.
When the default-on console-aware preference is disabled, no context
is attached. Text is marked sent only after it is included.
"""
from ..ai import settings as ai_settings
if not ai_settings.get_console_aware():
label = tr("Console context off")
return "", label
pieces = []
current_lengths = {}
for block, kind, full_text in self._pipeline_console_blocks():
key = id(block)
sent = min(self._console_sent_lengths.get(key, 0), len(full_text))
fresh = full_text[sent:]
if not fresh:
continue
anchor = ""
if kind == "stdout" and sent:
anchor = full_text[max(0, sent - 400):sent]
pieces.append((kind, anchor + fresh, len(fresh)))
current_lengths[key] = len(full_text)
if not pieces:
label = tr("Console context: no new output")
return "", label
tracebacks = [text for kind, text, _fresh in pieces
if kind == "traceback"]
stdout = "\n".join(text for kind, text, _fresh in pieces
if kind == "stdout")
traceback_text = "\n\n".join(
f"--- complete traceback ---\n{text}" for text in tracebacks)
remaining = max(0, AI_CONSOLE_CONTEXT_CHARS - len(traceback_text))
kept_stdout = stdout[-remaining:] if remaining else ""
dropped = max(0, len(stdout) - len(kept_stdout))
sections = []
if dropped:
sections.append(
f"[Console tail; {dropped:,} earlier characters dropped]")
elif stdout:
sections.append("[New console output]")
if kept_stdout:
sections.append(kept_stdout)
if traceback_text:
sections.append(traceback_text)
context = "\n".join(sections)
self._console_sent_lengths.update(current_lengths)
label = tr("Console context: {n} chars sent", n=f"{len(context):,}")
if dropped:
label += tr(", {n} dropped", n=f"{dropped:,}")
return context, label
[docs]
def begin_ai_entry(self, thinking: bool = True) -> "_StdoutBlock":
"""Open a "spaCR AI" entry of its own, thinking while it is empty.
The entry is a section like the others -- a "spaCR AI" heading in
spaCR teal (:func:`color_spacr_ai`) over its own block -- written
where the console stands now, directly under the error it explains.
Its heading carries the working dots, and :meth:`fill_ai_entry`
puts the answer in. Whatever is written meanwhile opens its own
section below it. GUI thread only.
:param thinking: show the working dots; ``False`` for an answer
already at hand, such as one saved on this computer.
:returns: the entry's block, to hand to :meth:`fill_ai_entry`.
"""
teal = color_spacr_ai()
dots = _WorkingDots(teal)
self._current_topic_label = None
bar = self.begin_topic(tr("spaCR AI"), accent=teal, trailing=dots)
block = _StdoutBlock(text_color=teal)
block.setProperty("consoleContextKind", "spacr_ai")
block.hide()
block._ai_dots = dots
block._ai_bar = bar
self._insert_entry(block)
self._current_stdout = None
self._last_entry_kind = "spacr_ai"
if thinking:
dots.start()
else:
dots.hide()
return block
[docs]
def fill_ai_entry(self, block: Optional["_StdoutBlock"],
text: str) -> "_StdoutBlock":
"""Put spaCR AI's answer (or why there is none) into its entry.
The working dots stop and the answer replaces them, its Markdown
drawn (:meth:`_StdoutBlock.show_markdown`). An entry that is gone,
because the console was cleared, is opened again at the end.
:param block: what :meth:`begin_ai_entry` returned, or ``None``.
:param text: the answer or the one-line note to show.
:returns: the block now showing it.
"""
try:
dots = getattr(block, "_ai_dots", None) if block else None
if dots is not None:
dots.stop()
if block is None or not block.parent():
raise RuntimeError("the entry is gone")
except RuntimeError:
block = self.begin_ai_entry(thinking=False)
block.show_markdown(text)
block.show()
self._scroll_to_bottom()
return block
def _ensure_stdout_block(self) -> None:
"""Open a new plain stdout block if the last entry was not one."""
if self._current_stdout is None or self._needs_topic("stdout"):
block = _StdoutBlock()
self._insert_entry(block)
self._current_stdout = block
self._last_entry_kind = "stdout"
def _start_stream(self, system: str) -> None:
"""Start the streaming worker for the pending conversation.
The thread is parented to the panel so its C++ lifetime is tied to the
panel rather than to a Python refcount -- an unparented ``QThread`` can
be collected between the worker returning and ``finished`` firing, which
aborts Qt.
:param system: system prompt for the request.
"""
provider = self._current_provider()
if provider is None:
return
if not self._hosted_is_current():
from ..ai.prompts import tool_approval_note
note = tool_approval_note(self._current_provider_name or "")
system = f"{system}\n\n{note}" if system else note
self._ai_buf = []
thread, worker = make_stream_thread(
provider, list(self._ai_messages), system=system,
parent=self,
)
worker.stage_changed.connect(self._on_stage)
worker.chunk_ready.connect(self._on_chunk)
worker.finished.connect(self._on_stream_finished)
self._ai_thread = thread
self._ai_worker = worker
thread.start()
[docs]
def cancel_ai(self) -> None:
"""Public — AppScreen calls this if the user cancels a stream."""
if self._ai_worker is not None:
self._ai_worker.cancel()
def _prune_retired(self) -> None:
"""Drop entries whose QThread has already exited (isRunning
returns False) OR whose C++ was already deleted by Qt's
deferred-delete queue. Both are safe to forget."""
alive = []
for thread, worker in self._retired:
try:
if thread.isRunning():
alive.append((thread, worker))
except RuntimeError:
pass
self._retired = alive
[docs]
def is_ai_streaming(self) -> bool:
"""Return True while an AI response is being streamed."""
return self._ai_thread is not None
[docs]
def shutdown(self) -> None:
"""Cancel any active stream and block until its QThread has
exited. Must be called before the panel (or its parent window)
is destroyed — otherwise Python drops the last reference to
the running QThread and Qt aborts with:
`QThread: Destroyed while thread '' is still running`.
The cancel path kills the CLI subprocess directly so the
stream reader unblocks immediately; this method then waits for
the worker's run() to return and the QThread to quit normally.
There is deliberately no ``QThread.terminate()`` fallback. It used
to be here, described as a last resort, and it was reached far more
often than "last resort" suggests: `spacr.qt.ai.worker` queued
`worker.finished -> thread.quit` to the GUI-affine QThread object,
so the event that stops the thread sat behind this method's own
`wait()` and the wait timed out on streams that had already
finished. Terminating a thread that is running Python is
`pthread_cancel`: if it dies holding the GIL the process stops
making progress with every thread still alive, and if it dies
inside Qt or PySide the heap is corrupt and the crash lands
somewhere unrelated later. `bridge.drain_thread` parks a thread
that will not stop instead, which keeps the "never destroy a
running QThread" rule without buying it with undefined behaviour.
"""
from ..bridge import drain_thread
worker = self._ai_worker
thread = self._ai_thread
try:
for p in ai_module.list_providers():
p.cancel_stream()
except Exception:
pass
if worker is not None:
try:
worker.cancel()
except Exception:
pass
drain_thread(thread, worker, timeout_ms=3000)
self._ai_thread = None
self._ai_worker = None
for pair in list(self._retired):
drain_thread(pair[0], pair[1], timeout_ms=1000)
self._retired.clear()
[docs]
def closeEvent(self, event) -> None:
"""Ensure the AI thread is drained before Qt destroys the panel.
:param event: the close event, passed to the base class after
:meth:`shutdown` drains the AI thread.
"""
self.shutdown()
super().closeEvent(event)
def _on_stage(self, _stage: str) -> None:
"""Ignore a stage change from the streaming worker.
:param _stage: the stage name. Nothing is shown for it yet -- this is
where a spinner would go.
"""
pass
def _take_down_the_waiting_heading(self) -> Optional[QWidget]:
"""Remove the "spaCR AI" heading opened before the reply, if it is empty.
A reply's heading is drawn when the question is asked, so the working
dots have somewhere to sit while the provider thinks. Anything written
in the meantime opens a heading of its own underneath it -- a failing
run writes its manifest, "run closed" and "✗ Failed" under
"spaCR output" -- and the reply then arrives below THAT and opens a
second "spaCR AI" heading. GitHub #117's console is the result, and
read top to bottom the empty first heading is the part that says the
AI was never asked.
So the heading MOVES rather than repeating: the empty one is taken
down, and its working indicator is handed back to be pinned to the
heading that replaces it, which keeps the indicator running from the
moment the question was asked. A reply that is not interrupted keeps
the heading it started under and nothing here runs.
:returns: the working indicator, detached and still running, to pass
as the replacement heading's ``trailing``; or ``None`` when there
was no empty heading to take down, or nothing was pinned to it.
"""
bar = self._pending_ai_topic
block = self._pending_ai_block
self._pending_ai_topic = None
self._pending_ai_block = None
if bar is None:
return None
try:
label = bar.text()
if block is not None and block.toPlainText().strip():
return None
except RuntimeError:
return None
dots = self._working_dots
if dots is not None:
dots.setParent(self)
for widget in (block, bar):
if widget is None:
continue
try:
self._entries.removeWidget(widget)
widget.setParent(None)
widget.deleteLater()
except RuntimeError:
pass
if self._current_stdout is block:
self._current_stdout = None
if self._current_topic_label == label:
self._current_topic_label = None
self._last_entry_kind = ""
return dots
def _on_chunk(self, chunk: str) -> None:
"""Append one streamed chunk to the AI reply block.
The block is recreated if it went away -- the error flow writes through
its own path and can clear it mid-stream.
A RECREATED BLOCK TAKES THE "spaCR AI" HEADING WITH IT. When a run
fails, the error flow opens the heading and its reply block at once,
and the run then writes its closing lines -- the manifest, "run
closed", "✗ Failed" -- under "spaCR output" before the provider's
first line arrives. Without a heading of its own the whole reply sat
under "spaCR output", in the AI's colour, and the "spaCR AI" heading
above it stayed empty: issue 117's console, which read as "the AI was
never asked". Drawing a second heading fixed the colour and left the
empty one, so the empty one is now taken down and its working
indicator moves to the heading the reply really starts under -- see
:meth:`_take_down_the_waiting_heading`.
:param chunk: the text just received.
"""
self._ai_buf.append(chunk)
if self._current_stdout is None or self._last_entry_kind != "ai":
ai_color = ai_color_for_provider(self._current_provider_name)
dots = self._take_down_the_waiting_heading()
self.begin_topic(tr("spaCR AI"), accent=ai_color, trailing=dots)
self._current_stdout = _StdoutBlock(text_color=ai_color)
self._insert_entry(self._current_stdout)
self._last_entry_kind = "ai"
else:
self._pending_ai_topic = None
self._pending_ai_block = None
self._current_stdout.append(chunk)
self._scroll_to_bottom()
def _on_stream_finished(self, ok: bool, final_text: str) -> None:
"""Close the reply block and retire the streaming thread.
The finished ``(thread, worker)`` pair is held in a list rather than
dropped, so Python cannot collect the ``QThread`` before its OS thread
has exited and Qt's ``deleteLater`` has run; already-dead entries are
pruned on the way in so the list cannot grow across a long session.
:param ok: whether the stream completed.
:param final_text: the assembled reply, or the error detail when
``ok`` is ``False``.
"""
self._prune_retired()
if self._working_dots is not None:
self._working_dots.stop()
self._working_dots = None
self._take_down_the_waiting_heading()
thread, worker = self._ai_thread, self._ai_worker
self._ai_thread = None
self._ai_worker = None
if thread is not None:
self._retired.append((thread, worker))
if ok:
self._ai_messages.append(
{"role": "assistant", "content": final_text}
)
if getattr(self, "_ai_error_traceback", ""):
self._ai_error_explanation = final_text or ""
if not self._ai_buf:
self.append_notice(
"(empty response — try again or switch provider)\n"
)
if self._hosted_is_current():
provider = self._current_provider()
allowance = getattr(provider, "last_allowance", None)
if allowance:
self._show_ai_usage(allowance)
else:
self.refresh_ai_usage()
else:
self.append_notice(
"[AI error] {detail}\n", detail=final_text)
if self._hosted_is_current():
self.refresh_ai_usage()
if self._current_stdout is not None:
self._current_stdout.append("\n")
self._ai_buf = []
self.ai_stream_finished.emit()
if ok:
self._offer_tool_approval()
def _offer_tool_approval(self) -> None:
"""Ask the user about tool calls the CLI blocked, and resume if allowed.
``claude -p`` cannot show its own approval prompt, so a command that
needs permission was refused silently and the reply asked the user to
approve a prompt that never appeared (GitHub #147). The provider keeps
the refused calls; this asks once in a dialog, and an allowed call is
run by resuming the same session with exactly that call allowed, for
that one reply.
"""
from ..ai.providers import describe_tool_call
provider = self._current_provider()
denials = [d for d in getattr(provider, "last_permission_denials", None) or []
if isinstance(d, dict)]
if not denials or not hasattr(provider, "allow_once"):
return
provider.last_permission_denials = []
session = str(getattr(provider, "last_session_id", "") or "")
calls = list(dict.fromkeys(describe_tool_call(d) for d in denials))
shown = "; ".join(calls)
if not session or not self._ask_tool_approval(calls):
self.append_notice(
"[AI] Not run: {calls}. Run it in your own terminal if you "
"need its output.\n", calls=shown)
return
provider.allow_once(denials, session)
self.append_notice("[AI] Allowed once: {calls}\n", calls=shown)
self._ai_messages.append({"role": "user", "content": (
"I approved these tool calls in spaCR's dialog. Run them now and "
"continue:\n" + "\n".join(f"- {call}" for call in calls))})
self._begin_ai_reply()
def _tool_approval_box(self, calls: List[str]) -> QMessageBox:
"""The dialog that asks whether the assistant may run ``calls`` once.
:param calls: one line per tool call, the command for a shell call.
:returns: the box, with :attr:`_allow_button` set to its allow button.
"""
box = QMessageBox(self)
box.setObjectName("aiToolApprovalBox")
box.setIcon(QMessageBox.Question)
box.setWindowTitle(tr("Allow the assistant to run this?"))
box.setText(tr("The assistant asked to run:") + "\n\n"
+ "\n".join(calls))
box.setInformativeText(tr(
"Allow once runs it a single time with your user permissions, "
"then the assistant continues. Allow it only if you trust the "
"command."))
self._allow_button = box.addButton(tr("Allow once"),
QMessageBox.AcceptRole)
deny = box.addButton(tr("Don't allow"), QMessageBox.RejectRole)
box.setDefaultButton(deny)
box.setEscapeButton(deny)
return box
def _ask_tool_approval(self, calls: List[str]) -> bool:
"""Show :meth:`_tool_approval_box` and return whether the user allowed it.
:param calls: one line per tool call.
:returns: ``True`` for "Allow once".
"""
box = self._tool_approval_box(calls)
box.exec()
allowed = box.clickedButton() is self._allow_button
box.deleteLater()
return allowed
[docs]
def ai_explanation_of(self, traceback_text: str) -> str:
"""spaCR AI's answer about ``traceback_text``, or ``""``.
Used by the bug reporter: when the AI is switched on it has usually
already diagnosed the crash by the time the user files, and including
that analysis saves the reader of the report from reproducing it.
Empty unless there is an answer and it is an answer to this error. The
console holds one conversation across a whole session, so without the
second condition a report about one crash would carry an explanation of
an earlier one, stated with equal confidence.
:param traceback_text: the traceback the report is about; compared,
whitespace-stripped, with the traceback the AI last explained.
"""
mine = getattr(self, "_ai_error_traceback", "") or ""
answer = getattr(self, "_ai_error_explanation", "") or ""
if not mine or not answer:
return ""
return answer if mine.strip() == (traceback_text or "").strip() else ""
[docs]
def open_error_flow(self, traceback_text: str, active_app: str = "",
show_raw: bool = True) -> None:
"""Send a traceback to the AI explainer and stream the reply inline.
:param traceback_text: raw traceback captured from the pipeline.
:param active_app: optional app label used in the framing prompt.
:param show_raw: when False, the raw traceback is not printed to the
console (only a short note); the AI still receives it in its
prompt, so the user can ask the AI to show the error.
"""
from ..ai.prompts import wrap_error_for_prompt, error_explainer_prompt
if self._current_provider() is None:
self.append_notice(
"[AI] Enable AI in the actions row + pick a provider first.\n"
)
return
prompt = wrap_error_for_prompt(
traceback_text, active_app or self._active_app_label
)
self._ai_error_traceback = traceback_text
self._ai_error_explanation = ""
self._ai_messages.append({"role": "user", "content": prompt})
self._append_user(
prompt if show_raw
else tr(
"An error occurred — asking spaCR AI to explain it. "
"(Ask the AI to \"show the raw error\" to see the traceback.)"
))
ai_color = ai_color_for_provider(self._current_provider_name)
self._working_dots = _WorkingDots(color=ai_color)
self._pending_ai_topic = self.begin_topic(
tr("spaCR AI"), accent=ai_color, trailing=self._working_dots)
self._working_dots.start()
self._current_stdout = _StdoutBlock(text_color=ai_color)
self._insert_entry(self._current_stdout)
self._pending_ai_block = self._current_stdout
self._last_entry_kind = "ai"
self._start_stream(system=error_explainer_prompt())