from __future__ import annotations import base64 import json import os import shutil import sqlite3 import threading import time import uuid from functools import wraps from pathlib import Path from urllib.parse import quote from typing import Any, Callable from flask import ( Flask, Response, g, jsonify, request, send_file, send_from_directory, session, stream_with_context, ) from .artifact_storage import EncryptedArtifactStore from .advanced import register_advanced_routes from .config import load_runtime_config from .crypto import ENCRYPTION_VERSION, EncryptionError, parse_master_key from .db import connect, initialise, json_text, transaction from .fileops import ( MAX_IMAGE_INLINE, FileValidationError, attachment_content, inspect_stored_file, ) from .models import build_profiles, normalise_preferences from .provider import ProviderError, build_payload, fake_stream, stream_together from .security import hash_password, new_token, verify_password from .tci import TCIError, execute_tci, fake_execute CHAT_TERMINAL = {"complete", "failed", "cancelled", "partial"} JOB_TERMINAL = {"successful", "failed", "cancelled", "partial", "blocked"} RUN_CANCEL: dict[str, threading.Event] = {} RUN_RESPONSES: dict[str, Any] = {} RUN_LOCK = threading.Lock() JOB_CANCEL: dict[str, threading.Event] = {} JOB_LOCK = threading.Lock() def uid(prefix: str) -> str: return f"{prefix}_{uuid.uuid4().hex}" def create_app(test_config: dict[str, Any] | None = None) -> Flask: runtime = load_runtime_config() if test_config: runtime.update(test_config) static_dir = Path(runtime["PROJECT_ROOT"]) / "static" app = Flask(__name__, static_folder=str(static_dir), static_url_path="") app.config.update(runtime) app.secret_key = str(runtime["SECRET_KEY"]) app.config.update( SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE="Lax", SESSION_COOKIE_SECURE=bool(runtime.get("COOKIE_SECURE", False)), MAX_CONTENT_LENGTH=int(runtime.get("MAX_UPLOAD_BYTES", 512 * 1024 * 1024)), ) Path(app.config["STORAGE_DIR"]).mkdir(parents=True, exist_ok=True) initialise(app.config["DATABASE"]) encryption_config_error = "" try: master_key = parse_master_key(app.config.get("MASTER_KEY")) except EncryptionError as exc: master_key = None encryption_config_error = str(exc) app.config["MASTER_KEY_BYTES"] = master_key app.config["ENCRYPTION_CONFIG_ERROR"] = encryption_config_error artifact_store = EncryptedArtifactStore( Path(app.config["STORAGE_DIR"]), master_key, int(app.config.get("MASTER_KEY_VERSION", 1)), ) if artifact_store.configured: migration_db = connect(app.config["DATABASE"]) try: app.config["ENCRYPTION_MIGRATION"] = artifact_store.migrate_legacy_files(migration_db) finally: migration_db.close() else: app.config["ENCRYPTION_MIGRATION"] = {"migrated": 0, "missing": 0} def get_db() -> sqlite3.Connection: if "db" not in g: g.db = connect(app.config["DATABASE"]) return g.db @app.teardown_appcontext def close_db(_error: BaseException | None) -> None: db = g.pop("db", None) if db is not None: db.close() def current_user() -> sqlite3.Row | None: user_id = session.get("user_id") if not user_id: return None return get_db().execute( "SELECT id,username,role FROM users WHERE id=?", (user_id,) ).fetchone() def error( code: str, message: str, status: int, retryable: bool = False, details: dict[str, Any] | None = None, ): return ( jsonify( { "error": { "code": code, "message": message, "retryable": retryable, "request_id": uid("req"), "details": details or {}, } } ), status, ) def require_auth(fn: Callable): @wraps(fn) def wrapped(*args, **kwargs): user = current_user() if not user: return error("KIMU_AUTH_REQUIRED", "Sign in is required.", 401) g.user = user return fn(*args, **kwargs) return wrapped def require_owner(fn: Callable): @wraps(fn) @require_auth def wrapped(*args, **kwargs): if g.user["role"] != "owner": return error("KIMU_OWNER_REQUIRED", "Owner access is required.", 403) return fn(*args, **kwargs) return wrapped def audit( db: sqlite3.Connection, action: str, target_type: str | None = None, target_id: str | None = None, detail: dict[str, Any] | None = None, user_id: str | None = None, ) -> None: db.execute( "INSERT INTO audit_log(user_id,action,target_type,target_id,detail_json) VALUES(?,?,?,?,?)", ( user_id or session.get("user_id"), action, target_type, target_id, json_text(detail or {}), ), ) def public_file(row: sqlite3.Row, include_inventory: bool = False) -> dict[str, Any]: try: private = artifact_store.private_metadata(row) metadata_error = None except EncryptionError as exc: private = {"warnings": [str(exc)], "zip_inventory": [], "text_content": ""} metadata_error = exc.code value = { "id": row["id"], "name": row["original_name"], "mime_type": row["mime_type"], "size_bytes": row["size_bytes"], "stored_size_bytes": row["stored_size_bytes"], "sha256": row["sha256"], "kind": row["kind"], "status": row["status"], "warnings": private.get("warnings") or [], "encrypted": int(row["encryption_version"] or 0) == ENCRYPTION_VERSION, "encryption_version": int(row["encryption_version"] or 0), "key_version": int(row["key_version"] or 0), "metadata_error": metadata_error, "created_at": row["created_at"], "download_url": f"/api/v1/files/{row['id']}/download", } if include_inventory: value["zip_inventory"] = private.get("zip_inventory") or [] value["text_preview"] = str(private.get("text_content") or "")[:12000] return value def public_job(row: sqlite3.Row) -> dict[str, Any]: return { "id": row["id"], "kind": row["kind"], "title": row["title"], "status": row["status"], "progress": row["progress"], "result": json.loads(row["result_json"] or "{}"), "error": ( {"code": row["error_code"], "message": row["error_message"]} if row["error_code"] else None ), "created_at": row["created_at"], "updated_at": row["updated_at"], } def message_attachments(db: sqlite3.Connection, message_id: str) -> list[dict[str, Any]]: rows = db.execute( """ SELECT f.* FROM message_attachments ma JOIN files f ON f.id=ma.file_id WHERE ma.message_id=? ORDER BY ma.position,f.created_at """, (message_id,), ).fetchall() return [public_file(row, include_inventory=row["kind"] == "zip") for row in rows] def public_message(db: sqlite3.Connection, row: sqlite3.Row) -> dict[str, Any]: value = dict(row) value["attachments"] = message_attachments(db, row["id"]) return value def append_run_event( db: sqlite3.Connection, run_id: str, event_type: str, data: dict[str, Any] ) -> int: seq = db.execute( "SELECT COALESCE(MAX(seq),0)+1 FROM run_events WHERE run_id=?", (run_id,) ).fetchone()[0] db.execute( "INSERT INTO run_events(run_id,seq,event_type,data_json) VALUES(?,?,?,?)", (run_id, seq, event_type, json_text(data)), ) return int(seq) def append_job_event( db: sqlite3.Connection, job_id: str, event_type: str, data: dict[str, Any] ) -> int: seq = db.execute( "SELECT COALESCE(MAX(seq),0)+1 FROM job_events WHERE job_id=?", (job_id,) ).fetchone()[0] db.execute( "INSERT INTO job_events(job_id,seq,event_type,data_json) VALUES(?,?,?,?)", (job_id, seq, event_type, json_text(data)), ) return int(seq) def record_change( db: sqlite3.Connection, user_id: str, entity_type: str, entity_id: str, action: str, version: int, payload: dict[str, Any], ) -> int: cursor = db.execute( """ INSERT INTO sync_changes( user_id,entity_type,entity_id,action,version,payload_json ) VALUES(?,?,?,?,?,?) """, (user_id, entity_type, entity_id, action, int(version), json_text(payload)), ) return int(cursor.lastrowid) def conversation_value(row: sqlite3.Row) -> dict[str, Any]: return { "id": row["id"], "title": row["title"], "model_key": row["model_key"], "created_at": row["created_at"], "updated_at": row["updated_at"], "version": int(row["version"] or 1), "deleted": bool(row["deleted_at"]), } def expected_version(data: dict[str, Any]) -> int | None: value = data.get("base_version") if value is None or value == "": return None try: return int(value) except (TypeError, ValueError): return -1 @app.before_request def csrf_guard(): if request.method in {"GET", "HEAD", "OPTIONS"}: return None if request.path in {"/api/v1/setup", "/api/v1/auth/login"}: return None if not session.get("user_id"): return None expected = session.get("csrf_token") supplied = request.headers.get("X-CSRF-Token", "") if not expected or supplied != expected: return error( "KIMU_CSRF_INVALID", "The request security token is missing or invalid.", 403, ) return None @app.after_request def security_headers(response: Response): response.headers.setdefault("X-Content-Type-Options", "nosniff") response.headers.setdefault("Referrer-Policy", "same-origin") response.headers.setdefault( "Permissions-Policy", "camera=(self), microphone=(self), geolocation=()" ) response.headers.setdefault( "Content-Security-Policy", "default-src 'self'; connect-src 'self'; img-src 'self' data: blob:; " "media-src 'self' blob:; style-src 'self' 'unsafe-inline'; script-src 'self'; " "worker-src 'self' blob:; manifest-src 'self'", ) if request.path.startswith("/api/"): response.headers.setdefault("Cache-Control", "no-store") return response @app.errorhandler(413) def content_too_large(_exc): return error( "KIMU_UPLOAD_TOO_LARGE", f"The upload exceeds the configured limit of {app.config['MAX_UPLOAD_BYTES']} bytes.", 413, ) @app.get("/api/v1/health/live") def health_live(): return jsonify( { "ok": True, "service": "kimu", "version": "5.2.0", "increment": 46, } ) @app.get("/api/v1/health/ready") def health_ready(): db = get_db() integrity = db.execute("PRAGMA integrity_check").fetchone()[0] users = db.execute("SELECT COUNT(*) FROM users").fetchone()[0] storage = Path(app.config["STORAGE_DIR"]) usage = shutil.disk_usage(storage) writable = os.access(storage, os.W_OK) secret_configured = ( str(app.config["SECRET_KEY"]) != "development-only-change-me" and len(str(app.config["SECRET_KEY"])) >= 12 ) encryption_ready = artifact_store.configured and not app.config["ENCRYPTION_CONFIG_ERROR"] plaintext_files = db.execute( "SELECT COUNT(*) FROM files WHERE encryption_version=0 AND deleted_at IS NULL" ).fetchone()[0] worker_count = int(app.config.get("WORKER_COUNT", 1)) single_worker_safe = worker_count == 1 ready = ( integrity == "ok" and writable and secret_configured and encryption_ready and plaintext_files == 0 and single_worker_safe ) if not current_user(): return jsonify( { "ready": ready, "service": "kimu", "version": "5.2.0", "increment": 46, } ) return jsonify( { "ready": ready, "database": { "integrity": integrity, "wal": db.execute("PRAGMA journal_mode").fetchone()[0], "schema": db.execute( "SELECT COALESCE(MAX(version),0) FROM schema_migrations" ).fetchone()[0], }, "storage": { "writable": writable, "free_bytes": usage.free, "total_bytes": usage.total, }, "security": { "secret_key_configured": secret_configured, "secure_cookie": bool(app.config["COOKIE_SECURE"]), "artifact_encryption": encryption_ready, "master_key_version": int(app.config.get("MASTER_KEY_VERSION", 1)), "plaintext_file_count": plaintext_files, "encryption_config_error": app.config["ENCRYPTION_CONFIG_ERROR"] or None, "startup_migration": app.config["ENCRYPTION_MIGRATION"], }, "setup_required": users == 0, "runtime": { "worker_count": worker_count, "single_worker_required": bool(app.config.get("SINGLE_WORKER_REQUIRED", True)), "single_worker_safe": single_worker_safe, "reason": None if single_worker_safe else "Process-local cancellation and realtime session state require KIMU_WORKERS=1.", }, "provider": { "fake": bool(app.config["FAKE_PROVIDER"]), "key_configured": bool(app.config["TOGETHER_API_KEY"]), "k27_configured": bool(app.config["K27_CONFIGURED"]), }, "audio": { "gateway_registered": bool(app.config.get("CONVERSATION_GATEWAY_REGISTERED")), "gateway_error": app.config.get("CONVERSATION_GATEWAY_ERROR"), "stt_model": app.config["WHISPER_MODEL_ID"], "tts_model": app.config["SONIC_MODEL_ID"], "full_duplex": True, "barge_in": True, }, "capabilities": { "chat": True, "files": True, "zip_inspection": True, "vision_attachments": True, "code_interpreter": bool( app.config["FAKE_PROVIDER"] or app.config["TOGETHER_API_KEY"] ), "jobs": True, "memory": True, "skills": True, "offline_drafts": True, "offline_queue": True, "incremental_sync": True, "encrypted_artifacts": encryption_ready, "tools_runtime": True, "conversation_mode": bool(app.config.get("CONVERSATION_GATEWAY_REGISTERED")), "website_builder": True, "swarm": True, }, } ) @app.get("/api/v1/auth/status") def auth_status(): db = get_db() count = db.execute("SELECT COUNT(*) FROM users").fetchone()[0] user = current_user() return jsonify( { "setup_required": count == 0, "authenticated": bool(user), "user": dict(user) if user else None, "csrf_token": session.get("csrf_token") if user else None, } ) @app.post("/api/v1/setup") def setup_owner(): db = get_db() if db.execute("SELECT COUNT(*) FROM users").fetchone()[0]: return error( "KIMU_SETUP_COMPLETE", "Owner setup has already been completed.", 409 ) data = request.get_json(silent=True) or {} username = str(data.get("username", "")).strip() password = str(data.get("password", "")) if len(username) < 3 or len(username) > 80: return error( "KIMU_USERNAME_INVALID", "Username must contain 3 to 80 characters.", 400, ) try: encoded = hash_password(password) except ValueError as exc: return error("KIMU_PASSWORD_WEAK", str(exc), 400) user_id = uid("usr") with transaction(db): db.execute( "INSERT INTO users(id,username,password_hash,role) VALUES(?,?,?,'owner')", (user_id, username, encoded), ) audit( db, "owner.created", "user", user_id, {"username": username}, user_id=user_id, ) session.clear() session["user_id"] = user_id session["csrf_token"] = new_token() return ( jsonify( { "ok": True, "user": {"id": user_id, "username": username, "role": "owner"}, "csrf_token": session["csrf_token"], } ), 201, ) @app.post("/api/v1/auth/login") def login(): data = request.get_json(silent=True) or {} username = str(data.get("username", "")).strip() password = str(data.get("password", "")) db = get_db() user = db.execute( "SELECT * FROM users WHERE username=? COLLATE NOCASE", (username,) ).fetchone() if not user or not verify_password(password, user["password_hash"]): time.sleep(0.25) audit( db, "auth.login_failed", "user", None, {"username": username[:80]}, user_id=None, ) return error( "KIMU_LOGIN_FAILED", "The username or password is incorrect.", 401 ) session.clear() session["user_id"] = user["id"] session["csrf_token"] = new_token() audit(db, "auth.login", "user", user["id"], {}, user_id=user["id"]) return jsonify( { "ok": True, "user": { "id": user["id"], "username": user["username"], "role": user["role"], }, "csrf_token": session["csrf_token"], } ) @app.post("/api/v1/auth/logout") @require_auth def logout(): audit(get_db(), "auth.logout", "user", g.user["id"]) session.clear() return jsonify({"ok": True}) @app.get("/api/v1/models") @require_auth def models(): profiles = build_profiles(app.config) db = get_db() rows = db.execute( """ SELECT model_key,preferences_json,version FROM model_preferences WHERE user_id=? """, (g.user["id"],), ).fetchall() prefs = {row["model_key"]: row for row in rows} return jsonify( { "models": [ { **profile.public(), "preferences": normalise_preferences( profile, json.loads(prefs[model_key]["preferences_json"]) if model_key in prefs else None, ), "preference_version": ( int(prefs[model_key]["version"]) if model_key in prefs else 0 ), } for model_key, profile in profiles.items() ] } ) @app.put("/api/v1/models//preferences") @require_auth def save_preferences(model_key: str): profiles = build_profiles(app.config) if model_key not in profiles: return error("KIMU_MODEL_UNKNOWN", "Unknown model profile.", 404) data = request.get_json(silent=True) or {} prefs = normalise_preferences(profiles[model_key], data) db = get_db() current = db.execute( """ SELECT preferences_json,version FROM model_preferences WHERE user_id=? AND model_key=? """, (g.user["id"], model_key), ).fetchone() current_version = int(current["version"]) if current else 0 base_version = expected_version(data) if base_version is not None and base_version != current_version: canonical = ( normalise_preferences( profiles[model_key], json.loads(current["preferences_json"]) ) if current else normalise_preferences(profiles[model_key], None) ) return error( "KIMU_SYNC_CONFLICT", "Model settings changed on another session.", 409, details={ "entity": "model_preferences", "model_key": model_key, "base_version": base_version, "server_version": current_version, "canonical": canonical, }, ) next_version = current_version + 1 db.execute( """ INSERT INTO model_preferences( user_id,model_key,preferences_json,version ) VALUES(?,?,?,?) ON CONFLICT(user_id,model_key) DO UPDATE SET preferences_json=excluded.preferences_json, version=excluded.version, updated_at=CURRENT_TIMESTAMP """, (g.user["id"], model_key, json_text(prefs), next_version), ) payload = { "model_key": model_key, "preferences": prefs, "version": next_version, } record_change( db, g.user["id"], "model_preferences", model_key, "updated", next_version, payload ) audit( db, "model.preferences_updated", "model", model_key, {"version": next_version}, ) return jsonify({"preferences": prefs, "version": next_version}) @app.get("/api/v1/conversations") @require_auth def list_conversations(): rows = get_db().execute( """ SELECT * FROM conversations WHERE user_id=? AND deleted_at IS NULL ORDER BY updated_at DESC LIMIT 200 """, (g.user["id"],), ).fetchall() return jsonify({"conversations": [conversation_value(row) for row in rows]}) @app.post("/api/v1/conversations") @require_auth def create_conversation(): data = request.get_json(silent=True) or {} model_key = str(data.get("model_key", "k2.6")) if model_key not in build_profiles(app.config): return error("KIMU_MODEL_UNKNOWN", "Unknown model profile.", 400) client_id = str(data.get("client_id") or "").strip() conversation_id = ( client_id if client_id.startswith("cnv_local_") and len(client_id) <= 96 else uid("cnv") ) title = str(data.get("title") or "New chat").strip()[:120] or "New chat" db = get_db() existing = db.execute( "SELECT * FROM conversations WHERE id=? AND user_id=?", (conversation_id, g.user["id"]), ).fetchone() if existing: return jsonify({"conversation": conversation_value(existing), "idempotent": True}) db.execute( "INSERT INTO conversations(id,user_id,title,model_key,version) VALUES(?,?,?,?,1)", (conversation_id, g.user["id"], title, model_key), ) row = db.execute( "SELECT * FROM conversations WHERE id=?", (conversation_id,) ).fetchone() payload = conversation_value(row) record_change( db, g.user["id"], "conversation", conversation_id, "created", 1, payload ) audit( db, "conversation.created", "conversation", conversation_id, {"version": 1} ) return jsonify({"conversation": payload}), 201 @app.get("/api/v1/conversations/") @require_auth def get_conversation(conversation_id: str): db = get_db() conversation = db.execute( """ SELECT * FROM conversations WHERE id=? AND user_id=? AND deleted_at IS NULL """, (conversation_id, g.user["id"]), ).fetchone() if not conversation: return error( "KIMU_CONVERSATION_NOT_FOUND", "Conversation not found.", 404 ) rows = db.execute( """ SELECT id,role,content,reasoning,status,created_at FROM messages WHERE conversation_id=? ORDER BY created_at,rowid """, (conversation_id,), ).fetchall() return jsonify( { "conversation": conversation_value(conversation), "messages": [public_message(db, row) for row in rows], } ) @app.patch("/api/v1/conversations/") @require_auth def update_conversation(conversation_id: str): data = request.get_json(silent=True) or {} profiles = build_profiles(app.config) db = get_db() conversation = db.execute( """ SELECT * FROM conversations WHERE id=? AND user_id=? AND deleted_at IS NULL """, (conversation_id, g.user["id"]), ).fetchone() if not conversation: return error( "KIMU_CONVERSATION_NOT_FOUND", "Conversation not found.", 404 ) base_version = expected_version(data) current_version = int(conversation["version"] or 1) if base_version is not None and base_version != current_version: return error( "KIMU_SYNC_CONFLICT", "Conversation changed on another session.", 409, details={ "entity": "conversation", "entity_id": conversation_id, "base_version": base_version, "server_version": current_version, "canonical": conversation_value(conversation), }, ) title = ( str(data.get("title", conversation["title"])).strip()[:120] or conversation["title"] ) model_key = str(data.get("model_key", conversation["model_key"])) if model_key not in profiles: return error("KIMU_MODEL_UNKNOWN", "Unknown model profile.", 400) next_version = current_version + 1 db.execute( """ UPDATE conversations SET title=?,model_key=?,version=?,updated_at=CURRENT_TIMESTAMP WHERE id=? """, (title, model_key, next_version, conversation_id), ) row = db.execute( "SELECT * FROM conversations WHERE id=?", (conversation_id,) ).fetchone() payload = conversation_value(row) record_change( db, g.user["id"], "conversation", conversation_id, "updated", next_version, payload, ) audit( db, "conversation.updated", "conversation", conversation_id, {"model_key": model_key, "version": next_version}, ) return jsonify({"conversation": payload}) @app.delete("/api/v1/conversations/") @require_auth def delete_conversation(conversation_id: str): data = request.get_json(silent=True) or {} db = get_db() conversation = db.execute( """ SELECT * FROM conversations WHERE id=? AND user_id=? AND deleted_at IS NULL """, (conversation_id, g.user["id"]), ).fetchone() if not conversation: return error( "KIMU_CONVERSATION_NOT_FOUND", "Conversation not found.", 404 ) base_version = expected_version(data) current_version = int(conversation["version"] or 1) if base_version is not None and base_version != current_version: return error( "KIMU_SYNC_CONFLICT", "Conversation changed before it could be archived.", 409, details={ "entity": "conversation", "entity_id": conversation_id, "base_version": base_version, "server_version": current_version, "canonical": conversation_value(conversation), }, ) next_version = current_version + 1 db.execute( """ UPDATE conversations SET deleted_at=CURRENT_TIMESTAMP,version=?, updated_at=CURRENT_TIMESTAMP WHERE id=? """, (next_version, conversation_id), ) payload = { **conversation_value(conversation), "version": next_version, "deleted": True, } record_change( db, g.user["id"], "conversation", conversation_id, "deleted", next_version, payload, ) audit( db, "conversation.archived", "conversation", conversation_id, {"version": next_version}, ) return jsonify({"ok": True, "version": next_version}) @app.get("/api/v1/conversations//export.txt") @require_auth def export_conversation(conversation_id: str): db = get_db() conversation = db.execute( """ SELECT * FROM conversations WHERE id=? AND user_id=? AND deleted_at IS NULL """, (conversation_id, g.user["id"]), ).fetchone() if not conversation: return error( "KIMU_CONVERSATION_NOT_FOUND", "Conversation not found.", 404 ) rows = db.execute( """ SELECT id,role,content,reasoning,created_at FROM messages WHERE conversation_id=? ORDER BY created_at,rowid """, (conversation_id,), ).fetchall() parts = [f"KIMU conversation: {conversation['title']}\n"] for row in rows: parts.append(f"\n[{row['created_at']}] {row['role'].upper()}\n") if row["reasoning"]: parts.append(f"Reasoning:\n{row['reasoning']}\n") parts.append(row["content"] + "\n") attachments = message_attachments(db, row["id"]) if attachments: parts.append( "Attachments: " + ", ".join( f"{item['name']} ({item['sha256'][:12]}…)" for item in attachments ) + "\n" ) audit(db, "conversation.exported", "conversation", conversation_id) filename = f"kimu-{conversation_id[:20]}.txt" return Response( "".join(parts), mimetype="text/plain", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) @app.get("/api/v1/files") @require_auth def list_files(): rows = get_db().execute( """ SELECT * FROM files WHERE user_id=? AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 500 """, (g.user["id"],), ).fetchall() return jsonify({"files": [public_file(row) for row in rows]}) @app.post("/api/v1/files") @require_auth def upload_files(): try: artifact_store.require_key() except EncryptionError as exc: return error(exc.code, str(exc), 503) uploads = request.files.getlist("files") if not uploads and "file" in request.files: uploads = [request.files["file"]] uploads = [item for item in uploads if item and item.filename] if not uploads: return error("KIMU_FILE_REQUIRED", "Choose at least one file.", 400) if len(uploads) > int(app.config["MAX_ATTACHMENTS_PER_MESSAGE"]): return error( "KIMU_TOO_MANY_FILES", f"Upload at most {app.config['MAX_ATTACHMENTS_PER_MESSAGE']} files at once.", 400, ) db = get_db() storage = Path(app.config["STORAGE_DIR"]) created: list[dict[str, Any]] = [] staged_paths: list[Path] = [] committed_paths: list[Path] = [] try: with transaction(db): for upload in uploads: file_id = uid("fil") staged = storage / f".{file_id}.upload" upload.save(staged) staged_paths.append(staged) if staged.stat().st_size > int(app.config["MAX_UPLOAD_BYTES"]): raise FileValidationError( "KIMU_UPLOAD_TOO_LARGE", "The file exceeds the configured upload limit.", ) inspection = inspect_stored_file( staged, upload.filename or "upload", upload.mimetype ) values = inspection.database_values() stored_name = file_id + ".kimuenc" final_path = storage / stored_name metadata_cipher = artifact_store.encode_private_metadata( file_id, values["sha256"], values["text_content"], values["zip_inventory_json"], values["warnings_json"], ) stored_size = artifact_store.encrypt_staged( staged, final_path, file_id, values["sha256"] ) committed_paths.append(final_path) staged.unlink(missing_ok=True) staged_paths.remove(staged) db.execute( """ INSERT INTO files( id,user_id,original_name,stored_name,mime_type,size_bytes, sha256,kind,status,text_content,zip_inventory_json,warnings_json, private_metadata_cipher,encryption_version,key_version,stored_size_bytes ) VALUES(?,?,?,?,?,?,?,?,'ready','','[]','[]',?,?,?,?) """, ( file_id, g.user["id"], values["original_name"], stored_name, values["mime_type"], values["size_bytes"], values["sha256"], values["kind"], metadata_cipher, ENCRYPTION_VERSION, int(app.config["MASTER_KEY_VERSION"]), stored_size, ), ) row = db.execute("SELECT * FROM files WHERE id=?", (file_id,)).fetchone() created.append(public_file(row, include_inventory=True)) audit( db, "file.uploaded", "file", file_id, { "name": values["original_name"], "size_bytes": values["size_bytes"], "sha256": values["sha256"], "kind": values["kind"], "encrypted": True, "key_version": int(app.config["MASTER_KEY_VERSION"]), }, ) db.execute( """ INSERT INTO sync_changes( user_id,entity_type,entity_id,action,version,payload_json ) VALUES(?, 'file', ?, 'created', 1, ?) """, (g.user["id"], file_id, json_text(created[-1])), ) except FileValidationError as exc: for path in staged_paths + committed_paths: path.unlink(missing_ok=True) return error(exc.code, str(exc), 400) except EncryptionError as exc: for path in staged_paths + committed_paths: path.unlink(missing_ok=True) return error(exc.code, str(exc), 503) except (OSError, sqlite3.DatabaseError): for path in staged_paths + committed_paths: path.unlink(missing_ok=True) return error( "KIMU_FILE_STORAGE_ERROR", "The file could not be committed to encrypted private storage.", 500, True, ) return jsonify({"files": created}), 201 @app.get("/api/v1/files/") @require_auth def get_file(file_id: str): row = get_db().execute( """ SELECT * FROM files WHERE id=? AND user_id=? AND deleted_at IS NULL """, (file_id, g.user["id"]), ).fetchone() if not row: return error("KIMU_FILE_NOT_FOUND", "File not found.", 404) return jsonify({"file": public_file(row, include_inventory=True)}) @app.get("/api/v1/files//download") @require_auth def download_file(file_id: str): db = get_db() row = db.execute( """ SELECT * FROM files WHERE id=? AND user_id=? AND deleted_at IS NULL """, (file_id, g.user["id"]), ).fetchone() if not row: return error("KIMU_FILE_NOT_FOUND", "File not found.", 404) path = artifact_store.body_path(row) if not path.is_file(): return error( "KIMU_FILE_BODY_MISSING", "The file metadata exists but the stored body is missing.", 500, ) try: if int(row["encryption_version"] or 0) > 0: artifact_store.require_key() except EncryptionError as exc: return error(exc.code, str(exc), 503) audit(db, "file.downloaded", "file", file_id, {"decrypted_in_stream": True}) ascii_name = "".join( character if 32 <= ord(character) < 127 and character not in {'"', '\\'} else "_" for character in row["original_name"] ) or "download" headers = { "Content-Disposition": ( f'attachment; filename="{ascii_name}"; ' f"filename*=UTF-8''{quote(row['original_name'])}" ), "Content-Length": str(row["size_bytes"]), "Accept-Ranges": "none", "X-KIMU-Encrypted-At-Rest": "1" if int(row["encryption_version"] or 0) else "0", } return Response( stream_with_context(artifact_store.iter_body(row)), mimetype=row["mime_type"], headers=headers, direct_passthrough=True, ) @app.delete("/api/v1/files/") @require_auth def delete_file(file_id: str): db = get_db() row = db.execute( """ SELECT * FROM files WHERE id=? AND user_id=? AND deleted_at IS NULL """, (file_id, g.user["id"]), ).fetchone() if not row: return error("KIMU_FILE_NOT_FOUND", "File not found.", 404) db.execute( """ UPDATE files SET status='deleted',deleted_at=CURRENT_TIMESTAMP WHERE id=? """, (file_id,), ) record_change( db, g.user["id"], "file", file_id, "deleted", 2, {"id": file_id, "name": row["original_name"], "deleted": True}, ) audit(db, "file.deleted", "file", file_id, {"name": row["original_name"]}) return jsonify({"ok": True}) def provider_message( db: sqlite3.Connection, row: sqlite3.Row, storage: Path, model_key: str ) -> dict[str, Any]: attachments = db.execute( """ SELECT f.* FROM message_attachments ma JOIN files f ON f.id=ma.file_id WHERE ma.message_id=? AND f.status='ready' ORDER BY ma.position """, (row["id"],), ).fetchall() if not attachments: item: dict[str, Any] = {"role": row["role"], "content": row["content"]} else: blocks: list[dict[str, Any]] = [ {"type": "text", "text": row["content"] or "Review the attachments."} ] for file_row in attachments: path = artifact_store.body_path(file_row) if not path.is_file(): continue private = artifact_store.private_metadata(file_row) body = ( artifact_store.body_bytes(file_row, MAX_IMAGE_INLINE) if file_row["kind"] == "image" and int(file_row["size_bytes"]) <= MAX_IMAGE_INLINE else None ) blocks.append( attachment_content( file_row, body_bytes=body, private_metadata=private ) ) item = {"role": row["role"], "content": blocks} if ( model_key == "k2.7-code" and row["role"] == "assistant" and row["reasoning"] ): item["reasoning_content"] = row["reasoning"] return item def run_worker( run_id: str, user_text: str, model_key: str, preferences: dict[str, Any], profile_id: str, ): db = connect(app.config["DATABASE"]) content = "" reasoning = "" cancel = RUN_CANCEL.get(run_id) or threading.Event() run = None try: run = db.execute("SELECT * FROM runs WHERE id=?", (run_id,)).fetchone() if not run: return def record_assistant_change() -> None: message_row = db.execute( """ SELECT id,role,content,reasoning,status,created_at FROM messages WHERE id=? """, (run["assistant_message_id"],), ).fetchone() if message_row: record_change( db, run["user_id"], "message", run["assistant_message_id"], "updated", 2, public_message(db, message_row), ) rows = db.execute( """ SELECT * FROM ( SELECT id,role,content,reasoning,status,created_at,rowid AS message_order FROM messages WHERE conversation_id=? AND status IN ('complete','queued') ORDER BY created_at DESC,rowid DESC LIMIT ? ) ORDER BY created_at,message_order """, ( run["conversation_id"], app.config["MAX_CONTEXT_MESSAGES"], ), ).fetchall() messages = [ provider_message( db, row, Path(app.config["STORAGE_DIR"]), model_key ) for row in rows if row["role"] != "assistant" or row["id"] != run["assistant_message_id"] ] payload = build_payload(profile_id, messages, preferences, model_key) def active_response(response) -> None: with RUN_LOCK: if response is None: RUN_RESPONSES.pop(run_id, None) else: RUN_RESPONSES[run_id] = response iterator = ( fake_stream(model_key, user_text) if app.config["FAKE_PROVIDER"] else stream_together( app.config["TOGETHER_BASE_URL"], app.config["TOGETHER_API_KEY"], payload, cancel_event=cancel, response_callback=active_response, ) ) db.execute( "UPDATE runs SET status='streaming',updated_at=CURRENT_TIMESTAMP WHERE id=?", (run_id,), ) append_run_event(db, run_id, "status", {"status": "streaming"}) for chunk in iterator: cancelled = cancel.is_set() or db.execute( "SELECT cancel_requested FROM runs WHERE id=?", (run_id,) ).fetchone()[0] if cancelled: db.execute( "UPDATE runs SET status='cancelled',updated_at=CURRENT_TIMESTAMP WHERE id=?", (run_id,), ) db.execute( """ UPDATE messages SET content=?,reasoning=?,status='cancelled' WHERE id=? """, (content, reasoning, run["assistant_message_id"]), ) append_run_event( db, run_id, "cancelled", {"status": "cancelled"} ) record_assistant_change() return if chunk["type"] == "content": content += chunk["text"] append_run_event( db, run_id, "content", {"delta": chunk["text"]} ) elif chunk["type"] == "reasoning": reasoning += chunk["text"] append_run_event( db, run_id, "reasoning", {"delta": chunk["text"]} ) elif chunk["type"] == "finish": append_run_event( db, run_id, "finish", {"reason": chunk["text"]} ) db.execute( """ UPDATE messages SET content=?,reasoning=?,status='streaming' WHERE id=? """, (content, reasoning, run["assistant_message_id"]), ) db.execute( """ UPDATE messages SET content=?,reasoning=?,status='complete' WHERE id=? """, (content.strip(), reasoning.strip(), run["assistant_message_id"]), ) db.execute( "UPDATE runs SET status='complete',updated_at=CURRENT_TIMESTAMP WHERE id=?", (run_id,), ) append_run_event(db, run_id, "done", {"status": "complete"}) record_assistant_change() audit( db, "chat.completed", "run", run_id, {"model_key": model_key}, user_id=run["user_id"], ) except ProviderError as exc: status = "partial" if content else "failed" if run: db.execute( """ UPDATE messages SET content=?,reasoning=?,status=? WHERE id=? """, (content, reasoning, status, run["assistant_message_id"]), ) db.execute( """ UPDATE runs SET status=?,error_code=?,error_message=?,updated_at=CURRENT_TIMESTAMP WHERE id=? """, (status, exc.code, str(exc), run_id), ) append_run_event( db, run_id, "error", { "code": exc.code, "message": str(exc), "retryable": exc.retryable, }, ) if run: record_assistant_change() except Exception: if run: db.execute( """ UPDATE messages SET content=?,reasoning=?,status='failed' WHERE id=? """, (content, reasoning, run["assistant_message_id"]), ) db.execute( """ UPDATE runs SET status='failed',error_code='KIMU_INTERNAL_ERROR', error_message='The chat run failed.',updated_at=CURRENT_TIMESTAMP WHERE id=? """, (run_id,), ) append_run_event( db, run_id, "error", { "code": "KIMU_INTERNAL_ERROR", "message": "The chat run failed.", "retryable": False, }, ) if run: record_assistant_change() finally: db.close() with RUN_LOCK: RUN_CANCEL.pop(run_id, None) response = RUN_RESPONSES.pop(run_id, None) if response is not None: try: response.close() except Exception: pass def create_chat_run_value( db: sqlite3.Connection, user_id: str, data: dict[str, Any] ) -> tuple[dict[str, Any] | None, tuple[Any, int] | None, bool]: operation_id = str(data.get("client_operation_id") or "").strip()[:160] if operation_id: existing_operation = db.execute( """ SELECT * FROM sync_operations WHERE user_id=? AND id=? """, (user_id, operation_id), ).fetchone() if existing_operation: if existing_operation["status"] == "applied": return json.loads(existing_operation["result_json"]), None, True saved_error = json.loads(existing_operation["error_json"] or "{}") return ( None, error( saved_error.get("code", "KIMU_SYNC_OPERATION_FAILED"), saved_error.get("message", "The queued operation failed."), 409 if existing_operation["status"] == "conflict" else 400, details=saved_error.get("details") or {}, ), True, ) conversation_id = str(data.get("conversation_id", "")) text = str(data.get("content", "")).strip() attachment_ids = [ str(value) for value in (data.get("attachment_ids") or []) if value ] if not text and not attachment_ids: return ( None, error( "KIMU_MESSAGE_EMPTY", "Enter a message or add an attachment before sending.", 400, ), False, ) if len(text) > app.config["MAX_MESSAGE_CHARS"]: return ( None, error( "KIMU_MESSAGE_TOO_LARGE", f"Message exceeds {app.config['MAX_MESSAGE_CHARS']} characters.", 413, ), False, ) if len(attachment_ids) > app.config["MAX_ATTACHMENTS_PER_MESSAGE"]: return ( None, error( "KIMU_TOO_MANY_ATTACHMENTS", f"Attach at most {app.config['MAX_ATTACHMENTS_PER_MESSAGE']} files.", 400, ), False, ) conversation = db.execute( """ SELECT * FROM conversations WHERE id=? AND user_id=? AND deleted_at IS NULL """, (conversation_id, user_id), ).fetchone() if not conversation and bool(data.get("create_if_missing")): requested_model = str(data.get("model_key") or "k2.6") if requested_model not in build_profiles(app.config): return None, error("KIMU_MODEL_UNKNOWN", "Unknown model profile.", 400), False if not conversation_id.startswith("cnv_local_") or len(conversation_id) > 96: return ( None, error( "KIMU_OFFLINE_CONVERSATION_ID_INVALID", "The queued conversation identifier is invalid.", 400, ), False, ) requested_title = ( str(data.get("conversation_title") or text or "New chat").strip()[:120] or "New chat" ) db.execute( """ INSERT INTO conversations(id,user_id,title,model_key,version) VALUES(?,?,?,?,1) """, (conversation_id, user_id, requested_title, requested_model), ) conversation = db.execute( "SELECT * FROM conversations WHERE id=?", (conversation_id,) ).fetchone() record_change( db, user_id, "conversation", conversation_id, "created", 1, conversation_value(conversation), ) audit( db, "conversation.created_offline", "conversation", conversation_id, {"operation_id": operation_id or None}, user_id=user_id, ) if not conversation: return ( None, error( "KIMU_CONVERSATION_NOT_FOUND", "Conversation not found.", 404 ), False, ) profiles = build_profiles(app.config) profile = profiles[conversation["model_key"]] if not profile.configured: return ( None, error( "KIMU_MODEL_NOT_CONFIGURED", f"{profile.display_name} is not configured for this installation.", 409, details={ "model_key": profile.key, "deployment": profile.deployment, }, ), False, ) file_rows: list[sqlite3.Row] = [] if attachment_ids: placeholders = ",".join("?" for _ in attachment_ids) file_rows = db.execute( f""" SELECT * FROM files WHERE user_id=? AND deleted_at IS NULL AND status='ready' AND id IN ({placeholders}) """, (user_id, *attachment_ids), ).fetchall() if len(file_rows) != len(set(attachment_ids)): return ( None, error( "KIMU_ATTACHMENT_INVALID", "One or more attachments are unavailable.", 400, ), False, ) if any(row["kind"] == "image" for row in file_rows) and not profile.supports_vision: return ( None, error( "KIMU_MODEL_NO_VISION", f"{profile.display_name} does not support image attachments.", 409, ), False, ) pref_row = db.execute( """ SELECT preferences_json FROM model_preferences WHERE user_id=? AND model_key=? """, (user_id, profile.key), ).fetchone() prefs = normalise_preferences( profile, json.loads(pref_row[0]) if pref_row else None ) user_message_id = uid("msg") assistant_message_id = uid("msg") run_id = uid("run") title = conversation["title"] title_source = text or (file_rows[0]["original_name"] if file_rows else "New chat") if title == "New chat": title = title_source[:72] + ("…" if len(title_source) > 72 else "") next_conversation_version = int(conversation["version"] or 1) + 1 with transaction(db): db.execute( """ INSERT INTO messages(id,conversation_id,role,content,status) VALUES(?,?, 'user',?,'complete') """, (user_message_id, conversation_id, text), ) for position, file_id in enumerate(attachment_ids): db.execute( """ INSERT INTO message_attachments(message_id,file_id,position) VALUES(?,?,?) """, (user_message_id, file_id, position), ) db.execute( """ INSERT INTO messages(id,conversation_id,role,status) VALUES(?,?, 'assistant','queued') """, (assistant_message_id, conversation_id), ) db.execute( """ INSERT INTO runs( id,user_id,conversation_id,assistant_message_id,model_key,status ) VALUES(?,?,?,?,?,'queued') """, (run_id, user_id, conversation_id, assistant_message_id, profile.key), ) db.execute( """ UPDATE conversations SET title=?,version=?,updated_at=CURRENT_TIMESTAMP WHERE id=? """, (title, next_conversation_version, conversation_id), ) append_run_event(db, run_id, "status", {"status": "queued"}) audit( db, "chat.started", "run", run_id, { "model_key": profile.key, "attachment_count": len(attachment_ids), "operation_id": operation_id or None, }, user_id=user_id, ) user_row = db.execute( """ SELECT id,role,content,reasoning,status,created_at FROM messages WHERE id=? """, (user_message_id,), ).fetchone() assistant_payload = { "id": assistant_message_id, "role": "assistant", "content": "", "reasoning": "", "status": "queued", "attachments": [], } result = { "run": { "id": run_id, "status": "queued", "assistant_message_id": assistant_message_id, }, "user_message": public_message(db, user_row), "assistant_message": assistant_payload, "title": title, "conversation_id": conversation_id, "conversation_version": next_conversation_version, } updated_conversation = db.execute( "SELECT * FROM conversations WHERE id=?", (conversation_id,) ).fetchone() record_change( db, user_id, "conversation", conversation_id, "updated", next_conversation_version, conversation_value(updated_conversation), ) record_change( db, user_id, "message", user_message_id, "created", 1, result["user_message"] ) record_change( db, user_id, "message", assistant_message_id, "created", 1, assistant_payload ) if operation_id: db.execute( """ INSERT INTO sync_operations( id,user_id,operation_type,status,request_json,result_json ) VALUES(?,?,'chat.send','applied',?,?) """, (operation_id, user_id, json_text(data), json_text(result)), ) cancel = threading.Event() with RUN_LOCK: RUN_CANCEL[run_id] = cancel threading.Thread( target=run_worker, args=(run_id, text, profile.key, prefs, profile.model_id), daemon=True, name=f"kimu-chat-{run_id[-8:]}", ).start() return result, None, False @app.post("/api/v1/chat/runs") @require_auth def create_run(): data = request.get_json(silent=True) or {} result, failure, idempotent = create_chat_run_value( get_db(), g.user["id"], data ) if failure is not None: return failure return jsonify({**result, "idempotent": idempotent}), 202 @app.get("/api/v1/chat/runs//events") @require_auth def run_events(run_id: str): owner = get_db().execute( "SELECT id FROM runs WHERE id=? AND user_id=?", (run_id, g.user["id"]), ).fetchone() if not owner: return error("KIMU_RUN_NOT_FOUND", "Chat run not found.", 404) try: last = int( request.args.get( "after", request.headers.get("Last-Event-ID", "0") ) or 0 ) except ValueError: last = 0 database = app.config["DATABASE"] @stream_with_context def generate(): nonlocal last db = connect(database) idle = 0 try: while idle < 600: rows = db.execute( """ SELECT seq,event_type,data_json FROM run_events WHERE run_id=? AND seq>? ORDER BY seq """, (run_id, last), ).fetchall() if rows: idle = 0 for row in rows: last = row["seq"] yield ( f"id: {last}\n" f"event: {row['event_type']}\n" f"data: {row['data_json']}\n\n" ) else: idle += 1 yield ": keepalive\n\n" status_row = db.execute( "SELECT status FROM runs WHERE id=?", (run_id,) ).fetchone() if not status_row: break if status_row[0] in CHAT_TERMINAL and not db.execute( "SELECT 1 FROM run_events WHERE run_id=? AND seq>?", (run_id, last), ).fetchone(): break time.sleep(0.5) finally: db.close() return Response( generate(), mimetype="text/event-stream", headers={ "X-Accel-Buffering": "no", "Cache-Control": "no-cache, no-transform", }, ) @app.post("/api/v1/chat/runs//cancel") @require_auth def cancel_run(run_id: str): db = get_db() cursor = db.execute( """ UPDATE runs SET cancel_requested=1,updated_at=CURRENT_TIMESTAMP WHERE id=? AND user_id=? AND status NOT IN ('complete','failed','cancelled','partial') """, (run_id, g.user["id"]), ) with RUN_LOCK: if run_id in RUN_CANCEL: RUN_CANCEL[run_id].set() response = RUN_RESPONSES.pop(run_id, None) if response is not None: try: response.close() except Exception: pass if cursor.rowcount: audit(db, "chat.cancel_requested", "run", run_id) return jsonify({"ok": bool(cursor.rowcount)}) @app.get("/api/v1/jobs") @require_auth def list_jobs(): rows = get_db().execute( """ SELECT * FROM jobs WHERE user_id=? ORDER BY updated_at DESC LIMIT 200 """, (g.user["id"],), ).fetchall() return jsonify({"jobs": [public_job(row) for row in rows]}) @app.get("/api/v1/jobs/") @require_auth def get_job(job_id: str): row = get_db().execute( "SELECT * FROM jobs WHERE id=? AND user_id=?", (job_id, g.user["id"]), ).fetchone() if not row: return error("KIMU_JOB_NOT_FOUND", "Task not found.", 404) return jsonify({"job": public_job(row)}) @app.get("/api/v1/jobs//events") @require_auth def job_events(job_id: str): owner = get_db().execute( "SELECT id FROM jobs WHERE id=? AND user_id=?", (job_id, g.user["id"]), ).fetchone() if not owner: return error("KIMU_JOB_NOT_FOUND", "Task not found.", 404) try: last = int(request.args.get("after", "0") or 0) except ValueError: last = 0 database = app.config["DATABASE"] @stream_with_context def generate(): nonlocal last db = connect(database) idle = 0 try: while idle < 600: rows = db.execute( """ SELECT seq,event_type,data_json FROM job_events WHERE job_id=? AND seq>? ORDER BY seq """, (job_id, last), ).fetchall() if rows: idle = 0 for row in rows: last = row["seq"] yield ( f"id: {last}\n" f"event: {row['event_type']}\n" f"data: {row['data_json']}\n\n" ) else: idle += 1 yield ": keepalive\n\n" status_row = db.execute( "SELECT status FROM jobs WHERE id=?", (job_id,) ).fetchone() if not status_row: break if status_row[0] in JOB_TERMINAL and not db.execute( "SELECT 1 FROM job_events WHERE job_id=? AND seq>?", (job_id, last), ).fetchone(): break time.sleep(0.5) finally: db.close() return Response( generate(), mimetype="text/event-stream", headers={ "X-Accel-Buffering": "no", "Cache-Control": "no-cache, no-transform", }, ) @app.post("/api/v1/jobs//cancel") @require_auth def cancel_job(job_id: str): db = get_db() cursor = db.execute( """ UPDATE jobs SET cancel_requested=1,updated_at=CURRENT_TIMESTAMP WHERE id=? AND user_id=? AND status NOT IN ('successful','failed','cancelled','partial','blocked') """, (job_id, g.user["id"]), ) with JOB_LOCK: if job_id in JOB_CANCEL: JOB_CANCEL[job_id].set() if cursor.rowcount: audit(db, "job.cancel_requested", "job", job_id) return jsonify({"ok": bool(cursor.rowcount)}) def code_worker( job_id: str, user_id: str, code: str, supplied_session: str | None, file_ids: list[str], ) -> None: db = connect(app.config["DATABASE"]) cancel = JOB_CANCEL.get(job_id) or threading.Event() try: db.execute( """ UPDATE jobs SET status='running',progress=10, updated_at=CURRENT_TIMESTAMP WHERE id=? """, (job_id,), ) append_job_event( db, job_id, "status", {"status": "running", "progress": 10} ) if cancel.is_set(): raise InterruptedError attached_files: list[dict[str, Any]] = [] if file_ids: placeholders = ",".join("?" for _ in file_ids) rows = db.execute( f""" SELECT * FROM files WHERE user_id=? AND status='ready' AND id IN ({placeholders}) """, (user_id, *file_ids), ).fetchall() for row in rows: private = artifact_store.private_metadata(row) text = str(private.get("text_content") or "") if text: attached_files.append( { "name": row["original_name"], "encoding": "string", "content": text[:2 * 1024 * 1024], } ) session_id = supplied_session if not session_id: row = db.execute( "SELECT session_id FROM tci_sessions WHERE user_id=?", (user_id,), ).fetchone() session_id = row["session_id"] if row else None db.execute( "UPDATE jobs SET progress=30,updated_at=CURRENT_TIMESTAMP WHERE id=?", (job_id,), ) append_job_event(db, job_id, "progress", {"progress": 30}) result = ( fake_execute(code, session_id) if app.config["FAKE_PROVIDER"] else execute_tci( app.config["TOGETHER_BASE_URL"], app.config["TOGETHER_API_KEY"], code, session_id=session_id, files=attached_files, ) ) if cancel.is_set() or db.execute( "SELECT cancel_requested FROM jobs WHERE id=?", (job_id,) ).fetchone()[0]: raise InterruptedError if result.get("session_id"): db.execute( """ INSERT INTO tci_sessions(user_id,session_id) VALUES(?,?) ON CONFLICT(user_id) DO UPDATE SET session_id=excluded.session_id, updated_at=CURRENT_TIMESTAMP """, (user_id, result["session_id"]), ) db.execute( """ UPDATE jobs SET status='successful',progress=100,result_json=?, updated_at=CURRENT_TIMESTAMP WHERE id=? """, (json_text(result), job_id), ) append_job_event( db, job_id, "done", {"status": "successful", "progress": 100, "result": result}, ) audit( db, "code_interpreter.completed", "job", job_id, {"session_id": result.get("session_id")}, user_id=user_id, ) except InterruptedError: db.execute( """ UPDATE jobs SET status='cancelled',updated_at=CURRENT_TIMESTAMP WHERE id=? """, (job_id,), ) append_job_event(db, job_id, "cancelled", {"status": "cancelled"}) except TCIError as exc: db.execute( """ UPDATE jobs SET status='failed',error_code=?,error_message=?, updated_at=CURRENT_TIMESTAMP WHERE id=? """, (exc.code, str(exc), job_id), ) append_job_event( db, job_id, "error", { "code": exc.code, "message": str(exc), "retryable": exc.retryable, }, ) except Exception: db.execute( """ UPDATE jobs SET status='failed', error_code='KIMU_CODE_INTERNAL_ERROR', error_message='Code Interpreter failed.', updated_at=CURRENT_TIMESTAMP WHERE id=? """, (job_id,), ) append_job_event( db, job_id, "error", { "code": "KIMU_CODE_INTERNAL_ERROR", "message": "Code Interpreter failed.", "retryable": False, }, ) finally: db.close() with JOB_LOCK: JOB_CANCEL.pop(job_id, None) @app.post("/api/v1/code/runs") @require_auth def create_code_run(): data = request.get_json(silent=True) or {} code = str(data.get("code", "")) if not code.strip(): return error("KIMU_CODE_EMPTY", "Enter Python code to run.", 400) if len(code) > 250000: return error( "KIMU_CODE_TOO_LARGE", "The code exceeds the 250,000-character limit.", 413, ) file_ids = [str(value) for value in data.get("file_ids", []) if value] session_id = str(data.get("session_id") or "").strip() or None if not app.config["FAKE_PROVIDER"] and not app.config["TOGETHER_API_KEY"]: return error( "KIMU_PROVIDER_KEY_MISSING", "Together API key is required for Code Interpreter.", 409, ) db = get_db() if file_ids: placeholders = ",".join("?" for _ in file_ids) count = db.execute( f""" SELECT COUNT(*) FROM files WHERE user_id=? AND status='ready' AND id IN ({placeholders}) """, (g.user["id"], *file_ids), ).fetchone()[0] if count != len(set(file_ids)): return error( "KIMU_ATTACHMENT_INVALID", "One or more Code Interpreter files are unavailable.", 400, ) job_id = uid("job") db.execute( """ INSERT INTO jobs(id,user_id,kind,title,status,progress,input_json) VALUES(?,?,'code-interpreter','Python execution','queued',0,?) """, ( job_id, g.user["id"], json_text( { "code_chars": len(code), "file_ids": file_ids, "session_id": session_id, } ), ), ) append_job_event( db, job_id, "status", {"status": "queued", "progress": 0} ) audit( db, "code_interpreter.started", "job", job_id, {"code_chars": len(code), "file_count": len(file_ids)}, ) cancel = threading.Event() with JOB_LOCK: JOB_CANCEL[job_id] = cancel threading.Thread( target=code_worker, args=(job_id, g.user["id"], code, session_id, file_ids), daemon=True, name=f"kimu-code-{job_id[-8:]}", ).start() row = db.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone() return jsonify({"job": public_job(row)}), 202 @app.delete("/api/v1/code/session") @require_auth def clear_code_session(): db = get_db() db.execute("DELETE FROM tci_sessions WHERE user_id=?", (g.user["id"],)) audit(db, "code_interpreter.session_cleared", "user", g.user["id"]) return jsonify({"ok": True}) @app.get("/api/v1/memories") @require_auth def list_memories(): rows = get_db().execute( """ SELECT * FROM memories WHERE user_id=? ORDER BY updated_at DESC LIMIT 500 """, (g.user["id"],), ).fetchall() return jsonify({"memories": [dict(row) for row in rows]}) @app.post("/api/v1/memories") @require_auth def create_memory(): data = request.get_json(silent=True) or {} title = str(data.get("title", "")).strip()[:160] content = str(data.get("content", "")).strip() scope = str(data.get("scope", "global")) if not title or not content: return error( "KIMU_MEMORY_INVALID", "Memory title and content are required.", 400 ) if scope not in {"global", "workspace", "temporary"}: return error("KIMU_MEMORY_SCOPE_INVALID", "Invalid memory scope.", 400) memory_id = uid("mem") db = get_db() db.execute( """ INSERT INTO memories(id,user_id,scope,title,content,source,expires_at) VALUES(?,?,?,?,?,'manual',?) """, ( memory_id, g.user["id"], scope, title, content, data.get("expires_at"), ), ) audit(db, "memory.created", "memory", memory_id, {"scope": scope}) row = db.execute( "SELECT * FROM memories WHERE id=?", (memory_id,) ).fetchone() return jsonify({"memory": dict(row)}), 201 @app.patch("/api/v1/memories/") @require_auth def update_memory(memory_id: str): db = get_db() row = db.execute( "SELECT * FROM memories WHERE id=? AND user_id=?", (memory_id, g.user["id"]), ).fetchone() if not row: return error("KIMU_MEMORY_NOT_FOUND", "Memory not found.", 404) data = request.get_json(silent=True) or {} scope = str(data.get("scope", row["scope"])) if scope not in {"global", "workspace", "temporary"}: return error("KIMU_MEMORY_SCOPE_INVALID", "Invalid memory scope.", 400) title = str(data.get("title", row["title"])).strip()[:160] or row["title"] content = str(data.get("content", row["content"])).strip() enabled = 1 if bool(data.get("enabled", row["enabled"])) else 0 db.execute( """ UPDATE memories SET scope=?,title=?,content=?,enabled=?,expires_at=?, updated_at=CURRENT_TIMESTAMP WHERE id=? """, ( scope, title, content, enabled, data.get("expires_at", row["expires_at"]), memory_id, ), ) audit(db, "memory.updated", "memory", memory_id) updated = db.execute( "SELECT * FROM memories WHERE id=?", (memory_id,) ).fetchone() return jsonify({"memory": dict(updated)}) @app.delete("/api/v1/memories/") @require_auth def delete_memory(memory_id: str): db = get_db() cursor = db.execute( "DELETE FROM memories WHERE id=? AND user_id=?", (memory_id, g.user["id"]), ) if not cursor.rowcount: return error("KIMU_MEMORY_NOT_FOUND", "Memory not found.", 404) audit(db, "memory.deleted", "memory", memory_id) return jsonify({"ok": True}) @app.get("/api/v1/skills") @require_auth def list_skills(): rows = get_db().execute( """ SELECT * FROM skills WHERE user_id=? ORDER BY updated_at DESC LIMIT 500 """, (g.user["id"],), ).fetchall() values = [] for row in rows: item = dict(row) item["input_schema"] = json.loads(item.pop("input_schema_json") or "{}") item["permitted_tools"] = json.loads( item.pop("permitted_tools_json") or "[]" ) values.append(item) return jsonify({"skills": values}) @app.post("/api/v1/skills") @require_auth def create_skill(): data = request.get_json(silent=True) or {} name = str(data.get("name", "")).strip()[:120] instructions = str(data.get("instructions", "")).strip() if not name or not instructions: return error( "KIMU_SKILL_INVALID", "Skill name and instructions are required.", 400, ) input_schema = data.get("input_schema") or {} permitted_tools = data.get("permitted_tools") or [] if not isinstance(input_schema, dict) or not isinstance( permitted_tools, list ): return error( "KIMU_SKILL_SCHEMA_INVALID", "Skill schema or tool list is invalid.", 400, ) skill_id = uid("skl") db = get_db() try: db.execute( """ INSERT INTO skills( id,user_id,name,description,instructions,input_schema_json, output_format,permitted_tools_json,risk_boundaries,status ) VALUES(?,?,?,?,?,?,?,?,?,'draft') """, ( skill_id, g.user["id"], name, str(data.get("description", "")).strip(), instructions, json_text(input_schema), str(data.get("output_format", "text"))[:80], json_text(permitted_tools), str(data.get("risk_boundaries", "")).strip(), ), ) except sqlite3.IntegrityError: return error( "KIMU_SKILL_NAME_CONFLICT", "A skill with that name and version already exists.", 409, ) audit(db, "skill.created", "skill", skill_id, {"name": name}) return jsonify({"id": skill_id, "name": name, "status": "draft"}), 201 @app.delete("/api/v1/skills/") @require_auth def delete_skill(skill_id: str): db = get_db() cursor = db.execute( "DELETE FROM skills WHERE id=? AND user_id=?", (skill_id, g.user["id"]), ) if not cursor.rowcount: return error("KIMU_SKILL_NOT_FOUND", "Skill not found.", 404) audit(db, "skill.deleted", "skill", skill_id) return jsonify({"ok": True}) def settings_defaults() -> dict[str, bool]: return { "haptics": True, "offline_queue": True, "save_history": True, "stream_responses": True, } @app.get("/api/v1/settings") @require_auth def get_settings(): row = get_db().execute( "SELECT settings_json,version FROM user_settings WHERE user_id=?", (g.user["id"],), ).fetchone() value = settings_defaults() if row: value.update(json.loads(row["settings_json"] or "{}")) return jsonify( {"settings": value, "version": int(row["version"]) if row else 0} ) @app.put("/api/v1/settings") @require_auth def save_settings(): supplied = request.get_json(silent=True) or {} allowed = { key: bool(supplied[key]) for key in ( "haptics", "offline_queue", "save_history", "stream_responses", ) if key in supplied } db = get_db() existing = db.execute( "SELECT settings_json,version FROM user_settings WHERE user_id=?", (g.user["id"],), ).fetchone() current_version = int(existing["version"]) if existing else 0 base_version = expected_version(supplied) if base_version is not None and base_version != current_version: canonical = settings_defaults() if existing: canonical.update(json.loads(existing["settings_json"] or "{}")) return error( "KIMU_SYNC_CONFLICT", "Application settings changed on another session.", 409, details={ "entity": "settings", "base_version": base_version, "server_version": current_version, "canonical": canonical, }, ) value = json.loads(existing["settings_json"] or "{}") if existing else {} value.update(allowed) next_version = current_version + 1 db.execute( """ INSERT INTO user_settings(user_id,settings_json,version) VALUES(?,?,?) ON CONFLICT(user_id) DO UPDATE SET settings_json=excluded.settings_json, version=excluded.version, updated_at=CURRENT_TIMESTAMP """, (g.user["id"], json_text(value), next_version), ) payload = {"settings": {**settings_defaults(), **value}, "version": next_version} record_change( db, g.user["id"], "settings", g.user["id"], "updated", next_version, payload ) audit( db, "settings.updated", "user", g.user["id"], {**allowed, "version": next_version}, ) return jsonify(payload) @app.get("/api/v1/sync/status") @require_auth def sync_status(): db = get_db() try: after = max(0, int(request.args.get("after", "0") or 0)) except ValueError: after = 0 cursor = db.execute( "SELECT COALESCE(MAX(seq),0) FROM sync_changes WHERE user_id=?", (g.user["id"],), ).fetchone()[0] pending_changes = db.execute( "SELECT COUNT(*) FROM sync_changes WHERE user_id=? AND seq>?", (g.user["id"], after), ).fetchone()[0] operations = db.execute( """ SELECT status,COUNT(*) AS count FROM sync_operations WHERE user_id=? GROUP BY status """, (g.user["id"],), ).fetchall() return jsonify( { "online": True, "cursor": int(cursor), "changes_after_cursor": int(pending_changes), "operations": {row["status"]: row["count"] for row in operations}, "conflict_protocol": "optimistic-version-v1", "idempotency": True, } ) @app.get("/api/v1/sync/pull") @require_auth def sync_pull(): try: after = max(0, int(request.args.get("after", "0") or 0)) limit = min(500, max(1, int(request.args.get("limit", "200") or 200))) except ValueError: return error("KIMU_SYNC_CURSOR_INVALID", "Sync cursor is invalid.", 400) rows = get_db().execute( """ SELECT seq,entity_type,entity_id,action,version,payload_json,created_at FROM sync_changes WHERE user_id=? AND seq>? ORDER BY seq LIMIT ? """, (g.user["id"], after, limit + 1), ).fetchall() has_more = len(rows) > limit rows = rows[:limit] changes = [ { "seq": int(row["seq"]), "entity_type": row["entity_type"], "entity_id": row["entity_id"], "action": row["action"], "version": int(row["version"]), "payload": json.loads(row["payload_json"] or "{}"), "created_at": row["created_at"], } for row in rows ] return jsonify( { "changes": changes, "cursor": changes[-1]["seq"] if changes else after, "has_more": has_more, } ) @app.get("/api/v1/sync/snapshot") @require_auth def sync_snapshot(): db = get_db() conversations = db.execute( """ SELECT * FROM conversations WHERE user_id=? ORDER BY updated_at DESC LIMIT 500 """, (g.user["id"],), ).fetchall() files = db.execute( """ SELECT * FROM files WHERE user_id=? AND deleted_at IS NULL ORDER BY created_at DESC LIMIT 500 """, (g.user["id"],), ).fetchall() cursor = db.execute( "SELECT COALESCE(MAX(seq),0) FROM sync_changes WHERE user_id=?", (g.user["id"],), ).fetchone()[0] return jsonify( { "cursor": int(cursor), "conversations": [conversation_value(row) for row in conversations], "files": [public_file(row) for row in files], } ) @app.get("/api/v1/admin/summary") @require_owner def admin_summary(): db = get_db() counts = { "users": db.execute("SELECT COUNT(*) FROM users").fetchone()[0], "conversations": db.execute( "SELECT COUNT(*) FROM conversations WHERE deleted_at IS NULL" ).fetchone()[0], "messages": db.execute("SELECT COUNT(*) FROM messages").fetchone()[0], "files": db.execute( "SELECT COUNT(*) FROM files WHERE deleted_at IS NULL" ).fetchone()[0], "jobs": db.execute("SELECT COUNT(*) FROM jobs").fetchone()[0], "memories": db.execute("SELECT COUNT(*) FROM memories").fetchone()[0], "skills": db.execute("SELECT COUNT(*) FROM skills").fetchone()[0], "workspaces": db.execute("SELECT COUNT(*) FROM workspaces").fetchone()[0], "tool_operations": db.execute("SELECT COUNT(*) FROM tool_operations").fetchone()[0], "website_projects": db.execute("SELECT COUNT(*) FROM website_projects").fetchone()[0], "swarms": db.execute("SELECT COUNT(*) FROM swarms").fetchone()[0], "audio_sessions": db.execute("SELECT COUNT(*) FROM audio_sessions").fetchone()[0], } audits = db.execute( """ SELECT id,user_id,action,target_type,target_id,detail_json,created_at FROM audit_log ORDER BY id DESC LIMIT 50 """ ).fetchall() ready_response = health_ready() ready_data = ready_response.get_json() return jsonify( { "counts": counts, "readiness": ready_data, "audit": [ { **dict(row), "detail": json.loads(row["detail_json"] or "{}"), } for row in audits ], "configuration": { "k26_model_id": app.config["K26_MODEL_ID"], "k27_model_id": app.config["K27_MODEL_ID"], "k27_configured": bool(app.config["K27_CONFIGURED"]), "fake_provider": bool(app.config["FAKE_PROVIDER"]), "storage_dir": str(app.config["STORAGE_DIR"]), "whisper_model_id": app.config["WHISPER_MODEL_ID"], "sonic_model_id": app.config["SONIC_MODEL_ID"], "conversation_gateway_registered": bool(app.config.get("CONVERSATION_GATEWAY_REGISTERED")), }, } ) @app.get("/") def index(): return send_from_directory(app.static_folder, "index.html") @app.get("/") def static_or_spa(path: str): candidate = Path(app.static_folder) / path if candidate.is_file(): return send_from_directory(app.static_folder, path) return send_from_directory(app.static_folder, "index.html") register_advanced_routes(app) return app