feat(supabase): isolate each app in its own Postgres schema + stack teardown

Rework the supabase stack so every app owns an isolated Postgres schema
(= program name) instead of sharing `public` with a table_prefix. This
gives a clean, knowable teardown and drops migration version tracking to
per-app, so version tokens can no longer collide across apps.

- stacks: add StackHandler.owns_data + teardown() (default no-op). Supabase
  build now creates+grants the app schema, tracks migrations in
  <schema>.schema_migrations, and applies each with search_path set to the
  schema (write unqualified names). teardown = drop schema <app> cascade.
  Fix _substrate_db_url default port 5432 -> 5433 (substrate's direct PG).
- deploy: derive ${supabase_app_schemas} from registered apps; substrate maps
  PGRST_DB_SCHEMAS: public,storage,graphql_public${supabase_app_schemas} so
  PostgREST exposes each app schema (restart substrate after add/remove).
- delete: generic owns_data -> teardown path behind new --purge-data flag,
  replacing the hardcoded supabase remnant note.
- scaffold + docs: new apps born schema-isolated (unqualified tables,
  db:{schema}, schema: in supabase.app.yaml).

cli/src/castle_cli/commands/delete.py

@@ -1,7 +1,15 @@
-"""castle delete — remove a program/service/job from the registry.
+"""castle delete — remove a program/deployment from the registry AND tear it down.
 
-Config-only by default (removes the castle.yaml entry; leaves source and any
-installed binary in place). Use --source to also delete the source directory.
+Deleting cascades: a program's referencing deployments are taken offline (stop +
+disable a service/job, uninstall a tool from PATH, drop a static route) before the
+config entries are removed, then the runtime is reconciled (`deploy()` prunes
+orphan units, regenerates the Caddyfile, reloads the gateway + tunnel) so nothing
+is left running or served. Use --source to also delete the source directory.
+
+Persistent *data* a program's stack created (e.g. a Supabase app's Postgres
+schema) is destroyed only with --purge-data — it survives an ordinary delete and
+is surfaced as a remnant otherwise. One remnant is still only surfaced, not
+auto-removed (pending a DNS-token decision): a public service's Cloudflare CNAME.
 """
 
 from __future__ import annotations
@@ -16,7 +24,7 @@ from castle_cli.config import load_config, save_config
 def run_delete(args: argparse.Namespace) -> int:
     config = load_config()
     name = args.name
-    resource = getattr(args, "resource", None)  # "program" | "service" | "job"
+    resource = getattr(args, "resource", None)  # "program" | "service" | "job" | ...
 
     # Resolve which sections this delete touches (scoped to one resource). Any
     # deployment resource name (service/job/tool/static/deployment) targets the
@@ -29,32 +37,45 @@ def run_delete(args: argparse.Namespace) -> int:
         print(f"Error: no{where} '{name}' in castle.yaml")
         return 1
 
-    where = [s for s, present in
-             (("program", in_programs), ("deployment", in_deployment)) if present]
+    where = [
+        s for s, present in (("program", in_programs), ("deployment", in_deployment)) if present
+    ]
 
-    # A program can't be removed while a deployment still references it. A ref
-    # named the same is removed in this call; any other referencing deployment
-    # would be left dangling, so refuse.
+    # Cascade: every deployment referencing this program is torn down and removed
+    # (a program and its 1:1 service/tool/static are one thing to the user). A
+    # deployment-only delete targets just the co-named deployment.
+    deployments_to_remove: list[str] = []
     if in_programs:
-        dangling = [
-            d for d, spec in config.deployments.items()
-            if spec.program == name and not (d == name and in_deployment)
+        deployments_to_remove = [
+            d for d, spec in config.deployments.items() if spec.program == name
         ]
-        if dangling:
-            print(
-                "Error: programs with active jobs or services cannot be removed.\n"
-                f"  Delete these first: {', '.join(dangling)}"
-            )
-            return 1
+    if in_deployment and name not in deployments_to_remove:
+        deployments_to_remove.append(name)
+
+    # Capture remnant facts BEFORE mutating config: public CNAMEs, and the program
+    # entry itself — its stack may own persistent data (a DB schema) that survives
+    # a code delete unless --purge-data drops it via the stack's `teardown`.
+    public_hosts = _public_hosts(config, deployments_to_remove)
+    from castle_core.stacks import get_handler
+
+    program_spec = config.programs.get(name) if in_programs else None
+    stack = program_spec.stack if program_spec else None
+    handler = get_handler(stack)
+    owns_data = bool(getattr(handler, "owns_data", False)) and program_spec is not None
 
     # Resolve source dir (from the program entry) for the optional --source removal.
     source_dir: Path | None = None
     if in_programs and config.programs[name].source:
         source_dir = Path(config.programs[name].source)
 
+    purge_data = getattr(args, "purge_data", False)
     print(f"Will remove '{name}' from castle.yaml ({', '.join(where)}).")
+    if deployments_to_remove:
+        print(f"Will tear down deployment(s): {', '.join(deployments_to_remove)}")
     if args.source and source_dir:
         print(f"Will ALSO delete source directory: {source_dir}")
+    if owns_data and purge_data:
+        print(f"Will ALSO destroy persistent data (stack: {stack}).")
 
     # Confirm unless --yes.
     if not args.yes:
@@ -67,14 +88,38 @@ def run_delete(args: argparse.Namespace) -> int:
             print("Aborted (no input). Re-run with --yes to confirm non-interactively.")
             return 1
 
-    # Remove registry entries.
+    # Take each deployment offline in its mode, then drop the config entry. Teardown
+    # is best-effort — the config is removed even if the runtime is already gone.
+    if deployments_to_remove:
+        import asyncio
+
+        from castle_core.lifecycle import deactivate
+
+        for d in deployments_to_remove:
+            try:
+                res = asyncio.run(deactivate(d, config, config.root))
+                if getattr(res, "message", None):
+                    print(f"  {res.message}")
+            except Exception as e:
+                print(f"  warning: teardown of '{d}' failed: {e}")
+            del config.deployments[d]
+
     if in_programs:
         del config.programs[name]
-    if in_deployment:
-        del config.deployments[name]
     save_config(config)
     print(f"Removed '{name}' from castle.yaml ({', '.join(where)}).")
 
+    # Converge the runtime: prune orphan units, regenerate the Caddyfile (dropping
+    # the static route), reload the gateway + tunnel.
+    if deployments_to_remove:
+        from castle_core.deploy import deploy
+
+        try:
+            deploy()
+            print("Reconciled runtime (castle deploy).")
+        except Exception as e:
+            print(f"warning: reconcile (castle deploy) failed — run 'castle deploy': {e}")
+
     # Optional: delete the source directory.
     if args.source and source_dir:
         if source_dir.exists():
@@ -83,16 +128,45 @@ def run_delete(args: argparse.Namespace) -> int:
         else:
             print(f"Source directory not found (already gone): {source_dir}")
 
-    # Warn about runtime artifacts we did NOT touch.
-    if in_deployment:
-        print(
-            f"\nNote: the systemd unit for '{name}' (if deployed) was NOT removed.\n"
-            f"  Run: castle service disable {name}"
-        )
-    if shutil.which(name):
-        print(
-            f"\nNote: '{name}' is still installed on PATH. To remove it:\n"
-            f"  castle uninstall {name}"
-        )
+    # Surface the remnants we don't yet auto-remove.
+    if public_hosts:
+        print("\nNote: public DNS record(s) still exist in Cloudflare (not auto-removed):")
+        for h in public_hosts:
+            print(f"  - {h}")
+        print("  Remove them in the Cloudflare dashboard for the public zone.")
+    # Persistent data the stack owns: destroy it on --purge-data, else surface it.
+    if owns_data:
+        if purge_data:
+            import asyncio
+
+            try:
+                res = asyncio.run(handler.teardown(name, program_spec, config.root))
+                print(f"\n{res.output}")
+                if res.status != "ok":
+                    print("  warning: data teardown reported an error (see above).")
+            except Exception as e:
+                print(f"\nwarning: data teardown failed: {e}")
+        else:
+            print(
+                f"\nNote: this program's persistent data (stack: {stack}) was left "
+                "intact.\n  Re-run with --purge-data to destroy it."
+            )
 
     return 0
+
+
+def _public_hosts(config, deployment_names: list[str]) -> list[str]:
+    """The Cloudflare CNAMEs (<subdomain>.<public_domain>) of any public deployments
+    being removed — surfaced so the operator can clean up DNS."""
+    gw = getattr(config, "gateway", None)
+    public_domain = getattr(gw, "public_domain", None) if gw else None
+    if not public_domain:
+        return []
+    hosts: list[str] = []
+    for d in deployment_names:
+        spec = config.deployments.get(d)
+        if spec is None or not getattr(spec, "public", False):
+            continue
+        sub = getattr(spec, "subdomain", None) or d
+        hosts.append(f"{sub}.{public_domain}")
+    return hosts

cli/src/castle_cli/main.py

@@ -30,7 +30,11 @@ def _build_program_group(subparsers: argparse._SubParsersAction) -> None:
     sub = prog.add_subparsers(dest="program_command")
 
     p = sub.add_parser("list", help="List programs")
-    p.add_argument("--kind", choices=["service", "job", "tool", "static", "reference"], help="Filter by derived kind")
+    p.add_argument(
+        "--kind",
+        choices=["service", "job", "tool", "static", "reference"],
+        help="Filter by derived kind",
+    )
     p.add_argument("--stack", help="Filter by stack")
     p.add_argument("--json", action="store_true", help="Output as JSON")
 
@@ -55,6 +59,11 @@ def _build_program_group(subparsers: argparse._SubParsersAction) -> None:
     p = sub.add_parser("delete", help="Remove a program from castle.yaml")
     _add_name(p, "Program name")
     p.add_argument("--source", action="store_true", help="Also delete the source directory")
+    p.add_argument(
+        "--purge-data",
+        action="store_true",
+        help="Also destroy the program's persistent data (e.g. a supabase app's DB schema)",
+    )
     p.add_argument("-y", "--yes", action="store_true", help="Skip confirmation")
 
     p = sub.add_parser("run", help="Run a program's declared run command")
@@ -87,7 +96,9 @@ def _build_tool_group(subparsers: argparse._SubParsersAction) -> None:
     p.add_argument("--json", action="store_true", help="Machine-readable output")
 
     sub.add_parser("install", help="Install a tool on PATH").add_argument("name", help="Tool name")
-    sub.add_parser("uninstall", help="Remove a tool from PATH").add_argument("name", help="Tool name")
+    sub.add_parser("uninstall", help="Remove a tool from PATH").add_argument(
+        "name", help="Tool name"
+    )
 
 
 def _add_service_create(sub: argparse._SubParsersAction, kind: str) -> None:
@@ -134,6 +145,7 @@ def _build_deployment_group(subparsers: argparse._SubParsersAction, kind: str) -
     _add_name(p, f"{kind.capitalize()} name")
     p.add_argument("-y", "--yes", action="store_true", help="Skip confirmation")
     p.add_argument("--source", action="store_true", help=argparse.SUPPRESS)
+    p.add_argument("--purge-data", action="store_true", help=argparse.SUPPRESS)
 
     cap = f"{kind.capitalize()} name"
     _add_name(sub.add_parser("deploy", help=f"Deploy this {kind} (unit + gateway)"), cap)
@@ -186,7 +198,11 @@ def build_parser() -> argparse.ArgumentParser:
 
     # Cross-resource overview
     p = subparsers.add_parser("list", help="List programs, services, jobs, and tools")
-    p.add_argument("--kind", choices=["service", "job", "tool", "static", "reference"], help="Filter by derived kind")
+    p.add_argument(
+        "--kind",
+        choices=["service", "job", "tool", "static", "reference"],
+        help="Filter by derived kind",
+    )
     p.add_argument("--stack", help="Filter by stack")
     p.add_argument("--json", action="store_true", help="Output as JSON")
 

cli/src/castle_cli/templates/scaffold.py

@@ -576,16 +576,19 @@ def _scaffold_supabase(project_dir: Path, name: str, description: str) -> None:
     /<name>/ by the gateway), and supabase.app.yaml (auth policy + wiring).
 
     The app owns its code and stays repo-durable; only its rows/blobs live on the
-    shared substrate. Tables are namespaced by the app id (they share the `public`
-    schema so no substrate reconfiguration is needed) and protected by RLS.
+    shared substrate. Each app is isolated in its **own Postgres schema** (the app
+    id) rather than sharing `public` — so `castle program build` creates+grants the
+    schema and tracks migrations per-app, and `castle delete --purge-data` drops
+    the whole schema cleanly. Rows are further protected by RLS.
     """
-    ident = name.replace("-", "_")  # a safe SQL/JS identifier
-    table = f"{ident}_entries"
+    ident = name.replace("-", "_")  # a safe SQL/JS identifier — and the app schema
+    table = "entries"  # unqualified; lives in the app's own schema (search_path)
 
     def sub(text: str) -> str:
         return (
             text.replace("__NAME__", name)
             .replace("__IDENT__", ident)
+            .replace("__SCHEMA__", ident)
             .replace("__TABLE__", table)
             .replace("__DESC__", description)
         )
@@ -606,8 +609,10 @@ substrate: supabase          # the shared castle service this app deploys agains
 auth: public
 # shared: [alice, bob]
 
-# Tables are namespaced by this prefix in the shared `public` schema.
-table_prefix: __IDENT___
+# This app is isolated in its own Postgres schema (created + exposed through
+# PostgREST by `castle program build`). The frontend selects it via
+# supabase-js `db: { schema }`.
+schema: __SCHEMA__
 """
         ),
     )
@@ -617,11 +622,14 @@ table_prefix: __IDENT___
         project_dir / "migrations" / "0001_init.sql",
         sub(
             """-- 0001_init: example table + RLS. Applied by `castle program build`
--- via the versioned migration runner (tracked in schema_migrations; only
--- unapplied migrations run). Forward-only — never edit an applied migration;
--- add a new numbered file instead.
+-- via the versioned migration runner (tracked per-app in
+-- <schema>.schema_migrations; only unapplied migrations run). Forward-only —
+-- never edit an applied migration; add a new numbered file instead.
+--
+-- The runner sets search_path to this app's own schema, so unqualified names
+-- land there (NOT in public). Keep tables unqualified for schema portability.
 
-create table if not exists public.__TABLE__ (
+create table if not exists __TABLE__ (
     id          bigint generated always as identity primary key,
     message     text not null check (char_length(message) <= 500),
     author      text,
@@ -632,13 +640,13 @@ create table if not exists public.__TABLE__ (
 -- static app shell or Storage objects. For a `public` app, anon read/write is
 -- intended (still row-gated). For `private`/`shared`, replace the policies below
 -- with owner checks (auth.uid()) AND gate the shell + use signed Storage URLs.
-alter table public.__TABLE__ enable row level security;
+alter table __TABLE__ enable row level security;
 
-drop policy if exists "__IDENT___public_read" on public.__TABLE__;
-create policy "__IDENT___public_read"  on public.__TABLE__ for select using (true);
+drop policy if exists "__IDENT___public_read" on __TABLE__;
+create policy "__IDENT___public_read"  on __TABLE__ for select using (true);
 
-drop policy if exists "__IDENT___public_write" on public.__TABLE__;
-create policy "__IDENT___public_write" on public.__TABLE__ for insert with check (true);
+drop policy if exists "__IDENT___public_write" on __TABLE__;
+create policy "__IDENT___public_write" on __TABLE__ for insert with check (true);
 """
         ),
     )
@@ -672,6 +680,7 @@ serve((_req: Request) => {
 window.APP = {
   SUPABASE_URL: "https://supabase.lan",
   SUPABASE_ANON_KEY: "PASTE_ANON_KEY_HERE",
+  SCHEMA: "__SCHEMA__",   // this app's isolated Postgres schema
   TABLE: "__TABLE__",
 };
 """
@@ -691,8 +700,8 @@ window.APP = {
   <script src="./config.js"></script>
   <script type="module">
     import { createClient } from "https://esm.sh/@supabase/supabase-js@2";
-    const { SUPABASE_URL, SUPABASE_ANON_KEY, TABLE } = window.APP;
-    const db = createClient(SUPABASE_URL, SUPABASE_ANON_KEY);
+    const { SUPABASE_URL, SUPABASE_ANON_KEY, SCHEMA, TABLE } = window.APP;
+    const db = createClient(SUPABASE_URL, SUPABASE_ANON_KEY, { db: { schema: SCHEMA } });
 
     const list = document.getElementById("list");
     async function refresh() {

core/src/castle_core/deploy.py

@@ -174,7 +174,9 @@ def _acme_preflight(config: CastleConfig, messages: list[str]) -> None:
             "won't get a wildcard cert (serving plain HTTP on the gateway port)."
         )
         return
-    token_env = _DNS_TOKEN_ENV.get(gw.acme_dns_provider or "cloudflare", "CLOUDFLARE_API_TOKEN")
+    token_env = _DNS_TOKEN_ENV.get(
+        gw.acme_dns_provider or "cloudflare", "CLOUDFLARE_API_TOKEN"
+    )
     svc = config.services.get(_GATEWAY_NAME)
     env = dict(svc.defaults.env) if (svc and svc.defaults and svc.defaults.env) else {}
     if token_env not in env:
@@ -216,11 +218,15 @@ def _write_tunnel_config(registry: NodeRegistry, messages: list[str]) -> None:
     # exact commands rather than silently assuming they're routed.
     tid = registry.node.tunnel_id
     for h in hosts:
-        messages.append(f"  public: {h}  (route once: cloudflared tunnel route dns {tid} {h})")
+        messages.append(
+            f"  public: {h}  (route once: cloudflared tunnel route dns {tid} {h})"
+        )
 
     tunnel_unit = unit_name(_TUNNEL_NAME)
     active = subprocess.run(
-        ["systemctl", "--user", "is-active", tunnel_unit], capture_output=True, text=True
+        ["systemctl", "--user", "is-active", tunnel_unit],
+        capture_output=True,
+        text=True,
     )
     if active.stdout.strip() == "active":
         subprocess.run(["systemctl", "--user", "restart", tunnel_unit], check=False)
@@ -278,15 +284,41 @@ def _public_url(
     return None
 
 
+def _supabase_app_schemas(config: CastleConfig) -> str:
+    """The ``${supabase_app_schemas}`` placeholder: each registered supabase app's
+    own schema, comma-prefixed and joined (or '' when there are none).
+
+    The substrate exposes app schemas through PostgREST by listing them in
+    PGRST_DB_SCHEMAS; its deployment maps that env to
+    ``public,storage,graphql_public${supabase_app_schemas}``. Comma-prefixing each
+    entry keeps the base list valid when zero apps are registered (no trailing
+    comma). Adding/removing a supabase app thus changes this list — the substrate
+    needs a restart (recreate) after `castle deploy` to pick it up.
+    """
+    from castle_core.stacks import app_schema
+
+    schemas = sorted(
+        app_schema(pn) for pn, ps in config.programs.items() if ps.stack == "supabase"
+    )
+    return "".join(f",{s}" for s in schemas)
+
+
 def _env_context(
-    name: str, config_key: str, port: int | None, public_url: str | None = None
+    name: str,
+    config_key: str,
+    port: int | None,
+    public_url: str | None = None,
+    supabase_app_schemas: str | None = None,
 ) -> dict[str, str]:
-    """Placeholder values for defaults.env: ${name}/${data_dir}/${port}/${public_url}."""
+    """Placeholder values for defaults.env: ${name}/${data_dir}/${port}/${public_url}/
+    ${supabase_app_schemas}."""
     ctx = {"name": name, "data_dir": str(DATA_DIR / config_key)}
     if port is not None:
         ctx["port"] = str(port)
     if public_url is not None:
         ctx["public_url"] = public_url
+    if supabase_app_schemas is not None:
+        ctx["supabase_app_schemas"] = supabase_app_schemas
     return ctx
 
 
@@ -313,9 +345,7 @@ def _write_secret_env_file(name: str, secret_env: dict[str, str]) -> Path | None
     return path
 
 
-def _resolve_description(
-    config: CastleConfig, spec: DeploymentBase
-) -> str | None:
+def _resolve_description(config: CastleConfig, spec: DeploymentBase) -> str | None:
     """Get description, falling through to program if referenced."""
     if spec.description:
         return spec.description
@@ -397,7 +427,8 @@ def _build_deployed(
     raw_env = dict(dep.defaults.env) if (dep.defaults and dep.defaults.env) else {}
     public_url = _public_url(config, name, expose, port)
     env, secret_env = resolve_env_split(
-        raw_env, _env_context(name, config_key, port, public_url)
+        raw_env,
+        _env_context(name, config_key, port, public_url, _supabase_app_schemas(config)),
     )
     secret_env_file = _write_secret_env_file(name, secret_env)
 

core/src/castle_core/stacks.py

@@ -87,6 +87,12 @@ def _source_dir(comp: ProgramSpec, root: Path) -> Path:
 class StackHandler:
     """Base class — subclasses implement each lifecycle action."""
 
+    # Whether this stack owns *persistent external state* (a database schema, a
+    # bucket, …) that outlives a code delete. Drives whether `castle delete`
+    # surfaces a data remnant / honors `--purge-data`. Overridden to True by the
+    # stacks whose `teardown` actually destroys something.
+    owns_data: bool = False
+
     async def build(self, name: str, comp: ProgramSpec, root: Path) -> ActionResult:
         raise NotImplementedError
 
@@ -127,6 +133,21 @@ class StackHandler:
     async def uninstall(self, name: str, comp: ProgramSpec, root: Path) -> ActionResult:
         raise NotImplementedError
 
+    async def teardown(self, name: str, comp: ProgramSpec, root: Path) -> ActionResult:
+        """Destroy the persistent external state this program's stack created
+        (database schema, blobs, …) — the irreversible counterpart to a code
+        delete. Distinct from `uninstall` (which only takes a program offline).
+
+        Default: nothing to tear down. Only stacks that set ``owns_data`` and own
+        durable state override this; `castle delete --purge-data` invokes it.
+        """
+        return ActionResult(
+            program=name,
+            action="teardown",
+            status="ok",
+            output=f"{name}: no persistent state to tear down.",
+        )
+
 
 class PythonHandler(StackHandler):
     """Handler for python-cli and python-fastapi stacks."""
@@ -362,8 +383,11 @@ def _substrate_db_url() -> str | None:
     """Best-effort Postgres URL for the shared substrate.
 
     Prefers an explicit ``SUPABASE_DB_URL``; otherwise builds one from the
-    generated ``SUPABASE_POSTGRES_PASSWORD`` secret against localhost:5432. Returns
-    None if neither is available (build then fails loud with guidance).
+    generated ``SUPABASE_POSTGRES_PASSWORD`` secret against the substrate's direct
+    Postgres port. The self-hosted substrate publishes Postgres on host **5433**
+    (5432 is taken by another Postgres on this node — see the substrate compose),
+    overridable via ``SUPABASE_DB_HOST_PORT``. Returns None if neither an explicit
+    URL nor the secret is available (build then fails loud with guidance).
     """
     explicit = os.environ.get("SUPABASE_DB_URL")
     if explicit:
@@ -373,25 +397,65 @@ def _substrate_db_url() -> str | None:
     pw_file = SECRETS_DIR / "SUPABASE_POSTGRES_PASSWORD"
     if pw_file.exists():
         pw = pw_file.read_text().strip()
-        return f"postgresql://postgres:{pw}@localhost:5432/postgres"
+        port = os.environ.get("SUPABASE_DB_HOST_PORT", "5433")
+        return f"postgresql://postgres:{pw}@localhost:{port}/postgres"
     return None
 
 
+def app_schema(name: str) -> str:
+    """The dedicated Postgres schema a supabase app owns.
+
+    Each app is isolated in its own schema (named after the program, ``-``→``_``
+    for a valid unquoted identifier) rather than sharing ``public``. That gives a
+    clean teardown (``drop schema … cascade``) and a per-app ``schema_migrations``
+    so migration version tokens never collide across apps.
+    """
+    return name.replace("-", "_")
+
+
+# The privilege grant that makes an app schema reachable through PostgREST — the
+# canonical Supabase "expose a custom schema" snippet. Idempotent; run before
+# migrations so `alter default privileges` also covers the tables they create.
+# Exposure ALSO requires the schema in the substrate's PGRST_DB_SCHEMAS — castle
+# derives that from the registered supabase apps (`${supabase_app_schemas}`), so a
+# newly-added app needs a `castle deploy` + substrate restart to become routable.
+def _schema_setup_sql(schema: str) -> str:
+    roles = "anon, authenticated, service_role"
+    return (
+        f'create schema if not exists "{schema}";\n'
+        f'grant usage on schema "{schema}" to {roles};\n'
+        f'grant all on all tables in schema "{schema}" to {roles};\n'
+        f'grant all on all routines in schema "{schema}" to {roles};\n'
+        f'grant all on all sequences in schema "{schema}" to {roles};\n'
+        f'alter default privileges in schema "{schema}" '
+        f"grant all on tables to {roles};\n"
+        f'alter default privileges in schema "{schema}" '
+        f"grant all on routines to {roles};\n"
+        f'alter default privileges in schema "{schema}" '
+        f"grant all on sequences to {roles};\n"
+    )
+
+
 class SupabaseHandler(StackHandler):
     """Stack handler for supabase apps (migrations + edge functions + static UI).
 
-    `build` runs the versioned migration runner against the shared substrate; the
-    static UI is served in place by the gateway (no process), so install/uninstall
-    are no-ops like a frontend.
+    Each app is isolated in its **own Postgres schema** (``app_schema(name)``): the
+    migration runner creates + grants that schema, tracks applied versions in
+    ``<schema>.schema_migrations``, and runs each migration with ``search_path``
+    set to it — so migration SQL is schema-agnostic and version tokens never
+    collide across apps. The static UI is served in place by the gateway (no
+    process), so install/uninstall are no-ops like a frontend. `teardown` drops the
+    schema (and thus every object the app created) in one shot.
     """
 
+    owns_data = True
+
     async def build(self, name: str, comp: ProgramSpec, root: Path) -> ActionResult:
-        """Apply unapplied migrations to the shared substrate (idempotent)."""
+        """Apply unapplied migrations into the app's own schema (idempotent)."""
         src = _source_dir(comp, root)
+        schema = app_schema(name)
         mig_dir = src / "migrations"
         files = sorted(mig_dir.glob("*.sql")) if mig_dir.is_dir() else []
-        if not files:
-            return ActionResult(name, "build", "ok", "No migrations to apply.")
 
         psql = shutil.which("psql")
         url = _substrate_db_url()
@@ -411,7 +475,8 @@ class SupabaseHandler(StackHandler):
                 "(scripts/gen-keys.py) so SUPABASE_POSTGRES_PASSWORD exists.",
             )
 
-        # Ensure the tracking table, then read applied versions.
+        # Ensure the app's schema exists, is PostgREST-exposable (grants), and has
+        # its own tracking table — then read applied versions from THAT schema.
         rc, out = await _run(
             [
                 psql,
@@ -419,15 +484,29 @@ class SupabaseHandler(StackHandler):
                 "-v",
                 "ON_ERROR_STOP=1",
                 "-c",
-                "create table if not exists public.schema_migrations "
+                _schema_setup_sql(schema),
+                "-c",
+                f'create table if not exists "{schema}".schema_migrations '
                 "(version text primary key, applied_at timestamptz default now())",
             ],
             src,
         )
         if rc != 0:
             return ActionResult(name, "build", "error", f"connect/init failed:\n{out}")
+
+        if not files:
+            return ActionResult(
+                name, "build", "ok", f"Schema '{schema}' ready. No migrations to apply."
+            )
+
         rc, out = await _run(
-            [psql, url, "-tA", "-c", "select version from public.schema_migrations"],
+            [
+                psql,
+                url,
+                "-tA",
+                "-c",
+                f'select version from "{schema}".schema_migrations',
+            ],
             src,
         )
         if rc != 0:
@@ -438,12 +517,19 @@ class SupabaseHandler(StackHandler):
 
         pending = plan_migrations(files, applied)
         if not pending:
-            return ActionResult(name, "build", "ok", "All migrations already applied.")
+            return ActionResult(
+                name,
+                "build",
+                "ok",
+                f"All migrations already applied (schema {schema}).",
+            )
 
         log = []
         for path in pending:
             version = _migration_version(path)
-            # File + version-insert in ONE transaction: a failed migration records
+            # search_path → the app schema, so migration SQL can write unqualified
+            # names and they land in the app's schema (not public). File +
+            # version-insert in ONE transaction: a failed migration records
             # nothing, so the next build safely retries it.
             rc, out = await _run(
                 [
@@ -452,10 +538,13 @@ class SupabaseHandler(StackHandler):
                     "-v",
                     "ON_ERROR_STOP=1",
                     "--single-transaction",
+                    "-c",
+                    f'set search_path to "{schema}", public',
                     "-f",
                     str(path),
                     "-c",
-                    f"insert into public.schema_migrations(version) values('{version}')",
+                    f'insert into "{schema}".schema_migrations(version) '
+                    f"values('{version}')",
                 ],
                 src,
             )
@@ -465,6 +554,45 @@ class SupabaseHandler(StackHandler):
             log.append(f"✓ {path.name}")
         return ActionResult(name, "build", "ok", "\n".join(log))
 
+    async def teardown(self, name: str, comp: ProgramSpec, root: Path) -> ActionResult:
+        """Drop the app's schema and everything in it (tables, its own
+        schema_migrations, functions) in one statement — total and knowable
+        because the app owns exactly one schema. The substrate's PGRST_DB_SCHEMAS
+        drops the (now-absent) schema on the next `castle deploy` + restart.
+        """
+        schema = app_schema(name)
+        psql = shutil.which("psql")
+        url = _substrate_db_url()
+        if not psql or not url:
+            return ActionResult(
+                name,
+                "teardown",
+                "error",
+                f"Cannot drop schema '{schema}': psql or substrate DB URL "
+                'unavailable. Drop it manually: drop schema "%s" cascade;' % schema,
+            )
+        cwd = src if (src := (root / comp.source if comp.source else None)) else root
+        rc, out = await _run(
+            [
+                psql,
+                url,
+                "-v",
+                "ON_ERROR_STOP=1",
+                "-c",
+                f'drop schema if exists "{schema}" cascade',
+            ],
+            cwd,
+        )
+        if rc != 0:
+            return ActionResult(name, "teardown", "error", f"drop failed:\n{out}")
+        return ActionResult(
+            name,
+            "teardown",
+            "ok",
+            f"Dropped schema '{schema}' (all tables + rows). Run `castle deploy` "
+            "and restart the substrate to prune it from PGRST_DB_SCHEMAS.",
+        )
+
     async def _deno(
         self, name: str, action: str, comp: ProgramSpec, root: Path, args: list[str]
     ) -> ActionResult:

docs/stacks/supabase.md

@@ -24,11 +24,16 @@ is **code + migrations that deploy against a shared backend**:
   deploys onto the substrate. Its rows/blobs live on the substrate; everything
   else rebuilds from git.
 
-Apps are isolated on the shared instance by **table prefix + RLS** in the shared
-`public` schema (and Storage buckets), under **one identity pool** — correct for a
-single-operator datalake. Substrate-per-app is deliberately not supported: ~14
-containers per app doesn't scale to "lots of small ideas," and a DB-backed app is
-a pet either way.
+Apps are isolated on the shared instance by their **own Postgres schema + RLS**
+(and Storage buckets), under **one identity pool** — correct for a single-operator
+datalake. Each app owns a schema named after the program (`my-app` → schema
+`my_app`); `castle program build` creates and grants it, tracks migrations in a
+per-app `<schema>.schema_migrations`, and PostgREST exposes it (castle derives the
+substrate's `PGRST_DB_SCHEMAS` from the registered apps). This gives a clean
+teardown — `castle delete --purge-data` runs `drop schema <app> cascade` — and
+means migration version tokens never collide across apps. Substrate-per-app is
+deliberately not supported: ~14 containers per app doesn't scale to "lots of small
+ideas," and a DB-backed app is a pet either way.
 
 ## Stack
 
@@ -66,33 +71,56 @@ gateway serves `public/` in place at `/my-app/` — no service, no process.
 name: my-app
 substrate: supabase        # the shared castle service this app deploys against
 auth: public               # public | private | shared: [handles]
-table_prefix: my_app_
+schema: my_app             # this app's isolated Postgres schema (frontend: db.schema)
 ```
 
 ## Migrations
 
 `migrations/*.sql` are **numbered, forward-only, and idempotent**. `castle program
-build my-app` runs the versioned migration runner: it ensures a
-`public.schema_migrations` table, reads applied versions, and applies only the
-**unapplied** files (in filename order), each in a single transaction with its
-version-insert — so a failed migration records nothing and the next build retries
+build my-app` runs the versioned migration runner: it creates + grants the app's
+schema, ensures a per-app `<schema>.schema_migrations` table, reads applied
+versions, and applies only the **unapplied** files (in filename order) with
+`search_path` set to the app schema — each in a single transaction with its
+version-insert, so a failed migration records nothing and the next build retries
 it. Never edit an applied migration; add a new numbered file.
 
+Because the runner sets `search_path` to the app's own schema, write **unqualified**
+names — they land in `<schema>`, not `public`:
+
 ```sql
 -- migrations/0001_init.sql
-create table if not exists public.my_app_entries (
+create table if not exists entries (
     id bigint generated always as identity primary key,
     message text not null,
     created_at timestamptz not null default now()
 );
-alter table public.my_app_entries enable row level security;
-create policy "my_app_read"  on public.my_app_entries for select using (true);
-create policy "my_app_write" on public.my_app_entries for insert with check (true);
+alter table entries enable row level security;
+create policy "my_app_read"  on entries for select using (true);
+create policy "my_app_write" on entries for insert with check (true);
 ```
 
 The runner connects via `SUPABASE_DB_URL`, or builds one from the generated
-`SUPABASE_POSTGRES_PASSWORD` secret against `localhost:5432`. `psql` must be on
-PATH; a missing URL or client fails loud with guidance.
+`SUPABASE_POSTGRES_PASSWORD` secret against the substrate's direct Postgres port
+(host **5433**, `SUPABASE_DB_HOST_PORT` to override). `psql` must be on PATH; a
+missing URL or client fails loud with guidance.
+
+The frontend selects the app schema through supabase-js:
+
+```js
+const db = createClient(SUPABASE_URL, SUPABASE_ANON_KEY, { db: { schema: SCHEMA } });
+```
+
+### Teardown
+
+An app's rows live only on the substrate, so an ordinary `castle delete my-app`
+leaves the schema intact (and says so). To destroy the data too:
+
+```bash
+castle delete my-app --purge-data      # drop schema my_app cascade
+```
+
+`castle deploy` then prunes the schema from `PGRST_DB_SCHEMAS`; **restart the
+`supabase` service** for PostgREST to pick up the added/removed schema list.
 
 ## Auth, RLS & the three privacy layers