Merge remote-tracking branch 'origin/growth/ci-release-hardening' into growth/ci-release-hardening

docs/guides/llm-integrations.md

@@ -28,6 +28,7 @@ The `semantica.llms` module provides a unified interface for connecting to Large
 ## When To Use / When Not To Use
 
 **Use LLM integrations for:**
+
 - Text generation, summarization, and question-answering tasks
 - Complex reasoning that requires natural language understanding
 - Structured data extraction from unstructured text
@@ -35,6 +36,7 @@ The `semantica.llms` module provides a unified interface for connecting to Large
 - Tasks where context, ambiguity, or domain knowledge matter
 
 **Deterministic tools may be better for:**
+
 - Pattern matching that regular expressions can handle
 - Simple rule-based classification with clear criteria
 - Mathematical calculations or statistical analysis
@@ -42,6 +44,7 @@ The `semantica.llms` module provides a unified interface for connecting to Large
 - Data transformations with known logic
 
 **A full LLM may be unnecessary for:**
+
 - Simple keyword search or exact string matching
 - Deterministic workflows with predefined decision trees
 - High-frequency, low-latency operations where inference overhead matters
@@ -143,6 +146,58 @@ risk_data = oai.generate_structured(
 
 The default model `gpt-3.5-turbo` is fine for classification and light extraction. Switch to `gpt-4o` for complex multi-step regulatory reasoning or document understanding.
 
+## Anthropic — Complex Reasoning and Structured Extraction
+
+**Anthropic** provides the Claude model family, built with an emphasis on careful, instruction-following behavior and strong performance on multi-step reasoning, long-document analysis, and code-related tasks. Claude models tend to be more cautious about ambiguous instructions than other providers. That matters when the cost of a confidently wrong answer is high.
+
+The `Anthropic` provider wraps the Claude API. Reach for it when the task involves reasoning through several dependent steps (not just single-turn extraction), when you're processing long source documents that need to stay in context, or when you need schema-validated structured output rather than best-effort JSON.
+
+Install with `pip install "semantica[llm-anthropic]"` (or just `pip install anthropic`) before using this provider.
+
+```python
+from semantica.llms import Anthropic
+
+claude = Anthropic(model="claude-sonnet-4-6", api_key="YOUR_ANTHROPIC_KEY")
+# api_key falls back to the ANTHROPIC_API_KEY environment variable
+
+# is_available() only confirms a client was constructed from some key.
+# It does not validate the key or check network reachability - an
+# invalid or expired key still passes this check and fails at generate().
+if not claude.is_available():
+    raise RuntimeError("Anthropic provider not configured - set ANTHROPIC_API_KEY")
+
+# Plain generation - multi-step reasoning over a contract clause
+verdict = claude.generate(
+    "A vendor contract has a 30-day termination-for-convenience clause "
+    "but a 90-day data-return obligation that survives termination. "
+    "If the customer terminates on day 1, when must vendor-held data "
+    "be returned? Answer with the date basis only.",
+    temperature=0.1,
+)
+print(verdict)
+# "Day 120 from termination notice. The 90-day return period runs from
+#  the termination date (day 30), not from the notice date."
+
+# Structured, schema-validated output
+from pydantic import BaseModel
+
+class ContractRisk(BaseModel):
+    clause: str
+    risk_level: str
+    days_to_deadline: int
+
+risk = claude.generate_typed(
+    "Extract the termination clause risk from: vendor contract, "
+    "30-day termination for convenience, 90-day post-termination "
+    "data return obligation.",
+    schema=ContractRisk,
+)
+print(risk.risk_level, risk.days_to_deadline)
+# "medium" 90
+```
+
+Model selection follows the same tier structure as the other providers: a Haiku model for high-volume classification where cost matters more than depth, a Sonnet model as the default for most extraction and reasoning tasks, an Opus model when a task genuinely needs the deepest reasoning available and latency/cost are secondary. Check Anthropic's docs for the current model identifiers, since they're versioned and change over time.
+
 ## LiteLLM — One Interface, 100+ Providers
 
 **LiteLLM** is a universal adapter that provides a single interface to over 100 different LLM providers, including Anthropic Claude, Azure OpenAI, AWS Bedrock, Google Vertex AI, and local Ollama instances. It acts as a translation layer, converting your unified API calls into provider-specific requests, enabling easy switching between providers without code changes.

explorer/src/workspaces/GraphWorkspace/useLoadGraph.ts

@@ -318,6 +318,20 @@ interface EdgeListResponse {
 
 const PAGE_LIMIT = 1000;
 
+/** Surface the server's `detail` message (e.g. auth/setup guidance) on non-OK responses. */
+async function fetchErrorDetail(response: Response): Promise<string> {
+  try {
+    const body: unknown = await response.json();
+    const detail = (body as { detail?: unknown } | null)?.detail;
+    if (typeof detail === "string" && detail.trim()) {
+      return ` — ${detail.trim()}`;
+    }
+  } catch {
+    // Non-JSON or unreadable body: fall back to the status-only message.
+  }
+  return "";
+}
+
 async function fetchAllNodes(
   signal: AbortSignal,
   onProgress?: (progress: GraphLoadProgress) => void,
@@ -335,7 +349,7 @@ async function fetchAllNodes(
 
     const response = await fetch(url.toString(), { signal });
     if (!response.ok) {
-      throw new Error(`Fetch failed: ${response.status}`);
+      throw new Error(`Fetch failed: ${response.status}${await fetchErrorDetail(response)}`);
     }
 
     const data: NodeListResponse = await response.json();
@@ -390,7 +404,7 @@ async function fetchAllEdges(
 
     const response = await fetch(url.toString(), { signal });
     if (!response.ok) {
-      throw new Error(`Fetch failed: ${response.status}`);
+      throw new Error(`Fetch failed: ${response.status}${await fetchErrorDetail(response)}`);
     }
 
     const data: EdgeListResponse = await response.json();

semantica/llms/__init__.py

@@ -10,9 +10,10 @@ Supported Providers:
     - OpenAI: OpenAI API (GPT-3.5, GPT-4, etc.)
     - HuggingFaceLLM: HuggingFace Transformers for local LLM inference
     - LiteLLM: Unified interface to 100+ LLM providers (OpenAI, Anthropic, Groq, Azure, Bedrock, Vertex AI, etc.)
+    - Anthropic: Anthropic Claude API (Claude sonnet, Opus, Haiku, etc.)
 
 Example Usage:
-    >>> from semantica.llms import Groq, OpenAI, HuggingFaceLLM, LiteLLM
+    >>> from semantica.llms import Groq, OpenAI, HuggingFaceLLM, LiteLLM, Anthropic
     >>> 
     >>> # Groq provider
     >>> groq = Groq(model="llama-3.1-8b-instant", api_key="your-key")
@@ -32,6 +33,10 @@ Example Usage:
     >>> # Or use other providers via LiteLLM
     >>> llm = LiteLLM(model="anthropic/claude-sonnet-4-20250514")
     >>> response = llm.generate("Hello, world!")
+    >>>
+    >>> # Anthropic provider
+    >>> claude = Anthropic(model="claude-sonnet-4-6", api_key="the-key")
+    >>> response = claude.generate("Hello, world!")
 
 Author: Semantica Contributors
 License: MIT
@@ -41,6 +46,7 @@ from .groq import Groq
 from .openai import OpenAI
 from .huggingface import HuggingFaceLLM
 from .litellm import LiteLLM
+from .anthropic import Anthropic
 
-__all__ = ["Groq", "OpenAI", "HuggingFaceLLM", "LiteLLM"]
+__all__ = ["Groq", "OpenAI", "HuggingFaceLLM", "LiteLLM", "Anthropic"]
 

semantica/llms/anthropic.py

@@ -0,0 +1,111 @@
+"""
+Anthropic LLM Provider
+
+Wrapper for Anthropic Claude API provider with clean interface
+"""
+
+from typing import Any, Dict, List, Optional, Union
+
+from ..semantic_extract.providers import AnthropicProvider
+from ..utils.exceptions import ProcessingError
+from ..utils.logging import get_logger
+
+logger = get_logger("llms.anthropic")
+
+
+class Anthropic:
+    """
+    Anthropic Claude LLM provider wrapper.
+
+    Provides clean interface to Anthropic's Claude API.
+
+    Example:
+        >>> from semantica.llms import Anthropic
+        >>> claude = Anthropic(model="claude-sonnet-4-6", api_key="the-key")
+        >>> response = claude.generate("What is API key?")
+    """
+
+    def __init__(
+        self,
+        model: str = "claude-sonnet-4-6",
+        api_key: Optional[str] = None,
+        **kwargs
+    ):
+        """
+        Initialize Anthropic provider.
+
+        Args:
+            model: Model name (default: claude-sonnet-4-6)
+            api_key: Anthropic API key (default: from ANTHROPIC_API_KEY env var)
+            **kwargs: Additional provider options
+        """
+        self.provider = AnthropicProvider(api_key=api_key, model=model, **kwargs)
+        self.model = model
+        self.api_key = api_key
+
+    def is_available(self) -> bool:
+        """Check if Anthropic provider is available."""
+        return self.provider.is_available()
+
+    def generate(self, prompt: str, **kwargs) -> str:
+        """
+        Generate text from prompt.
+
+        Args:
+            prompt: Input prompt text
+            **kwargs: Generation options (temperature, max_tokens, etc.)
+
+        Returns:
+            Generated text response
+
+        Raises:
+            ProcessingError: If provider is not available or generation fails
+        """
+        if not self.is_available():
+            raise ProcessingError(
+                "Anthropic provider not available. Set ANTHROPIC_API_KEY or pass api_key."
+            )
+        return self.provider.generate(prompt, **kwargs)
+
+    def generate_structured(self, prompt: str, **kwargs) -> Union[Dict[str, Any], List[Any]]:
+        """
+        Generates structured JSON output.
+
+        Args:
+            prompt: Input prompt text
+            **kwargs: Generation options
+
+        Returns:
+            Parsed JSON response. A dict for a top-level JSON object, or a
+            list if the model returns a top-level JSON array.
+
+        Raises:
+            ProcessingError: If provider is not available or generation fails
+        """
+        if not self.is_available():
+            raise ProcessingError(
+                "Anthropic provider not available. Set ANTHROPIC_API_KEY or pass api_key."
+            )
+        return self.provider.generate_structured(prompt, **kwargs)
+
+    def generate_typed(self, prompt: str, schema: Any, max_retries: int = 3, **kwargs) -> Any:
+        """
+        Generate output validated against a Pydantic schema.
+
+        Args:
+            prompt: Input prompt text
+            schema: Pydantic model class to validate the output against
+            max_retries: Number of retries if validation fails (default: 3)
+            **kwargs: Generation options
+
+        Returns:
+            An instance of `schema`, populated from the model's response
+
+        Raises:
+            ProcessingError: If provider is not available or generation fails
+        """
+        if not self.is_available():
+            raise ProcessingError(
+                "Anthropic provider not available. Set ANTHROPIC_API_KEY or pass api_key."
+            )
+        return self.provider.generate_typed(prompt, schema, max_retries=max_retries, **kwargs)

semantica/utils/progress_tracker.py

@@ -43,7 +43,7 @@ from contextlib import contextmanager
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
-from typing import Any, Callable, Dict, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, List, Optional, TextIO, Tuple, Union
 
 from .logging import get_logger
 
@@ -144,14 +144,37 @@ class ProgressDisplay(ABC):
 class ConsoleProgressDisplay(ProgressDisplay):
     """Console progress display with real-time updates."""
 
-    def __init__(self, use_emoji: bool = True, update_interval: float = 0.1):
+    def __init__(
+        self,
+        use_emoji: bool = True,
+        update_interval: float = 0.1,
+        stream: Optional[TextIO] = None,
+    ):
+        """Initialize the console display.
+
+        Args:
+            use_emoji: Whether to decorate output with emoji.
+            update_interval: Minimum seconds between redraws.
+            stream: Where progress is written. Defaults to ``sys.stderr``.
+
+                Progress is diagnostic output, so stderr is the correct stream
+                for it, and stdout must stay clean for programs that carry a
+                machine-readable protocol on it — the stdio MCP servers put
+                newline-delimited JSON-RPC there, and a progress bar on stdout
+                corrupts that stream.
+
+                Left as ``None``, the stream is resolved on each write rather
+                than captured here, so a later rebinding of ``sys.stderr``
+                (pytest capture, for instance) is honoured.
+        """
         self.use_emoji = use_emoji
-        
-        # Check if stdout supports emojis (especially on Windows)
+        self._stream = stream
+
+        # Check if the target stream supports emojis (especially on Windows)
         if self.use_emoji:
             try:
-                # Try encoding a test emoji with stdout's encoding
-                encoding = getattr(sys.stdout, "encoding", None)
+                # Try encoding a test emoji with the stream's encoding
+                encoding = getattr(self.stream, "encoding", None)
                 if encoding:
                     "🧠".encode(encoding)
             except (UnicodeEncodeError, LookupError, AttributeError):
@@ -162,6 +185,11 @@ class ConsoleProgressDisplay(ProgressDisplay):
         self.current_lines: Dict[str, str] = {}
         self.lock = threading.Lock()
 
+    @property
+    def stream(self) -> TextIO:
+        """The stream progress is written to; ``sys.stderr`` unless overridden."""
+        return self._stream if self._stream is not None else sys.stderr
+
     def _should_update(self) -> bool:
         """Check if enough time has passed for update."""
         now = time.time()
@@ -259,15 +287,16 @@ class ConsoleProgressDisplay(ProgressDisplay):
         return f"{base_msg}: {message}"
 
     def _safe_write(self, text: str) -> None:
-        """Safely write text to stdout handling encoding errors."""
+        """Safely write text to the progress stream handling encoding errors."""
+        stream = self.stream
         try:
-            sys.stdout.write(text)
+            stream.write(text)
         except UnicodeEncodeError:
             # Fallback: encode with replacement and write decoded
             # Use ascii as safe fallback if encoding is unknown or caused error
-            encoding = getattr(sys.stdout, "encoding", None) or "ascii"
+            encoding = getattr(stream, "encoding", None) or "ascii"
             safe_text = text.encode(encoding, errors="replace").decode(encoding)
-            sys.stdout.write(safe_text)
+            stream.write(safe_text)
 
     def update(self, item: ProgressItem) -> None:
         """Update console progress display."""
@@ -302,11 +331,11 @@ class ConsoleProgressDisplay(ProgressDisplay):
                     self._display_item_line(pipeline_item)
                     self._safe_write("\n")
                 
-                sys.stdout.flush()
+                self.stream.flush()
             else:
                 # Original single-item display
                 self._display_item_line(item)
-                sys.stdout.flush()
+                self.stream.flush()
     
     def _display_item_line(self, item: ProgressItem) -> None:
         """Display a single progress item line."""
@@ -468,13 +497,13 @@ class ConsoleProgressDisplay(ProgressDisplay):
                     f"Completed: {completed} | Failed: {failed} | Total Time: {total_time:.2f}s\n"
                 )
             self._safe_write("=" * 80 + "\n")
-            sys.stdout.flush()
+            self.stream.flush()
 
     def clear(self) -> None:
         """Clear console display."""
         with self.lock:
             self._safe_write("\r" + " " * 100 + "\r")
-            sys.stdout.flush()
+            self.stream.flush()
             self.current_lines.clear()
 
 

tests/deduplication/test_deduplication.py

@@ -291,8 +291,8 @@ class TestDeduplication(unittest.TestCase):
 class TestProgressTrackerEncoding(unittest.TestCase):
     """Regression tests for issue #531 — Unicode crash on cp1252 Windows consoles."""
 
-    def _make_cp1252_stdout(self):
-        """Return a stdout-like object that raises UnicodeEncodeError for non-cp1252 chars."""
+    def _make_cp1252_stream(self):
+        """Return a stream that raises UnicodeEncodeError for non-cp1252 chars."""
         class CP1252Writer:
             encoding = "cp1252"
             def write(self, text):
@@ -304,39 +304,39 @@ class TestProgressTrackerEncoding(unittest.TestCase):
     def test_safe_write_does_not_crash_on_cp1252(self):
         """_safe_write must not raise UnicodeEncodeError on a cp1252 console."""
         display = ConsoleProgressDisplay()
-        orig = sys.stdout
-        sys.stdout = self._make_cp1252_stdout()
+        orig = sys.stderr
+        sys.stderr = self._make_cp1252_stream()
         try:
             display._safe_write("🧠 Semantica - 📊 Current Progress\n")
         except UnicodeEncodeError:
-            self.fail("_safe_write raised UnicodeEncodeError on cp1252 stdout")
+            self.fail("_safe_write raised UnicodeEncodeError on a cp1252 stream")
         finally:
-            sys.stdout = orig
+            sys.stderr = orig
 
     def test_update_pipeline_header_does_not_crash_on_cp1252(self):
         """update() pipeline header write must not crash on a cp1252 console (issue #531)."""
         from semantica.utils.progress_tracker import ProgressItem
         display = ConsoleProgressDisplay()
         display.use_emoji = True  # force emoji path to exercise the fixed branch
-        orig = sys.stdout
-        sys.stdout = self._make_cp1252_stdout()
+        orig = sys.stderr
+        sys.stderr = self._make_cp1252_stream()
         try:
             display._safe_write("🧠 Semantica - 📊 Current Progress\n")
             display._safe_write("=" * 150 + "\n")
         except UnicodeEncodeError:
             self.fail("Pipeline header write raised UnicodeEncodeError on cp1252 stdout")
         finally:
-            sys.stdout = orig
+            sys.stderr = orig
 
     def test_emoji_detection_disables_on_cp1252(self):
-        """ConsoleProgressDisplay should auto-disable emoji when stdout is cp1252."""
-        orig = sys.stdout
-        sys.stdout = self._make_cp1252_stdout()
+        """ConsoleProgressDisplay should auto-disable emoji when the progress stream is cp1252."""
+        orig = sys.stderr
+        sys.stderr = self._make_cp1252_stream()
         try:
             display = ConsoleProgressDisplay()
-            self.assertFalse(display.use_emoji, "use_emoji should be False on cp1252 stdout")
+            self.assertFalse(display.use_emoji, "use_emoji should be False on a cp1252 progress stream")
         finally:
-            sys.stdout = orig
+            sys.stderr = orig
 
 
 class TestResultLimiting(unittest.TestCase):

tests/integrations/crewai/__init__.py

@@ -0,0 +1 @@
+# tests/integrations/crewai package

tests/integrations/langchain/__init__.py

@@ -0,0 +1 @@
+# tests/integrations/langchain package

tests/test_llm_anthropic.py

@@ -0,0 +1,97 @@
+"""Tests for the Anthropic LLM provider wrapper (semantica.llms.Anthropic)."""
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from semantica.llms import Anthropic
+from semantica.utils.exceptions import ProcessingError
+
+
+def test_construction_stores_model_and_api_key():
+    """Anthropic(...) should not crash and should remember what it was given."""
+    claude = Anthropic(model="claude-sonnet-4-6", api_key="fake-key")
+    assert claude.model == "claude-sonnet-4-6"
+    assert claude.api_key == "fake-key"
+
+
+def test_is_available_false_with_no_key(monkeypatch):
+    """Without a real key, is_available() must be a real False, not truthy junk.
+
+    api_key=None alone isn't enough to prove this: AnthropicProvider falls
+    back to the ANTHROPIC_API_KEY environment variable, so this test has to
+    clear it too or it would pass/fail depending on whoever's machine or CI
+    runner happens to run it.
+    """
+    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+    claude = Anthropic(api_key=None)
+    assert claude.is_available() is False
+
+
+def test_generate_raises_clear_error_when_unavailable(monkeypatch):
+    """generate() must fail loudly.
+
+    Clears ANTHROPIC_API_KEY for the same reason as test_is_available_false_with_no_key:
+    otherwise this test flakes depending on whether the runner's environment has a key set.
+    """
+    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+    claude = Anthropic(api_key=None)
+    with pytest.raises(ProcessingError, match="Anthropic provider not available"):
+        claude.generate("hello")
+
+
+def test_generate_forwards_to_the_real_provider_when_available():
+    """When available, generate() must actually call through to the real provider."""
+    claude = Anthropic(api_key="fake-key")
+
+    claude.provider = MagicMock()
+    claude.provider.is_available.return_value = True
+    claude.provider.generate.return_value = "a fake response"
+
+    result = claude.generate("hello", temperature=0.5)
+
+    assert result == "a fake response"
+    claude.provider.generate.assert_called_once_with("hello", temperature=0.5)
+
+
+def test_generate_structured_forwards_to_the_real_provider():
+    claude = Anthropic(api_key="fake-key")
+    claude.provider = MagicMock()
+    claude.provider.is_available.return_value = True
+    claude.provider.generate_structured.return_value = {"key": "value"}
+
+    result = claude.generate_structured("hello")
+
+    assert result == {"key": "value"}
+    claude.provider.generate_structured.assert_called_once_with("hello")
+
+
+def test_generate_typed_forwards_schema_and_max_retries():
+    claude = Anthropic(api_key="fake-key")
+    claude.provider = MagicMock()
+    claude.provider.is_available.return_value = True
+    fake_schema = object()
+    claude.provider.generate_typed.return_value = "typed result"
+
+    result = claude.generate_typed("hello", fake_schema, max_retries=5)
+
+    assert result == "typed result"
+    claude.provider.generate_typed.assert_called_once_with(
+        "hello", fake_schema, max_retries=5
+    )
+
+
+def test_generate_structured_raises_clear_error_when_unavailable(monkeypatch):
+    """generate_structured() must fail loudly, same as generate()."""
+    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+    claude = Anthropic(api_key=None)
+    with pytest.raises(ProcessingError, match="Anthropic provider not available"):
+        claude.generate_structured("hello")
+
+
+def test_generate_typed_raises_clear_error_when_unavailable(monkeypatch):
+    """generate_typed() must fail loudly, same as generate() and generate_structured()."""
+    monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False)
+    claude = Anthropic(api_key=None)
+    with pytest.raises(ProcessingError, match="Anthropic provider not available"):
+        claude.generate_typed("hello", object())

tests/utils/test_progress_stream.py

@@ -0,0 +1,184 @@
+"""Console progress must be written to stderr, never to stdout.
+
+Progress is diagnostic output. Writing it to stdout corrupts any program that
+carries a machine-readable protocol there — the stdio MCP servers put
+newline-delimited JSON-RPC on stdout, and a progress bar interleaved with a
+response body makes that response unparseable (#1134).
+
+Both servers currently defend against this by setting
+SEMANTICA_DISABLE_PROGRESS, and a console display is only attached when the
+console is interactive. Those are containments, not the fix: they give up
+progress output entirely, and every future entry point has to remember them.
+Writing to the correct stream in the first place is what these tests pin.
+"""
+
+import importlib
+import io
+import sys
+
+import pytest
+
+
+@pytest.fixture
+def progress_module():
+    """Import the real progress_tracker, bypassing a mocked sys.modules entry.
+
+    tests/test_extractors_dispatch.py assigns a MagicMock over
+    'semantica.utils.progress_tracker' at import time and never restores it, so
+    a plain module-level import here returns mocks when that file has already
+    run. Dropping the cached entry re-imports the real module.
+
+    Both bindings are restored afterwards: importing a submodule also rebinds it
+    as an attribute of its parent package, so restoring only the sys.modules
+    entry would leave `semantica.utils.progress_tracker` and
+    `sys.modules["semantica.utils.progress_tracker"]` pointing at different
+    objects for every test that follows.
+    """
+    name = "semantica.utils.progress_tracker"
+    attr = name.rsplit(".", 1)[1]
+    parent = importlib.import_module("semantica.utils")
+
+    missing = object()
+    saved_entry = sys.modules.get(name, missing)
+    saved_attr = getattr(parent, attr, missing)
+
+    sys.modules.pop(name, None)
+    try:
+        module = importlib.import_module(name)
+        assert hasattr(module, "__file__"), "expected the real module, got a stand-in"
+        yield module
+    finally:
+        if saved_entry is missing:
+            sys.modules.pop(name, None)
+        else:
+            sys.modules[name] = saved_entry
+
+        if saved_attr is missing:
+            if hasattr(parent, attr):
+                delattr(parent, attr)
+        else:
+            setattr(parent, attr, saved_attr)
+
+
+@pytest.fixture
+def display_cls(progress_module):
+    return progress_module.ConsoleProgressDisplay
+
+
+@pytest.fixture
+def make_item(progress_module):
+    def _make(**overrides):
+        defaults = dict(
+            module="kg",
+            submodule="Reasoner",
+            message="Inferring facts",
+            status="running",
+            total_items=10,
+            processed_items=3,
+        )
+        defaults.update(overrides)
+        return progress_module.ProgressItem(**defaults)
+
+    return _make
+
+
+class TestProgressStreamDefaults:
+    def test_defaults_to_stderr(self, display_cls):
+        assert display_cls().stream is sys.stderr
+
+    def test_default_is_not_stdout(self, display_cls):
+        """The whole point: stdout stays clean for protocol traffic."""
+        assert display_cls().stream is not sys.stdout
+
+    def test_stream_follows_rebinding(self, display_cls, monkeypatch):
+        """Resolved per write, so pytest capture and later rebinds are honoured."""
+        display = display_cls()
+        replacement = io.StringIO()
+        monkeypatch.setattr(sys, "stderr", replacement)
+        assert display.stream is replacement
+
+    def test_explicit_stream_overrides_the_default(self, display_cls):
+        buffer = io.StringIO()
+        assert display_cls(stream=buffer).stream is buffer
+
+
+class TestProgressWritesGoToTheStream:
+    def test_update_writes_to_the_configured_stream(self, display_cls, make_item):
+        buffer = io.StringIO()
+        display = display_cls(stream=buffer, use_emoji=False, update_interval=0.0)
+
+        display.update(make_item())
+
+        assert buffer.getvalue(), "progress should have been rendered"
+
+    def test_update_writes_nothing_to_stdout(self, display_cls, make_item, monkeypatch):
+        """Regression guard for #1134: stdout must stay untouched."""
+        fake_stdout = io.StringIO()
+        monkeypatch.setattr(sys, "stdout", fake_stdout)
+        buffer = io.StringIO()
+
+        display = display_cls(stream=buffer, use_emoji=False, update_interval=0.0)
+        display.update(make_item())
+        display.clear()
+
+        assert fake_stdout.getvalue() == "", (
+            f"console progress leaked to stdout: {fake_stdout.getvalue()!r}"
+        )
+
+    def test_default_display_writes_nothing_to_stdout(
+        self, display_cls, make_item, monkeypatch
+    ):
+        """Same guard without an explicit stream, i.e. the real default path."""
+        fake_stdout = io.StringIO()
+        fake_stderr = io.StringIO()
+        monkeypatch.setattr(sys, "stdout", fake_stdout)
+        monkeypatch.setattr(sys, "stderr", fake_stderr)
+
+        display = display_cls(use_emoji=False, update_interval=0.0)
+        display.update(make_item())
+
+        assert fake_stdout.getvalue() == ""
+        assert fake_stderr.getvalue(), "progress should have gone to stderr"
+
+    def test_clear_flushes_the_stream_not_stdout(self, display_cls, monkeypatch):
+        flushed = []
+
+        class RecordingStream(io.StringIO):
+            def flush(self):
+                flushed.append("stream")
+
+        class ExplodingStdout(io.StringIO):
+            def flush(self):  # pragma: no cover - fails the test if reached
+                raise AssertionError("progress must not flush stdout")
+
+        monkeypatch.setattr(sys, "stdout", ExplodingStdout())
+        display = display_cls(
+            stream=RecordingStream(), use_emoji=False, update_interval=0.0
+        )
+
+        display.clear()
+
+        assert flushed == ["stream"]
+
+
+class TestEncodingFallback:
+    def test_falls_back_when_the_stream_cannot_encode(self, display_cls):
+        """A cp1252-style console must not raise on emoji; it degrades instead."""
+
+        class AsciiOnly(io.StringIO):
+            encoding = "ascii"
+
+            def write(self, text):
+                text.encode("ascii")  # raises UnicodeEncodeError on emoji
+                return super().write(text)
+
+        buffer = AsciiOnly()
+        display = display_cls(stream=buffer, use_emoji=True, update_interval=0.0)
+
+        display._safe_write("progress \U0001f504 bar\n")
+
+        assert "progress" in buffer.getvalue()
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])