Import canonical Phast registry source

registry.py

@@ -0,0 +1,157 @@
+"""Durable SQLite registry for Phast app manifests."""
+from __future__ import annotations
+
+import os
+import re
+import sqlite3
+from contextlib import contextmanager
+from collections.abc import Iterator
+from datetime import datetime, timezone
+from pathlib import Path
+from urllib.parse import urlsplit
+
+SCHEMA_VERSION = 1
+_SLUG = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
+
+class RegistryUnavailable(RuntimeError):
+    pass
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+def _validate_slug(value: str) -> str:
+    value = value.strip()
+    if not _SLUG.fullmatch(value):
+        raise ValueError("slug must be 1-63 lowercase letters, numbers, or hyphens")
+    return value
+
+def _validate_name(value: str) -> str:
+    value = value.strip()
+    if not value or len(value) > 100:
+        raise ValueError("name must be 1-100 characters")
+    return value
+
+def _validate_base_url(value: str) -> str:
+    value = value.strip().rstrip("/")
+    parsed = urlsplit(value)
+    if parsed.scheme != "https" or not parsed.netloc or parsed.path or parsed.query or parsed.fragment:
+        raise ValueError("base_url must be an HTTPS origin without a path")
+    return value
+
+def _validate_route(value: str, field: str) -> str:
+    value = value.strip()
+    if not value.startswith("/") or value.startswith("//") or "?" in value or "#" in value:
+        raise ValueError(f"{field} must be an absolute path without query or fragment")
+    return value
+
+def _manifest(row: sqlite3.Row) -> dict:
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "kind": "phast.app",
+        "slug": row["slug"],
+        "name": row["name"],
+        "runtime": "python",
+        "base_url": row["base_url"],
+        "human": {"route": row["human_route"]},
+        "mcp": {"route": row["mcp_route"]},
+        "prephab": {"version": 1},
+        "created_at": row["created_at"],
+        "updated_at": row["updated_at"],
+    }
+
+class AppRegistry:
+    def __init__(self, database: str | Path):
+        self.database = Path(database)
+        self.database.parent.mkdir(parents=True, exist_ok=True)
+        self._initialize()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.database, timeout=5)
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA foreign_keys = ON")
+        return connection
+
+    @contextmanager
+    def _connection(self) -> Iterator[sqlite3.Connection]:
+        connection = self._connect()
+        try:
+            with connection:
+                yield connection
+        finally:
+            connection.close()
+
+    def _initialize(self) -> None:
+        with self._connection() as connection:
+            connection.execute("PRAGMA journal_mode = WAL")
+            connection.execute("""
+                CREATE TABLE IF NOT EXISTS apps (
+                    slug TEXT PRIMARY KEY,
+                    name TEXT NOT NULL,
+                    base_url TEXT NOT NULL UNIQUE,
+                    human_route TEXT NOT NULL,
+                    mcp_route TEXT NOT NULL,
+                    created_at TEXT NOT NULL,
+                    updated_at TEXT NOT NULL
+                )
+            """)
+
+    def list(self) -> list[dict]:
+        with self._connection() as connection:
+            rows = connection.execute("SELECT * FROM apps ORDER BY slug").fetchall()
+        return [_manifest(row) for row in rows]
+
+    def get(self, slug: str) -> dict | None:
+        slug = _validate_slug(slug)
+        with self._connection() as connection:
+            row = connection.execute("SELECT * FROM apps WHERE slug = ?", (slug,)).fetchone()
+        return _manifest(row) if row else None
+
+    def create(self, *, slug: str, name: str, base_url: str, human_route: str = "/", mcp_route: str = "/mcp") -> dict:
+        values = (
+            _validate_slug(slug), _validate_name(name), _validate_base_url(base_url),
+            _validate_route(human_route, "human_route"), _validate_route(mcp_route, "mcp_route"),
+        )
+        timestamp = _now()
+        try:
+            with self._connection() as connection:
+                connection.execute(
+                    "INSERT INTO apps (slug, name, base_url, human_route, mcp_route, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
+                    (*values, timestamp, timestamp),
+                )
+        except sqlite3.IntegrityError as error:
+            raise ValueError("an app with this slug or base_url already exists") from error
+        return self.get(values[0])
+
+    def update(self, slug: str, *, name: str | None = None, base_url: str | None = None,
+               human_route: str | None = None, mcp_route: str | None = None) -> dict:
+        slug = _validate_slug(slug)
+        changes = {
+            "name": _validate_name(name) if name is not None else None,
+            "base_url": _validate_base_url(base_url) if base_url is not None else None,
+            "human_route": _validate_route(human_route, "human_route") if human_route is not None else None,
+            "mcp_route": _validate_route(mcp_route, "mcp_route") if mcp_route is not None else None,
+        }
+        changes = {key: value for key, value in changes.items() if value is not None}
+        if not changes:
+            raise ValueError("at least one field must be updated")
+        assignments = ", ".join(f"{key} = ?" for key in changes)
+        try:
+            with self._connection() as connection:
+                cursor = connection.execute(
+                    f"UPDATE apps SET {assignments}, updated_at = ? WHERE slug = ?",
+                    (*changes.values(), _now(), slug),
+                )
+                if cursor.rowcount != 1:
+                    raise KeyError(f"unknown app: {slug}")
+        except sqlite3.IntegrityError as error:
+            raise ValueError("an app with this base_url already exists") from error
+        return self.get(slug)
+
+def registry_from_runtime() -> AppRegistry:
+    raw = os.environ.get("PHAST_DATA_DIR", "").strip()
+    if not raw:
+        raise RegistryUnavailable("PHAST_DATA_DIR is not configured; durable registry is offline")
+    data_dir = Path(raw)
+    if not data_dir.is_dir() or not os.path.ismount(data_dir):
+        raise RegistryUnavailable(f"{data_dir} is not a mounted Fly volume; durable registry is offline")
+    return AppRegistry(data_dir / "registry.sqlite3")