7 Commits

19 changed files with 1463 additions and 114 deletions
+8
View File
@@ -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.
+36 -33
View File
@@ -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`,
`/privacy`, `/status`, `/help`, `/exit`.
- CLI-Befehle: `/new`, `/sessions`, `/load <NR|ID>`, `/search <Text>`,
`/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`
- Feature-Branch ist als `7a14fe8` zu
`origin/feat/gemini-privacy-router` gepusht
- aktueller und geprüfter Branch: `main`
- stabiler Ausgangsstand: `main` bei `77b510b`
- 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
- 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,
beispielsweise Sitzungstitel/Suche oder Streaming. Noch keine Obsidian-
Integration, Tools, Sprache oder Serversteuerung beginnen.
Chatkomfort ist implementiert, gemergt und auf `main` erneut geprüft. Nach dem
Push stoppen und mit Pascal erst später ein einzelnes nächstes Paket festlegen.
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.
+12 -12
View File
@@ -2,22 +2,22 @@
## Ausgangsstand
Der Core-Chat und der Gemini-/Datenschutz-Meilenstein sind automatisiert durch
Codex und am 30.07.2026 manuell durch Pascal abgenommen. Der Featurestand ist zum
Merge freigegeben.
Der Core-Chat, Gemini-/Datenschutz-Meilenstein und das Chatkomfort-Paket sind
automatisiert geprüft. Pascal hat Core-Chat sowie Gemini/Datenschutz am
30.07.2026 manuell abgenommen.
Bestätigt sind echter Gemini-Erfolg, verdeckte Key-Speicherung, Nullkostenmodus,
bereinigter Status, Cloudzähler, `ask`- und `never`-Routing, erzwungener lokaler
Modus sowie das Startskript.
Chatkomfort umfasst sicheres Ollama-/Gemini-Streaming, Strg+C ohne Teilpersistenz,
medizinisches `never`, automatische und manuelle Sitzungstitel, nummerierte
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
- Streaming lokaler und Cloudantworten mit sauberem Abbruchverhalten
Vorher wie immer Branch, Arbeitsbaum, aktuellen Handoff und Tests prüfen.
Der Chatkomfort-Branch ist nach `main` gemergt und dort erneut geprüft. Den
finalen Stand aus `docs/CHATGPT_HANDOFF.md` übernehmen und mit Pascal ein
einzelnes nächstes Paket festlegen; noch keine Folgephase automatisch beginnen.
## Weiterhin nicht beginnen
+36 -18
View File
@@ -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,20 @@ Living-Mind-Frontend.
## Git
- aktueller Branch: `main`
- Feature-Commit: `7a14fe8`
- Merge-Commit: `564a3fd`
- Tests auf dem gemergten `main`: 66 bestanden
- `main` erfolgreich zu `origin/main` gepusht
- Grundgerüstcommit: `6d04171`
- Basis des Feature-Branches: `e6eb995`
- Toolchain: `fa90a5c`
- Laufzeit/Modell: `10d7cd7`
- Chatkern: `ec9b2f9`
- Tests: `b153daa`
- Dokumentation: `5d567e5`
- Merge nach `main`: `f3d6dfb`
- Basis: `main` bei `77b510b`
- medizinisches Routing: `9148193`
- Streaming: `00a5a6c`
- Titelmigration: `112ae6d`
- Sitzungsnavigation: `43c1f76`
- Dokumentation: `328dcd9`
- Feature-Branch zu `origin/feat/chat-comfort` gepusht
- Merge nach `main`: `cbba102`; 91 Tests und Ruff auf `main` erneut bestanden
- vorheriger Gemini-/Datenschutz-Merge auf `main`: `564a3fd`
## Bekannte Einschränkung
Das Startskript startet Ollama bei Bedarf lokal. Antworten werden noch nicht
gestreamt; Sitzungen haben noch keine Titel, Suche oder Löschfunktion.
Das Startskript startet Ollama bei Bedarf lokal. Eine Löschfunktion ist nicht
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
+114 -2
View File
@@ -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
+259 -12
View File
@@ -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],
+115 -8
View File
@@ -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"{session.provider}/{session.model} | {session.message_count} Nachrichten"
f"{number}. {session.id[:8]} | {session.title} | {date} | "
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)
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}")
+164 -13
View File
@@ -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)
VALUES (?, ?, ?, ?, ?)
INSERT INTO sessions
(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"],
)
+11
View File
@@ -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."""
...
+69 -1
View File
@@ -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)
+74
View File
@@ -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."
)
+7 -2
View File
@@ -42,8 +42,13 @@ _NEVER_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
),
(
re.compile(
r"(?i)\b(diagnose|gesundheit|krankheit|medikament|patient|arztbericht|"
r"psychotherapie|blutwert)\b"
r"(?i)\b(symptom(?:e|en)?|blut(?:e|est|et|en|ung|ungen|wert|werte)?|"
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",
),
+76 -2
View File
@@ -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__":
+131 -1
View File
@@ -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"
+59
View File
@@ -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())
+59 -1
View File
@@ -4,16 +4,33 @@ 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))
if type(self).request_payload["stream"]:
chunks = [
{
"message": {"role": "assistant", "content": "Lokale "},
"done": False,
}
]
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"},
@@ -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()
+20
View File
@@ -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,
+91
View File
@@ -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()
+116 -3
View File
@@ -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.store = SQLiteSessionStore(database)
self.database = Path(self.temporary_directory.name) / "sessions.sqlite3"
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()