Sync modernized KAPO runtime from control plane

Dockerfile

@@ -1,10 +1,10 @@
-FROM python:3.11-slim
-WORKDIR /app
-ENV PYTHONUNBUFFERED=1
-ENV PYTHONUTF8=1
-COPY requirements.txt /app/requirements.txt
-RUN pip install --no-cache-dir -r /app/requirements.txt
-RUN mkdir -p /data/kapo_runtime/current /data/kapo_runtime/overlay
-COPY brain_server /app/brain_server
-COPY bootstrap_space_runtime.py /app/bootstrap_space_runtime.py
-CMD ["python", "/app/bootstrap_space_runtime.py"]
+FROM python:3.11-slim
+WORKDIR /app
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONUTF8=1
+COPY requirements.txt /app/requirements.txt
+RUN pip install --no-cache-dir -r /app/requirements.txt
+RUN mkdir -p /data/kapo_runtime/current /data/kapo_runtime/overlay
+COPY brain_server /app/brain_server
+COPY bootstrap_space_runtime.py /app/bootstrap_space_runtime.py
+CMD ["python", "/app/bootstrap_space_runtime.py"]

README.md

@@ -1,5 +1,5 @@
 ---
-title: hf_debugger_main
+title: ai_coder_main
 emoji: 🤖
 colorFrom: blue
 colorTo: indigo
@@ -7,14 +7,14 @@ sdk: docker
 pinned: false
 ---
 
-# hf_debugger_main
+# ai_coder_main
 
 Generated Hugging Face deployment package from KAPO Control Center.
 
-Model profile: hf-debugger-qwen25-7b-instruct
-Model repo: Qwen/Qwen2.5-1.5B-Instruct
+Model profile: hf-coder-qwen25-coder-7b-instruct
+Model repo: Qwen/Qwen2.5-Coder-1.5B-Instruct
 Model file: not set
-Roles: fallback
+Roles: coding, planner, fallback
 Languages: ar, en
 
 This is a Docker-oriented Hugging Face Space package.

bootstrap_space_runtime.py

@@ -6,28 +6,40 @@ import sys
 from pathlib import Path
 
 DEFAULT_ENV = {
-    "REMOTE_BRAIN_ONLY": "1",
-    "KAGGLE_AUTO_BOOTSTRAP": "0",
-    "BRAIN_AUTO_NGROK": "0",
-    "BRAIN_AUTO_PUBLISH_URL_ON_STARTUP": "0",
-    "BRAIN_REUSE_PUBLIC_URL_ON_RESTART": "0",
-    "HF_SPACE_DOCKER": "1",
-    "KAPO_COMPUTE_PROFILE": "cpu",
-    "HF_ACCELERATOR": "cpu",
-    "KAPO_HF_TRANSFORMERS_RUNTIME": "1",
-    "KAPO_LAZY_MODEL_STARTUP": "1",
-    "KAPO_LAZY_EMBED_STARTUP": "1",
-    "MODEL_PROFILE_ID": "hf-debugger-qwen25-7b-instruct",
-    "MODEL_REPO": "Qwen/Qwen2.5-1.5B-Instruct",
-    "BRAIN_ROLES": "fallback",
-    "BRAIN_LANGUAGES": "ar,en",
-    "BRAIN_PLATFORM_NAME": "hf_debugger_main",
-    "BRAIN_TEMPLATE": "hf-space-cpu",
-    "BRAIN_PROVIDER": "huggingface",
-    "FIREBASE_ENABLED": "1",
-    "FIREBASE_PROJECT_ID": "citadel4travels",
-    "FIREBASE_NAMESPACE": "kapo",
-    "BRAIN_REUSE_PUBLIC_URL_ON_RESTART": "0",
+  "BRAIN_AUTO_NGROK": "0",
+  "BRAIN_AUTO_PUBLISH_URL_ON_STARTUP": "0",
+  "BRAIN_LANGUAGES": "ar,en",
+  "BRAIN_PLATFORM_NAME": "ai_coder_main",
+  "BRAIN_PROVIDER": "huggingface",
+  "BRAIN_REUSE_PUBLIC_URL_ON_RESTART": "0",
+  "BRAIN_ROLES": "coding,planner,fallback",
+  "BRAIN_TEMPLATE": "hf-space-cpu",
+  "BRAIN_TUNNEL_PROVIDER": "none",
+  "EXECUTOR_URL": "https://hartford-resorts-resulted-dis.trycloudflare.com",
+  "FIREBASE_ENABLED": "0",
+  "FIREBASE_NAMESPACE": "kapo",
+  "FIREBASE_PROJECT_ID": "citadel4travels",
+  "GOOGLE_DRIVE_BOOTSTRAP_URL": "https://drive.google.com/uc?export=download&id=19jyBWsQ9ciJVPi2PUigu5ti3gJ24A6TG",
+  "HF_ACCELERATOR": "cpu",
+  "HF_SPACE_DOCKER": "1",
+  "KAGGLE_AUTO_BOOTSTRAP": "0",
+  "KAPO_BOOTSTRAP_URL": "https://drive.google.com/uc?export=download&id=19jyBWsQ9ciJVPi2PUigu5ti3gJ24A6TG",
+  "KAPO_CLOUDFLARE_QUEUE_NAME": "kapo-task-events",
+  "KAPO_COMPUTE_PROFILE": "cpu",
+  "KAPO_CONTROL_PLANE_URL": "https://kapo-control-plane.kaboalby2015.workers.dev",
+  "KAPO_HF_INFERENCE_API": "1",
+  "KAPO_HF_TRANSFORMERS_RUNTIME": "0",
+  "KAPO_LAZY_EMBED_STARTUP": "1",
+  "KAPO_LAZY_MODEL_STARTUP": "1",
+  "KAPO_PATCH_BUNDLE_URL": "https://drive.google.com/uc?export=download&id=16rIe05GZihhAz7ba8E-WibKaJKbh9eu1",
+  "KAPO_PATCH_MANIFEST_URL": "https://drive.google.com/uc?export=download&id=1jLuPMCA3hp9qstZZtpBNzTK0XOmLrV8b",
+  "KAPO_REMOTE_ENV_PASSWORD_B64": "Nmk4ZWItQXdNQzgzTU10Rndoem5XYTFXaG41OGdxQW0",
+  "KAPO_REMOTE_ENV_URL_B64": "aHR0cHM6Ly9kcml2ZS5nb29nbGUuY29tL3VjP2V4cG9ydD1kb3dubG9hZCZpZD0xb3Itd2NFbjNmYzBpekg1NGVVeEpFRXljOEFLQ2swcHo",
+  "KAPO_SHARED_STATE_BACKEND": "google_drive",
+  "MODEL_PROFILE_ID": "hf-coder-qwen25-coder-7b-instruct",
+  "MODEL_REPO": "Qwen/Qwen2.5-Coder-1.5B-Instruct",
+  "REMOTE_BRAIN_ONLY": "1",
+  "SPACE_PUBLIC_URL": "https://MrA7A1-AiCoder.hf.space"
 }
 
 
@@ -64,8 +76,13 @@ def main() -> None:
     for key, value in DEFAULT_ENV.items():
         os.environ.setdefault(str(key), str(value))
     space_host = str(os.getenv('SPACE_HOST', '')).strip().rstrip('/')
+    space_id = str(os.getenv('SPACE_ID', '')).strip()
+    public_url = ''
     if space_host:
         public_url = space_host if space_host.startswith('http://') or space_host.startswith('https://') else f'https://{space_host}'
+    elif space_id and '/' in space_id:
+        public_url = 'https://' + space_id.replace('/', '-').lower() + '.hf.space'
+    if public_url:
         os.environ['BRAIN_PUBLIC_URL'] = public_url.rstrip('/')
     runtime_root.mkdir(parents=True, exist_ok=True)
     overlay_root.mkdir(parents=True, exist_ok=True)

brain_server/api/deps.py

@@ -7,12 +7,26 @@ from typing import Any
 
 import yaml
 from dotenv import load_dotenv
+from shared.remote_env import load_remote_env_if_configured
 
 CONFIG_CACHE: dict[str, Any] | None = None
 _LOGGING_READY = False
 PLACEHOLDER_RE = re.compile(r"^\$\{[A-Z0-9_]+\}$")
 
 
+def _load_env_stack() -> None:
+    candidates = [
+        ".env",
+        "kapo.env",
+        ".env.runtime",
+    ]
+    for candidate in candidates:
+        try:
+            load_dotenv(candidate, override=True)
+        except Exception:
+            continue
+
+
 def _normalize_config_paths(cfg: dict[str, Any]) -> dict[str, Any]:
     if os.name != "nt":
         return cfg
@@ -53,7 +67,8 @@ def load_config() -> dict:
     if CONFIG_CACHE is not None:
         return CONFIG_CACHE
 
-    load_dotenv()
+    _load_env_stack()
+    load_remote_env_if_configured(override=True, logger_name="kapo.brain.remote_env")
     config_path = os.path.join(os.path.dirname(__file__), "..", "config", "config.yaml")
     with open(config_path, "r", encoding="utf-8") as handle:
         raw = handle.read()

brain_server/api/firebase_store.py

@@ -1,28 +1,63 @@
-"""Optional Firebase mirror for brain runtime state."""
+"""Shared-state store for brain runtime via Google Drive, local files, or Firebase."""
 
 from __future__ import annotations
 
 import json
 import logging
 import os
+import threading
 import time
 from pathlib import Path
 from typing import Any
 
+from shared.google_drive_state import GoogleDriveStateClient
+
 
 class FirebaseStore:
     def __init__(self, component: str, logger_name: str = "kapo.brain.firebase") -> None:
         self.component = component
         self.logger = logging.getLogger(logger_name)
         self._db = None
-        self._cache: dict[str, tuple[float, Any]] = {}
+        self._lock = threading.Lock()
+        self._read_cache: dict[str, tuple[float, Any]] = {}
+        self._list_cache: dict[str, tuple[float, list[dict[str, Any]]]] = {}
+        self._write_cache: dict[str, tuple[float, str]] = {}
+        self._quota_backoff_until: float = 0.0
+        self._drive = GoogleDriveStateClient(self.logger)
+
+    def backend(self) -> str:
+        configured = str(os.getenv("KAPO_SHARED_STATE_BACKEND", "")).strip().lower()
+        if configured in {"google_drive", "drive", "gdrive"}:
+            return "google_drive"
+        if configured in {"file", "files"}:
+            return "file"
+        if configured in {"firebase", "firestore"}:
+            return "firebase"
+        if configured in {"disabled", "off", "none"}:
+            return "disabled"
+        if self._drive.enabled():
+            return "google_drive"
+        if str(os.getenv("FIREBASE_ENABLED", "0")).strip().lower() in {"1", "true", "yes", "on"}:
+            return "firebase"
+        return "file"
 
     def enabled(self) -> bool:
-        return str(os.getenv("FIREBASE_ENABLED", "0")).strip().lower() in {"1", "true", "yes", "on"}
+        return self.backend() != "disabled"
 
     def namespace(self) -> str:
         return str(os.getenv("FIREBASE_NAMESPACE", "kapo")).strip() or "kapo"
 
+    def storage_root(self) -> Path:
+        configured = str(os.getenv("KAPO_SHARED_STATE_DIR", "")).strip()
+        if configured:
+            root = Path(configured).expanduser()
+            if not root.is_absolute():
+                root = Path.cwd().resolve() / root
+        else:
+            root = (Path.cwd().resolve() / "data" / "local" / "shared_state").resolve()
+        root.mkdir(parents=True, exist_ok=True)
+        return root
+
     def _service_payload(self) -> dict[str, Any] | None:
         raw = str(os.getenv("FIREBASE_SERVICE_ACCOUNT_JSON", "")).strip()
         if not raw:
@@ -38,40 +73,41 @@ class FirebaseStore:
         return str(os.getenv("FIREBASE_SERVICE_ACCOUNT_PATH", "")).strip()
 
     def _client(self):
-        if not self.enabled():
-            return None
-        if self._db is not None:
-            return self._db
-        try:
-            import firebase_admin
-            from firebase_admin import credentials, firestore
-
-            if not firebase_admin._apps:
-                payload = self._service_payload()
-                if payload:
-                    cred = credentials.Certificate(payload)
-                else:
-                    service_path = self._service_path()
-                    if not service_path:
-                        return None
-                    path_obj = Path(service_path).expanduser()
-                    if not path_obj.exists() or not path_obj.is_file():
-                        self.logger.warning(
-                            "Firebase service account path is unavailable on this runtime: %s",
-                            service_path,
-                        )
-                        return None
-                    cred = credentials.Certificate(str(path_obj.resolve()))
-                options = {}
-                project_id = str(os.getenv("FIREBASE_PROJECT_ID", "")).strip()
-                if project_id:
-                    options["projectId"] = project_id
-                firebase_admin.initialize_app(cred, options or None)
-            self._db = firestore.client()
-            return self._db
-        except Exception:
-            self.logger.exception("Failed to initialize Firebase client")
+        if self.backend() != "firebase":
             return None
+        with self._lock:
+            if self._db is not None:
+                return self._db
+            try:
+                import firebase_admin
+                from firebase_admin import credentials, firestore
+
+                if not firebase_admin._apps:
+                    payload = self._service_payload()
+                    if payload:
+                        cred = credentials.Certificate(payload)
+                    else:
+                        service_path = self._service_path()
+                        if not service_path:
+                            return None
+                        path_obj = Path(service_path).expanduser()
+                        if not path_obj.exists() or not path_obj.is_file():
+                            self.logger.warning(
+                                "Firebase service account path is unavailable on this runtime: %s",
+                                service_path,
+                            )
+                            return None
+                        cred = credentials.Certificate(str(path_obj.resolve()))
+                    options = {}
+                    project_id = str(os.getenv("FIREBASE_PROJECT_ID", "")).strip()
+                    if project_id:
+                        options["projectId"] = project_id
+                    firebase_admin.initialize_app(cred, options or None)
+                self._db = firestore.client()
+                return self._db
+            except Exception:
+                self.logger.exception("Failed to initialize Firebase client")
+                return None
 
     def _collection(self, name: str) -> str:
         return f"{self.namespace()}_{name}"
@@ -81,45 +117,177 @@ class FirebaseStore:
         text = str(value or "").strip() or default
         return "".join(ch if ch.isalnum() or ch in {"-", "_", "."} else "_" for ch in text)[:180]
 
+    @staticmethod
+    def _payload_hash(payload: Any) -> str:
+        return json.dumps(payload, ensure_ascii=False, sort_keys=True, default=str)
+
+    @staticmethod
+    def _is_quota_error(exc: Exception) -> bool:
+        text = str(exc or "").lower()
+        return "resourceexhausted" in text or "quota exceeded" in text or "429" in text
+
+    def _quota_backoff_active(self) -> bool:
+        return time.time() < self._quota_backoff_until
+
+    def _activate_quota_backoff(self, seconds: float | None = None) -> None:
+        delay = float(seconds or os.getenv("FIREBASE_QUOTA_BACKOFF_SEC", "120") or 120)
+        self._quota_backoff_until = max(self._quota_backoff_until, time.time() + max(5.0, delay))
+
+    def _should_skip_write(self, key: str, payload: Any, min_interval_sec: float) -> bool:
+        now = time.time()
+        payload_hash = self._payload_hash(payload)
+        last = self._write_cache.get(key)
+        if last and last[1] == payload_hash and (now - last[0]) < min_interval_sec:
+            return True
+        self._write_cache[key] = (now, payload_hash)
+        return False
+
+    def _file_collection_dir(self, collection: str) -> Path:
+        path = self.storage_root() / self._safe_id(collection, "collection")
+        path.mkdir(parents=True, exist_ok=True)
+        return path
+
+    def _file_doc_path(self, collection: str, doc_id: str) -> Path:
+        return self._file_collection_dir(collection) / f"{self._safe_id(doc_id)}.json"
+
+    def _write_json_atomic(self, path: Path, payload: dict[str, Any]) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        tmp = path.with_suffix(f"{path.suffix}.tmp")
+        tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
+        tmp.replace(path)
+
     def get_document(self, collection: str, doc_id: str, ttl_sec: float = 12.0) -> dict[str, Any]:
-        db = self._client()
-        if db is None:
-            return {}
         safe_doc = self._safe_id(doc_id)
         cache_key = f"{collection}:{safe_doc}"
         now = time.time()
-        cached = self._cache.get(cache_key)
+        cached = self._read_cache.get(cache_key)
         if cached and (now - cached[0]) < ttl_sec:
             return dict(cached[1] or {})
+        backend = self.backend()
+        if backend == "google_drive":
+            payload = self._drive.get_document(collection, safe_doc)
+            self._read_cache[cache_key] = (now, payload)
+            return dict(payload or {})
+        if backend == "file":
+            path = self._file_doc_path(collection, safe_doc)
+            if not path.exists():
+                return {}
+            try:
+                payload = json.loads(path.read_text(encoding="utf-8"))
+            except Exception:
+                self.logger.warning("Failed to read shared-state file %s", path, exc_info=True)
+                return {}
+            self._read_cache[cache_key] = (now, payload)
+            return dict(payload or {})
+        if self._quota_backoff_active():
+            return dict(cached[1] or {}) if cached else {}
+        db = self._client()
+        if db is None:
+            return dict(cached[1] or {}) if cached else {}
         try:
             snapshot = db.collection(self._collection(collection)).document(safe_doc).get()
             payload = snapshot.to_dict() if snapshot.exists else {}
-            self._cache[cache_key] = (now, payload)
+            self._read_cache[cache_key] = (now, payload)
             return dict(payload or {})
-        except Exception:
-            self.logger.exception("Failed to read Firebase document %s/%s", collection, safe_doc)
-            return {}
+        except Exception as exc:
+            if self._is_quota_error(exc):
+                self._activate_quota_backoff()
+                self.logger.warning("Firebase quota exceeded while reading %s/%s; using cache/backoff", collection, safe_doc)
+            else:
+                self.logger.exception("Failed to read Firebase document %s/%s", collection, safe_doc)
+            return dict(cached[1] or {}) if cached else {}
 
-    def set_document(self, collection: str, doc_id: str, payload: dict[str, Any], merge: bool = True) -> bool:
+    def set_document(self, collection: str, doc_id: str, payload: dict[str, Any], merge: bool = True, min_interval_sec: float = 5.0) -> bool:
+        safe_doc = self._safe_id(doc_id)
+        cache_key = f"{collection}:{safe_doc}"
+        body = dict(payload or {})
+        body["component"] = self.component
+        body["updated_at"] = time.time()
+        if self._should_skip_write(cache_key, body, min_interval_sec):
+            return True
+        backend = self.backend()
+        if backend == "google_drive":
+            stored = self._drive.set_document(collection, safe_doc, body, merge=merge)
+            if stored:
+                self._read_cache.pop(cache_key, None)
+                stale_prefix = f"{collection}:list:"
+                for key in list(self._list_cache.keys()):
+                    if key.startswith(stale_prefix):
+                        self._list_cache.pop(key, None)
+            return stored
+        if backend == "file":
+            try:
+                path = self._file_doc_path(collection, safe_doc)
+                existing = {}
+                if merge and path.exists():
+                    existing = json.loads(path.read_text(encoding="utf-8"))
+                combined = {**existing, **body} if merge else body
+                combined.setdefault("id", safe_doc)
+                self._write_json_atomic(path, combined)
+                self._read_cache.pop(cache_key, None)
+                stale_prefix = f"{collection}:list:"
+                for key in list(self._list_cache.keys()):
+                    if key.startswith(stale_prefix):
+                        self._list_cache.pop(key, None)
+                return True
+            except Exception:
+                self.logger.warning("Failed to write shared-state file %s/%s", collection, safe_doc, exc_info=True)
+                return False
+        if self._quota_backoff_active():
+            return False
         db = self._client()
         if db is None:
             return False
-        safe_doc = self._safe_id(doc_id)
         try:
-            body = dict(payload or {})
-            body["component"] = self.component
-            body["updated_at"] = time.time()
+            if self._should_skip_write(cache_key, body, min_interval_sec):
+                return True
             db.collection(self._collection(collection)).document(safe_doc).set(body, merge=merge)
-            self._cache.pop(f"{collection}:{safe_doc}", None)
+            self._read_cache.pop(cache_key, None)
+            stale_prefix = f"{collection}:list:"
+            for key in list(self._list_cache.keys()):
+                if key.startswith(stale_prefix):
+                    self._list_cache.pop(key, None)
             return True
-        except Exception:
-            self.logger.exception("Failed to write Firebase document %s/%s", collection, safe_doc)
+        except Exception as exc:
+            if self._is_quota_error(exc):
+                self._activate_quota_backoff()
+                self.logger.warning("Firebase quota exceeded while writing %s/%s; write skipped", collection, safe_doc)
+            else:
+                self.logger.exception("Failed to write Firebase document %s/%s", collection, safe_doc)
             return False
 
-    def list_documents(self, collection: str, limit: int = 200) -> list[dict[str, Any]]:
+    def list_documents(self, collection: str, limit: int = 200, ttl_sec: float = 30.0) -> list[dict[str, Any]]:
+        cache_key = f"{collection}:list:{max(1, int(limit))}"
+        now = time.time()
+        cached = self._list_cache.get(cache_key)
+        if cached and (now - cached[0]) < ttl_sec:
+            return [dict(item) for item in (cached[1] or [])]
+        backend = self.backend()
+        if backend == "google_drive":
+            items = self._drive.list_documents(collection, limit=max(1, int(limit)))
+            self._list_cache[cache_key] = (now, [dict(item) for item in items])
+            return items
+        if backend == "file":
+            items: list[dict[str, Any]] = []
+            try:
+                paths = sorted(
+                    self._file_collection_dir(collection).glob("*.json"),
+                    key=lambda path: path.stat().st_mtime,
+                    reverse=True,
+                )
+                for path in paths[: max(1, int(limit))]:
+                    payload = json.loads(path.read_text(encoding="utf-8"))
+                    payload.setdefault("id", path.stem)
+                    items.append(payload)
+            except Exception:
+                self.logger.warning("Failed to list shared-state files for %s", collection, exc_info=True)
+            self._list_cache[cache_key] = (now, [dict(item) for item in items])
+            return items
+        if self._quota_backoff_active():
+            return [dict(item) for item in ((cached[1] if cached else []) or [])]
         db = self._client()
         if db is None:
-            return []
+            return [dict(item) for item in ((cached[1] if cached else []) or [])]
         try:
             docs = db.collection(self._collection(collection)).limit(max(1, int(limit))).stream()
             items: list[dict[str, Any]] = []
@@ -127,20 +295,62 @@ class FirebaseStore:
                 payload = doc.to_dict() or {}
                 payload.setdefault("id", doc.id)
                 items.append(payload)
+            self._list_cache[cache_key] = (now, [dict(item) for item in items])
             return items
-        except Exception:
-            self.logger.exception("Failed to list Firebase collection %s", collection)
-            return []
+        except Exception as exc:
+            if self._is_quota_error(exc):
+                self._activate_quota_backoff()
+                self.logger.warning("Firebase quota exceeded while listing %s; using cache/backoff", collection)
+            else:
+                self.logger.exception("Failed to list Firebase collection %s", collection)
+            return [dict(item) for item in ((cached[1] if cached else []) or [])]
 
     def delete_document(self, collection: str, doc_id: str) -> bool:
+        safe_doc = self._safe_id(doc_id)
+        backend = self.backend()
+        if backend == "google_drive":
+            deleted = self._drive.delete_document(collection, safe_doc)
+            if deleted:
+                self._read_cache.pop(f"{collection}:{safe_doc}", None)
+                self._write_cache.pop(f"{collection}:{safe_doc}", None)
+                stale_prefix = f"{collection}:list:"
+                for key in list(self._list_cache.keys()):
+                    if key.startswith(stale_prefix):
+                        self._list_cache.pop(key, None)
+            return deleted
+        if backend == "file":
+            try:
+                path = self._file_doc_path(collection, safe_doc)
+                if path.exists():
+                    path.unlink()
+                self._read_cache.pop(f"{collection}:{safe_doc}", None)
+                self._write_cache.pop(f"{collection}:{safe_doc}", None)
+                stale_prefix = f"{collection}:list:"
+                for key in list(self._list_cache.keys()):
+                    if key.startswith(stale_prefix):
+                        self._list_cache.pop(key, None)
+                return True
+            except Exception:
+                self.logger.warning("Failed to delete shared-state file %s/%s", collection, safe_doc, exc_info=True)
+                return False
+        if self._quota_backoff_active():
+            return False
         db = self._client()
         if db is None:
             return False
-        safe_doc = self._safe_id(doc_id)
         try:
             db.collection(self._collection(collection)).document(safe_doc).delete()
-            self._cache.pop(f"{collection}:{safe_doc}", None)
+            self._read_cache.pop(f"{collection}:{safe_doc}", None)
+            self._write_cache.pop(f"{collection}:{safe_doc}", None)
+            stale_prefix = f"{collection}:list:"
+            for key in list(self._list_cache.keys()):
+                if key.startswith(stale_prefix):
+                    self._list_cache.pop(key, None)
             return True
-        except Exception:
-            self.logger.exception("Failed to delete Firebase document %s/%s", collection, safe_doc)
+        except Exception as exc:
+            if self._is_quota_error(exc):
+                self._activate_quota_backoff()
+                self.logger.warning("Firebase quota exceeded while deleting %s/%s; delete skipped", collection, safe_doc)
+            else:
+                self.logger.exception("Failed to delete Firebase document %s/%s", collection, safe_doc)
             return False

brain_server/api/main.py

@@ -1,4 +1,5 @@
 """FastAPI entrypoint for the Brain Server."""
+import base64
 import gc
 import logging
 import os
@@ -28,6 +29,7 @@ try:
     from api.routes_analyze import router as analyze_router
     from api.routes_execute import router as execute_router
     from api.routes_plan import router as plan_router
+    from shared.google_drive_state import GoogleDriveStateClient
 except ImportError:
     from . import deps as deps_module
     from .deps import get_executor_headers, get_logger, load_config
@@ -35,6 +37,7 @@ except ImportError:
     from .routes_analyze import router as analyze_router
     from .routes_execute import router as execute_router
     from .routes_plan import router as plan_router
+    from shared.google_drive_state import GoogleDriveStateClient
 
 logger = get_logger("kapo.brain.main")
 
@@ -72,9 +75,11 @@ MODEL_ERROR = None
 MODEL_META = {"repo_id": None, "filename": None, "path": None}
 EMBED_MODEL = None
 FIREBASE = FirebaseStore("brain", logger_name="kapo.brain.firebase")
+DRIVE_STATE = GoogleDriveStateClient(logger)
 FIREBASE_RUNTIME_CACHE: dict[str, tuple[float, Any]] = {}
 RUNTIME_LOG_BUFFER: deque[dict[str, Any]] = deque(maxlen=200)
 LAST_BRAIN_URL_REPORT: dict[str, Any] = {"url": "", "ts": 0.0}
+RUNTIME_STATE_THREAD_STARTED = False
 
 DEFAULT_MODEL_REPO = "QuantFactory/aya-expanse-8b-GGUF"
 DEFAULT_MODEL_FILE = "aya-expanse-8b.Q4_K_M.gguf"
@@ -125,6 +130,13 @@ def _configured_public_url() -> str:
     return str(os.getenv("BRAIN_PUBLIC_URL", "")).strip().rstrip("/")
 
 
+def _prefer_configured_public_url() -> bool:
+    provider = str(os.getenv("BRAIN_PROVIDER", "") or os.getenv("BRAIN_TEMPLATE", "")).strip().lower()
+    if "huggingface" in provider or "hf-space" in provider:
+        return True
+    return _feature_enabled("BRAIN_FORCE_CONFIGURED_PUBLIC_URL", default=False)
+
+
 def _reuse_public_url_on_restart() -> bool:
     return _feature_enabled("BRAIN_REUSE_PUBLIC_URL_ON_RESTART", default=True)
 
@@ -137,6 +149,10 @@ def _internal_restart_in_progress() -> bool:
     return _feature_enabled("KAPO_INTERNAL_RESTART", default=False)
 
 
+def _fast_restart_enabled() -> bool:
+    return _feature_enabled("KAPO_FAST_INTERNAL_RESTART", default=True)
+
+
 def _executor_connect_timeout() -> float:
     return float(os.getenv("EXECUTOR_CONNECT_TIMEOUT_SEC", "3.0") or 3.0)
 
@@ -158,6 +174,29 @@ def _executor_roundtrip_allowed(feature_name: str, default: bool = True) -> bool
     return _feature_enabled(feature_name, default=effective_default)
 
 
+def _remote_runtime_reads_enabled() -> bool:
+    return _feature_enabled("KAPO_REMOTE_STATE_READS", default=False)
+
+
+def _shared_state_backend() -> str:
+    return str(os.getenv("KAPO_SHARED_STATE_BACKEND", "")).strip().lower()
+
+
+def _drive_bootstrap_configured() -> bool:
+    return bool(
+        str(os.getenv("GOOGLE_DRIVE_BOOTSTRAP_URL", "") or os.getenv("KAPO_BOOTSTRAP_URL", "") or "").strip()
+    )
+
+
+def _bootstrap_shared_state() -> None:
+    if _drive_bootstrap_configured() or _shared_state_backend() in {"google_drive", "drive", "gdrive"}:
+        DRIVE_STATE.ensure_bootstrap_loaded(force=False)
+
+
+def _startup_self_update_enabled() -> bool:
+    return _feature_enabled("KAPO_STARTUP_SELF_UPDATE", default=True)
+
+
 def _should_report_brain_url(public_url: str) -> bool:
     normalized = str(public_url or "").strip().rstrip("/")
     if not normalized:
@@ -291,6 +330,19 @@ def _apply_executor_settings(settings: dict[str, Any]) -> None:
         "FIREBASE_SERVICE_ACCOUNT_PATH",
         "FIREBASE_SERVICE_ACCOUNT_JSON",
         "FIREBASE_NAMESPACE",
+        "KAPO_SHARED_STATE_BACKEND",
+        "GOOGLE_DRIVE_SHARED_STATE_FOLDER_ID",
+        "GOOGLE_DRIVE_SHARED_STATE_PREFIX",
+        "GOOGLE_DRIVE_BOOTSTRAP_URL",
+        "KAPO_BOOTSTRAP_URL",
+        "GOOGLE_DRIVE_ACCESS_TOKEN",
+        "GOOGLE_DRIVE_REFRESH_TOKEN",
+        "GOOGLE_DRIVE_CLIENT_SECRET_JSON",
+        "GOOGLE_DRIVE_CLIENT_SECRET_JSON_BASE64",
+        "GOOGLE_DRIVE_CLIENT_SECRET_PATH",
+        "GOOGLE_DRIVE_TOKEN_EXPIRES_AT",
+        "KAPO_CONTROL_PLANE_URL",
+        "KAPO_CLOUDFLARE_QUEUE_NAME",
     ):
         value = settings.get(key)
         if value not in (None, ""):
@@ -299,6 +351,8 @@ def _apply_executor_settings(settings: dict[str, Any]) -> None:
 
 
 def _apply_firebase_runtime_settings() -> None:
+    if not _remote_runtime_reads_enabled():
+        return
     if not FIREBASE.enabled():
         return
     shared = FIREBASE.get_document("settings", "global")
@@ -310,7 +364,6 @@ def _apply_firebase_runtime_settings() -> None:
     mappings = {
         "executor_public_url": "EXECUTOR_PUBLIC_URL",
         "executor_url": "EXECUTOR_URL",
-        "current_brain_url": "BRAIN_PUBLIC_URL",
         "model_repo": "MODEL_REPO",
         "model_file": "MODEL_FILE",
         "model_profile_id": "MODEL_PROFILE_ID",
@@ -318,6 +371,9 @@ def _apply_firebase_runtime_settings() -> None:
         "brain_roles": "BRAIN_ROLES",
         "brain_languages": "BRAIN_LANGUAGES",
     }
+    current_brain_url = str(merged.get("current_brain_url") or "").strip()
+    if current_brain_url:
+        os.environ["KAPO_EXECUTOR_CURRENT_BRAIN_URL"] = current_brain_url
     for key, env_name in mappings.items():
         value = merged.get(key)
         if value not in (None, ""):
@@ -339,6 +395,8 @@ def _firebase_collection_cache_key(name: str) -> str:
 
 
 def _firebase_list_documents_cached(collection: str, ttl_sec: float = 30.0, limit: int = 200) -> list[dict[str, Any]]:
+    if not _remote_runtime_reads_enabled():
+        return []
     if not FIREBASE.enabled():
         return []
     key = _firebase_collection_cache_key(collection)
@@ -409,6 +467,10 @@ def _firebase_prompt_body(role_name: str, language: str = "en") -> str:
 
 
 def _prepare_runtime_environment() -> None:
+    try:
+        _bootstrap_shared_state()
+    except Exception:
+        logger.warning("Shared-state bootstrap preload failed", exc_info=True)
     if not _is_kaggle_runtime():
         return
 
@@ -444,6 +506,10 @@ def _prepare_runtime_environment() -> None:
     os.environ["TOOLS_DB_PATH"] = str(data_dir / "tools.db")
     os.environ["FAISS_INDEX_PATH"] = str(data_dir / "faiss.index")
     os.environ["REMOTE_BRAIN_ONLY"] = str(os.getenv("REMOTE_BRAIN_ONLY", "1") or "1")
+    saved_executor_url = _load_saved_executor_url()
+    current_executor_url = str(os.getenv("EXECUTOR_URL", "") or "").strip()
+    if saved_executor_url and not current_executor_url:
+        os.environ["EXECUTOR_URL"] = saved_executor_url
     deps_module.CONFIG_CACHE = None
 
 
@@ -494,9 +560,6 @@ def _remember_public_url(public_url: str) -> None:
 
 
 def _load_saved_public_url() -> str:
-    configured = _configured_public_url()
-    if configured:
-        return configured
     try:
         value = _public_url_state_path().read_text(encoding="utf-8").strip().rstrip("/")
         return value
@@ -511,6 +574,140 @@ def _ngrok_api_state_path() -> Path:
     return state_dir / "ngrok_api_url.txt"
 
 
+def _executor_url_state_path() -> Path:
+    runtime_root = Path(_sync_target_root()).resolve()
+    state_dir = runtime_root / "data" / "local" / "brain_runtime"
+    state_dir.mkdir(parents=True, exist_ok=True)
+    return state_dir / "executor_url.txt"
+
+
+def _remember_executor_url(executor_url: str) -> None:
+    value = str(executor_url or "").strip().rstrip("/")
+    if not value:
+        return
+    os.environ["EXECUTOR_URL"] = value
+    try:
+        _executor_url_state_path().write_text(value, encoding="utf-8")
+    except Exception:
+        logger.warning("Failed to persist executor URL", exc_info=True)
+
+
+def _load_saved_executor_url() -> str:
+    configured = str(os.getenv("EXECUTOR_URL", "")).strip().rstrip("/")
+    if configured:
+        return configured
+    try:
+        return _executor_url_state_path().read_text(encoding="utf-8").strip().rstrip("/")
+    except Exception:
+        return ""
+
+
+def _brain_state_id() -> str:
+    public_url = str(os.getenv("BRAIN_PUBLIC_URL") or _load_saved_public_url() or "").strip().rstrip("/")
+    if public_url:
+        return re.sub(r"[^A-Za-z0-9._-]+", "_", public_url)
+    runtime_root = str(os.getenv("KAPO_RUNTIME_ROOT") or _sync_target_root() or "brain_runtime").strip()
+    return re.sub(r"[^A-Za-z0-9._-]+", "_", runtime_root)
+
+
+def _applied_runtime_version_path() -> Path:
+    runtime_root = Path(_sync_target_root()).resolve()
+    state_dir = runtime_root / "data" / "local" / "brain_runtime"
+    state_dir.mkdir(parents=True, exist_ok=True)
+    return state_dir / "applied_version.json"
+
+
+def _load_applied_runtime_version() -> dict[str, Any]:
+    try:
+        return dict(json.loads(_applied_runtime_version_path().read_text(encoding="utf-8")) or {})
+    except Exception:
+        return {}
+
+
+def _write_applied_runtime_version(payload: dict[str, Any]) -> None:
+    _applied_runtime_version_path().write_text(
+        json.dumps(dict(payload or {}), ensure_ascii=False, indent=2, sort_keys=True),
+        encoding="utf-8",
+    )
+
+
+def _download_remote_zip(url: str, destination: Path) -> Path:
+    response = requests.get(str(url).strip(), timeout=120)
+    response.raise_for_status()
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    destination.write_bytes(response.content)
+    return destination
+
+
+def _apply_zip_overlay(zip_path: Path, target_root: Path) -> None:
+    with zipfile.ZipFile(zip_path, "r") as archive:
+        archive.extractall(target_root)
+
+
+def _load_patch_manifest_from_bootstrap() -> dict[str, Any]:
+    bootstrap = DRIVE_STATE.ensure_bootstrap_loaded(force=True) or {}
+    manifest_url = str(bootstrap.get("patch_manifest_url") or "").strip()
+    if not manifest_url:
+        return dict(bootstrap)
+    try:
+        response = requests.get(manifest_url, timeout=30)
+        response.raise_for_status()
+        payload = dict(response.json() or {})
+        merged = {**bootstrap, **payload}
+        return merged
+    except Exception:
+        logger.warning("Failed to load patch manifest from bootstrap URL", exc_info=True)
+        return dict(bootstrap)
+
+
+def _run_startup_self_update() -> None:
+    if not _startup_self_update_enabled():
+        return
+    manifest = _load_patch_manifest_from_bootstrap()
+    target_version = str(manifest.get("version") or "").strip()
+    target_hash = str(manifest.get("build_hash") or "").strip()
+    if not target_version and not target_hash:
+        return
+    current = _load_applied_runtime_version()
+    if (
+        str(current.get("version") or "").strip() == target_version
+        and str(current.get("build_hash") or "").strip() == target_hash
+        and target_version
+    ):
+        return
+    runtime_root = Path(_sync_target_root()).resolve()
+    temp_dir = runtime_root / "data" / "local" / "brain_runtime" / "updates"
+    patch_url = str(manifest.get("patch_bundle_url") or "").strip()
+    full_url = str(manifest.get("full_package_url") or "").strip()
+    applied_mode = ""
+    source_url = ""
+    try:
+        if patch_url:
+            zip_path = _download_remote_zip(patch_url, temp_dir / "patch_bundle.zip")
+            _apply_zip_overlay(zip_path, runtime_root)
+            applied_mode = "patch_bundle"
+            source_url = patch_url
+        elif full_url:
+            zip_path = _download_remote_zip(full_url, temp_dir / "full_package.zip")
+            _apply_zip_overlay(zip_path, runtime_root)
+            applied_mode = "full_package"
+            source_url = full_url
+        else:
+            return
+        _write_applied_runtime_version(
+            {
+                "version": target_version,
+                "build_hash": target_hash,
+                "applied_at": time.time(),
+                "mode": applied_mode,
+                "source_url": source_url,
+            }
+        )
+        logger.info("Applied startup self-update (%s) version=%s", applied_mode, target_version or target_hash)
+    except Exception:
+        logger.warning("Startup self-update failed", exc_info=True)
+
+
 def _remember_ngrok_api_url(api_url: str) -> None:
     value = str(api_url or "").strip().rstrip("/")
     if not value:
@@ -661,15 +858,57 @@ def _create_ngrok_tunnel(api_url: str, port: int) -> str | None:
     return str(payload.get("public_url") or "").strip().rstrip("/") or None
 
 
+def _publish_brain_presence(public_url: str, *, source: str = "runtime") -> None:
+    normalized = str(public_url or "").strip().rstrip("/")
+    if not normalized:
+        return
+    payload = {
+        "url": normalized,
+        "status": "healthy",
+        "source": source,
+        "platform": "kaggle" if _is_kaggle_runtime() else str(os.getenv("BRAIN_PROVIDER") or "remote"),
+        "role": os.getenv("BRAIN_PRIMARY_ROLE", "fallback"),
+        "roles": [part.strip() for part in os.getenv("BRAIN_ROLES", "supervisor,chat,coding,planner,arabic,fallback").split(",") if part.strip()],
+        "languages": [part.strip() for part in os.getenv("BRAIN_LANGUAGES", "ar,en").split(",") if part.strip()],
+        "model_profile_id": os.getenv("MODEL_PROFILE_ID") or os.getenv("SUPERVISOR_MODEL_PROFILE_ID") or DEFAULT_MODEL_PROFILE_ID,
+        "model_repo": MODEL_META.get("repo_id") or os.getenv("MODEL_REPO") or DEFAULT_MODEL_REPO,
+        "model_file": MODEL_META.get("filename") or os.getenv("MODEL_FILE") or DEFAULT_MODEL_FILE,
+        "updated_at": time.time(),
+    }
+    FIREBASE.set_document("brains", normalized, payload)
+    FIREBASE.set_document(
+        "runtime",
+        "brains_last_report",
+        {
+            "brain_url": normalized,
+            "source": source,
+            "updated_at": time.time(),
+            "provider": payload["platform"],
+            "model_profile_id": payload["model_profile_id"],
+        },
+    )
+    FIREBASE.set_document(
+        "tunnels",
+        f"brain_{normalized}",
+        {
+            "kind": "brain",
+            "public_url": normalized,
+            "provider": "ngrok" if "ngrok" in normalized else payload["platform"],
+            "updated_at": time.time(),
+        },
+    )
+
+
 def _report_brain_url(public_url: str) -> None:
+    _publish_brain_presence(public_url, source="report_attempt")
     executor_url = os.getenv("EXECUTOR_URL", "").strip().rstrip("/")
     if not executor_url:
         return
     if not _should_report_brain_url(public_url):
         return
     last_error: Exception | None = None
-    connect_timeout = max(1.0, float(os.getenv("BRAIN_REPORT_CONNECT_TIMEOUT_SEC", "3.0") or 3.0))
-    read_timeout = max(2.0, float(os.getenv("BRAIN_REPORT_READ_TIMEOUT_SEC", "5.0") or 5.0))
+    connect_timeout = max(1.0, float(os.getenv("BRAIN_REPORT_CONNECT_TIMEOUT_SEC", "4.0") or 4.0))
+    read_timeout = max(5.0, float(os.getenv("BRAIN_REPORT_READ_TIMEOUT_SEC", "15.0") or 15.0))
     retries = max(1, int(os.getenv("BRAIN_REPORT_RETRIES", "2") or 2))
     for _ in range(retries):
         try:
@@ -689,6 +928,7 @@ def _report_brain_url(public_url: str) -> None:
                 timeout=(connect_timeout, read_timeout),
             )
             response.raise_for_status()
+            _publish_brain_presence(public_url, source="executor_report")
             return
         except Exception as exc:
             last_error = exc
@@ -700,20 +940,25 @@ def _report_brain_url(public_url: str) -> None:
 
 
 def _pull_executor_settings() -> dict[str, Any]:
-    executor_url = os.getenv("EXECUTOR_URL", "").strip().rstrip("/")
+    if _shared_state_backend() in {"google_drive", "drive", "gdrive"} or _drive_bootstrap_configured():
+        return {}
+    executor_url = _load_saved_executor_url()
     if not executor_url:
         return {}
     try:
         response = requests.get(
             f"{executor_url}/share/settings",
             headers=_brain_headers(),
-            timeout=15,
+            timeout=(
+                max(1.5, float(os.getenv("EXECUTOR_CONNECT_TIMEOUT_SEC", "3.0") or 3.0)),
+                max(2.0, float(os.getenv("EXECUTOR_SETTINGS_READ_TIMEOUT_SEC", "5.0") or 5.0)),
+            ),
         )
         if response.status_code == 200:
             return response.json()
-        logger.warning("Executor settings request failed: %s", response.text)
-    except Exception:
-        logger.warning("Failed to pull executor settings", exc_info=True)
+        logger.warning("Executor settings request failed: %s", response.text[:400])
+    except Exception as exc:
+        logger.warning("Failed to pull executor settings (%s)", exc)
     return {}
 
 
@@ -731,7 +976,7 @@ def start_ngrok(token: str | None = None) -> str | None:
             return saved_public_url
 
     configured_public_url = _configured_public_url()
-    if configured_public_url:
+    if configured_public_url and _prefer_configured_public_url():
         _remember_public_url(configured_public_url)
         _report_brain_url(configured_public_url)
         FIREBASE.set_document("brains", configured_public_url, {"url": configured_public_url, "status": "healthy", "source": "configured_public_url"})
@@ -790,15 +1035,26 @@ def _report_known_public_url() -> str | None:
 def _bootstrap_executor_handshake(start_tunnel: bool = False) -> None:
     executor_url = os.getenv("EXECUTOR_URL", "").strip()
     if not executor_url:
+        if start_tunnel:
+            public_url = start_ngrok(os.getenv("NGROK_AUTHTOKEN") or None)
+            if public_url:
+                logger.info("Brain public URL started locally without executor handshake: %s", public_url)
+            else:
+                logger.info("Brain started without publishing a public URL")
+            return
         logger.info("Skipping executor handshake: EXECUTOR_URL not configured")
         return
 
     settings = _pull_executor_settings()
     _apply_executor_settings(settings)
 
-    public_url = _report_known_public_url()
-    if not public_url and start_tunnel:
+    public_url = None
+    if start_tunnel:
         public_url = start_ngrok(os.getenv("NGROK_AUTHTOKEN") or None)
+        if not public_url:
+            public_url = _report_known_public_url()
+    else:
+        public_url = _report_known_public_url()
     if public_url:
         logger.info("Brain public URL reported to executor: %s", public_url)
     else:
@@ -807,48 +1063,51 @@ def _bootstrap_executor_handshake(start_tunnel: bool = False) -> None:
 
 @app.on_event("startup")
 async def startup_event():
+    global RUNTIME_STATE_THREAD_STARTED
     try:
-        settings = _pull_executor_settings()
-        _apply_executor_settings(settings)
-    except Exception:
-        logger.exception("Executor settings bootstrap failed")
-    try:
-        _apply_firebase_runtime_settings()
+        _bootstrap_shared_state()
     except Exception:
-        logger.exception("Firebase runtime bootstrap failed")
+        logger.exception("Shared-state bootstrap failed")
     try:
         _prepare_runtime_environment()
     except Exception:
         logger.exception("Runtime environment bootstrap failed")
-    if str(os.getenv("KAPO_LAZY_MODEL_STARTUP", "1")).strip().lower() not in {"1", "true", "yes", "on"}:
+    try:
+        _run_startup_self_update()
+    except Exception:
+        logger.exception("Startup self-update bootstrap failed")
+    internal_restart = _internal_restart_in_progress()
+    fast_restart = internal_restart and _fast_restart_enabled()
+    if not fast_restart:
+        try:
+            settings = _pull_executor_settings()
+            _apply_executor_settings(settings)
+        except Exception:
+            logger.exception("Executor settings bootstrap failed")
+        try:
+            _apply_firebase_runtime_settings()
+        except Exception:
+            logger.exception("Firebase runtime bootstrap failed")
+    else:
+        logger.info("Fast internal restart enabled; skipping executor/Firebase startup bootstrap")
+    if not fast_restart:
         _load_default_model()
     try:
-        if str(os.getenv("KAPO_LAZY_EMBED_STARTUP", "1")).strip().lower() not in {"1", "true", "yes", "on"}:
+        if not fast_restart:
             _load_embed_model()
     except Exception:
         logger.exception("Embedding model startup failed")
     try:
-        start_tunnel = _auto_publish_public_url_on_startup() and not _internal_restart_in_progress()
+        start_tunnel = _auto_publish_public_url_on_startup() and not internal_restart
         _bootstrap_executor_handshake(start_tunnel=start_tunnel)
     except Exception:
         logger.exception("Executor handshake startup failed")
     finally:
         os.environ["KAPO_INTERNAL_RESTART"] = "0"
-    FIREBASE.set_document(
-        "brains",
-        os.getenv("BRAIN_PUBLIC_URL") or os.getenv("KAPO_RUNTIME_ROOT") or "brain_runtime",
-        {
-            "url": os.getenv("BRAIN_PUBLIC_URL", ""),
-            "runtime_root": os.getenv("KAPO_RUNTIME_ROOT", ""),
-            "sync_root": os.getenv("KAPO_SYNC_ROOT", ""),
-            "model_repo": os.getenv("MODEL_REPO", DEFAULT_MODEL_REPO),
-            "model_file": os.getenv("MODEL_FILE", DEFAULT_MODEL_FILE),
-            "model_profile_id": os.getenv("MODEL_PROFILE_ID", DEFAULT_MODEL_PROFILE_ID),
-            "roles": os.getenv("BRAIN_ROLES", "supervisor,chat,coding,planner,arabic,fallback"),
-            "languages": os.getenv("BRAIN_LANGUAGES", "ar,en"),
-            "status": "starting",
-        },
-    )
+    _persist_runtime_state_snapshot(reason="startup")
+    if not RUNTIME_STATE_THREAD_STARTED:
+        RUNTIME_STATE_THREAD_STARTED = True
+        threading.Thread(target=_runtime_state_pulse, daemon=True).start()
 
 
 class ModelLoadRequest(BaseModel):
@@ -1461,6 +1720,7 @@ def _restart_process(delay_sec: float = 1.0) -> None:
         target_root = _sync_target_root()
         os.chdir(target_root)
         os.environ["KAPO_INTERNAL_RESTART"] = "1"
+        os.environ.setdefault("KAPO_FAST_INTERNAL_RESTART", "1")
         if _reuse_public_url_on_restart():
             current_public_url = _load_saved_public_url()
             if current_public_url:
@@ -1515,6 +1775,15 @@ async def runtime_errors(limit: int = 50, level: str = "WARNING"):
     return {"status": "ok", "count": len(items[-limit:]), "items": items[-limit:]}
 
 
+@app.get("/runtime/modernization")
+async def runtime_modernization():
+    try:
+        return {"status": "ok", "modernization": _runtime_modernization_snapshot()}
+    except Exception as exc:
+        logger.warning("Failed to build runtime modernization snapshot", exc_info=True)
+        return {"status": "degraded", "detail": str(exc), "modernization": {}}
+
+
 @app.get("/model/status")
 async def model_status():
     return {
@@ -1668,7 +1937,7 @@ async def chat(req: ChatRequest):
 
 @app.post("/init-connection")
 async def init_connection(payload: ConnectionInit):
-    os.environ["EXECUTOR_URL"] = payload.executor_url
+    _remember_executor_url(payload.executor_url)
     FIREBASE.set_document("runtime", "executor", {"executor_url": payload.executor_url})
     public_url = _report_known_public_url()
     if not public_url:
@@ -1686,15 +1955,17 @@ async def system_publish_url(req: PublishUrlRequest | None = None):
         FIREBASE.set_document("brains", explicit_public_url, {"url": explicit_public_url, "status": "healthy", "source": "explicit_publish"})
         return {"status": "published", "brain_public_url": explicit_public_url, "mode": "explicit"}
 
-    public_url = _report_known_public_url()
-    if public_url:
-        return {"status": "published", "brain_public_url": public_url, "mode": "saved"}
-
     if not payload.start_tunnel:
+        public_url = _report_known_public_url()
+        if public_url:
+            return {"status": "published", "brain_public_url": public_url, "mode": "saved"}
         return {"status": "skipped", "brain_public_url": None, "mode": "none"}
 
     public_url = start_ngrok(payload.ngrok_token)
-    return {"status": "published" if public_url else "error", "brain_public_url": public_url, "mode": "ngrok"}
+    if public_url:
+        return {"status": "published", "brain_public_url": public_url, "mode": "ngrok"}
+    public_url = _report_known_public_url()
+    return {"status": "published" if public_url else "error", "brain_public_url": public_url, "mode": "saved" if public_url else "none"}
 
 
 @app.get("/system/files")
@@ -1825,8 +2096,7 @@ async def system_restart(req: RestartRequest | None = None):
     }
 
 
-@app.get("/health")
-async def health(executor_url: str | None = None, check_executor: bool = False):
+def _health_payload(check_executor: bool = False, executor_url: str | None = None) -> dict[str, Any]:
     cfg = load_config()
     base_exec_url = (executor_url or os.getenv("EXECUTOR_URL", "")).strip().rstrip("/")
     exec_ok = False
@@ -1853,7 +2123,7 @@ async def health(executor_url: str | None = None, check_executor: bool = False):
                 exec_error = str(exc)
 
     faiss_path = cfg.get("FAISS_INDEX_PATH")
-    payload = {
+    return {
         "status": "ok",
         "model_loaded": MODEL is not None,
         "model_error": MODEL_ERROR,
@@ -1867,11 +2137,88 @@ async def health(executor_url: str | None = None, check_executor: bool = False):
         "sync_root": _sync_target_root(),
         "timestamp": time.time(),
     }
+
+
+def _decode_b64_text(value: str) -> str:
+    raw = str(value or "").strip()
+    if not raw:
+        return ""
+    padding = "=" * (-len(raw) % 4)
+    try:
+        return base64.urlsafe_b64decode((raw + padding).encode("utf-8")).decode("utf-8").strip()
+    except Exception:
+        return ""
+
+
+def _runtime_modernization_snapshot() -> dict[str, Any]:
+    bootstrap = DRIVE_STATE.ensure_bootstrap_loaded(force=False) or {}
+    patch_manifest_url = str(
+        os.getenv("KAPO_PATCH_MANIFEST_URL", "")
+        or bootstrap.get("patch_manifest_url")
+        or ""
+    ).strip()
+    remote_env_url = _decode_b64_text(os.getenv("KAPO_REMOTE_ENV_URL_B64", "")) or str(os.getenv("KAPO_REMOTE_ENV_URL", "") or "").strip()
+    public_url = str(os.getenv("BRAIN_PUBLIC_URL", "") or LAST_BRAIN_URL_REPORT.get("url") or _load_saved_public_url() or "").strip()
+    applied_version = _load_applied_runtime_version()
+    control_plane_url = str(os.getenv("KAPO_CONTROL_PLANE_URL", "") or "").strip()
+    queue_name = str(os.getenv("KAPO_CLOUDFLARE_QUEUE_NAME", "") or "").strip()
+    return {
+        "shared_state_backend": _shared_state_backend(),
+        "firebase_enabled": str(os.getenv("FIREBASE_ENABLED", "0")).strip().lower() in {"1", "true", "yes", "on"},
+        "drive": {
+            "folder_id": str(os.getenv("GOOGLE_DRIVE_SHARED_STATE_FOLDER_ID", "") or os.getenv("GOOGLE_DRIVE_STORAGE_FOLDER_ID", "") or "").strip(),
+            "prefix": str(os.getenv("GOOGLE_DRIVE_SHARED_STATE_PREFIX", "") or os.getenv("GOOGLE_DRIVE_STORAGE_PREFIX", "") or "").strip(),
+            "bootstrap_loaded": bool(bootstrap),
+        },
+        "bootstrap": {
+            "configured": bool(str(os.getenv("KAPO_BOOTSTRAP_URL", "") or os.getenv("GOOGLE_DRIVE_BOOTSTRAP_URL", "") or "").strip()),
+            "url": str(os.getenv("KAPO_BOOTSTRAP_URL", "") or os.getenv("GOOGLE_DRIVE_BOOTSTRAP_URL", "") or "").strip(),
+            "loaded": bool(bootstrap),
+            "version": str(bootstrap.get("version") or "").strip(),
+            "updated_at": bootstrap.get("updated_at"),
+        },
+        "patch": {
+            "configured": bool(patch_manifest_url),
+            "manifest_url": patch_manifest_url,
+            "startup_self_update_enabled": _startup_self_update_enabled(),
+            "applied_version": applied_version,
+        },
+        "remote_env": {
+            "configured": bool(remote_env_url),
+            "loaded": str(os.getenv("KAPO_REMOTE_ENV_LOADED", "0")).strip().lower() in {"1", "true", "yes", "on"},
+            "url": remote_env_url,
+        },
+        "control_plane": {
+            "configured": bool(control_plane_url),
+            "url": control_plane_url,
+            "queue_name": queue_name,
+            "queue_configured": bool(queue_name),
+        },
+        "transport": {
+            "provider": str(os.getenv("BRAIN_TUNNEL_PROVIDER", "ngrok") or "ngrok").strip().lower(),
+            "auto_ngrok": _ngrok_bootstrap_enabled(),
+            "reuse_public_url_on_restart": _reuse_public_url_on_restart(),
+            "public_url": public_url,
+            "executor_url": str(os.getenv("EXECUTOR_URL", "") or "").strip(),
+        },
+        "runtime": {
+            "root": str(_sync_target_root()),
+            "model_profile_id": str(os.getenv("MODEL_PROFILE_ID", "") or "").strip(),
+            "model_loaded": MODEL is not None,
+            "embed_loaded": EMBED_MODEL is not None,
+        },
+    }
+
+
+def _persist_runtime_state_snapshot(reason: str = "periodic") -> dict[str, Any]:
+    payload = _health_payload(check_executor=False)
+    public_url = os.getenv("BRAIN_PUBLIC_URL") or _load_saved_public_url() or ""
+    state_id = _brain_state_id()
     FIREBASE.set_document(
         "brains",
-        os.getenv("BRAIN_PUBLIC_URL") or _load_saved_public_url() or os.getenv("KAPO_RUNTIME_ROOT") or "brain_runtime",
+        public_url or state_id,
         {
-            "url": os.getenv("BRAIN_PUBLIC_URL") or _load_saved_public_url() or "",
+            "url": public_url,
             "health": payload,
             "status": "healthy" if payload["model_loaded"] else "degraded",
             "model_repo": os.getenv("MODEL_REPO", DEFAULT_MODEL_REPO),
@@ -1879,8 +2226,40 @@ async def health(executor_url: str | None = None, check_executor: bool = False):
             "model_profile_id": os.getenv("MODEL_PROFILE_ID", DEFAULT_MODEL_PROFILE_ID),
             "roles": os.getenv("BRAIN_ROLES", "supervisor,chat,coding,planner,arabic,fallback"),
             "languages": os.getenv("BRAIN_LANGUAGES", "ar,en"),
+            "updated_at": time.time(),
+            "source": reason,
         },
+        min_interval_sec=5.0,
     )
+    FIREBASE.set_document(
+        "brain_runtime",
+        state_id,
+        {
+            "brain_url": public_url,
+            "reason": reason,
+            "health": payload,
+            "recent_logs": list(RUNTIME_LOG_BUFFER)[-25:],
+            "updated_at": time.time(),
+        },
+        min_interval_sec=5.0,
+    )
+    return payload
+
+
+def _runtime_state_pulse() -> None:
+    interval = max(20.0, float(os.getenv("BRAIN_STATE_SYNC_INTERVAL_SEC", "60") or 60))
+    while True:
+        try:
+            _persist_runtime_state_snapshot(reason="background_pulse")
+        except Exception:
+            logger.warning("Background runtime state sync failed", exc_info=True)
+        time.sleep(interval)
+
+
+@app.get("/health")
+async def health(executor_url: str | None = None, check_executor: bool = False):
+    payload = _health_payload(check_executor=check_executor, executor_url=executor_url)
+    _persist_runtime_state_snapshot(reason="health_endpoint")
     return payload
 
 

brain_server/requirements.txt

@@ -15,6 +15,7 @@ huggingface_hub>=0.33.5,<2.0
 pyngrok==7.1.6
 firebase-admin==6.5.0
 starlette>=0.40.0,<1.0
+websockets==12.0
 
 sentence-transformers
 transformers

requirements.txt

@@ -1,18 +1,9 @@
-numpy<2.0.0
 fastapi>=0.115.2,<1.0
 uvicorn==0.30.1
 pydantic==2.7.3
 python-dotenv==1.0.1
+PyYAML>=6.0,<7.0
 requests==2.32.3
-python-multipart>=0.0.18
-sqlalchemy==2.0.30
-sentence-transformers==3.0.1
-faiss-cpu==1.8.0
 python-json-logger==2.0.7
-langgraph==0.0.50
 huggingface_hub>=0.33.5,<2.0
-firebase-admin==6.5.0
 starlette>=0.40.0,<1.0
-transformers>=4.46.0,<5.0
-accelerate>=0.34.0,<1.0
-datasets>=2.21.0,<4.0

shared/__init__.py

@@ -0,0 +1 @@
+"""Shared utilities for KAPO runtimes."""

shared/google_drive_state.py

@@ -0,0 +1,409 @@
+"""Google Drive backed shared-state utilities."""
+
+from __future__ import annotations
+
+import base64
+import json
+import logging
+import os
+import sqlite3
+import threading
+import time
+from pathlib import Path
+from typing import Any
+
+import requests
+
+
+def _safe_name(value: str, default: str = "item") -> str:
+    text = str(value or "").strip() or default
+    return "".join(ch if ch.isalnum() or ch in {"-", "_", "."} else "_" for ch in text)[:180]
+
+
+class GoogleDriveStateClient:
+    def __init__(self, logger: logging.Logger) -> None:
+        self.logger = logger
+        self._folder_cache: dict[str, str] = {}
+        self._file_cache: dict[str, str] = {}
+        self._bootstrap_lock = threading.Lock()
+        self._bootstrap_loaded_at = 0.0
+        self._bootstrap_payload: dict[str, Any] = {}
+
+    def _bootstrap_url(self) -> str:
+        return str(os.getenv("GOOGLE_DRIVE_BOOTSTRAP_URL", "") or os.getenv("KAPO_BOOTSTRAP_URL", "") or "").strip()
+
+    def _bootstrap_ttl(self) -> float:
+        return float(os.getenv("GOOGLE_DRIVE_BOOTSTRAP_TTL_SEC", "300") or 300)
+
+    def _apply_bootstrap_payload(self, payload: dict[str, Any]) -> None:
+        mappings = {
+            "shared_state_backend": "KAPO_SHARED_STATE_BACKEND",
+            "shared_state_folder_id": "GOOGLE_DRIVE_SHARED_STATE_FOLDER_ID",
+            "shared_state_prefix": "GOOGLE_DRIVE_SHARED_STATE_PREFIX",
+            "storage_folder_id": "GOOGLE_DRIVE_STORAGE_FOLDER_ID",
+            "storage_prefix": "GOOGLE_DRIVE_STORAGE_PREFIX",
+            "access_token": "GOOGLE_DRIVE_ACCESS_TOKEN",
+            "refresh_token": "GOOGLE_DRIVE_REFRESH_TOKEN",
+            "client_secret_json": "GOOGLE_DRIVE_CLIENT_SECRET_JSON",
+            "client_secret_json_base64": "GOOGLE_DRIVE_CLIENT_SECRET_JSON_BASE64",
+            "client_secret_path": "GOOGLE_DRIVE_CLIENT_SECRET_PATH",
+            "token_expires_at": "GOOGLE_DRIVE_TOKEN_EXPIRES_AT",
+            "firebase_enabled": "FIREBASE_ENABLED",
+            "executor_url": "EXECUTOR_URL",
+            "executor_public_url": "EXECUTOR_PUBLIC_URL",
+            "ngrok_authtoken": "NGROK_AUTHTOKEN",
+            "brain_public_url": "BRAIN_PUBLIC_URL",
+            "brain_roles": "BRAIN_ROLES",
+            "brain_languages": "BRAIN_LANGUAGES",
+        }
+        for key, env_name in mappings.items():
+            value = payload.get(key)
+            if value not in (None, ""):
+                os.environ[env_name] = str(value)
+
+    def ensure_bootstrap_loaded(self, force: bool = False) -> dict[str, Any]:
+        bootstrap_url = self._bootstrap_url()
+        if not bootstrap_url:
+            return {}
+        now = time.time()
+        if not force and self._bootstrap_payload and (now - self._bootstrap_loaded_at) < self._bootstrap_ttl():
+            return dict(self._bootstrap_payload)
+        with self._bootstrap_lock:
+            now = time.time()
+            if not force and self._bootstrap_payload and (now - self._bootstrap_loaded_at) < self._bootstrap_ttl():
+                return dict(self._bootstrap_payload)
+            try:
+                response = requests.get(bootstrap_url, timeout=20)
+                response.raise_for_status()
+                payload = dict(response.json() or {})
+                self._bootstrap_payload = payload
+                self._bootstrap_loaded_at = time.time()
+                self._apply_bootstrap_payload(payload)
+                return dict(payload)
+            except Exception:
+                self.logger.warning("Failed to load Google Drive bootstrap manifest", exc_info=True)
+                return dict(self._bootstrap_payload)
+
+    def enabled(self) -> bool:
+        self.ensure_bootstrap_loaded()
+        return bool(self.root_folder_id())
+
+    def root_folder_id(self) -> str:
+        return str(
+            os.getenv("GOOGLE_DRIVE_SHARED_STATE_FOLDER_ID", "")
+            or os.getenv("GOOGLE_DRIVE_STORAGE_FOLDER_ID", "")
+            or ""
+        ).strip()
+
+    def prefix(self) -> str:
+        return str(
+            os.getenv("GOOGLE_DRIVE_SHARED_STATE_PREFIX", "")
+            or os.getenv("GOOGLE_DRIVE_STORAGE_PREFIX", "")
+            or "kapo/shared_state"
+        ).strip("/ ")
+
+    def _control_plane_token(self, label: str) -> str:
+        candidates = [
+            Path.cwd().resolve() / "data" / "local" / "control_plane.db",
+            Path.cwd().resolve() / "data" / "drive_cache" / "control_plane.db",
+        ]
+        for db_path in candidates:
+            if not db_path.exists():
+                continue
+            try:
+                conn = sqlite3.connect(db_path)
+                conn.row_factory = sqlite3.Row
+                row = conn.execute(
+                    "SELECT token FROM ngrok_tokens WHERE label = ? AND enabled = 1",
+                    (label,),
+                ).fetchone()
+                conn.close()
+                token = str(row["token"]) if row else ""
+                if token:
+                    return token
+            except Exception:
+                self.logger.warning("Failed to read Google Drive token from %s", db_path, exc_info=True)
+        return ""
+
+    def _client_secret_payload(self) -> dict[str, Any] | None:
+        raw = str(os.getenv("GOOGLE_DRIVE_CLIENT_SECRET_JSON", "")).strip()
+        raw_b64 = str(os.getenv("GOOGLE_DRIVE_CLIENT_SECRET_JSON_BASE64", "")).strip()
+        path = str(os.getenv("GOOGLE_DRIVE_CLIENT_SECRET_PATH", "")).strip()
+        if raw:
+            try:
+                return json.loads(raw)
+            except Exception:
+                self.logger.warning("Invalid GOOGLE_DRIVE_CLIENT_SECRET_JSON", exc_info=True)
+        if raw_b64:
+            try:
+                return json.loads(base64.b64decode(raw_b64).decode("utf-8"))
+            except Exception:
+                self.logger.warning("Invalid GOOGLE_DRIVE_CLIENT_SECRET_JSON_BASE64", exc_info=True)
+        if path:
+            try:
+                return json.loads(Path(path).expanduser().read_text(encoding="utf-8"))
+            except Exception:
+                self.logger.warning("Failed to load Google Drive client secret path %s", path, exc_info=True)
+        client_id = str(os.getenv("GOOGLE_DRIVE_CLIENT_ID", "")).strip()
+        client_secret = str(os.getenv("GOOGLE_DRIVE_CLIENT_SECRET", "")).strip()
+        token_uri = str(os.getenv("GOOGLE_DRIVE_TOKEN_URI", "https://oauth2.googleapis.com/token")).strip()
+        if client_id and client_secret:
+            return {
+                "installed": {
+                    "client_id": client_id,
+                    "client_secret": client_secret,
+                    "token_uri": token_uri,
+                }
+            }
+        return None
+
+    def _oauth_client(self) -> dict[str, str]:
+        payload = self._client_secret_payload() or {}
+        installed = payload.get("installed") or payload.get("web") or {}
+        return {
+            "client_id": str(installed.get("client_id") or "").strip(),
+            "client_secret": str(installed.get("client_secret") or "").strip(),
+            "token_uri": str(installed.get("token_uri") or "https://oauth2.googleapis.com/token").strip(),
+        }
+
+    def _refresh_token(self) -> str:
+        return str(os.getenv("GOOGLE_DRIVE_REFRESH_TOKEN", "") or self._control_plane_token("google_drive_refresh_token") or "").strip()
+
+    def _access_token(self) -> str:
+        token = str(os.getenv("GOOGLE_DRIVE_ACCESS_TOKEN", "") or self._control_plane_token("google_drive_access_token") or "").strip()
+        expires_at = float(str(os.getenv("GOOGLE_DRIVE_TOKEN_EXPIRES_AT", "0") or "0") or 0)
+        if token and (not expires_at or expires_at > (time.time() + 60)):
+            return token
+        refreshed = self._refresh_access_token()
+        return refreshed or token
+
+    def _refresh_access_token(self) -> str:
+        refresh_token = self._refresh_token()
+        client = self._oauth_client()
+        if not refresh_token or not client["client_id"] or not client["client_secret"]:
+            return ""
+        response = requests.post(
+            client["token_uri"],
+            data={
+                "grant_type": "refresh_token",
+                "refresh_token": refresh_token,
+                "client_id": client["client_id"],
+                "client_secret": client["client_secret"],
+            },
+            timeout=30,
+        )
+        response.raise_for_status()
+        payload = response.json()
+        access_token = str(payload.get("access_token") or "").strip()
+        expires_in = float(payload.get("expires_in") or 0)
+        if access_token:
+            os.environ["GOOGLE_DRIVE_ACCESS_TOKEN"] = access_token
+            os.environ["GOOGLE_DRIVE_TOKEN_EXPIRES_AT"] = str(time.time() + expires_in) if expires_in else "0"
+        return access_token
+
+    def _request(self, method: str, url: str, *, headers: dict[str, str] | None = None, timeout: int = 60, **kwargs) -> requests.Response:
+        token = self._access_token()
+        if not token:
+            raise ValueError("Google Drive access token is not configured")
+        base_headers = dict(headers or {})
+        response: requests.Response | None = None
+        for _ in range(2):
+            response = requests.request(
+                method,
+                url,
+                headers={**base_headers, "Authorization": f"Bearer {token}"},
+                timeout=timeout,
+                **kwargs,
+            )
+            if response.status_code != 401:
+                response.raise_for_status()
+                return response
+            token = self._refresh_access_token()
+            if not token:
+                break
+        assert response is not None
+        response.raise_for_status()
+        return response
+
+    @staticmethod
+    def _escape_query(value: str) -> str:
+        return str(value or "").replace("\\", "\\\\").replace("'", "\\'")
+
+    def _find_folder(self, name: str, parent_id: str) -> str:
+        query = (
+            "mimeType = 'application/vnd.google-apps.folder' and trashed = false "
+            f"and name = '{self._escape_query(name)}'"
+        )
+        if parent_id:
+            query += f" and '{self._escape_query(parent_id)}' in parents"
+        response = self._request(
+            "GET",
+            "https://www.googleapis.com/drive/v3/files",
+            params={"q": query, "fields": "files(id,name)", "pageSize": 10, "supportsAllDrives": "true"},
+        )
+        files = response.json().get("files", [])
+        return str(files[0].get("id") or "").strip() if files else ""
+
+    def _create_folder(self, name: str, parent_id: str) -> str:
+        payload: dict[str, Any] = {"name": name, "mimeType": "application/vnd.google-apps.folder"}
+        if parent_id:
+            payload["parents"] = [parent_id]
+        response = self._request(
+            "POST",
+            "https://www.googleapis.com/drive/v3/files?supportsAllDrives=true",
+            headers={"Content-Type": "application/json; charset=UTF-8"},
+            json=payload,
+        )
+        return str(response.json().get("id") or "").strip()
+
+    def _ensure_folder_path(self, relative_path: str) -> str:
+        current_parent = self.root_folder_id()
+        parts = [part for part in f"{self.prefix()}/{relative_path}".replace("\\", "/").split("/") if part.strip()]
+        for part in parts:
+            safe_part = _safe_name(part, "folder")
+            cache_key = f"{current_parent}:{safe_part}"
+            cached = self._folder_cache.get(cache_key)
+            if cached:
+                current_parent = cached
+                continue
+            folder_id = self._find_folder(safe_part, current_parent)
+            if not folder_id:
+                folder_id = self._create_folder(safe_part, current_parent)
+            self._folder_cache[cache_key] = folder_id
+            current_parent = folder_id
+        return current_parent
+
+    def _find_file(self, folder_id: str, file_name: str) -> str:
+        cache_key = f"{folder_id}:{file_name}"
+        cached = self._file_cache.get(cache_key)
+        if cached:
+            return cached
+        query = (
+            "trashed = false "
+            f"and name = '{self._escape_query(file_name)}' "
+            f"and '{self._escape_query(folder_id)}' in parents"
+        )
+        response = self._request(
+            "GET",
+            "https://www.googleapis.com/drive/v3/files",
+            params={"q": query, "fields": "files(id,name)", "pageSize": 10, "supportsAllDrives": "true"},
+        )
+        files = response.json().get("files", [])
+        file_id = str(files[0].get("id") or "").strip() if files else ""
+        if file_id:
+            self._file_cache[cache_key] = file_id
+        return file_id
+
+    def _create_file(self, folder_id: str, file_name: str) -> str:
+        response = self._request(
+            "POST",
+            "https://www.googleapis.com/drive/v3/files?supportsAllDrives=true",
+            headers={"Content-Type": "application/json; charset=UTF-8"},
+            json={"name": file_name, "parents": [folder_id]},
+        )
+        file_id = str(response.json().get("id") or "").strip()
+        if file_id:
+            self._file_cache[f"{folder_id}:{file_name}"] = file_id
+        return file_id
+
+    def _upload_json(self, file_id: str, payload: dict[str, Any]) -> None:
+        self._request(
+            "PATCH",
+            f"https://www.googleapis.com/upload/drive/v3/files/{file_id}?uploadType=media&supportsAllDrives=true",
+            headers={"Content-Type": "application/json; charset=UTF-8"},
+            data=json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8"),
+        )
+
+    def _download_json(self, file_id: str) -> dict[str, Any]:
+        response = self._request(
+            "GET",
+            f"https://www.googleapis.com/drive/v3/files/{file_id}?alt=media&supportsAllDrives=true",
+        )
+        return dict(response.json() or {})
+
+    def get_document(self, collection: str, doc_id: str) -> dict[str, Any]:
+        if not self.enabled():
+            return {}
+        folder_id = self._ensure_folder_path(_safe_name(collection, "collection"))
+        file_name = f"{_safe_name(doc_id)}.json"
+        file_id = self._find_file(folder_id, file_name)
+        if not file_id:
+            return {}
+        try:
+            payload = self._download_json(file_id)
+            payload.setdefault("id", _safe_name(doc_id))
+            return payload
+        except Exception:
+            self.logger.warning("Failed to download shared-state file %s/%s", collection, doc_id, exc_info=True)
+            return {}
+
+    def set_document(self, collection: str, doc_id: str, payload: dict[str, Any], *, merge: bool = True) -> bool:
+        if not self.enabled():
+            return False
+        safe_doc = _safe_name(doc_id)
+        try:
+            existing = self.get_document(collection, safe_doc) if merge else {}
+            combined = {**existing, **dict(payload or {})} if merge else dict(payload or {})
+            combined.setdefault("id", safe_doc)
+            folder_id = self._ensure_folder_path(_safe_name(collection, "collection"))
+            file_name = f"{safe_doc}.json"
+            file_id = self._find_file(folder_id, file_name)
+            if not file_id:
+                file_id = self._create_file(folder_id, file_name)
+            self._upload_json(file_id, combined)
+            return True
+        except Exception:
+            self.logger.warning("Failed to upload shared-state file %s/%s", collection, safe_doc, exc_info=True)
+            return False
+
+    def list_documents(self, collection: str, *, limit: int = 200) -> list[dict[str, Any]]:
+        if not self.enabled():
+            return []
+        try:
+            folder_id = self._ensure_folder_path(_safe_name(collection, "collection"))
+            query = f"trashed = false and '{self._escape_query(folder_id)}' in parents"
+            response = self._request(
+                "GET",
+                "https://www.googleapis.com/drive/v3/files",
+                params={
+                    "q": query,
+                    "fields": "files(id,name,modifiedTime)",
+                    "pageSize": max(1, int(limit)),
+                    "supportsAllDrives": "true",
+                    "orderBy": "modifiedTime desc",
+                },
+            )
+            items: list[dict[str, Any]] = []
+            for file in response.json().get("files", []):
+                file_id = str(file.get("id") or "").strip()
+                file_name = str(file.get("name") or "").strip()
+                if not file_id or not file_name.endswith(".json"):
+                    continue
+                self._file_cache[f"{folder_id}:{file_name}"] = file_id
+                payload = self._download_json(file_id)
+                payload.setdefault("id", file_name[:-5])
+                items.append(payload)
+            return items
+        except Exception:
+            self.logger.warning("Failed to list shared-state collection %s", collection, exc_info=True)
+            return []
+
+    def delete_document(self, collection: str, doc_id: str) -> bool:
+        if not self.enabled():
+            return False
+        safe_doc = _safe_name(doc_id)
+        try:
+            folder_id = self._ensure_folder_path(_safe_name(collection, "collection"))
+            file_name = f"{safe_doc}.json"
+            file_id = self._find_file(folder_id, file_name)
+            if not file_id:
+                return True
+            self._request(
+                "DELETE",
+                f"https://www.googleapis.com/drive/v3/files/{file_id}?supportsAllDrives=true",
+            )
+            self._file_cache.pop(f"{folder_id}:{file_name}", None)
+            return True
+        except Exception:
+            self.logger.warning("Failed to delete shared-state file %s/%s", collection, safe_doc, exc_info=True)
+            return False

shared/remote_env.py

@@ -0,0 +1,102 @@
+"""Encrypted remote .env loader and publisher helpers."""
+
+from __future__ import annotations
+
+import base64
+import io
+import json
+import logging
+import os
+from typing import Any
+
+import requests
+from cryptography.fernet import Fernet
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
+from dotenv import dotenv_values
+
+
+def _decode_maybe_b64(value: str) -> str:
+    text = str(value or "").strip()
+    if not text:
+        return ""
+    try:
+        padded = text + "=" * (-len(text) % 4)
+        decoded = base64.urlsafe_b64decode(padded.encode("utf-8")).decode("utf-8")
+        if decoded:
+            return decoded
+    except Exception:
+        pass
+    return text
+
+
+def _derive_key(password: str, salt: bytes) -> bytes:
+    kdf = PBKDF2HMAC(
+        algorithm=hashes.SHA256(),
+        length=32,
+        salt=salt,
+        iterations=390000,
+    )
+    return base64.urlsafe_b64encode(kdf.derive(password.encode("utf-8")))
+
+
+def build_remote_env_bundle(env_text: str, password: str, metadata: dict[str, Any] | None = None) -> dict[str, Any]:
+    salt = os.urandom(16)
+    key = _derive_key(password, salt)
+    token = Fernet(key).encrypt((env_text or "").encode("utf-8"))
+    return {
+        "schema": "kapo-remote-env-v1",
+        "salt": base64.urlsafe_b64encode(salt).decode("utf-8"),
+        "token": token.decode("utf-8"),
+        "metadata": dict(metadata or {}),
+    }
+
+
+def decrypt_remote_env_bundle(payload: dict[str, Any], password: str) -> str:
+    salt_raw = str(payload.get("salt") or "").strip()
+    token = str(payload.get("token") or "").strip()
+    if not salt_raw or not token:
+        raise ValueError("Remote env payload is incomplete")
+    salt = base64.urlsafe_b64decode(salt_raw.encode("utf-8"))
+    key = _derive_key(password, salt)
+    return Fernet(key).decrypt(token.encode("utf-8")).decode("utf-8")
+
+
+def _remote_env_url() -> str:
+    return _decode_maybe_b64(
+        str(os.getenv("KAPO_REMOTE_ENV_URL", "") or os.getenv("KAPO_REMOTE_ENV_URL_B64", "") or "").strip()
+    )
+
+
+def _remote_env_password() -> str:
+    return _decode_maybe_b64(
+        str(os.getenv("KAPO_REMOTE_ENV_PASSWORD", "") or os.getenv("KAPO_REMOTE_ENV_PASSWORD_B64", "") or "").strip()
+    )
+
+
+def load_remote_env_if_configured(*, override: bool = True, logger_name: str = "kapo.remote_env") -> dict[str, str]:
+    logger = logging.getLogger(logger_name)
+    if str(os.getenv("KAPO_REMOTE_ENV_LOADED", "")).strip().lower() in {"1", "true", "yes", "on"}:
+        return {}
+    url = _remote_env_url()
+    password = _remote_env_password()
+    if not url or not password:
+        return {}
+    try:
+        response = requests.get(url, timeout=30)
+        response.raise_for_status()
+        payload = dict(response.json() or {})
+        env_text = decrypt_remote_env_bundle(payload, password)
+        parsed = {
+            str(key): str(value)
+            for key, value in dotenv_values(stream=io.StringIO(env_text)).items()
+            if key and value is not None
+        }
+        for key, value in parsed.items():
+            if override or key not in os.environ or not str(os.getenv(key) or "").strip():
+                os.environ[key] = value
+        os.environ["KAPO_REMOTE_ENV_LOADED"] = "1"
+        return parsed
+    except Exception:
+        logger.warning("Failed to load encrypted remote env bundle", exc_info=True)
+        return {}