docs: confirm chat comfort push

CHANGELOG.md

@@ -24,6 +24,10 @@ Alle wesentlichen Projektänderungen werden hier in verständlicher Form dokumen
 - Strikter Nullkostenmodus, lokale Schutzlimits und inhaltsfreies Provider-Audit.
 - Bereinigte CLI-Befehle `/provider`, `/privacy` und `/status`.
 - Sicheres PowerShell-Startskript ohne Systemänderungen.
+- Gemeinsames Streaming für Ollama und Gemini mit atomarer Speicherung,
+  Gedankenfilter, lesbarer Terminalausgabe und sicherem Strg+C-Abbruch.
+- Automatische lokale Sitzungstitel, `/rename`, nummerierte `/sessions`,
+  `/load <NR|ID>` und begrenzte `/search`-Suche.
 
 ### Changed
 
@@ -42,6 +46,8 @@ Alle wesentlichen Projektänderungen werden hier in verständlicher Form dokumen
 - Allgemeine unpersönliche Technikfragen werden präziser als `allowed`
   klassifiziert; persönliche, interne und geheime Inhalte behalten Vorrang.
 - Privacy-Ausgabe unterscheidet lokalen Modus von technischem Fallback.
+- SQLite-Sitzungen werden verlustfrei und idempotent um Titel und zuletzt
+  verwendeten Provider migriert.
 
 ### Security
 
@@ -51,5 +57,7 @@ Alle wesentlichen Projektänderungen werden hier in verständlicher Form dokumen
   Nicht-Billing deaktiviert; Javis aktiviert niemals Billing.
 - Cloudkontext wird lokal klassifiziert und minimiert; 429, Netzwerk-, Timeout-,
   Key- und Modellfehler fallen auf Ollama zurück.
+- Gesundheits-, Symptom-, Verletzungs-, Medikamenten-, Diagnose-, Behandlungs-
+  und medizinische Notfallfragen werden vor Cloudclient-Erstellung strikt lokal behandelt.
 - Gemini-/Datenschutzablauf wurde am 30.07.2026 zusätzlich von Pascal manuell
   erfolgreich abgenommen; kein Keywert erschien in der Ausgabe.

docs/CHATGPT_HANDOFF.md

@@ -58,8 +58,8 @@ Rootserver:
 - Modelle liegen außerhalb von Git unter `D:\Javis-Data\ollama-models`.
 - Der lokale Ollama-Dienst bindet nur an `127.0.0.1:11434`; PATH, Registry,
   Autostart und Windows-Dienste blieben unverändert.
-- CLI-Befehle: `/new`, `/sessions`, `/load <ID>`, `/clear`, `/provider`,
+- CLI-Befehle: `/new`, `/sessions`, `/load <NR|ID>`, `/search <Text>`,
-  `/privacy`, `/status`, `/help`, `/exit`.
+  `/rename <Titel>`, `/clear`, `/provider`, `/privacy`, `/status`, `/help`, `/exit`.
 - Sitzungsdaten liegen über `JAVIS_DATA_DIR` außerhalb von Git; der bestätigte
   Smoke-Test nutzte `D:\Javis-Data\runtime\core-chat-smoke`.
 - Kein Obsidian-, Laptop- oder Rootserverzugriff wurde implementiert.
@@ -102,8 +102,13 @@ Rootserver:
 - Allgemeine unpersönliche Technikfragen zu RAM/SSD, SQLite, HTTP, Git und
   ähnlichen Begriffen werden gezielter `allowed`; `never` und `ask` behalten
   Vorrang bei Secrets, interner Infrastruktur und persönlichem Kontext.
+- Gesundheits-, Symptom-, Blutungs-, Verletzungs-, Medikamenten-, Diagnose-,
+  Behandlungs- und medizinische Notfallfragen sind zwingend lokal `never`.
 - `/privacy` trennt Inhaltsklassifikation, tatsächlichen Provider, Unterdrückung
   durch Modus `local` und technischen Provider-Fallback.
+- Ollama, Gemini, Hybridrouter, Chatservice und CLI streamen sichtbare Antwortteile.
+  Verdeckte Gedanken werden nicht angefordert oder ausgegeben; Terminal-Markdown
+  wird lesbar bereinigt. Strg+C verwirft Teilantworten atomar und kehrt zum Chat zurück.
 
 ## Aktuelle Architektur
 
@@ -111,7 +116,10 @@ Rootserver:
 - plattformunabhängige Pfade und zentrale Konfiguration
 - Modell-Provider als kleine austauschbare Schnittstelle
 - Ollama-Provider akzeptiert nur lokale HTTP-Loopback-Adressen
-- SQLite-Sitzungsspeicher mit atomaren Benutzer-/Assistentenpaaren
+- SQLite-Sitzungsspeicher mit atomaren Benutzer-/Assistentenpaaren, neutralem
+  Starttitel, lokaler sicherer Titelerzeugung und zuletzt genutztem Provider
+- Streaming wird erst nach vollständigem Abschluss atomar gespeichert; bei Abbruch
+  bleibt weder die Benutzerfrage noch eine unvollständige Antwort im Verlauf.
 - Laufzeitdaten außerhalb von Git über `JAVIS_DATA_DIR` oder sicheren Plattformstandard
 - Hybridprovider bevorzugt Gemini für cloudgeeignete Inhalte und verwendet
   Ollama bei Datenschutz, Ablehnung, Offline-, Quota- und Providerfehlern
@@ -162,6 +170,11 @@ Rootserver:
 - Lesender Ollama-Status prüft nur Loopback-Erreichbarkeit und Modellliste.
 - Sicheres PowerShell-Startskript und kompakte Startdokumentation ergänzt; keine
   PATH-, Registry-, Autostart-, Dienst- oder Richtlinienänderung.
+- Medizinische Inhalte vor Cloudclient-Erstellung strikt auf `never` gesetzt;
+  bestehende SQLite-Daten transaktional und wiederholbar um Titel/Provider migriert.
+- Echtes Ollama-/Gemini-Streaming, providerübergreifendes Fallback vor der ersten
+  Ausgabe und sicheren Streaming-Abbruch nach Teilausgabe implementiert.
+- CLI-Streaming mit lesbarer Markdown-Bereinigung und sicherem Strg+C ergänzt.
 
 ## Aktuelle Tests
 
@@ -176,11 +189,15 @@ Letzter bestätigter Projektstand:
 - uv-Lock und `uv sync --dev`: bestanden
 - Python in `.venv`: 3.12.13
 - Ruff in `.venv`: 0.16.0
-- Unit-Tests: 66 bestanden; neue Technikklassifikation und Privacy-Ausgabe abgedeckt
+- Unit-Tests: 91 bestanden; Navigation, Streaming, Migration, Datenschutz,
+  ID-Präfixe, begrenzte Suche und SQL-Sonderzeichen abgedeckt
 - PowerShell-Syntax des Startskripts: erfolgreich geparst
 - Ruff Lint: bestanden
 - Ruff Formatprüfung: bestanden
 - lokaler CLI-Smoke ohne Secretzugriff: bestanden
+- echter Ollama-CLI-Streaming-Smoke mit Neustart/Laden: bestanden
+- echter Gemini-Streaming-Smoke mit öffentlichem Inhalt: bestanden; ein Abschnitt,
+  20 sichtbare Zeichen, 14 Eingabe- und 4 Ausgabetoken; kein Keywert ausgegeben
 - Obsidian-Vault unverändert: 1 Markdown-, 11 Gesamtdateien, letzter
   Schreibzeitpunkt weiterhin 30.07.2026 15:42:31
 - Rootserver und Ubuntu-Laptop wurden nicht kontaktiert
@@ -218,22 +235,19 @@ Abnahmestatus:
 
 ## Git-Stand
 
-- aktueller und stabiler Branch: `main`
+- aktueller und geprüfter Branch: `main`
-- Feature-Branch ist als `7a14fe8` zu
+- stabiler Ausgangsstand: `main` bei `77b510b`
-  `origin/feat/gemini-privacy-router` gepusht
+- medizinischer Datenschutz-Fix: `9148193`
+- Provider- und CLI-Streaming: `00a5a6c`
+- SQLite-Titel und Migration: `112ae6d`
+- Sitzungsnavigation: `43c1f76`
+- Dokumentation: `328dcd9`
+- Chatkomfort-Merge nach `main`: `cbba102`
+- 91 Unit-Tests, Ruff und Strukturprüfung auf dem gemergten `main` bestanden
+- Feature-Branch erfolgreich zu `origin/feat/chat-comfort` gepusht
 - konfliktfreier Merge nach `main`: `564a3fd`
 - 66 Unit-Tests, Ruff und Strukturprüfung auf dem gemergten `main` bestanden
-- `main` erfolgreich zu `origin/main` gepusht; Remote-Tracking ist synchron
+- `main` erfolgreich bis `3361eae` zu `origin/main` gepusht und synchron verifiziert
-- Grundgerüstcommit: `6d04171`
-- letzter Commit vor der Handoff-Verdichtung: `0e82748`
-- Handoff-Verdichtung auf `main`: `e6eb995`
-- Python-Entwicklungsumgebung auf dem Feature-Branch: `fa90a5c`
-- Laufzeit-/Modelldokumentation: `10d7cd7`
-- Chatkern und CLI: `ec9b2f9`
-- Unit-Tests: `b153daa`
-- Feature-Dokumentation: `5d567e5`
-- lokaler Merge nach `main`: `f3d6dfb`
-- geprüfter und gepushter Abschlussstand vor diesem Handoff-Update: `0a550e6`
 - Secret-Provider: `570e7fa`
 - Datenschutz/Nullkosten: `286d96d`
 - isolierter Gemini-Provider: `f3e1183`
@@ -247,23 +261,11 @@ Abnahmestatus:
 - `origin` verwendet HTTPS
 - kein Force-Push und keine umgeschriebene Historie
 
-## Kompakter Changelog
-
-- `49a1dd3`: erster Gitea-Verbindungstest
-- `6d04171`: sicheres Javis-Grundgerüst
-- `0e82748`: Werkzeug-, Token- und Handoff-Regeln
-- `e6eb995`: einmalige Archivierung und Verdichtung des Handoffs
-- aktuell: isolierte Python-Toolchain und reproduzierbare Entwicklungsumgebung
-- aktuell: portable Ollama-Laufzeit und `qwen3:8b` außerhalb von Git verifiziert
-- aktuell: lokaler CLI-Chat und persistente SQLite-Sitzungen vollständig abgenommen
-- aktuell: Gemini-Free-Hybridrouting, Datenschutz und Nullkostenmodus
-  automatisiert und durch Pascal manuell vollständig abgenommen
-
 ## Offene Entscheidungen und Fehler
 
 - endgültiger Produkt-/Repositoryname bleibt offen
 - Startskript startet Ollama bei Bedarf pro Prozess; kein Autostart oder Dienst
-- Antworten werden noch nicht gestreamt; Sitzungen besitzen noch keine Titel oder Suche
+- Antworten werden gestreamt; Titel, Navigation und Suche sind implementiert
 - genauer späterer Obsidian-Schreibbereich ist nicht freigegeben
 - normaler Secret-Provider ist festgelegt: Betriebssystem-Keyring; eine
   Klartext-XML wird nicht für API-Schlüssel verwendet
@@ -287,9 +289,10 @@ Abnahmestatus:
 
 ## Nächster sinnvoller Auftrag
 
-Ein separates, kleines Paket für Chatkomfort und Robustheit planen,
+Chatkomfort ist implementiert, gemergt, auf `main` erneut geprüft und gepusht.
-beispielsweise Sitzungstitel/Suche oder Streaming. Noch keine Obsidian-
+Jetzt stoppen und mit Pascal erst später ein einzelnes nächstes Paket festlegen.
-Integration, Tools, Sprache oder Serversteuerung beginnen.
+Medizinische Antwortqualität bleibt ein späteres Sicherheits-/Systemprompt-Thema.
+Noch keine Obsidian-Integration oder Tools beginnen.
 
 Keinen API-Key in Chat, Git, Obsidian, Logs oder Dokumentation übernehmen.
 

docs/NEXT_SESSION.md

@@ -2,22 +2,22 @@
 
 ## Ausgangsstand
 
-Der Core-Chat und der Gemini-/Datenschutz-Meilenstein sind automatisiert durch
+Der Core-Chat, Gemini-/Datenschutz-Meilenstein und das Chatkomfort-Paket sind
-Codex und am 30.07.2026 manuell durch Pascal abgenommen. Der Featurestand ist zum
+automatisiert geprüft. Pascal hat Core-Chat sowie Gemini/Datenschutz am
-Merge freigegeben.
+30.07.2026 manuell abgenommen.
 
-Bestätigt sind echter Gemini-Erfolg, verdeckte Key-Speicherung, Nullkostenmodus,
+Chatkomfort umfasst sicheres Ollama-/Gemini-Streaming, Strg+C ohne Teilpersistenz,
-bereinigter Status, Cloudzähler, `ask`- und `never`-Routing, erzwungener lokaler
+medizinisches `never`, automatische und manuelle Sitzungstitel, nummerierte
-Modus sowie das Startskript.
+Listen, Laden per Nummer/ID-Präfix und begrenzte Suche in Titeln/Benutzertexten.
 
-## Nächstes sinnvolles Paket
+91 Unit-Tests, Ruff, Struktur-, Secret-, Gitignore- und Auditprüfungen sowie
+echte Ollama- und Gemini-Streaming-Smokes sind erfolgreich.
 
-Nur ein kleines Chatkomfort-/Robustheitsthema auswählen, zum Beispiel:
+## Nächster sinnvoller Schritt
 
-- automatische, lokal erzeugte Sitzungstitel und Sitzungssuche oder
+Der Chatkomfort-Branch ist nach `main` gemergt und dort erneut geprüft. Den
-- Streaming lokaler und Cloudantworten mit sauberem Abbruchverhalten
+finalen Stand aus `docs/CHATGPT_HANDOFF.md` übernehmen und mit Pascal ein
-
+einzelnes nächstes Paket festlegen; noch keine Folgephase automatisch beginnen.
-Vorher wie immer Branch, Arbeitsbaum, aktuellen Handoff und Tests prüfen.
 
 ## Weiterhin nicht beginnen
 

docs/PROJECT_STATUS.md

@@ -1,6 +1,6 @@
 # Projektstatus
 
-Stand: Gemini-Free-Feature automatisiert geprüft, 30.07.2026.
+Stand: Chatkomfort-Feature vollständig automatisiert geprüft, 30.07.2026.
 
 ## Erreicht
 
@@ -20,10 +20,15 @@ Stand: Gemini-Free-Feature automatisiert geprüft, 30.07.2026.
 - zwei echte deutsche Modellantworten erfolgreich vollständig auf der RTX 3060 erzeugt
 - kalter Test 36,19 s, warmer Kurztest 0,56 s; geladene GPU-Belegung 6.842 MiB
 - lokaler Ollama-Provider, zentrale Konfiguration und SQLite-Sitzungsspeicher implementiert
-- CLI mit `/new`, `/sessions`, `/load`, `/clear`, `/help` und `/exit` implementiert
+- CLI streamt Antworten und unterstützt `/new`, nummerierte `/sessions`,
+  `/load <NR|ID>`, `/search`, `/rename`, `/clear`, `/help` und `/exit`
+- bestehende SQLite-Datenbanken werden automatisch, transaktional und
+  wiederholbar um Sitzungstitel und zuletzt verwendeten Provider migriert
+- lokale Titelbildung ohne Modellaufruf; offensichtliche Secrets und medizinische
+  Inhalte werden nicht wörtlich in automatische Titel übernommen
 - echte CLI-Modellantwort sowie Laden und Fortsetzen derselben Sitzung in einem
   neuen Prozess bestätigt
-- 13 Unit-Tests, Ruff und Strukturprüfung erfolgreich
+- 91 Unit-Tests, Ruff, Struktur-, Gitignore-, Secret- und Auditprüfung erfolgreich
 - lokaler Core-Chat am 30.07.2026 zusätzlich von Pascal manuell erfolgreich
   geprüft: deutsche Antworten, Sitzungskontext, `/sessions`, `/exit`, Neustart,
   `/load <UUID>`, vollständiger Verlauf und Fortsetzung
@@ -38,6 +43,22 @@ Stand: Gemini-Free-Feature automatisiert geprüft, 30.07.2026.
 - für produktive Nutzung freigegeben: nicht separat erklärt; keine spätere Tool-,
   Obsidian- oder Autonomiefunktion
 
+## Chatkomfort-Meilenstein
+
+- Branch `feat/chat-comfort`
+- Ollama und Gemini verwenden denselben optionalen Streamingvertrag; Provider
+  ohne Streaming bleiben kompatibel
+- keine versteckten Gedanken; Terminal-Markdown wird lesbar bereinigt
+- vollständige Antwort wird genau einmal gespeichert; Strg+C und Streamfehler
+  hinterlassen keine unvollständige Nachricht
+- medizinische Fragen werden vor jeder Cloudclient-Erstellung zwingend `never`
+- Sitzungsliste zeigt Nummer, kurze ID, Titel, Datum, Nachrichtenanzahl und Provider
+- Laden per letzter Listennummer oder eindeutigem ID-Präfix; Mehrdeutigkeit wird erklärt
+- Suche ist auf 20 Ergebnisse begrenzt, parametrisiert und durchsucht nur Titel
+  sowie Benutzernachrichten
+- echter Ollama-CLI-Streaming-/Neustart-Smoke und echter kleiner
+  Gemini-Streaming-Smoke mit öffentlichem Inhalt erfolgreich
+
 ## Gemini-/Datenschutz-Meilenstein
 
 - Branch `feat/gemini-privacy-router`
@@ -67,7 +88,7 @@ Stand: Gemini-Free-Feature automatisiert geprüft, 30.07.2026.
 - `/privacy` trennt lokalen Modus von einem technischen Provider-Fallback
 - sicheres `scripts/start-javis.ps1` mit relativer Repositoryprüfung,
   Loopback-Ollama und überschreibbaren externen Pfaden
-- 66 Unit-Tests aktuell erfolgreich; PowerShell-Syntax, Lockfile, Sync,
+- 91 Unit-Tests aktuell erfolgreich; PowerShell-Syntax, Lockfile, Sync,
   Strukturprüfung, Secret-Scan, Gitignore, Ruff-Format und Lint sauber
 
 ## Nicht implementiert
@@ -78,23 +99,21 @@ Living-Mind-Frontend.
 ## Git
 
 - aktueller Branch: `main`
-- Feature-Commit: `7a14fe8`
+- Basis: `main` bei `77b510b`
-- Merge-Commit: `564a3fd`
+- medizinisches Routing: `9148193`
-- Tests auf dem gemergten `main`: 66 bestanden
+- Streaming: `00a5a6c`
-- `main` erfolgreich zu `origin/main` gepusht
+- Titelmigration: `112ae6d`
-- Grundgerüstcommit: `6d04171`
+- Sitzungsnavigation: `43c1f76`
-- Basis des Feature-Branches: `e6eb995`
+- Dokumentation: `328dcd9`
-- Toolchain: `fa90a5c`
+- Feature-Branch zu `origin/feat/chat-comfort` gepusht
-- Laufzeit/Modell: `10d7cd7`
+- Merge nach `main`: `cbba102`; 91 Tests und Ruff auf `main` erneut bestanden
-- Chatkern: `ec9b2f9`
+- `main` bis `3361eae` zu `origin/main` gepusht und synchron verifiziert
-- Tests: `b153daa`
+- vorheriger Gemini-/Datenschutz-Merge auf `main`: `564a3fd`
-- Dokumentation: `5d567e5`
-- Merge nach `main`: `f3d6dfb`
 
 ## Bekannte Einschränkung
 
-Das Startskript startet Ollama bei Bedarf lokal. Antworten werden noch nicht
+Das Startskript startet Ollama bei Bedarf lokal. Eine Löschfunktion ist nicht
-gestreamt; Sitzungen haben noch keine Titel, Suche oder Löschfunktion.
+Bestandteil dieses Pakets.
 `qwen3:8b` ist nicht automatisch für die GTX 1050 des späteren Zielhosts geeignet.
 Der Gemini-Key liegt ausschließlich im Betriebssystem-Keyring; sein Wert wurde
 von Codex weder gelesen noch ausgegeben. Pascal hat den echten Cloud- und

src/javis/core/chat_service.py

@@ -2,16 +2,55 @@
 
 from __future__ import annotations
 
+import re
+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):
     """The active provider cannot safely continue the stored session."""
 
 
+_SENSITIVE_TITLE_PATTERN = re.compile(
+    r"(?ix)"
+    r"\b(?:api[-_ ]?key|passwort|password|token|secret|iban|kreditkarte|"
+    r"private[-_ ]?key|ssh[-_ ]?key)\b|"
+    r"-----BEGIN [A-Z ]+PRIVATE KEY-----|"
+    r"\b[A-Z]{2}\d{2}(?:[ ]?[A-Z0-9]){11,30}\b|"
+    r"\b[\w.+-]+@[\w.-]+\.[A-Z]{2,}\b"
+)
+_MEDICAL_TITLE_PATTERN = re.compile(
+    r"(?ix)\b(?:arzt|ärzt|blut(?:e|en|ung)?|diagnos|gesundheit|krank|"
+    r"medikament|notfall|schmerz|symptom|therap|verletz|wunde)\w*\b"
+)
+
+
+def derive_session_title(text: str, *, limit: int = 60) -> str:
+    """Create a short local title without retaining obvious sensitive values."""
+
+    normalized = " ".join(text.split())
+    if _SENSITIVE_TITLE_PATTERN.search(normalized):
+        return "Sensible Anfrage"
+    if _MEDICAL_TITLE_PATTERN.search(normalized):
+        return "Gesundheitsfrage"
+    cleaned = re.sub(r"[`*_#~]+", "", normalized).strip(" -:;,.!?")
+    if not cleaned:
+        return "Neue Unterhaltung"
+    if len(cleaned) <= limit:
+        return cleaned
+    shortened = cleaned[: limit - 1].rsplit(" ", 1)[0].rstrip(" -:;,.!?")
+    if not shortened:
+        shortened = cleaned[: limit - 1].rstrip()
+    return f"{shortened}…"
+
+
 @dataclass(frozen=True, slots=True)
 class LoadedSession:
     session: ChatSession
@@ -33,6 +72,24 @@ class ChatService:
     def list_sessions(self) -> list[ChatSession]:
         return self.store.list_sessions()
 
+    def search_sessions(self, text: str) -> list[ChatSession]:
+        normalized = text.strip()
+        if not normalized:
+            raise ValueError("Für die Suche fehlt ein Text.")
+        return self.store.search_sessions(normalized)
+
+    def resolve_session(self, reference: str) -> LoadedSession:
+        session = self.store.resolve_unique_prefix(reference)
+        return self.load_session(session.id)
+
+    def rename_session(self, session_id: str, title: str) -> ChatSession:
+        normalized = " ".join(title.split()).strip()
+        if not normalized:
+            raise ValueError("Der Sitzungstitel darf nicht leer sein.")
+        if len(normalized) > 60:
+            normalized = f"{normalized[:59].rstrip()}…"
+        return self.store.rename_session(session_id, normalized)
+
     def load_session(self, session_id: str) -> LoadedSession:
         session = self.store.get_session(session_id)
         supports_session = getattr(self.provider, "supports_session", None)
@@ -56,8 +113,63 @@ class ChatService:
         loaded = self.load_session(session_id)
         messages = [*loaded.messages, ChatMessage("user", normalized)]
         response = self.provider.chat(messages)
-        self.store.append_exchange(session_id, normalized, response)
+        self.store.append_exchange(
+            session_id,
+            normalized,
+            response,
+            title_if_first=derive_session_title(normalized),
+            last_provider=self._last_provider_name(),
+        )
         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,
+            title_if_first=derive_session_title(normalized),
+            last_provider=self._last_provider_name(),
+        )
+
     def clear_session(self, session_id: str) -> None:
         self.store.clear_messages(session_id)
+
+    def _last_provider_name(self) -> str:
+        route = getattr(self.provider, "last_route", None)
+        provider_name = getattr(route, "provider", None)
+        return provider_name if isinstance(provider_name, str) else self.provider.name

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_cloud_error(decision, error)
-        self._record(
-            ProviderEvent(
-                provider="gemini",
-                model=self._cloud_model,
-                success=False,
-                error_category=error_category,
-                fallback=False,
-                privacy_policy=decision.policy.value,
-            )
-        )
         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
@@ -13,6 +14,7 @@ from javis.config.settings import ConfigurationError, Settings
 from javis.core.chat_service import ChatService, SessionProviderMismatchError
 from javis.core.provider_router import HybridProvider
 from javis.memory.sqlite_store import (
+    AmbiguousSessionReferenceError,
     ChatSession,
     SessionNotFoundError,
     SessionStoreError,
@@ -27,12 +29,15 @@ 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:
   /new              neue Sitzung beginnen
   /sessions         gespeicherte Sitzungen auflisten
-  /load <ID>        frühere Sitzung laden und fortsetzen
+  /load <NR|ID>     Sitzung per Listennummer oder eindeutigem ID-Anfang laden
+  /search <TEXT>    Titel und eigene Nachrichten durchsuchen
+  /rename <TITEL>   aktive Sitzung umbenennen
   /clear            Nachrichten der aktiven Sitzung leeren
   /provider         aktiven Providermodus anzeigen
   /provider <MODUS> Modus auto, local oder gemini setzen
@@ -137,19 +142,56 @@ def _show_sessions(sessions: list[ChatSession], output: OutputFunction) -> None:
     if not sessions:
         output("Keine gespeicherten Sitzungen vorhanden.")
         return
-    for session in sessions:
+    for number, session in enumerate(sessions, start=1):
+        date = session.updated_at[:10]
         output(
-            f"{session.id} | {session.updated_at} | "
+            f"{number}. {session.id[:8]} | {session.title} | {date} | "
-            f"{session.provider}/{session.model} | {session.message_count} Nachrichten"
+            f"{session.message_count} Nachrichten | {session.last_provider}"
         )
 
 
+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:
@@ -169,6 +211,7 @@ def run_chat(
         return 2
 
     output("Lokaler Javis-Chat. /help zeigt die Befehle.")
+    shown_sessions: list[ChatSession] = []
 
     while True:
         try:
@@ -194,10 +237,33 @@ def run_chat(
             continue
         if entered == "/sessions":
             try:
-                _show_sessions(service.list_sessions(), output)
+                shown_sessions = service.list_sessions()
+                _show_sessions(shown_sessions, output)
             except SessionStoreError as exc:
                 output(f"Fehler: {exc}")
             continue
+        if entered.startswith("/search"):
+            parts = entered.split(maxsplit=1)
+            if len(parts) != 2 or not parts[1].strip():
+                output("Verwendung: /search <Text>")
+                continue
+            try:
+                shown_sessions = service.search_sessions(parts[1])
+                _show_sessions(shown_sessions, output)
+            except (SessionStoreError, ValueError) as exc:
+                output(f"Fehler: {exc}")
+            continue
+        if entered.startswith("/rename"):
+            parts = entered.split(maxsplit=1)
+            if len(parts) != 2 or not parts[1].strip():
+                output("Verwendung: /rename <neuer Titel>")
+                continue
+            try:
+                renamed = service.rename_session(active_id, parts[1])
+                output(f"Sitzung umbenannt: {renamed.title}")
+            except (SessionNotFoundError, SessionStoreError, ValueError) as exc:
+                output(f"Fehler: {exc}")
+            continue
         if entered == "/clear":
             try:
                 service.clear_session(active_id)
@@ -247,12 +313,24 @@ def run_chat(
                 output("Verwendung: /load <Sitzungs-ID>")
                 continue
             try:
-                loaded = service.load_session(parts[1].strip())
+                reference = parts[1].strip()
+                if reference.isdecimal():
+                    number = int(reference)
+                    if not shown_sessions:
+                        output("Zuerst /sessions oder /search anzeigen.")
+                        continue
+                    if number < 1 or number > len(shown_sessions):
+                        output("Diese Listennummer ist nicht vorhanden.")
+                        continue
+                    loaded = service.load_session(shown_sessions[number - 1].id)
+                else:
+                    loaded = service.resolve_session(reference)
                 active_id = loaded.session.id
                 output(f"Sitzung geladen: {active_id}")
                 _show_history(service, active_id, output)
             except (
                 SessionNotFoundError,
+                AmbiguousSessionReferenceError,
                 SessionProviderMismatchError,
                 SessionStoreError,
             ) as exc:
@@ -262,12 +340,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)
+            for chunk in stream:
-            output(f"Javis: {response}")
+                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/memory/sqlite_store.py

@@ -21,6 +21,10 @@ class SessionNotFoundError(SessionStoreError):
     """A requested session ID does not exist."""
 
 
+class AmbiguousSessionReferenceError(SessionStoreError):
+    """A short session reference matches multiple sessions."""
+
+
 @dataclass(frozen=True, slots=True)
 class ChatSession:
     id: str
@@ -28,9 +32,14 @@ class ChatSession:
     updated_at: str
     provider: str
     model: str
+    title: str
+    last_provider: str
     message_count: int = 0
 
 
+DEFAULT_SESSION_TITLE = "Neue Sitzung"
+
+
 def _utc_now() -> str:
     return datetime.now(UTC).isoformat(timespec="seconds")
 
@@ -63,16 +72,21 @@ class SQLiteSessionStore:
 
     def _initialize(self) -> None:
         with self._connection() as connection:
-            connection.executescript(
+            connection.execute(
                 """
                 CREATE TABLE IF NOT EXISTS sessions (
                     id TEXT PRIMARY KEY,
                     created_at TEXT NOT NULL,
                     updated_at TEXT NOT NULL,
                     provider TEXT NOT NULL,
-                    model TEXT NOT NULL
+                    model TEXT NOT NULL,
-                );
+                    title TEXT NOT NULL DEFAULT 'Neue Sitzung',
-
+                    last_provider TEXT NOT NULL DEFAULT ''
+                )
+                """
+            )
+            connection.execute(
+                """
                 CREATE TABLE IF NOT EXISTS messages (
                     id INTEGER PRIMARY KEY AUTOINCREMENT,
                     session_id TEXT NOT NULL,
@@ -82,10 +96,31 @@ class SQLiteSessionStore:
                     created_at TEXT NOT NULL,
                     FOREIGN KEY (session_id) REFERENCES sessions(id) ON DELETE CASCADE,
                     UNIQUE (session_id, position)
-                );
+                )
-
+                """
+            )
+            columns = {
+                row["name"] for row in connection.execute("PRAGMA table_info(sessions)").fetchall()
+            }
+            if "title" not in columns:
+                connection.execute(
+                    "ALTER TABLE sessions ADD COLUMN title TEXT NOT NULL DEFAULT 'Neue Sitzung'"
+                )
+            if "last_provider" not in columns:
+                connection.execute(
+                    "ALTER TABLE sessions ADD COLUMN last_provider TEXT NOT NULL DEFAULT ''"
+                )
+            connection.execute(
+                """
+                UPDATE sessions
+                SET last_provider = provider
+                WHERE last_provider = ''
+                """
+            )
+            connection.execute(
+                """
                 CREATE INDEX IF NOT EXISTS idx_messages_session_position
-                    ON messages(session_id, position);
+                    ON messages(session_id, position)
                 """
             )
 
@@ -96,14 +131,31 @@ class SQLiteSessionStore:
             with self._connection() as connection:
                 connection.execute(
                     """
-                    INSERT INTO sessions (id, created_at, updated_at, provider, model)
+                    INSERT INTO sessions
-                    VALUES (?, ?, ?, ?, ?)
+                        (id, created_at, updated_at, provider, model, title, last_provider)
+                    VALUES (?, ?, ?, ?, ?, ?, ?)
                     """,
-                    (session_id, now, now, provider, model),
+                    (
+                        session_id,
+                        now,
+                        now,
+                        provider,
+                        model,
+                        DEFAULT_SESSION_TITLE,
+                        provider,
+                    ),
                 )
         except sqlite3.Error as exc:
             raise SessionStoreError("Die Sitzung konnte nicht gespeichert werden.") from exc
-        return ChatSession(session_id, now, now, provider, model)
+        return ChatSession(
+            session_id,
+            now,
+            now,
+            provider,
+            model,
+            DEFAULT_SESSION_TITLE,
+            provider,
+        )
 
     def get_session(self, session_id: str) -> ChatSession:
         try:
@@ -140,6 +192,84 @@ class SQLiteSessionStore:
             raise SessionStoreError("Die Sitzungen konnten nicht aufgelistet werden.") from exc
         return [self._session_from_row(row) for row in rows]
 
+    def search_sessions(self, text: str, *, limit: int = 20) -> list[ChatSession]:
+        normalized = text.strip()
+        if not normalized:
+            return []
+        escaped = normalized.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        pattern = f"%{escaped}%"
+        try:
+            with self._connection() as connection:
+                rows = connection.execute(
+                    """
+                    SELECT s.*, COUNT(m.id) AS message_count
+                    FROM sessions AS s
+                    LEFT JOIN messages AS m ON m.session_id = s.id
+                    WHERE s.title LIKE ? ESCAPE '\\' COLLATE NOCASE
+                       OR EXISTS (
+                            SELECT 1
+                            FROM messages AS searched
+                            WHERE searched.session_id = s.id
+                              AND searched.role = 'user'
+                              AND searched.content LIKE ? ESCAPE '\\' COLLATE NOCASE
+                       )
+                    GROUP BY s.id
+                    ORDER BY s.updated_at DESC, s.created_at DESC
+                    LIMIT ?
+                    """,
+                    (pattern, pattern, max(1, min(limit, 100))),
+                ).fetchall()
+        except sqlite3.Error as exc:
+            raise SessionStoreError("Die Sitzungssuche ist fehlgeschlagen.") from exc
+        return [self._session_from_row(row) for row in rows]
+
+    def resolve_unique_prefix(self, prefix: str) -> ChatSession:
+        normalized = prefix.strip()
+        if not normalized:
+            raise SessionNotFoundError("Leerer Sitzungsverweis.")
+        escaped = normalized.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+        try:
+            with self._connection() as connection:
+                rows = connection.execute(
+                    """
+                    SELECT id
+                    FROM sessions
+                    WHERE id LIKE ? ESCAPE '\\'
+                    ORDER BY id
+                    LIMIT 2
+                    """,
+                    (f"{escaped}%",),
+                ).fetchall()
+        except sqlite3.Error as exc:
+            raise SessionStoreError("Der Sitzungsverweis konnte nicht geprüft werden.") from exc
+        if not rows:
+            raise SessionNotFoundError(f"Keine Sitzung beginnt mit '{normalized}'.")
+        if len(rows) > 1:
+            raise AmbiguousSessionReferenceError(
+                f"Der Sitzungsanfang '{normalized}' ist nicht eindeutig."
+            )
+        return self.get_session(rows[0]["id"])
+
+    def rename_session(self, session_id: str, title: str) -> ChatSession:
+        now = _utc_now()
+        try:
+            with self._connection() as connection:
+                cursor = connection.execute(
+                    """
+                    UPDATE sessions
+                    SET title = ?, updated_at = ?
+                    WHERE id = ?
+                    """,
+                    (title, now, session_id),
+                )
+                if cursor.rowcount == 0:
+                    raise SessionNotFoundError(f"Sitzung '{session_id}' wurde nicht gefunden.")
+        except SessionNotFoundError:
+            raise
+        except sqlite3.Error as exc:
+            raise SessionStoreError("Die Sitzung konnte nicht umbenannt werden.") from exc
+        return self.get_session(session_id)
+
     def get_messages(self, session_id: str) -> list[ChatMessage]:
         self.get_session(session_id)
         try:
@@ -162,6 +292,9 @@ class SQLiteSessionStore:
         session_id: str,
         user_content: str,
         assistant_content: str,
+        *,
+        title_if_first: str | None = None,
+        last_provider: str | None = None,
     ) -> None:
         now = _utc_now()
         try:
@@ -194,8 +327,24 @@ class SQLiteSessionStore:
                     ),
                 )
                 connection.execute(
-                    "UPDATE sessions SET updated_at = ? WHERE id = ?",
+                    """
-                    (now, session_id),
+                    UPDATE sessions
+                    SET updated_at = ?,
+                        title = CASE
+                            WHEN ? = 0 AND title = ? THEN COALESCE(?, title)
+                            ELSE title
+                        END,
+                        last_provider = COALESCE(?, last_provider)
+                    WHERE id = ?
+                    """,
+                    (
+                        now,
+                        next_position,
+                        DEFAULT_SESSION_TITLE,
+                        title_if_first,
+                        last_provider,
+                        session_id,
+                    ),
                 )
         except SessionNotFoundError:
             raise
@@ -232,5 +381,7 @@ class SQLiteSessionStore:
             updated_at=row["updated_at"],
             provider=row["provider"],
             model=row["model"],
+            title=row["title"],
+            last_provider=row["last_provider"],
             message_count=row["message_count"],
         )

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."
+            )

src/javis/security/privacy.py

@@ -42,8 +42,13 @@ _NEVER_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
     ),
     (
         re.compile(
-            r"(?i)\b(diagnose|gesundheit|krankheit|medikament|patient|arztbericht|"
+            r"(?i)\b(symptom(?:e|en)?|blut(?:e|est|et|en|ung|ungen|wert|werte)?|"
-            r"psychotherapie|blutwert)\b"
+            r"verletz(?:ung|ungen|t)|wunde[n]?|medikament(?:e|en)?|diagnos(?:e|en|tiziert)|"
+            r"behandl(?:ung|ungen|e|en)|medizinisch(?:e[rsn]?|er)?\s+notfall|"
+            r"notaufnahme|rettungsdienst|gesundheit|krankheit(?:en)?|patient(?:en)?|"
+            r"arzt|ärzt(?:in|lich(?:e[rsn]?|er)?)|arztbericht|krankenhaus|"
+            r"psychotherapie|schmerz(?:en)?|fieber|atemnot|bewusstlos|ohnmacht|"
+            r"vergiftung|allergisch(?:e[rsn]?|er)?\s+reaktion)\b"
         ),
         "Gesundheitsinformation",
     ),

tests/unit/test_chat_service.py

@@ -2,9 +2,9 @@ import tempfile
 import unittest
 from pathlib import Path
 
-from javis.core.chat_service import ChatService
+from javis.core.chat_service import ChatService, derive_session_title
 from javis.memory.sqlite_store import SQLiteSessionStore
-from javis.providers.base import ChatMessage, LocalModelProvider
+from javis.providers.base import ChatMessage, LocalModelProvider, ProviderUnavailableError
 
 
 class RecordingProvider:
@@ -19,6 +19,19 @@ class RecordingProvider:
         return f"Antwort {len(self.calls)}"
 
 
+class StreamingProvider(RecordingProvider):
+    def __init__(self, *, fail_after_first: bool = False) -> None:
+        super().__init__()
+        self.fail_after_first = fail_after_first
+
+    def stream_chat(self, messages: list[ChatMessage]):
+        self.calls.append(messages)
+        yield "Teil "
+        if self.fail_after_first:
+            raise ProviderUnavailableError("Streamfehler")
+        yield "Antwort"
+
+
 class ChatServiceTests(unittest.TestCase):
     def setUp(self) -> None:
         self.temporary_directory = tempfile.TemporaryDirectory()
@@ -51,6 +64,67 @@ class ChatServiceTests(unittest.TestCase):
                 "Antwort 2",
             ],
         )
+        self.assertEqual(loaded.session.title, "Hallo")
+
+    def test_title_is_local_short_and_redacts_obvious_sensitive_content(self) -> None:
+        self.assertEqual(
+            derive_session_title("Erkläre SQLite-Transaktionen für Anfänger"),
+            "Erkläre SQLite-Transaktionen für Anfänger",
+        )
+        self.assertLessEqual(derive_session_title("Wort " * 30).__len__(), 60)
+        self.assertEqual(
+            derive_session_title("Mein API-Key ist ABC123 und funktioniert nicht"),
+            "Sensible Anfrage",
+        )
+        self.assertEqual(
+            derive_session_title("Wenn ich blute, sollte ich zum Arzt?"),
+            "Gesundheitsfrage",
+        )
+
+    def test_manual_title_is_normalized_and_limited(self) -> None:
+        session = self.service.new_session()
+
+        renamed = self.service.rename_session(session.id, "  Mein   Titel  ")
+        shortened = self.service.rename_session(session.id, "x" * 80)
+
+        self.assertEqual(renamed.title, "Mein Titel")
+        self.assertEqual(len(shortened.title), 60)
+        self.assertTrue(shortened.title.endswith("…"))
+
+    def test_streaming_response_is_persisted_exactly_once_after_completion(self) -> None:
+        provider = StreamingProvider()
+        service = ChatService(self.service.store, provider)
+        session = service.new_session()
+
+        chunks = list(service.stream_send(session.id, "Hallo"))
+
+        self.assertEqual(chunks, ["Teil ", "Antwort"])
+        self.assertEqual(
+            [message.content for message in service.load_session(session.id).messages],
+            ["Hallo", "Teil Antwort"],
+        )
+        self.assertEqual(service.load_session(session.id).session.message_count, 2)
+
+    def test_closed_stream_does_not_persist_partial_response(self) -> None:
+        provider = StreamingProvider()
+        service = ChatService(self.service.store, provider)
+        session = service.new_session()
+        stream = service.stream_send(session.id, "Hallo")
+
+        self.assertEqual(next(stream), "Teil ")
+        stream.close()
+
+        self.assertEqual(service.load_session(session.id).messages, [])
+
+    def test_provider_error_during_stream_does_not_persist_partial_response(self) -> None:
+        provider = StreamingProvider(fail_after_first=True)
+        service = ChatService(self.service.store, provider)
+        session = service.new_session()
+
+        with self.assertRaises(ProviderUnavailableError):
+            list(service.stream_send(session.id, "Hallo"))
+
+        self.assertEqual(service.load_session(session.id).messages, [])
 
 
 if __name__ == "__main__":

tests/unit/test_cli.py

@@ -1,3 +1,4 @@
+import sqlite3
 import tempfile
 import unittest
 from pathlib import Path
@@ -19,6 +20,22 @@ class _FakeProvider:
     def chat(self, messages: list[ChatMessage]) -> str:
         return f"Echo: {messages[-1].content}"
 
+    def stream_chat(self, messages: list[ChatMessage]):
+        yield "Echo: "
+        yield messages[-1].content
+
+
+class _MarkdownProvider(_FakeProvider):
+    def stream_chat(self, messages: list[ChatMessage]):
+        yield "## **Ant"
+        yield "wort**\n`SQLite`"
+
+
+class _InterruptingProvider(_FakeProvider):
+    def stream_chat(self, messages: list[ChatMessage]):
+        yield "angefangene Antwort"
+        raise KeyboardInterrupt
+
 
 class _FakeHybridProvider(_FakeProvider):
     name = "hybrid"
@@ -59,10 +76,53 @@ class CliTests(unittest.TestCase):
             )
 
         self.assertEqual(result, 0)
-        self.assertTrue(any(line == "Javis: Echo: Hallo" for line in output))
+        self.assertIn("Javis: Echo: Hallo\n", "".join(output))
         self.assertTrue(any("Nachrichten" in line for line in output))
         self.assertEqual(output[-1], "Chat beendet.")
 
+    def test_streaming_cleans_terminal_markdown_but_persists_raw_answer(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
+            service = ChatService(store, _MarkdownProvider())
+            inputs = iter(["Frage", "/exit"])
+            output: list[str] = []
+            streamed: list[str] = []
+
+            result = run_chat(
+                service,
+                input_fn=lambda _prompt: next(inputs),
+                output=output.append,
+                stream_output=streamed.append,
+            )
+            session = service.list_sessions()[0]
+            messages = service.load_session(session.id).messages
+
+        self.assertEqual(result, 0)
+        self.assertEqual("".join(streamed), "Javis: Antwort\nSQLite\n")
+        self.assertEqual(messages[-1].content, "## **Antwort**\n`SQLite`")
+
+    def test_keyboard_interrupt_drops_partial_answer_and_returns_to_prompt(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
+            service = ChatService(store, _InterruptingProvider())
+            inputs = iter(["Frage", "/exit"])
+            output: list[str] = []
+            streamed: list[str] = []
+
+            result = run_chat(
+                service,
+                input_fn=lambda _prompt: next(inputs),
+                output=output.append,
+                stream_output=streamed.append,
+            )
+            session = service.list_sessions()[0]
+            messages = service.load_session(session.id).messages
+
+        self.assertEqual(result, 0)
+        self.assertEqual(messages, [])
+        self.assertTrue(any("nicht gespeichert" in line for line in output))
+        self.assertEqual(output[-1], "Chat beendet.")
+
     def test_provider_privacy_and_status_commands_are_sanitized(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
@@ -108,6 +168,76 @@ class CliTests(unittest.TestCase):
         )
         self.assertIn("Gemini-Schlüssel vorhanden: ja", output)
 
+    def test_numbered_sessions_load_rename_and_search(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            store = SQLiteSessionStore(Path(directory) / "sessions.sqlite3")
+            service = ChatService(store, _FakeProvider())
+            first = service.new_session()
+            service.send(first.id, "Erste Frage")
+            second = service.new_session()
+            service.send(second.id, "Zweite Frage")
+            shown_before = service.list_sessions()
+            target = shown_before[1]
+            inputs = iter(
+                [
+                    "/sessions",
+                    "/load 2",
+                    "/rename Gefundene Sitzung",
+                    "/search Gefundene",
+                    "/exit",
+                ]
+            )
+            output: list[str] = []
+
+            result = run_chat(
+                service,
+                session_id=first.id,
+                input_fn=lambda _prompt: next(inputs),
+                output=output.append,
+            )
+            renamed_title = store.get_session(target.id).title
+
+        self.assertEqual(result, 0)
+        self.assertTrue(any(line.startswith("1. ") and "Frage" in line for line in output))
+        self.assertIn(f"Sitzung geladen: {target.id}", output)
+        self.assertEqual(renamed_title, "Gefundene Sitzung")
+        self.assertTrue(any("Gefundene Sitzung" in line for line in output))
+
+    def test_load_accepts_unique_prefix_and_rejects_ambiguous_prefix(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            database = Path(directory) / "sessions.sqlite3"
+            store = SQLiteSessionStore(database)
+            service = ChatService(store, _FakeProvider())
+            active = service.new_session()
+            with sqlite3.connect(database) as connection:
+                connection.executemany(
+                    """
+                    INSERT INTO sessions
+                        (id, created_at, updated_at, provider, model, title, last_provider)
+                    VALUES (?, '2026-01-01', '2026-01-01',
+                            'ollama', 'test-model', ?, 'ollama')
+                    """,
+                    (
+                        ("shared-111", "Eins"),
+                        ("shared-222", "Zwei"),
+                        ("unique-333", "Drei"),
+                    ),
+                )
+            connection.close()
+            inputs = iter(["/load unique", "/load shared", "/exit"])
+            output: list[str] = []
+
+            result = run_chat(
+                service,
+                session_id=active.id,
+                input_fn=lambda _prompt: next(inputs),
+                output=output.append,
+            )
+
+        self.assertEqual(result, 0)
+        self.assertIn("Sitzung geladen: unique-333", output)
+        self.assertTrue(any("nicht eindeutig" in line for line in output))
+
     def test_gemini_configuration_defaults_to_no(self) -> None:
         with tempfile.TemporaryDirectory() as directory:
             path = Path(directory) / "javis.toml"

tests/unit/test_gemini_provider.py

@@ -23,6 +23,7 @@ class _FakeModels:
     def __init__(self, result: object) -> None:
         self.result = result
         self.call: dict[str, Any] | None = None
+        self.stream_call: dict[str, Any] | None = None
 
     def generate_content(self, **kwargs: object) -> object:
         self.call = kwargs
@@ -30,6 +31,14 @@ class _FakeModels:
             raise self.result
         return self.result
 
+    def generate_content_stream(self, **kwargs: object):
+        self.stream_call = kwargs
+        if isinstance(self.result, BaseException):
+            raise self.result
+        if isinstance(self.result, list):
+            return iter(self.result)
+        return iter((self.result,))
+
 
 class _FakeClient:
     def __init__(self, result: object) -> None:
@@ -56,6 +65,23 @@ def _response(text: str = "Cloud-Antwort") -> SimpleNamespace:
     )
 
 
+def _stream_response(
+    visible_text: str,
+    *,
+    thought_text: str | None = None,
+    with_usage: bool = False,
+) -> SimpleNamespace:
+    parts = []
+    if thought_text:
+        parts.append(SimpleNamespace(text=thought_text, thought=True))
+    parts.append(SimpleNamespace(text=visible_text, thought=False))
+    usage = SimpleNamespace(prompt_token_count=12, candidates_token_count=7) if with_usage else None
+    return SimpleNamespace(
+        candidates=[SimpleNamespace(content=SimpleNamespace(parts=parts))],
+        usage_metadata=usage,
+    )
+
+
 class GeminiProviderTests(unittest.TestCase):
     def _provider(
         self, result: object, **overrides: object
@@ -99,6 +125,39 @@ class GeminiProviderTests(unittest.TestCase):
         self.assertEqual([content.role for content in call["contents"]], ["user", "model", "user"])
         self.assertEqual(call["config"].max_output_tokens, 1024)
         self.assertIsNone(call["config"].tools)
+        self.assertFalse(call["config"].thinking_config.include_thoughts)
+
+    def test_streaming_yields_only_visible_text_and_tracks_usage(self) -> None:
+        provider, factory = self._provider(
+            [
+                _stream_response("Cloud ", thought_text="verstecktes Denken"),
+                _stream_response("Antwort", with_usage=True),
+            ]
+        )
+
+        chunks = list(provider.stream_chat([ChatMessage("user", "Was ist SQLite?")]))
+
+        self.assertEqual(chunks, ["Cloud ", "Antwort"])
+        self.assertNotIn("verstecktes Denken", "".join(chunks))
+        self.assertEqual(provider.last_usage.input_tokens, 12)
+        self.assertEqual(provider.last_usage.output_tokens, 7)
+        self.assertEqual(
+            factory.client.models.stream_call["model"],
+            "gemini-3.6-flash",
+        )
+
+    def test_streaming_error_is_sanitized(self) -> None:
+        provider, _ = self._provider(
+            errors.ClientError(
+                429,
+                {"message": "quota test-key-not-a-real-secret"},
+            )
+        )
+
+        with self.assertRaises(ProviderRateLimitError) as raised:
+            list(provider.stream_chat([ChatMessage("user", "Was ist Python?")]))
+
+        self.assertNotIn("test-key-not-a-real-secret", str(raised.exception))
 
     def test_missing_key_stops_before_client_creation(self) -> None:
         factory = _ClientFactory(_response())

tests/unit/test_ollama_provider.py

@@ -4,22 +4,39 @@ import unittest
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from typing import ClassVar
 
-from javis.providers.base import ChatMessage
+from javis.providers.base import ChatMessage, ResponseAbortedError
 from javis.providers.ollama import OllamaProvider
 
 
 class _OllamaHandler(BaseHTTPRequestHandler):
     request_payload: ClassVar[dict[str, object]] = {}
+    abort_stream: ClassVar[bool] = False
 
     def do_POST(self) -> None:
         length = int(self.headers["Content-Length"])
         type(self).request_payload = json.loads(self.rfile.read(length))
-        body = json.dumps(
+        if type(self).request_payload["stream"]:
-            {
+            chunks = [
-                "message": {"role": "assistant", "content": "Lokale Antwort"},
+                {
-                "done": True,
+                    "message": {"role": "assistant", "content": "Lokale "},
-            }
+                    "done": False,
-        ).encode()
+                }
+            ]
+            if not type(self).abort_stream:
+                chunks.append(
+                    {
+                        "message": {"role": "assistant", "content": "Antwort"},
+                        "done": True,
+                    }
+                )
+            body = b"".join(json.dumps(chunk).encode() + b"\n" for chunk in chunks)
+        else:
+            body = json.dumps(
+                {
+                    "message": {"role": "assistant", "content": "Lokale Antwort"},
+                    "done": True,
+                }
+            ).encode()
         self.send_response(200)
         self.send_header("Content-Type", "application/json")
         self.send_header("Content-Length", str(len(body)))
@@ -39,6 +56,9 @@ class _OllamaHandler(BaseHTTPRequestHandler):
 
 
 class OllamaProviderTests(unittest.TestCase):
+    def setUp(self) -> None:
+        _OllamaHandler.abort_stream = False
+
     def test_provider_uses_local_chat_endpoint(self) -> None:
         server = ThreadingHTTPServer(("127.0.0.1", 0), _OllamaHandler)
         thread = threading.Thread(target=server.serve_forever, daemon=True)
@@ -62,6 +82,44 @@ class OllamaProviderTests(unittest.TestCase):
         self.assertTrue(status.reachable)
         self.assertTrue(status.model_available)
 
+    def test_provider_streams_visible_text_without_thinking(self) -> None:
+        server = ThreadingHTTPServer(("127.0.0.1", 0), _OllamaHandler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            provider = OllamaProvider(
+                "test-model",
+                f"http://127.0.0.1:{server.server_port}",
+                2,
+            )
+            chunks = list(provider.stream_chat([ChatMessage("user", "Hallo")]))
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()
+
+        self.assertEqual(chunks, ["Lokale ", "Antwort"])
+        self.assertTrue(_OllamaHandler.request_payload["stream"])
+        self.assertFalse(_OllamaHandler.request_payload["think"])
+
+    def test_provider_rejects_incomplete_stream(self) -> None:
+        _OllamaHandler.abort_stream = True
+        server = ThreadingHTTPServer(("127.0.0.1", 0), _OllamaHandler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            provider = OllamaProvider(
+                "test-model",
+                f"http://127.0.0.1:{server.server_port}",
+                2,
+            )
+            with self.assertRaises(ResponseAbortedError):
+                list(provider.stream_chat([ChatMessage("user", "Hallo")]))
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join()
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_privacy.py

@@ -63,6 +63,26 @@ class PrivacyRouterTests(unittest.TestCase):
             CloudPolicy.NEVER,
         )
 
+    def test_medical_questions_are_always_never(self) -> None:
+        for question in (
+            "Wenn ich blute, sollte ich dann zum Arzt?",
+            "Welche Symptome hat eine Grippe?",
+            "Wie nehme ich dieses Medikament ein?",
+            "Was bedeutet diese Diagnose?",
+            "Wie sollte eine Verletzung behandelt werden?",
+        ):
+            with self.subTest(question=question):
+                self.assertEqual(
+                    self.router.classify(question).policy,
+                    CloudPolicy.NEVER,
+                )
+
+    def test_technical_terms_cannot_override_medical_never(self) -> None:
+        decision = self.router.classify(
+            "Wie speichere ich meine Symptome und Diagnose mit Python in SQLite?"
+        )
+        self.assertEqual(decision.policy, CloudPolicy.NEVER)
+
     def test_unknown_content_is_conservatively_ask(self) -> None:
         self.assertEqual(
             self.router.classify("Erkläre das bitte genauer.").policy,

tests/unit/test_provider_router.py

@@ -16,6 +16,7 @@ from javis.providers.base import (
     ProviderRateLimitError,
     ProviderTimeoutError,
     ProviderUsage,
+    ResponseAbortedError,
 )
 from javis.security.privacy import CloudPolicy, PrivacyRouter
 
@@ -28,11 +29,15 @@ class _RecordingProvider:
         *,
         answer: str = "Antwort",
         error: Exception | None = None,
+        stream_chunks: tuple[str, ...] | None = None,
+        stream_error: Exception | None = None,
     ) -> None:
         self.name = name
         self.model = model
         self.answer = answer
         self.error = error
+        self.stream_chunks = stream_chunks
+        self.stream_error = stream_error
         self.calls: list[list[ChatMessage]] = []
         self.last_usage = ProviderUsage(11, 5)
 
@@ -42,6 +47,17 @@ class _RecordingProvider:
             raise self.error
         return self.answer
 
+    def stream_chat(self, messages: list[ChatMessage]):
+        self.calls.append(messages)
+        if self.stream_chunks is None:
+            if self.error:
+                raise self.error
+            yield self.answer
+        else:
+            yield from self.stream_chunks
+        if self.stream_error:
+            raise self.stream_error
+
 
 class HybridProviderTests(unittest.TestCase):
     def setUp(self) -> None:
@@ -131,6 +147,23 @@ class HybridProviderTests(unittest.TestCase):
         self.assertTrue(router.last_route.fallback)
         self.assertIn("lokale Antwort", self.notices[-1])
 
+    def test_medical_never_routes_locally_before_cloud_construction(self) -> None:
+        constructed = False
+
+        def cloud_factory() -> _RecordingProvider:
+            nonlocal constructed
+            constructed = True
+            return self.cloud
+
+        router = self._router(cloud_provider_factory=cloud_factory, mode="gemini")
+
+        answer = router.chat([ChatMessage("user", "Wenn ich blute, sollte ich dann zum Arzt?")])
+
+        self.assertEqual(answer, "Lokal")
+        self.assertFalse(constructed)
+        self.assertFalse(self.approvals)
+        self.assertEqual(router.last_route.privacy_policy, CloudPolicy.NEVER)
+
     def test_ask_defaults_to_local_when_approval_is_denied(self) -> None:
         router = self._router()
 
@@ -266,6 +299,64 @@ class HybridProviderTests(unittest.TestCase):
         self.assertFalse(router.last_route.cloud_suppressed_by_local_mode)
         self.assertTrue(router.last_route.technical_fallback)
 
+    def test_allowed_cloud_response_streams_visible_chunks(self) -> None:
+        self.cloud.stream_chunks = ("Cloud ", "Stream")
+        router = self._router()
+
+        chunks = list(router.stream_chat([ChatMessage("user", "Wie funktioniert SQLite?")]))
+
+        self.assertEqual(chunks, ["Cloud ", "Stream"])
+        self.assertEqual(router.last_route.provider, "gemini")
+        self.assertFalse(router.last_route.fallback)
+
+    def test_cloud_stream_error_before_output_falls_back_locally(self) -> None:
+        self.cloud.stream_chunks = ()
+        self.cloud.stream_error = CloudNetworkError("offline")
+        self.local.stream_chunks = ("Lokaler ", "Ersatz")
+        router = self._router()
+
+        chunks = list(router.stream_chat([ChatMessage("user", "Wie funktioniert Python?")]))
+
+        self.assertEqual(chunks, ["Lokaler ", "Ersatz"])
+        self.assertTrue(router.last_route.fallback)
+        self.assertTrue(router.last_route.technical_fallback)
+        self.assertIn("lokale Antwort", self.notices[-1])
+
+    def test_cloud_stream_error_after_output_never_appends_local_answer(self) -> None:
+        self.cloud.stream_chunks = ("Teilantwort",)
+        self.cloud.stream_error = CloudNetworkError("offline")
+        self.local.stream_chunks = ("Lokaler Ersatz",)
+        router = self._router()
+        stream = router.stream_chat([ChatMessage("user", "Wie funktioniert Python?")])
+
+        self.assertEqual(next(stream), "Teilantwort")
+        with self.assertRaises(ResponseAbortedError):
+            next(stream)
+
+        self.assertFalse(self.local.calls)
+        self.assertFalse(router.last_route.fallback)
+        self.assertTrue(router.last_route.technical_fallback)
+
+    def test_medical_never_streams_locally_before_cloud_construction(self) -> None:
+        constructed = False
+        self.local.stream_chunks = ("Lokal",)
+
+        def cloud_factory() -> _RecordingProvider:
+            nonlocal constructed
+            constructed = True
+            return self.cloud
+
+        router = self._router(cloud_provider_factory=cloud_factory, mode="gemini")
+
+        chunks = list(
+            router.stream_chat([ChatMessage("user", "Welche Diagnose passt zu meinen Schmerzen?")])
+        )
+
+        self.assertEqual(chunks, ["Lokal"])
+        self.assertFalse(constructed)
+        self.assertFalse(self.approvals)
+        self.assertEqual(router.last_route.privacy_policy, CloudPolicy.NEVER)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_sqlite_store.py

@@ -1,15 +1,20 @@
+import sqlite3
 import tempfile
 import unittest
 from pathlib import Path
 
-from javis.memory.sqlite_store import SessionNotFoundError, SQLiteSessionStore
+from javis.memory.sqlite_store import (
+    AmbiguousSessionReferenceError,
+    SessionNotFoundError,
+    SQLiteSessionStore,
+)
 
 
 class SQLiteSessionStoreTests(unittest.TestCase):
     def setUp(self) -> None:
         self.temporary_directory = tempfile.TemporaryDirectory()
-        database = Path(self.temporary_directory.name) / "sessions.sqlite3"
+        self.database = Path(self.temporary_directory.name) / "sessions.sqlite3"
-        self.store = SQLiteSessionStore(database)
+        self.store = SQLiteSessionStore(self.database)
 
     def tearDown(self) -> None:
         self.temporary_directory.cleanup()
@@ -21,6 +26,8 @@ class SQLiteSessionStoreTests(unittest.TestCase):
         self.assertEqual(loaded.id, created.id)
         self.assertEqual(loaded.provider, "ollama")
         self.assertEqual(loaded.model, "test-model")
+        self.assertEqual(loaded.title, "Neue Sitzung")
+        self.assertEqual(loaded.last_provider, "ollama")
         self.assertEqual(self.store.list_sessions()[0].id, created.id)
 
     def test_messages_keep_exchange_order(self) -> None:
@@ -53,6 +60,112 @@ class SQLiteSessionStoreTests(unittest.TestCase):
         self.assertEqual(self.store.get_messages(session.id), [])
         self.assertEqual(self.store.get_session(session.id).message_count, 0)
 
+    def test_first_exchange_sets_title_and_last_provider_atomically(self) -> None:
+        session = self.store.create_session("hybrid", "auto")
+
+        self.store.append_exchange(
+            session.id,
+            "Frage",
+            "Antwort",
+            title_if_first="Kurzer Titel",
+            last_provider="gemini",
+        )
+        self.store.append_exchange(
+            session.id,
+            "Noch eine Frage",
+            "Noch eine Antwort",
+            title_if_first="Darf Titel nicht ersetzen",
+            last_provider="ollama",
+        )
+
+        loaded = self.store.get_session(session.id)
+        self.assertEqual(loaded.title, "Kurzer Titel")
+        self.assertEqual(loaded.last_provider, "ollama")
+
+    def test_old_database_is_migrated_idempotently_without_data_loss(self) -> None:
+        database = Path(self.temporary_directory.name) / "legacy.sqlite3"
+        with sqlite3.connect(database) as connection:
+            connection.executescript(
+                """
+                CREATE TABLE sessions (
+                    id TEXT PRIMARY KEY,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL,
+                    provider TEXT NOT NULL,
+                    model TEXT NOT NULL
+                );
+                CREATE TABLE messages (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    session_id TEXT NOT NULL,
+                    position INTEGER NOT NULL,
+                    role TEXT NOT NULL,
+                    content TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    UNIQUE (session_id, position)
+                );
+                INSERT INTO sessions VALUES
+                    ('legacy', '2026-01-01', '2026-01-02', 'ollama', 'old-model');
+                INSERT INTO messages
+                    (session_id, position, role, content, created_at)
+                VALUES ('legacy', 0, 'user', 'Alte Nachricht', '2026-01-02');
+                """
+            )
+        connection.close()
+
+        migrated = SQLiteSessionStore(database)
+        migrated_again = SQLiteSessionStore(database)
+
+        session = migrated_again.get_session("legacy")
+        self.assertEqual(session.title, "Neue Sitzung")
+        self.assertEqual(session.last_provider, "ollama")
+        self.assertEqual(
+            [message.content for message in migrated.get_messages("legacy")],
+            ["Alte Nachricht"],
+        )
+
+    def test_rename_and_search_only_titles_and_user_messages(self) -> None:
+        first = self.store.create_session("ollama", "test-model")
+        second = self.store.create_session("ollama", "test-model")
+        self.store.rename_session(first.id, "SQLite Hilfe")
+        self.store.append_exchange(first.id, "Transaktion erklären", "Privates Lösungswort")
+        self.store.append_exchange(second.id, "Andere Frage", "Nur SQLite in Antwort")
+
+        self.assertEqual(self.store.get_session(first.id).title, "SQLite Hilfe")
+        self.assertEqual(
+            [session.id for session in self.store.search_sessions("Transaktion")],
+            [first.id],
+        )
+        self.assertEqual(self.store.search_sessions("Lösungswort"), [])
+        self.assertEqual(self.store.search_sessions("SQLite"), [self.store.get_session(first.id)])
+
+    def test_search_treats_sql_wildcards_and_injection_as_plain_text(self) -> None:
+        session = self.store.create_session("ollama", "test-model")
+        self.store.rename_session(session.id, "100% SQLite_Name")
+
+        self.assertEqual(self.store.search_sessions("%")[0].id, session.id)
+        self.assertEqual(self.store.search_sessions("_")[0].id, session.id)
+        self.assertEqual(self.store.search_sessions("' OR 1=1 --"), [])
+
+    def test_unique_and_ambiguous_session_prefixes(self) -> None:
+        with sqlite3.connect(self.database) as connection:
+            connection.executemany(
+                """
+                INSERT INTO sessions
+                    (id, created_at, updated_at, provider, model, title, last_provider)
+                VALUES (?, '2026-01-01', '2026-01-01', 'ollama', 'model', ?, 'ollama')
+                """,
+                (
+                    ("abc111", "Eins"),
+                    ("abc222", "Zwei"),
+                    ("unique333", "Drei"),
+                ),
+            )
+        connection.close()
+
+        self.assertEqual(self.store.resolve_unique_prefix("unique").id, "unique333")
+        with self.assertRaises(AmbiguousSessionReferenceError):
+            self.store.resolve_unique_prefix("abc")
+
 
 if __name__ == "__main__":
     unittest.main()