feat: stream provider responses

src/javis/core/chat_service.py

@@ -2,10 +2,15 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator
 from dataclasses import dataclass
 
 from javis.memory.sqlite_store import ChatSession, SQLiteSessionStore
-from javis.providers.base import ChatMessage, LocalModelProvider
+from javis.providers.base import (
+    ChatMessage,
+    InvalidProviderResponseError,
+    LocalModelProvider,
+)
 
 
 class SessionProviderMismatchError(RuntimeError):
@@ -59,5 +64,43 @@ class ChatService:
         self.store.append_exchange(session_id, normalized, response)
         return response
 
+    def stream_send(self, session_id: str, text: str) -> Iterator[str]:
+        normalized = text.strip()
+        if not normalized:
+            raise ValueError("Eine leere Nachricht wird nicht gesendet.")
+        loaded = self.load_session(session_id)
+        messages = [*loaded.messages, ChatMessage("user", normalized)]
+        stream_method = getattr(self.provider, "stream_chat", None)
+        stream = (
+            stream_method(messages)
+            if callable(stream_method)
+            else iter((self.provider.chat(messages),))
+        )
+        chunks: list[str] = []
+        completed = False
+        try:
+            for chunk in stream:
+                if not isinstance(chunk, str):
+                    raise InvalidProviderResponseError(
+                        "Der Provider lieferte einen ungültigen Streaming-Abschnitt."
+                    )
+                if not chunk:
+                    continue
+                chunks.append(chunk)
+                yield chunk
+            completed = True
+        finally:
+            if not completed:
+                close_stream = getattr(stream, "close", None)
+                if callable(close_stream):
+                    close_stream()
+
+        response = "".join(chunks).strip()
+        if not response:
+            raise InvalidProviderResponseError(
+                "Der Provider lieferte keine verwendbare Streaming-Antwort."
+            )
+        self.store.append_exchange(session_id, normalized, response)
+
     def clear_session(self, session_id: str) -> None:
         self.store.clear_messages(session_id)

src/javis/core/provider_router.py

@@ -2,17 +2,19 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from dataclasses import dataclass
 
 from javis.memory.usage_store import ProviderEvent, SQLiteUsageStore, UsageStoreError
 from javis.providers.base import (
     ChatMessage,
     InvalidApiKeyError,
+    InvalidProviderResponseError,
     LocalModelProvider,
     MissingApiKeyError,
     ProviderError,
     ProviderUsage,
+    ResponseAbortedError,
 )
 from javis.security.privacy import CloudPolicy, PrivacyDecision, PrivacyRouter
 
@@ -165,6 +167,245 @@ class HybridProvider:
         )
         return response
 
+    def stream_chat(self, messages: list[ChatMessage]) -> Iterator[str]:
+        current_text = messages[-1].content if messages else ""
+        decision = self._privacy_router.classify(
+            current_text,
+            cloud_requested=self.mode == "gemini",
+        )
+
+        if self.mode == "local":
+            yield from self._stream_local(
+                messages,
+                decision,
+                "lokaler Modus",
+                fallback=False,
+                cloud_suppressed_by_local_mode=True,
+            )
+            return
+        if decision.policy is CloudPolicy.NEVER:
+            yield from self._stream_local(
+                messages,
+                decision,
+                decision.reason,
+                fallback=True,
+            )
+            return
+
+        approved = decision.policy is CloudPolicy.ALLOWED
+        if decision.policy is CloudPolicy.ASK:
+            approved = self._approval_callback(decision)
+            if not approved:
+                yield from self._stream_local(
+                    messages,
+                    decision,
+                    "Cloudfreigabe abgelehnt",
+                    fallback=True,
+                )
+                return
+
+        unavailable_reason = self._cloud_unavailable_reason()
+        if unavailable_reason:
+            yield from self._stream_local(
+                messages,
+                decision,
+                unavailable_reason,
+                fallback=True,
+            )
+            return
+        if len(current_text) > self._max_cloud_input_chars:
+            yield from self._stream_local(
+                messages,
+                decision,
+                "Nachricht überschreitet das lokale Cloud-Größenlimit",
+                fallback=True,
+            )
+            return
+
+        cloud_messages = self._privacy_router.minimal_context(
+            messages,
+            current_policy=decision.policy,
+            approved=approved,
+            max_chars=self._max_cloud_input_chars,
+            max_messages=self._max_cloud_context_messages,
+        )
+        if not cloud_messages:
+            yield from self._stream_local(
+                messages,
+                decision,
+                "kein freigegebener Cloudkontext",
+                fallback=True,
+            )
+            return
+
+        emitted = False
+        try:
+            cloud_provider = self._cloud_provider_factory()
+            for chunk in self._provider_chunks(cloud_provider, cloud_messages):
+                emitted = True
+                yield chunk
+        except InvalidApiKeyError as exc:
+            self._cloud_session_disabled = True
+            if emitted:
+                self._raise_interrupted_cloud(decision, exc)
+            yield from self._stream_cloud_failure(
+                messages,
+                decision,
+                exc,
+                disable_notice=True,
+            )
+            return
+        except (MissingApiKeyError, ProviderError) as exc:
+            if emitted:
+                self._raise_interrupted_cloud(decision, exc)
+            yield from self._stream_cloud_failure(messages, decision, exc)
+            return
+
+        if not emitted:
+            error = InvalidProviderResponseError(
+                "Gemini hat keine verwendbare Streaming-Antwort geliefert."
+            )
+            yield from self._stream_cloud_failure(messages, decision, error)
+            return
+
+        usage = getattr(cloud_provider, "last_usage", ProviderUsage())
+        self._record(
+            ProviderEvent(
+                provider=cloud_provider.name,
+                model=cloud_provider.model,
+                success=True,
+                error_category=None,
+                fallback=False,
+                privacy_policy=decision.policy.value,
+                input_tokens=usage.input_tokens,
+                output_tokens=usage.output_tokens,
+            )
+        )
+        self.last_route = RouteStatus(
+            provider=cloud_provider.name,
+            privacy_policy=decision.policy,
+            fallback=False,
+            cloud_suppressed_by_local_mode=False,
+            technical_fallback=False,
+            reason="Cloudstream erfolgreich",
+        )
+
+    @staticmethod
+    def _provider_chunks(
+        provider: LocalModelProvider,
+        messages: list[ChatMessage],
+    ) -> Iterator[str]:
+        stream_method = getattr(provider, "stream_chat", None)
+        chunks = (
+            stream_method(messages) if callable(stream_method) else iter((provider.chat(messages),))
+        )
+        completed = False
+        try:
+            for chunk in chunks:
+                if not isinstance(chunk, str):
+                    raise InvalidProviderResponseError(
+                        "Der Provider lieferte einen ungültigen Streaming-Abschnitt."
+                    )
+                if chunk:
+                    yield chunk
+            completed = True
+        finally:
+            if not completed:
+                close = getattr(chunks, "close", None)
+                if callable(close):
+                    close()
+
+    def _stream_cloud_failure(
+        self,
+        messages: list[ChatMessage],
+        decision: PrivacyDecision,
+        error: ProviderError,
+        *,
+        disable_notice: bool = False,
+    ) -> Iterator[str]:
+        self._record_cloud_error(decision, error)
+        reason = "Gemini nicht verfügbar"
+        if disable_notice:
+            reason = "Gemini-Schlüssel abgelehnt; Cloud für diese Sitzung deaktiviert"
+        yield from self._stream_local(
+            messages,
+            decision,
+            reason,
+            fallback=True,
+            technical_fallback=True,
+        )
+
+    def _raise_interrupted_cloud(
+        self,
+        decision: PrivacyDecision,
+        error: ProviderError,
+    ) -> None:
+        self._record_cloud_error(decision, error)
+        self.last_route = RouteStatus(
+            provider="gemini",
+            privacy_policy=decision.policy,
+            fallback=False,
+            cloud_suppressed_by_local_mode=False,
+            technical_fallback=True,
+            reason="Cloudstream nach Teilausgabe abgebrochen",
+        )
+        raise ResponseAbortedError(
+            "Der Gemini-Stream wurde nach einer Teilausgabe abgebrochen; "
+            "es wurde kein lokaler Ersatz angehängt."
+        ) from error
+
+    def _stream_local(
+        self,
+        messages: list[ChatMessage],
+        decision: PrivacyDecision,
+        reason: str,
+        *,
+        fallback: bool,
+        cloud_suppressed_by_local_mode: bool = False,
+        technical_fallback: bool = False,
+    ) -> Iterator[str]:
+        if fallback:
+            self._notice_callback(f"{reason} – lokale Antwort mit Ollama.")
+        emitted = False
+        try:
+            for chunk in self._provider_chunks(self.local_provider, messages):
+                emitted = True
+                yield chunk
+        except ProviderError as exc:
+            self._record(
+                ProviderEvent(
+                    provider=self.local_provider.name,
+                    model=self.local_provider.model,
+                    success=False,
+                    error_category=type(exc).__name__,
+                    fallback=fallback,
+                    privacy_policy=decision.policy.value,
+                )
+            )
+            raise
+        if not emitted:
+            raise InvalidProviderResponseError(
+                "Ollama hat keine verwendbare Streaming-Antwort geliefert."
+            )
+        self._record(
+            ProviderEvent(
+                provider=self.local_provider.name,
+                model=self.local_provider.model,
+                success=True,
+                error_category=None,
+                fallback=fallback,
+                privacy_policy=decision.policy.value,
+            )
+        )
+        self.last_route = RouteStatus(
+            provider=self.local_provider.name,
+            privacy_policy=decision.policy,
+            fallback=fallback,
+            cloud_suppressed_by_local_mode=cloud_suppressed_by_local_mode,
+            technical_fallback=technical_fallback,
+            reason=reason,
+        )
+
     def _cloud_unavailable_reason(self) -> str | None:
         if not self._cloud_enabled:
             return "Gemini ist lokal nicht aktiviert"
@@ -188,17 +429,7 @@ class HybridProvider:
         *,
         disable_notice: bool = False,
     ) -> str:
-        error_category = type(error).__name__
-        self._record(
-            ProviderEvent(
-                provider="gemini",
-                model=self._cloud_model,
-                success=False,
-                error_category=error_category,
-                fallback=False,
-                privacy_policy=decision.policy.value,
-            )
-        )
+        self._record_cloud_error(decision, error)
         reason = "Gemini nicht verfügbar"
         if disable_notice:
             reason = "Gemini-Schlüssel abgelehnt; Cloud für diese Sitzung deaktiviert"
@@ -210,6 +441,22 @@ class HybridProvider:
             technical_fallback=True,
         )
 
+    def _record_cloud_error(
+        self,
+        decision: PrivacyDecision,
+        error: ProviderError,
+    ) -> None:
+        self._record(
+            ProviderEvent(
+                provider="gemini",
+                model=self._cloud_model,
+                success=False,
+                error_category=type(error).__name__,
+                fallback=False,
+                privacy_policy=decision.policy.value,
+            )
+        )
+
     def _local(
         self,
         messages: list[ChatMessage],

src/javis/interface/cli.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import argparse
 import getpass
+import re
 import sys
 from collections.abc import Callable, Sequence
 from pathlib import Path
@@ -27,6 +28,7 @@ from javis.security.secrets import SecretProvider, SecretStoreError
 
 InputFunction = Callable[[str], str]
 OutputFunction = Callable[[str], None]
+StreamOutputFunction = Callable[[str], None]
 StatusFunction = Callable[[], list[str]]
 
 HELP_TEXT = """Befehle:
@@ -144,12 +146,48 @@ def _show_sessions(sessions: list[ChatSession], output: OutputFunction) -> None:
         )
 
 
+class TerminalMarkdownRenderer:
+    """Turn common Markdown markers into readable incremental terminal text."""
+
+    _tail_size = 3
+
+    def __init__(self) -> None:
+        self._pending = ""
+
+    @staticmethod
+    def _clean(text: str) -> str:
+        text = re.sub(r"(?m)^[ \t]{0,3}#{1,6}[ \t]+", "", text)
+        for marker in ("```", "**", "__", "~~", "`"):
+            text = text.replace(marker, "")
+        return text
+
+    def feed(self, chunk: str) -> str:
+        combined = self._pending + chunk
+        if len(combined) <= self._tail_size:
+            self._pending = combined
+            return ""
+        visible = combined[: -self._tail_size]
+        self._pending = combined[-self._tail_size :]
+        return self._clean(visible)
+
+    def finish(self) -> str:
+        visible = self._clean(self._pending)
+        self._pending = ""
+        return visible
+
+
+def _terminal_write(text: str) -> None:
+    sys.stdout.write(text)
+    sys.stdout.flush()
+
+
 def run_chat(
     service: ChatService,
     *,
     session_id: str | None = None,
     input_fn: InputFunction = input,
     output: OutputFunction = print,
+    stream_output: StreamOutputFunction | None = None,
     status_fn: StatusFunction | None = None,
 ) -> int:
     try:
@@ -262,12 +300,41 @@ def run_chat(
             output("Unbekannter Befehl. /help zeigt die verfügbaren Befehle.")
             continue
 
+        stream = service.stream_send(active_id, entered)
+        writer = stream_output or (_terminal_write if output is print else output)
+        renderer = TerminalMarkdownRenderer()
+        started = False
         try:
-            response = service.send(active_id, entered)
-            output(f"Javis: {response}")
+            for chunk in stream:
+                visible = renderer.feed(chunk)
+                if not visible:
+                    continue
+                if not started:
+                    writer("Javis: ")
+                    started = True
+                writer(visible)
+            remaining = renderer.finish()
+            if remaining:
+                if not started:
+                    writer("Javis: ")
+                    started = True
+                writer(remaining)
+            if started:
+                writer("\n")
+        except KeyboardInterrupt:
+            close = getattr(stream, "close", None)
+            if callable(close):
+                close()
+            if started:
+                writer("\n")
+            output("Generierung abgebrochen. Die unvollständige Antwort wurde nicht gespeichert.")
         except (ProviderError, SessionStoreError, SessionProviderMismatchError) as exc:
+            if started:
+                writer("\n")
             output(f"Fehler: {exc}")
         except ValueError as exc:
+            if started:
+                writer("\n")
             output(f"Fehler: {exc}")
 
 

src/javis/providers/base.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Iterator
 from dataclasses import dataclass
 from typing import Protocol, runtime_checkable
 
@@ -70,3 +71,13 @@ class LocalModelProvider(Protocol):
     def chat(self, messages: list[ChatMessage]) -> str:
         """Return one complete assistant response."""
         ...
+
+
+@runtime_checkable
+class StreamingModelProvider(Protocol):
+    name: str
+    model: str
+
+    def stream_chat(self, messages: list[ChatMessage]) -> Iterator[str]:
+        """Yield only visible assistant text chunks."""
+        ...

src/javis/providers/gemini.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
 from typing import Any
 
 import httpx
@@ -49,6 +49,7 @@ class GeminiProvider:
         self._config = types.GenerateContentConfig(
             candidate_count=1,
             max_output_tokens=max_output_tokens,
+            thinking_config=types.ThinkingConfig(include_thoughts=False),
         )
         self._client = client_factory(
             api_key=api_key,
@@ -104,6 +105,73 @@ class GeminiProvider:
         )
         return text.strip()
 
+    def stream_chat(self, messages: list[ChatMessage]) -> Iterator[str]:
+        contents = [
+            types.Content(
+                role="model" if message.role == "assistant" else "user",
+                parts=[types.Part.from_text(text=message.content)],
+            )
+            for message in messages
+        ]
+        self.last_usage = ProviderUsage()
+        visible_text_received = False
+        try:
+            responses = self._client.models.generate_content_stream(
+                model=self.model,
+                contents=contents,
+                config=self._config,
+            )
+            for response in responses:
+                usage = getattr(response, "usage_metadata", None)
+                if usage is not None:
+                    self.last_usage = ProviderUsage(
+                        input_tokens=self._token_count(usage, "prompt_token_count"),
+                        output_tokens=self._token_count(
+                            usage,
+                            "candidates_token_count",
+                        ),
+                    )
+                for text in self._visible_text_parts(response):
+                    visible_text_received = True
+                    yield text
+        except errors.APIError as exc:
+            self._raise_api_error(exc)
+        except httpx.TimeoutException as exc:
+            raise ProviderTimeoutError("Gemini hat nicht rechtzeitig geantwortet.") from exc
+        except httpx.NetworkError as exc:
+            raise CloudNetworkError(
+                "Gemini ist wegen eines Netzwerkfehlers nicht erreichbar."
+            ) from exc
+        except (OSError, ConnectionError) as exc:
+            raise CloudNetworkError(
+                "Gemini ist wegen eines Netzwerkfehlers nicht erreichbar."
+            ) from exc
+
+        if not visible_text_received:
+            raise InvalidProviderResponseError(
+                "Gemini hat keine verwendbare Streaming-Antwort geliefert."
+            )
+
+    @staticmethod
+    def _visible_text_parts(response: object) -> Iterator[str]:
+        candidates = getattr(response, "candidates", None)
+        if isinstance(candidates, list) and candidates:
+            for candidate in candidates:
+                content = getattr(candidate, "content", None)
+                parts = getattr(content, "parts", None)
+                if not isinstance(parts, list):
+                    continue
+                for part in parts:
+                    if getattr(part, "thought", False):
+                        continue
+                    text = getattr(part, "text", None)
+                    if isinstance(text, str) and text:
+                        yield text
+            return
+        text = getattr(response, "text", None)
+        if isinstance(text, str) and text:
+            yield text
+
     @staticmethod
     def _token_count(usage: object, name: str) -> int | None:
         value = getattr(usage, name, None)

src/javis/providers/ollama.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import json
 import socket
+from collections.abc import Iterator
 from dataclasses import dataclass
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
@@ -107,3 +108,76 @@ class OllamaProvider:
                 "Ollama hat keine verwendbare Textantwort geliefert."
             )
         return content.strip()
+
+    def stream_chat(self, messages: list[ChatMessage]) -> Iterator[str]:
+        payload = {
+            "model": self.model,
+            "messages": [
+                {"role": message.role, "content": message.content} for message in messages
+            ],
+            "stream": True,
+            "think": False,
+        }
+        request = Request(
+            self._endpoint,
+            data=json.dumps(payload).encode("utf-8"),
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        completed = False
+        visible_text_received = False
+        try:
+            with urlopen(request, timeout=self._timeout_seconds) as response:
+                for raw_line in response:
+                    if not raw_line.strip():
+                        continue
+                    try:
+                        result = json.loads(raw_line)
+                    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
+                        raise InvalidProviderResponseError(
+                            "Ollama hat einen ungültigen Streaming-Abschnitt geliefert."
+                        ) from exc
+                    error = result.get("error")
+                    if isinstance(error, str) and error:
+                        if "not found" in error.lower():
+                            raise ModelNotInstalledError(
+                                f"Das lokale Modell '{self.model}' ist nicht installiert."
+                            )
+                        raise ProviderUnavailableError("Ollama hat den Stream abgelehnt.")
+                    content = result.get("message", {}).get("content")
+                    if content is not None and not isinstance(content, str):
+                        raise InvalidProviderResponseError(
+                            "Ollama hat ungültigen sichtbaren Text geliefert."
+                        )
+                    if content:
+                        visible_text_received = True
+                        yield content
+                    if result.get("done") is True:
+                        completed = True
+                        break
+        except HTTPError as exc:
+            details = exc.read().decode("utf-8", errors="replace")
+            if exc.code == 404 or "not found" in details.lower():
+                raise ModelNotInstalledError(
+                    f"Das lokale Modell '{self.model}' ist nicht installiert."
+                ) from exc
+            raise ProviderUnavailableError(f"Ollama meldet HTTP-Fehler {exc.code}.") from exc
+        except TimeoutError as exc:
+            raise ProviderTimeoutError(
+                "Die Modellantwort hat das Zeitlimit überschritten."
+            ) from exc
+        except URLError as exc:
+            if isinstance(exc.reason, (TimeoutError, socket.timeout)):
+                raise ProviderTimeoutError(
+                    "Die Modellantwort hat das Zeitlimit überschritten."
+                ) from exc
+            raise ProviderUnavailableError(
+                "Ollama ist unter der konfigurierten lokalen Adresse nicht erreichbar."
+            ) from exc
+
+        if not completed:
+            raise ResponseAbortedError("Der Ollama-Stream wurde vorzeitig beendet.")
+        if not visible_text_received:
+            raise InvalidProviderResponseError(
+                "Ollama hat keine verwendbare Streaming-Antwort geliefert."
+            )