F..................       [100%]
=================================== FAILURES ===================================
_ Phase15CQtWorkspaceIntegrationTests.test_buffer_store_is_bounded_snapshot_only_and_payload_safe _
tests/test_phase15c_qt_workspace_integration.py:653: in test_buffer_store_is_bounded_snapshot_only_and_payload_safe
    self.assertIn("class SourceBufferStore(QObject):", source)
E   AssertionError: 'class SourceBufferStore(QObject):' not found in '"""Persistent per-document Qt buffers for the closed SourceBundle."""\n\nfrom __future__ import annotations\n\nfrom dataclasses import dataclass\nimport re\nfrom typing import Any, Iterable\n\nfrom PySide6.QtCore import QObject, QSignalBlocker, QTimer, Signal\nfrom PySide6.QtGui import QTextCursor, QTextDocument\nfrom PySide6.QtWidgets import QPlainTextDocumentLayout\n\nfrom .qt_buffer_retention import BufferRetentionMixin\n\n\nMAX_SOURCE_BUFFERS = 64\nDEFER_SOURCE_CHARS = 256 * 1024\nSOURCE_SLICE_CHARS = 64 * 1024\n_PARAGRAPH_BREAK = re.compile(r"[\\r\\n\\u2029]")\n\n\ndef _slice_end(text: str, start: int) -> int:\n    """Append complete paragraphs: repeatedly laying out a long line is quadratic.\n\n    A single unusually long paragraph remains one Qt operation. The slice size\n    is a scheduling target, not a hard latency or memory guarantee.\n    """\n    limit = min(len(text), start + SOURCE_SLICE_CHARS)\n    if limit == len(text):\n        return limit\n    end = max(text.rfind(char, start, limit) for char in "\\r\\n\\u2029")\n    if end < start:\n        match = _PARAGRAPH_BREAK.search(text, limit)\n        end = match.start() if match else len(text) - 1\n    end += 1\n    if text[end - 1] == "\\r" and text[end:end + 1] == "\\n":\n        end += 1\n    return end\n\n\n@dataclass(slots=True)\nclass _SourceLoad:\n    document_id: str\n    text: str\n    document: QTextDocument\n    cursor: QTextCursor\n    offset: int = 0\n\n\ndef detach_source_views(editors):\n    """Detach shared/folded views before their buffer store is released."""\n    for editor in editors:\n        editor.verticalScrollBar().setValue(0)\n    for editor in editors:\n        empty = QTextDocument(editor)\n        empty.setDocumentLayout(QPlainTextDocumentLayout(empty))\n        with QSignalBlocker(editor):\n            editor.bind_text_document(empty)\n            editor.setReadOnly(True)\n\n\n@dataclass(slots=True)\nclass _SourceBuffer:\n    document: QTextDocument\n    source_key: int\n    snapshot_text: str | None = None\n\n    def invalidate_text(self, position, removed, added):\n        if removed or added:\n            self.snapshot_text = None\n\n\nclass SourceBufferStore(BufferRetentionMixin, QObject):\n    """Publish complete GUI-owned buffers, preparing large sources on demand."""\n\n    ready = Signal(str)\n\n    def __init__(self, parent: QObject, *, document_template=None) -> None:\n        super().__init__(parent)\n        self._document_template = document_template\n        self._buffers: dict[str, _SourceBuffer] = {}\n        self._sources: dict[str, str] = {}\n        self._normalized: dict[str, str] = {}\n        self._requested: set[str] = set()\n        self._active_document_id: str | None = None\n        self._selection_key = None\n        self.selection_generation = 0\n        self._load: _SourceLoad | None = None\n        self._closed = False\n        self._initialize_retention()\n        self._load_timer = QTimer(self)\n        self._load_timer.setSingleShot(True)\n        self._load_timer.setInterval(1)\n        self._load_timer.timeout.connect(self._advance_load)\n        self._empty_document = QTextDocument(self)\n        self._empty_document.setDocumentLayout(\n            QPlainTextDocumentLayout(self._empty_document)\n        )\n\n    @property\n    def empty_document(self) -> QTextDocument:\n        """Return the source editor\'s read-only zero-workspace document."""\n\n        return self._empty_document\n\n    @property\n    def document_ids(self) -> tuple[str, ...]:\n        return tuple(self._buffers)\n\n    def document_for(self, document_id: str) -> QTextDocument:\n        try:\n            buffer = self._buffers.pop(document_id)\n            self._buffers[document_id] = buffer\n            return buffer.document\n        except KeyError as exc:\n            raise KeyError(\n                f"source buffer is unavailable: {document_id}"\n            ) from exc\n\n    def is_modified(self, document_id: str) -> bool:\n        buffer = self._buffers.get(document_id)\n        return buffer is not None and buffer.document.isModified()\n\n    def is_ready(self, document_id: str | None) -> bool:\n        return document_id in self._buffers\n\n    def source_text(self, document_id: str) -> str:\n        """Return complete Qt-normalized text, including unvisited sources."""\n        if self.is_ready(document_id):\n            return self.text_for_document(self.document_for(document_id))\n        if document_id not in self._normalized:\n            text = (\n                self._sources[document_id].replace("\\r\\n", "\\n")\n                .replace("\\r", "\\n").replace("\\u2028", "\\n")\n                .replace("\\u2029", "\\n").replace("\\u00a0", " ")\n            )\n            self._remember_normalized(document_id, text)\n            return text\n        self._normalized[document_id] = self._normalized.pop(document_id)\n        return self._normalized[document_id]\n\n    def text_for_document(self, document: QTextDocument) -> str:\n        """Share one immutable Qt-normalized snapshot between current consumers."""\n        for buffer in self._buffers.values():\n            if buffer.document is document:\n                if buffer.snapshot_text is None:\n                    buffer.snapshot_text = document.toPlainText()\n                    self._schedule_retention()\n                return buffer.snapshot_text\n        return document.toPlainText()\n\n    def reconcile(\n        self,\n        documents: Iterable[Any],\n        *,\n        skip_document_id: str | None = None,\n        active_document_id: str | None = None,\n    ) -> None:\n        if self._closed:\n            return\n        records = tuple(documents)\n        if records and not 1 <= len(records) <= MAX_SOURCE_BUFFERS:\n            raise ValueError("source buffers accept at most 64 documents")\n        identifiers = tuple(item.document_id for item in records)\n        if len(identifiers) != len(set(identifiers)):\n            raise ValueError("source buffer document IDs must be unique")\n        self._active_document_id = active_document_id\n        sources = {item.document_id: item.text for item in records}\n        selection_key = (active_document_id, id(sources.get(active_document_id)))\n        if selection_key != self._selection_key:\n            self.selection_generation += 1\n            self._selection_key = selection_key\n            self._schedule_retention()\n        for identifier in tuple(self._normalized):\n            if sources.get(identifier) is not self._sources.get(identifier):\n                self._normalized.pop(identifier)\n        self._sources = sources\n        dirty_ids = {item.document_id for item in records if item.dirty}\n        if dirty_ids != self._dirty_ids:\n            self._dirty_ids = dirty_ids\n            self._schedule_retention()\n        for keys in self._pins.values():\n            keys.intersection_update(identifiers)\n        self._requested.intersection_update(identifiers)\n        for document_id in tuple(self._buffers):\n            if document_id in identifiers:\n                continue\n            self._discard_buffer(document_id)\n        for item in records:\n            key = id(item.text)\n            buffer = self._buffers.get(item.document_id)\n            if (buffer is not None and item.document_id != skip_document_id\n                    and buffer.source_key != key and len(item.text) > DEFER_SOURCE_CHARS):\n                self._discard_buffer(item.document_id)\n                buffer = None\n            if buffer is None:\n                if len(item.text) > DEFER_SOURCE_CHARS:\n                    if active_document_id is None:\n                        self._requested.add(item.document_id)\n                    continue\n                document = self._new_document()\n                document.setPlainText(item.text)\n                # Controller ``dirty`` is disk-save authority.  Qt\'s modified\n                # bit is reserved for text not yet synchronized back into the\n                # immutable workspace snapshot.\n                document.setModified(False)\n                self._buffers[item.document_id] = _SourceBuffer(\n                    document,\n                    key,\n                )\n                document.contentsChange.connect(\n                    self._buffers[item.document_id].invalidate_text\n                )\n                self._schedule_retention()\n                continue\n            if (\n                item.document_id != skip_document_id\n                and buffer.source_key != key\n            ):\n                undo_enabled = buffer.document.isUndoRedoEnabled()\n                buffer.document.setUndoRedoEnabled(False)\n                buffer.document.setPlainText(item.text)\n                buffer.document.setUndoRedoEnabled(undo_enabled)\n                buffer.document.setModified(False)\n                buffer.source_key = key\n                self._schedule_retention()\n        desired = self._requested | {active_document_id}\n        if self._load is not None and (\n            self._load.document_id not in desired\n            or self._sources.get(self._load.document_id) is not self._load.text\n            or (active_document_id is not None\n                and not self.is_ready(active_document_id)\n                and self._load.document_id != active_document_id)\n        ):\n            self._cancel_load()\n        self._start_next_load()\n\n    def prepare(self, document_ids: Iterable[str]) -> None:\n        """Request buffers for a reviewed bundle edit; build only one at a time."""\n        if self._closed:\n            return\n        identifiers = tuple(document_ids)\n        self.pin_documents("preparation", identifiers)\n        self._requested.update(identifier for identifier in identifiers if identifier in self._sources)\n        self._start_next_load()\n\n    def cancel_preparation(self) -> None:\n        self._requested.clear()\n        self.pin_documents("preparation", ())\n        if self._load is not None and self._load.document_id != self._active_document_id:\n            self._cancel_load()\n\n    def _new_document(self) -> QTextDocument:\n        document = QTextDocument(self)\n        document.setDocumentLayout(QPlainTextDocumentLayout(document))\n        if self._document_template is not None:\n            template = self._document_template()\n            document.setDefaultFont(template.defaultFont())\n            document.setDefaultTextOption(template.defaultTextOption())\n        return document\n\n    def _discard_buffer(self, document_id: str) -> None:\n        buffer = self._buffers.pop(document_id)\n        buffer.document.contentsChange.disconnect(buffer.invalidate_text)\n        # Views detach before Qt processes this deferred deletion.\n        buffer.document.deleteLater()\n\n    def _cancel_load(self) -> None:\n        self._load_timer.stop()\n        if self._load is not None:\n            self._load.document.deleteLater()\n            self._load = None\n\n    def _start_next_load(self) -> None:\n        if self._closed or self._load is not None:\n            return\n        self._requested.difference_update(self._buffers)\n        candidates = (self._active_document_id, *self._sources)\n        identifier = next((key for key in candidates if key in self._sources\n                           and key not in self._buffers\n                           and (key == self._active_document_id or key in self._requested)), None)\n        if identifier is None:\n            return\n        document = self._new_document()\n        document.setUndoRedoEnabled(False)\n        self._load = _SourceLoad(identifier, self._sources[identifier], document, QTextCursor(document))\n        self._load_timer.start()\n\n    def _advance_load(self) -> None:\n        load = self._load\n        if load is None or self._closed:\n            return\n        if self._sources.get(load.document_id) is not load.text:\n            self._cancel_load()\n            self._start_next_load()\n            return\n        end = _slice_end(load.text, load.offset)\n        load.cursor.insertText(load.text[load.offset:end])\n        load.offset = end\n        if end < len(load.text):\n            self._load_timer.start()\n            return\n        load.document.setUndoRedoEnabled(True)\n        load.document.setModified(False)\n        buffer = _SourceBuffer(\n            load.document, id(load.text), self._normalized.pop(load.document_id, None)\n        )\n        self._buffers[load.document_id] = buffer\n        load.document.contentsChange.connect(buffer.invalidate_text)\n        self._requested.discard(load.document_id)\n        self._load = None\n        self.ready.emit(load.document_id)\n        self._schedule_retention()\n        self._start_next_load()\n\n    def mark_synchronized(\n        self,\n        document_id: str,\n        source_key: int,\n        *,\n        dirty: bool,\n    ) -> None:\n        buffer = self._buffers.get(document_id)\n        if buffer is None:\n            return\n        if dirty:\n            self._dirty_ids.add(document_id)\n        else:\n            self._dirty_ids.discard(document_id)\n        changed = buffer.source_key != int(source_key)\n        buffer.source_key = int(source_key)\n        buffer.document.setModified(False)\n        if changed:\n            self._schedule_retention()\n\n    def close(self) -> None:\n        if self._closed:\n            return\n        self._closed = True\n        self._close_retention()\n        self._cancel_load()\n        for buffer in self._buffers.values():\n            buffer.document.contentsChange.disconnect(buffer.invalidate_text)\n            buffer.document.deleteLater()\n        self._buffers.clear()\n        self._sources.clear()\n        self._document_template = None\n        self._normalized.clear()\n        self._requested.clear()\n        self._empty_document.deleteLater()\n\n\n__all__ = ["MAX_SOURCE_BUFFERS", "SourceBufferStore"]\n'
1 failed, 18 passed, 47 subtests passed in 2.09s
