""" KI-Server — Einstiegspunkt. FastAPI + APScheduler + Router-Integration. """ import asyncio import logging from contextlib import asynccontextmanager from pathlib import Path from apscheduler.schedulers.asyncio import AsyncIOScheduler from fastapi import FastAPI from api.admin import router as admin_router from api.chat import entitaeten_embeddings_sicherstellen from api.chat import router as chat_router from api.drive import router as drive_router from api.gedaechtnis import router as gedaechtnis_router from api.health import ollama_testen, postgres_testen, redis_testen # API-Router from api.health import router as health_router from api.session import router as session_router from api.websocket import aktive_verbindungen from api.websocket import router as websocket_router from config import ( AKTIVES_PAAR_USER_ID, ASSISTANT_USER_ID, NMCP_ABGLEICH_INTERVALL_SEKUNDEN, PIXIE_AKTIV, POSTGRES_URL, graph_run_lock, redis_client, shutdown_event, ) from graph.builder import build_agent_graph, build_character_graph, build_human_graph from memory.pipeline_log import ( init_buffer as pipeline_log_init, ) from memory.pipeline_log import ( writer_loop as pipeline_log_writer, ) from memory.ziele import ziele_embeddings_sicherstellen from schema_bootstrap import apply_agent_schemas from services.event_consumer import event_consumer_loop from services.prompt_consumer import prompt_consumer_loop from services.shadow_delivery import shadow_delivery_loop from tools.core_schema import ( CORE_SCHEMA_PATH, LOCK_TIMEOUT_MS, SchemaBootstrapError, SchemaFileMissingError, SchemaLockTimeoutError, apply_core_schema, redact_connection, ) logger = logging.getLogger("ki_server") scheduler = AsyncIOScheduler() # Pfad und Sperrfrist der Migration stehen beim Kernschema, das der Start mit # dem Aufbau von null teilt; die Begruendung der Frist steht dort an der Konstante. SQL_INIT_PFAD: Path = CORE_SCHEMA_PATH MIGRATION_SPERRFRIST_MS: int = LOCK_TIMEOUT_MS def schema_migrieren(postgres_url: str) -> None: """Migration: führt db/init.sql gegen die bestehende Live-Datenbank aus. db/init.sql ist Single Source of Truth für das Postgres-Schema und enthält alle Statements idempotent (CREATE ... IF NOT EXISTS, ALTER ... IF NOT EXISTS, INSERT ... WHERE NOT EXISTS, DO-Blöcke mit pg_constraint-Check). Den Rumpf traegt `tools.core_schema.apply_core_schema`, der bei jeder Stoerung wirft; diese Huelle schluckt fuer den Start, damit ein Neuladen nicht an der Migration scheitert. EVA --- E: postgres_url muss erreichbar sein. db/init.sql muss am erwarteten Pfad (server/../db/init.sql) liegen. V: `apply_core_schema` — autocommit=True, damit DO-Blöcke und unabhängige Statements korrekt transaktioniert werden, und `lock_timeout` (`MIGRATION_SPERRFRIST_MS`), damit eine fremde Sperre den Start nicht unbegrenzt anhält. A: info-Log mit Pfad und einer groben Statement-Zahl (Semikolon-Zählung). Bei Fehler warning-Log und Rückkehr ohne Re-Raise — Fail-Mode wie bisher in P0. Eine abgelaufene Sperrfrist ist ein `error`: Das Schema ist dann nicht migriert, und die Zeile nennt den Grund. Kein Grund traegt die Verbindungszeichenkette. """ try: apply_core_schema(postgres_url) anzahl_stmts: int = SQL_INIT_PFAD.read_text(encoding="utf-8").count(";") except SchemaFileMissingError: logger.warning(f"Migration: db/init.sql nicht gefunden unter {SQL_INIT_PFAD}") return except SchemaLockTimeoutError as fehler: logger.error( f"Migration: db/init.sql NICHT ausgeführt — nach " f"{MIGRATION_SPERRFRIST_MS} ms keine Tabellensperre ({fehler.reason}). " f"Eine andere Verbindung haelt eine Transaktion offen; das Schema " f"bleibt auf dem vorigen Stand bis zum naechsten Start." ) return except SchemaBootstrapError as fehler: logger.warning(f"Migration: db/init.sql fehlgeschlagen — {fehler.reason}") return except Exception as fehler: grund: str = redact_connection(str(fehler), postgres_url) logger.warning(f"Migration: db/init.sql fehlgeschlagen — {grund}") return logger.info( f"Migration: db/init.sql ausgeführt " f"({anzahl_stmts} Statements geparst, Pfad {SQL_INIT_PFAD})." ) # ───────────────────────────────────────────── # Lifespan (Start / Stop) # ───────────────────────────────────────────── @asynccontextmanager async def lifespan(app: FastAPI): # Health-Check-Spam aus dem Access-Log filtern class HealthCheckFilter(logging.Filter): def filter(self, record): return "/health" not in record.getMessage() logging.getLogger("uvicorn.access").addFilter(HealthCheckFilter()) logger.info("Server startet...") # Das Paar, fuer das die Hintergrundarbeit dieses Laufs stattfindet. Steht # hier, weil es sonst von aussen nicht feststellbar ist: Wird die Umgebung # fuer eine Messreihe umgestellt und greift die Umstellung nicht — etwa # weil `docker compose restart` den Behaelter neu startet, ohne ihn neu zu # erzeugen —, laeuft die Reihe dreissig Turns lang gegen das falsche Paar, # und nichts meldet es. Ein Wert, den niemand ablesen kann, ist keine # Konfiguration, sondern eine Annahme. logger.info( f"Aktives Paar: ({AKTIVES_PAAR_USER_ID}, {ASSISTANT_USER_ID}) — " f"Pixie bedient beide Seiten dieses Paares, sonst niemanden" ) # Event-Loop-Referenz für synchrone Endpoints (broadcast_threadsafe) app.state.loop = asyncio.get_running_loop() # Pipeline-Log-Buffer initialisieren (Forensik-Infrastruktur). # Muss vor dem ersten Helper-Aufruf passieren, der spätestens im # Enricher des ersten Konversations-Turns kommt. pipeline_log_init(app.state.loop) # Model-Service-Worker starten (Phase 2: EmbedWorker) from services.model_services import model_service await model_service.startup() # Verbindungstests ollama_ok: bool = ollama_testen() redis_ok: bool = redis_testen() postgres_ok: bool = postgres_testen() if not all([redis_ok, postgres_ok]): logger.error("Kritische Verbindung fehlgeschlagen — degradierter Modus.") # Schema-Migrationen (idempotent) if postgres_ok: schema_migrieren(POSTGRES_URL) if not ollama_ok: logger.warning("Ollama nicht bereit — Antworten nicht möglich bis Modell geladen.") # Embedding-Repair: Entitäten ohne Embedding nachträglich versorgen if postgres_ok and ollama_ok: await entitaeten_embeddings_sicherstellen() await ziele_embeddings_sicherstellen(POSTGRES_URL) # Epic 11: Agent-Discovery — derselbe Weg wie der Aufbau von null: # discover_agents, dann setup je Agent der Registry; ein werfendes setup # bricht den Start ab. from agents import AgentRegistry apply_agent_schemas() logger.info(f"Agent-Discovery: {len(AgentRegistry.alle())} Agenten registriert") # NMCP-Handshake: Naht herstellen und Kompatibilitaet pruefen. # # Das ist der einzige Zeitpunkt mit einem echten Verweigerungsrecht — # zur Laufzeit kann niemand nein sagen, ohne das Gespraech anzuhalten. # Die Ablehnung trifft den einzelnen Dienst und NICHT das System: Ein # fehlerhaft angemeldetes Plugin darf den Start nicht verhindern. from agents.nmcp import anmelden, gesamtbild_pruefen _befunde = {} for _agent in AgentRegistry.alle().values(): try: _befunde[_agent.name] = anmelden(_agent) except (TypeError, ValueError) as _fehler: logger.error( f"NMCP: Dienst '{_agent.name}' nicht anmeldbar " f"({type(_fehler).__name__}: {_fehler}) — nicht eingebunden" ) _nicht = [b.name for b in _befunde.values() if not b.eingebunden] _ohne_zweifel = [ b.name for b in _befunde.values() if b.eingebunden and not b.zweifel_erlaubt ] logger.info( f"NMCP-Handshake: {len(_befunde)} geprueft, " f"{len(_befunde) - len(_nicht)} eingebunden, " f"{len(_nicht)} verweigert, {len(_ohne_zweifel)} ohne Zweifelsfaelle" ) # Eine verweigerte Einbindung muss zur LAUFZEIT sichtbar bleiben, nicht # nur hier. Eine Startmeldung ist nach zehn Minuten aus dem Blick, und # danach verhaelt sich der fehlende Dienst wie einer, den niemand # braucht — genau der stille Zustand, gegen den der Quotenabgleich # gebaut ist. `/health` rechnet den Befund dafuer neu, statt diesen # Startzustand mitzuschleppen: ein Schnappschuss altert. if _nicht: logger.error(f"NMCP: NICHT eingebunden — {_nicht}") if _ohne_zweifel: logger.warning( f"NMCP: ohne Zweifelsfaelle (vierter Ausgang fehlt) — {_ohne_zweifel}" ) for _meldung in gesamtbild_pruefen(AgentRegistry.alle()): logger.warning(f"NMCP-Gesamtbild [{_meldung.regel}]: {_meldung.text}") # Objekt-Merkmale einbetten (Scheibe 12, Teil C). Nur die eine Seite der # Naehe-Rechnung; der Empfang liest die Vektoren noch nicht. if ollama_ok: from agents.object_feature import feature_vectors_ensure await feature_vectors_ensure(AgentRegistry.alle()) else: logger.error("Objekt-Merkmal: Ollama nicht bereit — keine Merkmalsvektoren in diesem Lauf") # Periodische Pixie-Aufgaben registrieren (aus Agent periodic_task()) if PIXIE_AKTIV: import time as _time for _agent in AgentRegistry.alle().values(): _task = _agent.periodic_task() # Ein Agent, der keine periodische Aufgabe (mehr) meldet, darf keinen # Zeitplan-Eintrag zuruecklassen. Ein solcher Zombie bleibt faellig, # wird bei jedem Heartbeat als Kandidat gesammelt und waechst seit dem # Aging (Chat 113) bis zum Deckel — er gewaenne dann jeden Zyklus, # ohne je zu laufen. Abschalten muss den Kandidaten entfernen, nicht # nur seine Wirkung. if not _task: _zombie = f"pixie:schedule:{_agent.name}" if redis_client.delete(_zombie): logger.info( f"Pixie: Zeitplan entfernt — {_agent.name} meldet keine " f"periodische Aufgabe mehr (Kandidat abgemeldet)" ) continue if _task: _key = f"pixie:schedule:{_task.name}" _neu: bool = not redis_client.exists(_key) # Takt und Prioritaet folgen der Konfiguration, next_run nicht. # Bis Chat 111 wurde der Eintrag nur beim ersten Mal geschrieben # — eine Aenderung an config.py erreichte das laufende System # dann nie, und niemand sah warum. next_run bleibt stehen, sonst # rutschte jeder Neustart den Takt nach vorn. if _neu: redis_client.hset(_key, mapping={ "priority": str(_task.priority), "interval": str(_task.interval), "next_run": str(_time.time()), "description": _task.description, }) logger.info( f"Pixie: Periodische Aufgabe registriert — {_task.name} " f"(Prio {_task.priority}, alle {_task.interval}s)" ) else: _alt_intervall = redis_client.hget(_key, "interval") _alt_prio = redis_client.hget(_key, "priority") redis_client.hset(_key, mapping={ "priority": str(_task.priority), "interval": str(_task.interval), "description": _task.description, }) if (_alt_intervall and str(_alt_intervall) != str(_task.interval)) or \ (_alt_prio and str(_alt_prio) != str(_task.priority)): logger.info( f"Pixie: Zeitplan angeglichen — {_task.name}: " f"Takt {_alt_intervall}s -> {_task.interval}s, " f"Prio {_alt_prio} -> {_task.priority}" ) else: logger.debug("main: Periodic-Task-Discovery uebersprungen (PIXIE_AKTIV=False)") # Scheduler starten — Pixie-Heartbeat (kompetitives Scheduling) from config import PIXIE_CPU_INTERVALL_SEKUNDEN, PIXIE_INTERVALL_SEKUNDEN if PIXIE_AKTIV: from services.model_services.spur import SPUR_CPU, SPUR_LLM from services.pixie.scheduler import pixie_heartbeat # Zwei Spuren, zwei Jobs, zwei Sperren. Je `max_instances=1`, damit # innerhalb einer Spur weiterhin genau einer laeuft — darauf beruht # die Arbeitsliste der Promotion. Ein Agent gehoert zu genau einer # Spur, also kann er auch nicht doppelt laufen. async def _pixie_job_llm() -> None: await pixie_heartbeat(app.state, SPUR_LLM) async def _pixie_job_cpu() -> None: await pixie_heartbeat(app.state, SPUR_CPU) scheduler.add_job( _pixie_job_llm, trigger = "interval", seconds = PIXIE_INTERVALL_SEKUNDEN, id = "pixie_heartbeat_llm", name = "Pixie Heartbeat (LLM-Spur)", max_instances = 1, coalesce = True, ) scheduler.add_job( _pixie_job_cpu, trigger = "interval", seconds = PIXIE_CPU_INTERVALL_SEKUNDEN, id = "pixie_heartbeat_cpu", name = "Pixie Heartbeat (CPU-Spur)", max_instances = 1, coalesce = True, ) # Der Quotenabgleich braucht einen Aufrufer, sonst ist er selbst # eine Deklaration ohne Leser. Der Takt ist bewusst lang: Der # Abgleich braucht 30 bis 100 Aeusserungen, bevor er ueberhaupt # urteilen darf — haeufiger zu pruefen erzeugt nur "keine Aussage". async def _nmcp_abgleich() -> None: """Haelt die angemeldeten Quoten gegen die gezaehlten Zustellungen.""" from agents import AgentRegistry from agents.nmcp_quote import abgleich_protokollieren abgleich_protokollieren(AgentRegistry.alle()) scheduler.add_job( _nmcp_abgleich, trigger = "interval", seconds = NMCP_ABGLEICH_INTERVALL_SEKUNDEN, id = "nmcp_quotenabgleich", name = "NMCP-Quotenabgleich", max_instances = 1, coalesce = True, ) scheduler.start() logger.info( f"Scheduler gestartet — LLM-Spur {PIXIE_INTERVALL_SEKUNDEN}s, " f"CPU-Spur {PIXIE_CPU_INTERVALL_SEKUNDEN}s." ) else: logger.debug("main: Pixie-Heartbeat-Job uebersprungen (PIXIE_AKTIV=False)") logger.info("Pixie-Master-Switch: AKTIV=False — alle Pixie-Pfade ruhen") # Graphen kompilieren compiled_human, human_graph = build_human_graph( redis_client = redis_client, postgres_url = POSTGRES_URL, ) app.state.conversation_graph = compiled_human app.state.human_graph = human_graph logger.info("HumanGraph initialisiert.") compiled_agent, agent_graph = build_agent_graph( redis_client = redis_client, postgres_url = POSTGRES_URL, ) app.state.agent_graph = agent_graph app.state.compiled_agent_graph = compiled_agent logger.info("AgentGraph initialisiert.") compiled_character, character_graph = build_character_graph( redis_client = redis_client, postgres_url = POSTGRES_URL, ) app.state.character_graph = character_graph app.state.compiled_character_graph = compiled_character logger.info("CharacterGraph initialisiert.") # ── NEU: Shadow Delivery Service starten ── if PIXIE_AKTIV: delivery_task = asyncio.create_task( shadow_delivery_loop( redis_client = redis_client, websocket_map = aktive_verbindungen, graph_run_lock = graph_run_lock, compiled_agent_graph = compiled_agent, agent_graph = agent_graph, ) ) logger.info("Shadow Delivery Service gestartet.") else: delivery_task = None logger.debug("main: Shadow Delivery Service uebersprungen (PIXIE_AKTIV=False)") # ── Event-Consumer starten ── consumer_task = asyncio.create_task( event_consumer_loop( redis_client = redis_client, character_graph = character_graph, compiled_character = compiled_character, websocket_map = aktive_verbindungen, graph_run_lock = graph_run_lock, ) ) logger.info("Event-Consumer gestartet.") # ── Prompt-Consumer starten ── # Er faehrt Pfad 1 hinter der Eingangs-Queue: Der Endpunkt nimmt nur an, # und mehrere Aeusserungen innerhalb des Fensters werden zu einem Prompt. prompt_task = asyncio.create_task( prompt_consumer_loop( human_graph = app.state.human_graph, conversation_graph = app.state.conversation_graph, ) ) logger.info("Prompt-Consumer gestartet.") # Pipeline-Log-Writer als viertes Hintergrund-Task. pipeline_log_task: asyncio.Task = asyncio.create_task( pipeline_log_writer(POSTGRES_URL, shutdown_event) ) logger.info("Lifespan: Pipeline-Log-Writer gestartet.") yield # ── Shutdown: Event setzen, dann aufräumen ── shutdown_event.set() if PIXIE_AKTIV: scheduler.shutdown(wait=False) if delivery_task is not None: delivery_task.cancel() consumer_task.cancel() # Der Prompt-Consumer faehrt Pfad 1 hinter der Eingangs-Queue. Er fehlte # hier bis zum 25.08.2026 als einziger der vier Aufgaben und lief beim # Herunterfahren weiter, bis der Prozess starb — sichtbar geworden, weil # `prompt_task` als einzige Task-Variable nie gelesen wurde. prompt_task.cancel() # Pipeline-Log-Writer sauber beenden: shutdown_event ist bereits gesetzt, # der Writer sieht das im nächsten Loop-Tick und führt einen Final-Flush # durch. Wir warten bis zu 30 Sekunden — generös, damit auch ein voller # Buffer noch in die DB kommt. # # Bewusste Abweichung von der Bestandspraxis (delivery_task, consumer_task # werden nur per .cancel() ohne await beendet). Pipeline-Log ist Forensik; # Datenverlust beim Shutdown wäre die schmerzlichste Stelle. Muster gilt # als Vorbild für REFAC-SHUTDOWN-DISZIPLIN (siehe Backlog). try: await asyncio.wait_for(pipeline_log_task, timeout=30.0) logger.info("Lifespan: Pipeline-Log-Writer beendet (regulär).") except asyncio.TimeoutError: logger.warning( "Lifespan: Pipeline-Log-Writer hat Timeout überschritten — Cancel." ) pipeline_log_task.cancel() try: await pipeline_log_task except (asyncio.CancelledError, Exception): pass # Model-Service-Worker sauber beenden await model_service.shutdown() logger.info("Server gestoppt.") # ───────────────────────────────────────────── # FastAPI App # ───────────────────────────────────────────── app = FastAPI( title = "KI-Server", description = "Persönlicher KI-Assistent mit Gedächtnis und Hintergrund-Rauschen", version = "0.3.0", lifespan = lifespan, ) app.include_router(health_router) app.include_router(chat_router) app.include_router(gedaechtnis_router) app.include_router(session_router) app.include_router(websocket_router) app.include_router(admin_router) app.include_router(drive_router)