"""Storage HUD — a fast, lazy file browser over Supabase Storage. The bucket 'jarvis-files' (on Ahmed's Supabase) is the WRITABLE shared file store that any Jarvis / Claude on any device can read, save, and delete from. This module is the engine side of the HUD's "pull up the storage" panel. DESIGN RULE (the whole point): the LIST is METADATA-ONLY. `list()` / `show()` fetch names + type + size + updated timestamp and NOTHING ELSE — no file bodies. A file is downloaded ONLY when Ahmed actually opens it (`open_file` / `download`), and even then it's cached to disk and re-fetched ONLY when the remote copy is newer. This is deliberately unlike the memory graph, which was slow because it bulk-loaded everything. Ports the proven REST logic from jarvis-tools-service/storage.py (stdlib urllib, no new deps). Endpoints under {SUPABASE_URL}/storage/v1/...: POST /object/list/{bucket} → metadata list (name, updated_at, metadata.size…) GET /object/{bucket}/{name} → the file body (only on open) POST /object/{bucket}/{name} → upload (x-upsert) DELETE /object/{bucket}/{name} Creds come from the process env, else the Ultron repo's .env (the single source of truth — SUPABASE_URL + SUPABASE_SECRET, the service_role key). MCP_STORAGE=0 disables the whole thing. Engine → HUD event (voice.events.emit): storage {files:[{name,type,size,updated}]} — open / refresh the panel """ from __future__ import annotations import mimetypes import os import subprocess import sys import urllib.error import urllib.parse import urllib.request from datetime import datetime from pathlib import Path from voice.events import emit _PROJECT = Path(__file__).resolve().parent.parent # claude-voice/ _ULTRON_ENV = Path(os.environ.get("ULTRON_ENV", str(_PROJECT.parent / ".env"))) # Ultron/.env BUCKET = os.environ.get("STORAGE_BUCKET", "jarvis-files") # Local cache for opened files — download-on-click lands here and is reused # until the remote copy is newer (see download()). CACHE = Path(os.environ.get( "JARVIS_STORAGE_CACHE", str(Path.home() / "Library" / "Application Support" / "Jarvis" / "storage"))) _PLACEHOLDER = ".emptyFolderPlaceholder" # --------------------------------------------------------------------------- # Credentials — process env first, then the Ultron .env (mirrors ultron_db_control) # --------------------------------------------------------------------------- def _creds() -> tuple[str, str] | None: """(base_url, service_key) from env, falling back to the Ultron repo .env.""" url = os.environ.get("SUPABASE_URL", "") key = os.environ.get("SUPABASE_SECRET", "") if not (url and key) and _ULTRON_ENV.exists(): try: for line in _ULTRON_ENV.read_text().splitlines(): line = line.strip() if line.startswith("#") or "=" not in line: continue k, _, v = line.partition("=") k, v = k.strip(), v.strip() if k == "SUPABASE_URL" and not url: url = v elif k == "SUPABASE_SECRET" and not key: key = v except Exception: # noqa: BLE001 — unreadable .env just means disabled pass if not (url and key): return None return url.rstrip("/"), key def enabled() -> bool: """On when the Supabase service creds are available and not disabled.""" if os.environ.get("MCP_STORAGE", "1") == "0": return False return _creds() is not None # --------------------------------------------------------------------------- # Low-level REST # --------------------------------------------------------------------------- def _do(method: str, path: str, body: bytes | None = None, headers: dict | None = None, timeout: int = 30) -> bytes: creds = _creds() if not creds: raise RuntimeError("storage not configured (no SUPABASE_URL/SECRET)") base, key = creds req = urllib.request.Request(f"{base}/storage/v1{path}", data=body, method=method) req.add_header("Authorization", f"Bearer {key}") req.add_header("apikey", key) for k, v in (headers or {}).items(): req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=timeout) as r: return r.read() except urllib.error.HTTPError as e: raise RuntimeError(f"HTTP {e.code}: " f"{e.read()[:200].decode('utf-8', 'ignore')}") def _raw_list() -> list[dict]: """Raw Supabase list rows (METADATA ONLY — never the file bodies).""" import json data = _do("POST", f"/object/list/{BUCKET}", json.dumps({"prefix": "", "limit": 1000, "sortBy": {"column": "name", "order": "asc"}} ).encode(), {"Content-Type": "application/json"}) items = json.loads(data) return [i for i in items if i.get("name") and i.get("name") != _PLACEHOLDER] def _kind(name: str, meta: dict) -> str: """A short type tag for the row icon: the extension if any, else the mime subtype (e.g. 'pdf', 'png', 'plain'). Lowercased, no dot.""" ext = Path(name).suffix.lstrip(".").lower() if ext: return ext mime = (meta or {}).get("mimetype") or mimetypes.guess_type(name)[0] or "" return mime.split("/")[-1] if "/" in mime else "" def _epoch(iso: str) -> float: """Parse an ISO-8601 timestamp (with 'Z' / fractional seconds) to epoch seconds. 0.0 on anything unparseable.""" if not iso: return 0.0 try: return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp() except Exception: # noqa: BLE001 return 0.0 # Supabase Storage keys must be ASCII from a limited set — it rejects '%' AND any # non-ASCII (an Arabic filename fails with InvalidKey). So the STORED key is a # reversible escaping of the display name: every byte that isn't a safe keep-char # becomes _, and '_' itself is escaped so round-trips are exact. Ahmed only # ever sees the decoded original name; Supabase only ever sees the safe key. _KEEP = frozenset( "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-() ") _HEX = frozenset("0123456789abcdef") def _enc_key(name: str) -> str: name = str(name).strip().lstrip("/") return "".join(chr(b) if chr(b) in _KEEP else f"_{b:02x}" for b in name.encode("utf-8")) def _dec_name(key: str) -> str: b = bytearray() i, n = 0, len(key) while i < n: if key[i] == "_" and i + 3 <= n and key[i+1] in _HEX and key[i+2] in _HEX: b.append(int(key[i+1:i+3], 16)) i += 3 else: b.append(ord(key[i]) & 0xFF) i += 1 return b.decode("utf-8", "replace") # --------------------------------------------------------------------------- # Public API — list / download / upload / delete / show # --------------------------------------------------------------------------- def list() -> list[dict]: # noqa: A001 — intentional public name per HUD contract """Metadata-only listing: [{name, type, size, updated}] sorted by name. NEVER fetches file content.""" out: list[dict] = [] for i in _raw_list(): meta = i.get("metadata") or {} size = meta.get("size") if size is None: size = meta.get("contentLength") display = _dec_name(i["name"]) # key → original (e.g. Arabic) name out.append({ "name": display, "type": _kind(display, meta), "size": int(size) if isinstance(size, (int, float)) else 0, "updated": i.get("updated_at") or meta.get("lastModified") or "", }) return out def _remote_updated(name: str) -> str: """The remote updated_at for one file (from the metadata list).""" key = _enc_key(name) for i in _raw_list(): if i.get("name") == key: meta = i.get("metadata") or {} return i.get("updated_at") or meta.get("lastModified") or "" return "" def download(name: str, updated: str | None = None) -> str: """Download ONE file to the local cache and return its path. Re-downloads only when the remote copy is newer than the cached one (compared via the remote updated_at vs the cached file's mtime). This is the ONLY path that ever pulls bytes.""" name = str(name).strip().lstrip("/") if not name: raise ValueError("download: need a file name") dest = CACHE / name remote_iso = updated if updated is not None else _remote_updated(name) remote_ts = _epoch(remote_iso) if dest.is_file() and remote_ts and dest.stat().st_mtime >= remote_ts - 1: return str(dest) # cache is up to date — no fetch data = _do("GET", f"/object/{BUCKET}/{urllib.parse.quote(_enc_key(name), safe='/')}") dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(data) if remote_ts: # stamp mtime so the next check is exact try: os.utime(dest, (remote_ts, remote_ts)) except OSError: pass return str(dest) def open_file(name: str) -> str: """Download (cached) then open the file with the OS default app so Ahmed can view/edit it. Returns the local path.""" path = download(name) try: if sys.platform == "darwin": subprocess.Popen(["open", path]) elif os.name == "nt": os.startfile(path) # type: ignore[attr-defined] # noqa: S606 else: subprocess.Popen(["xdg-open", path]) except Exception as e: # noqa: BLE001 — a failed open must not crash the app print(f" [storage] open failed: {str(e)[:120]}") return path def upload(local_path: str, name: str | None = None) -> str: """Upload a LOCAL file into the bucket (upsert). `name` defaults to the file's own basename. Returns the stored name.""" lp = os.path.expanduser(str(local_path).strip()) if not lp or not os.path.isfile(lp): raise FileNotFoundError(f"no file at {lp!r}") name = (str(name).strip().lstrip("/") if name else "") or os.path.basename(lp) mime = mimetypes.guess_type(name)[0] or "application/octet-stream" with open(lp, "rb") as f: body = f.read() _do("POST", f"/object/{BUCKET}/{urllib.parse.quote(_enc_key(name), safe='/')}", body, {"Content-Type": mime, "x-upsert": "true"}) return name def delete(name: str) -> str: """Remove a file from the bucket. Returns the deleted name.""" name = str(name).strip().lstrip("/") if not name: raise ValueError("delete: need a file name") _do("DELETE", f"/object/{BUCKET}/{urllib.parse.quote(_enc_key(name), safe='/')}") # drop the stale local cache copy too try: (CACHE / name).unlink(missing_ok=True) except OSError: pass return name def show() -> list[dict]: """List (metadata only) and push a `storage` event so the HUD opens/refreshes the file panel. Returns the same list for the caller. Never raises — a viewer refresh must not crash the assistant.""" try: files = list() emit("storage", files=files) return files except Exception as e: # noqa: BLE001 print(f" [storage] panel refresh failed: {str(e)[:120]}") emit("storage", files=[]) return [] # --------------------------------------------------------------------------- # In-process MCP server (Claude sees mcp__storage__) # --------------------------------------------------------------------------- def build_server(): import asyncio from claude_agent_sdk import tool, create_sdk_mcp_server def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": msg}]} @tool("storage_show", "Pull up Ahmed's file storage on the HUD and list what's there. Use " "for 'pull up the storage', 'show my files', 'open storage', 'what " "files do I have'. Opens a fast panel (names + types only — nothing is " "downloaded) and returns the list. He can then open/delete files there, " "or you can with storage_open / storage_delete.", {}) async def storage_show(args: dict) -> dict: try: files = await asyncio.to_thread(show) if not files: return _text("Storage is empty.") lines = [f"- {f['name']}" + (f" ({f['size']} bytes)" if f.get("size") else "") for f in files] return _text(f"{len(files)} file(s) in storage (panel is up):\n" + "\n".join(lines)) except Exception as e: # noqa: BLE001 return _text(f"storage_show failed: {str(e)[:180]}") @tool("storage_open", "Open ONE stored file by name — downloads it (cached; only re-fetched " "if it changed) and opens it in its default app so Ahmed can view or " "edit it. `name` is the exact file name from storage_show.", {"name": str}) async def storage_open(args: dict) -> dict: name = str(args.get("name", "")).strip() if not name: return _text("storage_open failed: need a file name.") try: await asyncio.to_thread(open_file, name) return _text(f"Opened {name}.") except Exception as e: # noqa: BLE001 return _text(f"storage_open failed: {str(e)[:180]}") @tool("storage_upload", "Upload a LOCAL file (by absolute path on this Mac) into Ahmed's " "storage so it syncs everywhere. `local_path` = the path; `name` = " "optional name to store it as (defaults to the file's own name). " "Refreshes the HUD panel.", {"local_path": str, "name": str}) async def storage_upload(args: dict) -> dict: lp = str(args.get("local_path", "")).strip() if not lp: return _text("storage_upload failed: need a local_path.") try: name = await asyncio.to_thread( upload, lp, str(args.get("name", "")).strip() or None) await asyncio.to_thread(show) # refresh the panel return _text(f"Uploaded {name} to storage.") except Exception as e: # noqa: BLE001 return _text(f"storage_upload failed: {str(e)[:180]}") @tool("storage_delete", "Delete a file from Ahmed's storage by name (from storage_show). " "Refreshes the HUD panel.", {"name": str}) async def storage_delete(args: dict) -> dict: name = str(args.get("name", "")).strip() if not name: return _text("storage_delete failed: need a file name.") try: await asyncio.to_thread(delete, name) await asyncio.to_thread(show) # refresh the panel return _text(f"Deleted {name} from storage.") except Exception as e: # noqa: BLE001 return _text(f"storage_delete failed: {str(e)[:180]}") return create_sdk_mcp_server( name="storage", version="1.0.0", tools=[storage_show, storage_open, storage_upload, storage_delete], )