feat(hub): leaving a group, a cap on public ones, and hosting as a precondition

Leaving is its own endpoint rather than a relaxation of the owner's removal
check — an authorization rule with an exception in it is the one that gets read
wrong later. The owner cannot leave: the group would be left with nobody able
to admit, edit or delete it, which is the answer removal and account deletion
already give.

Public groups are capped at ten live ones per owner. They are the ones that
cost other people something — listed in Discover, joinable by anyone — so a
script that opens hundreds fills the directory for everybody. Private groups
are invisible to non-members and are not capped. Hub staff are exempt; the cap
is anti-spam, not a rule about running an instance. Creation is the only place
it can be checked, and deliberately so, because PATCH refuses to change
visibility at all.

A group is now listed only once a node has announced that it hosts it. Before
that it has no files, no key and nothing to connect to, so showing it to a
member produces a name they cannot open and cannot be told why; its owner still
sees it while they set the node up. `meshbay-hub prune-groups` collects the
ones that never got a node, meant for cron, with --dry-run. The migration
backfills hosted_at from created_at: without that the first run would have
deleted every live group.

Presence rides on the group list itself, read from the signaling registry the
hub already keeps — no poll, no timer. It says a node is connected *to the hub*,
which is not a promise that this browser can reach it and not something a
dishonest hub could not fake; the client downgrades it on a connection it tried
and failed, which is the evidence that concerns the reader.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

10 files changed, 863 insertions, 14 deletions

diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index 8e3197c..8283276 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -2,7 +2,7 @@
 
 from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
-from sqlalchemy import select
+from sqlalchemy import func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.api.deps import get_current_user, require_user_scope
@@ -26,10 +26,16 @@ async def my_groups(
     db: AsyncSession = Depends(get_db),
 ):
     """List groups the current user belongs to."""
+    from meshbay_hub.api.revocation import get_online_nodes_for_group
+
     result = await db.execute(
         select(Group)
         .join(GroupMember, Group.id == GroupMember.group_id)
-        .where(GroupMember.user_id == current_user.id, Group.status == "active")
+        .where(GroupMember.user_id == current_user.id, Group.status == "active",
+               # A group nobody hosts yet is the owner's business alone. Someone
+               # added to it before a node exists would see a name they cannot
+               # open and cannot be told why.
+               or_(Group.hosted_at.is_not(None), Group.admin_id == current_user.id))
         .order_by(Group.name)
     )
     groups = result.scalars().all()
@@ -48,6 +54,15 @@ async def my_groups(
                 "created_at": g.created_at.isoformat(),
                 "is_admin": g.admin_id == current_user.id,
                 "description": g.description or "",
+                # Presence, from the socket registry the hub already keeps for
+                # signaling — so the sidebar gets it on the request it already
+                # makes, with no poll and no timer. It says a node serving this
+                # group is connected *to the hub*; it does not promise this
+                # browser can reach it, and a hub is free to lie about it. The
+                # client downgrades to offline on its own failed connection,
+                # which is the evidence that actually concerns the user.
+                "node_online": bool(get_online_nodes_for_group(g.id)),
+                "hosted": g.hosted_at is not None,
             }
             for g in groups
         ]
@@ -88,7 +103,12 @@ async def list_public_groups(
     include_federated: bool = True,
 ):
     """List/search public groups — local and optionally federated. No auth required."""
-    query = select(Group).where(Group.visibility == "public", Group.status == "active")
+    # Unhosted groups are absent from the directory: until a node announces it,
+    # a group has no files, no key and nothing to connect to, so listing it only
+    # produces a dead end. Its owner still sees it in /mine while they set it up.
+    query = select(Group).where(Group.visibility == "public",
+                                Group.status == "active",
+                                Group.hosted_at.is_not(None))
     if q:
         query = query.where(Group.name.ilike(f"%{q}%"))
     result = await db.execute(
@@ -246,10 +266,50 @@ async def join_group(
 class GroupCreateRequest(BaseModel):
     name:        str
     visibility:  str = "private"   # public|private
-    join_policy: str = "invite"    # open|request|invite
+    join_policy: str = "invite"    # open (public groups) | invite (private)
     description: str | None = None
 
 
+MAX_PUBLIC_GROUPS = 10
+
+
+async def _check_public_group_quota(db: AsyncSession, user: User) -> None:
+    """Refuse an eleventh live public group from the same owner.
+
+    Public groups are the ones that cost other people something: they appear in
+    Discover and anyone may join them, so a script that opens hundreds fills the
+    directory for everybody. Private groups are invisible to anyone not invited
+    and are not capped.
+
+    Counted: public, still active, owned by this user. A group suspended by
+    moderation does not hold a slot — the owner is already being dealt with, and
+    keeping the slot occupied would punish them twice. Deleting one frees a slot,
+    since the row is gone.
+
+    Creation is the only place this can be checked, and deliberately so: PATCH
+    refuses to change visibility at all, so a private group cannot be flipped
+    public behind the cap. **If visibility ever becomes editable, this check has
+    to move with it.**
+    """
+    if user.role in ("admin", "moderator"):
+        return   # the cap is an anti-spam measure, not a rule about operating an instance
+
+    count = (await db.execute(
+        select(func.count())
+        .select_from(Group)
+        .where(Group.admin_id == user.id,
+               Group.visibility == "public",
+               Group.status == "active")
+    )).scalar_one()
+
+    if count >= MAX_PUBLIC_GROUPS:
+        raise HTTPException(
+            status_code=409,
+            detail=f"You already run {count} public groups, which is the limit of "
+                   f"{MAX_PUBLIC_GROUPS}. Delete one you no longer use, or create "
+                   f"this one as private — private groups are not limited.")
+
+
 @router.post("", status_code=201)
 async def create_group(
     body: GroupCreateRequest,
@@ -257,6 +317,19 @@ async def create_group(
     current_user: User = Depends(require_user_scope),
     db: AsyncSession = Depends(get_db),
 ):
+    if body.visibility == "public":
+        # A public group that admits nobody is a contradiction: it is listed in
+        # the directory, so people find it and then discover they cannot get in.
+        # Admission by request was considered and dropped — between strangers the
+        # only channel is the hub, so the one-time code would travel through the
+        # very party it exists to keep out, and would protect nothing.
+        if body.join_policy != "open":
+            raise HTTPException(
+                status_code=422,
+                detail="A public group is open to join. Make it private if you "
+                       "want to choose who comes in.")
+        await _check_public_group_quota(db, current_user)
+
     desc = (body.description or "")[:512] if body.description else None
     group = Group(
         name=body.name,
@@ -324,6 +397,55 @@ async def remove_group_member(
     return {"status": "removed", "group_id": group_id, "username": username}
 
 
+@router.post("/{group_id}/leave")
+async def leave_group(
+    group_id: str,
+    request: Request,
+    current_user: User = Depends(require_user_scope),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Leave a group you are a member of.
+
+    Deliberately separate from `DELETE /{group_id}/members/{username}`, which is
+    the owner removing somebody else and is refused to everyone else. Reusing it
+    would have meant relaxing that check for the self case, and an authorization
+    rule with an exception in it is the kind that gets read wrong later.
+
+    The owner cannot leave: the group would be left with no one able to admit a
+    member, edit it or delete it. That is the same answer the removal endpoint
+    already gives, and the same one account deletion gives while you still own
+    groups — hand the group over (not yet possible) or delete it.
+
+    This is only the hub's half, exactly as for removal: membership is gone, so
+    signaling will not reach a node and the next token will not name this group.
+    The node keeps what it holds — the identity it pinned, the keypair bundle,
+    and the files uploaded — until its operator unpins them, and whoever left
+    still holds the group key they were served, so the operator should rotate it
+    (`meshbay-node gek-init`) if that matters.
+    """
+    group = await db.get(Group, group_id)
+    if not group:
+        raise HTTPException(status_code=404, detail="Group not found")
+
+    if group.admin_id == current_user.id:
+        raise HTTPException(
+            status_code=409,
+            detail="You own this group, so you cannot leave it — it would be left "
+                   "with nobody able to manage it. Delete the group instead.")
+
+    membership = await db.get(GroupMember, (group_id, current_user.id))
+    if not membership:
+        raise HTTPException(status_code=404, detail="You are not a member of this group")
+
+    await db.delete(membership)
+    db.add(IPLog(user_id=current_user.id, event="group_leave",
+                 ip_address=client_ip(request),
+                 detail=f"left {group.name}"))
+    await db.commit()
+    return {"status": "left", "group_id": group_id}
+
+
 class GroupUpdateRequest(BaseModel):
     description: str | None = None
 
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index 1f6d7f0..d555f9f 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -31,11 +31,12 @@ import json
 import logging
 import time
 import uuid
+from datetime import datetime, timezone
 from typing import Any
 
 from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect
 from pydantic import BaseModel
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import jwt
@@ -67,6 +68,36 @@ def get_online_nodes_for_group(group_id: str) -> list[str]:
     return [nid for nid, gids in _node_groups.items() if group_id in gids]
 
 
+
+async def _mark_hosted(group_ids: list[str]) -> None:
+    """Stamp the first time a node announced it hosts each of these groups.
+
+    `group_ids` is already narrowed to what this node may claim — the caller
+    derives it from the database and a node can only shrink the set, never widen
+    it (finding C2) — so being announced here is evidence the group has a host.
+
+    Set once. A node going offline does not un-host a group, and re-stamping on
+    every reconnection would make `hosted_at` a "last seen" field, which is what
+    the in-memory registry is already for.
+    """
+    from meshbay_hub.db.engine import get_session_factory
+    from meshbay_hub.db.models import Group
+
+    if not group_ids:
+        return
+    try:
+        async with get_session_factory()() as db:
+            await db.execute(
+                update(Group)
+                .where(Group.id.in_(group_ids), Group.hosted_at.is_(None))
+                .values(hosted_at=datetime.now(timezone.utc)))
+            await db.commit()
+    except Exception as e:
+        # A group that stays unhosted in the table is visible to its owner and
+        # collected later; failing the socket over it would take the node down.
+        log.warning("Could not mark groups hosted: %s", e)
+
+
 async def broadcast_revocation(token: str) -> int:
     """Push a signed revocation token to all connected nodes. Returns count sent."""
     payload = json.dumps({"type": "revocation", "token": token})
@@ -236,6 +267,7 @@ async def node_websocket(ws: WebSocket):
         node_id = resolved_id
         _connected_nodes[node_id] = ws
         _node_groups[node_id] = group_ids
+        await _mark_hosted(group_ids)
         log.info("Node WS connected: %s (user=%s, groups=%d)",
                  node_id[:8], user_id[:8], len(group_ids))
         await ws.send_text(json.dumps({"type": "auth_ok", "node_id": node_id}))
diff --git a/packages/meshbay-hub/src/meshbay_hub/daemon.py b/packages/meshbay-hub/src/meshbay_hub/daemon.py
index ff66fee..4af26f5 100644
--- a/packages/meshbay-hub/src/meshbay_hub/daemon.py
+++ b/packages/meshbay-hub/src/meshbay_hub/daemon.py
@@ -1,6 +1,7 @@
-"""Entry point for the meshbay-hub systemd service."""
+"""Entry point for the meshbay-hub systemd service, and its few CLI chores."""
 
 import argparse
+import asyncio
 import logging
 import sys
 from pathlib import Path
@@ -15,6 +16,25 @@ def main() -> None:
     parser.add_argument("--config", type=Path, default=None)
     parser.add_argument("--log-level", default="INFO",
                         choices=["DEBUG", "INFO", "WARNING", "ERROR"])
+    # Optional on purpose: the systemd unit runs `meshbay-hub --config …` with no
+    # subcommand and must go on starting the server.
+    sub = parser.add_subparsers(dest="command")
+
+    prune = sub.add_parser(
+        "prune-groups",
+        help="delete groups no node ever hosted (meant for cron)")
+    prune.add_argument("--days", type=int, default=None,
+                       help="grace period since creation (default 7)")
+    prune.add_argument("--dry-run", action="store_true",
+                       help="list what would go, delete nothing")
+    # Accepted after the subcommand too. Everyone writes the cron line as
+    # `prune-groups --config …`, and argparse only takes an option before the
+    # subcommand unless the subparser declares it as well. SUPPRESS so that
+    # leaving it out here does not overwrite a value given before it.
+    prune.add_argument("--config", type=Path, default=argparse.SUPPRESS)
+    prune.add_argument("--log-level", default=argparse.SUPPRESS,
+                       choices=["DEBUG", "INFO", "WARNING", "ERROR"])
+
     args = parser.parse_args()
 
     logging.basicConfig(
@@ -24,6 +44,9 @@ def main() -> None:
 
     cfg = load_config(args.config)
 
+    if args.command == "prune-groups":
+        sys.exit(asyncio.run(_prune_groups(cfg, args.days, args.dry_run)))
+
     uvicorn.run(
         "meshbay_hub.app:create_app",
         factory=True,
@@ -34,5 +57,41 @@ def main() -> None:
     )
 
 
+async def _prune_groups(cfg, days: int | None, dry_run: bool) -> int:
+    """Collect groups that were created and never given a node.
+
+    A group with no host has no files, no key and nothing to connect to, and is
+    invisible to everyone but its owner — so it is litter rather than data. The
+    grace period is counted from creation and `hosted_at` is never cleared, so a
+    node being offline today cannot make a live group look abandoned.
+
+    Deliberately a command rather than a loop inside the server: deleting other
+    people's groups on a timer nobody asked for is the kind of thing an operator
+    should schedule knowingly, and `--dry-run` lets them see the list first.
+    """
+    from meshbay_hub.db.engine import close_db, get_session_factory, init_db
+    from meshbay_hub.tasks.cleanup import UNHOSTED_GRACE_DAYS, prune_unhosted_groups
+
+    grace = UNHOSTED_GRACE_DAYS if days is None else days
+    # init_db also runs create_all. That is what the server does at startup, so
+    # it adds no risk here — and it creates missing *tables* only, never a
+    # missing column, which is why deploys run alembic separately.
+    await init_db(cfg.db.url)
+    try:
+        async with get_session_factory()() as db:
+            gone = await prune_unhosted_groups(db, grace_days=grace, dry_run=dry_run)
+    finally:
+        await close_db()
+
+    verb = "would delete" if dry_run else "deleted"
+    if not gone:
+        print(f"No group older than {grace} day(s) is still unhosted.")
+        return 0
+    print(f"{verb} {len(gone)} group(s) unhosted for more than {grace} day(s):")
+    for gid, name in gone:
+        print(f"  {gid}  {name}")
+    return 0
+
+
 if __name__ == "__main__":
     main()
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/3dc91cd4ea52_group_hosted_at.py b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/3dc91cd4ea52_group_hosted_at.py
new file mode 100644
index 0000000..57aa9a7
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/3dc91cd4ea52_group_hosted_at.py
@@ -0,0 +1,37 @@
+"""group hosted_at
+
+Records the first time a node announced that it hosts a group. Groups without
+it are shown to their owner only, and `meshbay-hub prune-groups` collects the
+ones that never got a node.
+
+Backfilled to created_at for every existing group: they predate the rule, and
+starting the clock on them retroactively would delete live groups on the first
+run of the reaper.
+
+Revision ID: 3dc91cd4ea52
+Revises: d1f47a90c3b2
+Create Date: 2026-08-16 03:20:00.575130
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '3dc91cd4ea52'
+down_revision: Union[str, Sequence[str], None] = 'd1f47a90c3b2'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column("groups",
+                  sa.Column("hosted_at", sa.DateTime(timezone=True), nullable=True))
+    # Existing groups are grandfathered in rather than left to expire.
+    op.execute("UPDATE groups SET hosted_at = created_at WHERE hosted_at IS NULL")
+
+
+def downgrade() -> None:
+    op.drop_column("groups", "hosted_at")
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/models.py b/packages/meshbay-hub/src/meshbay_hub/db/models.py
index 7d5a3e3..a220a64 100644
--- a/packages/meshbay-hub/src/meshbay_hub/db/models.py
+++ b/packages/meshbay-hub/src/meshbay_hub/db/models.py
@@ -99,6 +99,11 @@ class Group(Base):
     description: Mapped[str | None] = mapped_column(String(512))
     status:      Mapped[str] = mapped_column(String(16), default="active")    # active|suspended|revoked
     created_at:  Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
+    # First time a node registered on /v1/nodes/ws announcing that it hosts this
+    # group. Until then the group has no files, no key and nobody to serve it, so
+    # it is shown to its owner only and is what `prune-groups` collects. Set once
+    # and never cleared: a node going offline does not un-host a group.
+    hosted_at:   Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
 
     members:     Mapped[list["GroupMember"]] = relationship(back_populates="group")
 
diff --git a/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
index 7674c6e..c387100 100644
--- a/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
+++ b/packages/meshbay-hub/src/meshbay_hub/tasks/cleanup.py
@@ -1,13 +1,13 @@
-"""Scheduled cleanup tasks — IP log purge (1-year retention)."""
+"""Scheduled cleanup tasks — IP log purge, and unhosted group collection."""
 
 import asyncio
 import logging
 from datetime import datetime, timedelta, timezone
 
-from sqlalchemy import delete
+from sqlalchemy import delete, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from meshbay_hub.db.models import IPLog
+from meshbay_hub.db.models import Group, GroupMember, IPLog
 
 log = logging.getLogger(__name__)
 
@@ -38,3 +38,44 @@ async def cleanup_loop(get_session):
             await asyncio.sleep(CLEANUP_INTERVAL_HOURS * 3600)
     except asyncio.CancelledError:
         return
+
+
+# ── Groups that never got a node ──────────────────────────────────────────────
+
+UNHOSTED_GRACE_DAYS = 7
+
+
+async def find_unhosted_groups(db: AsyncSession, grace_days: int = UNHOSTED_GRACE_DAYS):
+    """Groups created more than `grace_days` ago that no node has ever announced.
+
+    `hosted_at` is set the first time a node registers claiming the group and is
+    never cleared, so this finds groups that were created and then abandoned —
+    not ones whose node happens to be offline today. That distinction is the
+    whole reason the column exists rather than a check against the live socket
+    registry, which would delete every group during a hub restart.
+    """
+    cutoff = datetime.now(timezone.utc) - timedelta(days=grace_days)
+    result = await db.execute(
+        select(Group).where(Group.hosted_at.is_(None), Group.created_at < cutoff))
+    return list(result.scalars().all())
+
+
+async def prune_unhosted_groups(db: AsyncSession, grace_days: int = UNHOSTED_GRACE_DAYS,
+                                dry_run: bool = False) -> list[tuple[str, str]]:
+    """Delete abandoned groups. Returns [(id, name)] of what was (or would be) removed.
+
+    Memberships go with the group — there is no cascade configured, and leaving
+    orphan rows behind would keep the group in everyone's /mine query through the
+    join. Nothing on a node is touched: the hub does not command those machines,
+    and by definition no node ever claimed this group anyway.
+    """
+    doomed = await find_unhosted_groups(db, grace_days)
+    if not doomed or dry_run:
+        return [(g.id, g.name) for g in doomed]
+
+    ids = [g.id for g in doomed]
+    await db.execute(delete(GroupMember).where(GroupMember.group_id.in_(ids)))
+    await db.execute(delete(Group).where(Group.id.in_(ids)))
+    await db.commit()
+    log.info("Pruned %d group(s) that no node ever hosted", len(doomed))
+    return [(g.id, g.name) for g in doomed]
diff --git a/packages/meshbay-hub/tests/test_group_hosting.py b/packages/meshbay-hub/tests/test_group_hosting.py
new file mode 100644
index 0000000..ad5f57d
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_group_hosting.py
@@ -0,0 +1,271 @@
+"""
+Groups nobody hosts, and the admission policy a public group is allowed.
+
+A group created before its node exists has no files, no key and nothing to
+connect to. Showing it in the directory produces a dead end for whoever clicks
+it, so it stays with its owner until a node announces it — and if none ever
+does, `prune-groups` collects it.
+
+The distinction that matters is *ever hosted* against *online now*. `hosted_at`
+is stamped once and never cleared, so a node being offline today cannot make a
+live group look abandoned. Checking the live socket registry instead would have
+deleted every group during a hub restart.
+"""
+
+import base64
+import hashlib
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import select
+
+from meshbay_hub.db.models import Group, GroupMember
+from meshbay_hub.tasks.cleanup import find_unhosted_groups, prune_unhosted_groups
+
+
+def _auth_key(password: str, username: str) -> str:
+    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
+    return base64.b64encode(
+        hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
+
+
+async def _user(client, username, password="a-long-enough-passphrase"):
+    await client.post("/v1/users/register", json={
+        "username": username, "email": f"{username}@example.com",
+        "auth_key": _auth_key(password, username)})
+    r = await client.post("/v1/users/login", json={
+        "username": username, "auth_key": _auth_key(password, username)})
+    return {"Authorization": f"Bearer {r.json()['access_token']}"}
+
+
+async def _group(client, owner, name, visibility="private", join_policy=None):
+    body = {"name": name, "visibility": visibility}
+    if join_policy is not None:
+        body["join_policy"] = join_policy
+    elif visibility == "public":
+        body["join_policy"] = "open"
+    return await client.post("/v1/groups", json=body, headers=owner)
+
+
+async def _mark_hosted(db_session, group_id, when=None):
+    g = await db_session.get(Group, group_id)
+    g.hosted_at = when or datetime.now(timezone.utc)
+    await db_session.commit()
+
+
+# ── Visibility before a node exists ───────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_the_owner_sees_their_unhosted_group(client):
+    """They have to: it is the page they set the node up from."""
+    owner = await _user(client, "setup1")
+    r = await _group(client, owner, "not-yet")
+    assert r.status_code == 201
+
+    mine = await client.get("/v1/groups/mine", headers=owner)
+    entry = next(g for g in mine.json()["groups"] if g["name"] == "not-yet")
+    assert entry["hosted"] is False
+
+
+@pytest.mark.asyncio
+async def test_a_member_does_not_see_an_unhosted_group(client):
+    """A name they cannot open, with no way to say why, is worse than nothing."""
+    owner = await _user(client, "setup2")
+    member = await _user(client, "early_bird")
+    gid = (await _group(client, owner, "premature")).json()["group_id"]
+    await client.post(f"/v1/groups/{gid}/members/early_bird", json={}, headers=owner)
+
+    mine = await client.get("/v1/groups/mine", headers=member)
+    assert [g["name"] for g in mine.json()["groups"]] == []
+
+
+@pytest.mark.asyncio
+async def test_a_member_sees_it_once_a_node_has_announced_it(client, db_session):
+    owner = await _user(client, "setup3")
+    member = await _user(client, "patient")
+    gid = (await _group(client, owner, "ready")).json()["group_id"]
+    await client.post(f"/v1/groups/{gid}/members/patient", json={}, headers=owner)
+
+    await _mark_hosted(db_session, gid)
+
+    mine = await client.get("/v1/groups/mine", headers=member)
+    entry = next(g for g in mine.json()["groups"] if g["id"] == gid)
+    assert entry["hosted"] is True
+
+
+@pytest.mark.asyncio
+async def test_the_public_directory_hides_unhosted_groups(client, db_session):
+    owner = await _user(client, "setup4")
+    hidden = (await _group(client, owner, "pub-unhosted", "public")).json()["group_id"]
+    shown = (await _group(client, owner, "pub-hosted", "public")).json()["group_id"]
+    await _mark_hosted(db_session, shown)
+
+    listed = (await client.get("/v1/groups")).json()["groups"]
+    ids = [g["id"] for g in listed]
+    assert shown in ids
+    assert hidden not in ids
+
+
+@pytest.mark.asyncio
+async def test_a_group_stays_visible_when_its_node_goes_offline(client, db_session):
+    """`hosted_at` records that a node existed, not that one is answering now."""
+    owner = await _user(client, "setup5")
+    gid = (await _group(client, owner, "quiet-node", "public")).json()["group_id"]
+    await _mark_hosted(db_session, gid)
+
+    listed = (await client.get("/v1/groups")).json()["groups"]
+    assert gid in [g["id"] for g in listed], "no node is online, and that is not the question"
+
+
+# ── Collecting the abandoned ones ─────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_fresh_unhosted_group_is_left_alone(client, db_session):
+    owner = await _user(client, "reaper1")
+    gid = (await _group(client, owner, "brand-new")).json()["group_id"]
+
+    assert await find_unhosted_groups(db_session) == []
+    assert await db_session.get(Group, gid) is not None
+
+
+@pytest.mark.asyncio
+async def test_an_unhosted_group_past_the_grace_period_is_collected(client, db_session):
+    owner = await _user(client, "reaper2")
+    gid = (await _group(client, owner, "abandoned")).json()["group_id"]
+
+    g = await db_session.get(Group, gid)
+    g.created_at = datetime.now(timezone.utc) - timedelta(days=8)
+    await db_session.commit()
+
+    gone = await prune_unhosted_groups(db_session)
+    assert [name for _, name in gone] == ["abandoned"]
+    assert await db_session.get(Group, gid) is None
+
+
+@pytest.mark.asyncio
+async def test_an_old_group_that_was_hosted_is_never_collected(client, db_session):
+    """The whole point of the column: age alone must not condemn a group."""
+    owner = await _user(client, "reaper3")
+    gid = (await _group(client, owner, "long-lived")).json()["group_id"]
+
+    g = await db_session.get(Group, gid)
+    g.created_at = datetime.now(timezone.utc) - timedelta(days=400)
+    g.hosted_at = datetime.now(timezone.utc) - timedelta(days=399)
+    await db_session.commit()
+
+    assert await find_unhosted_groups(db_session) == []
+
+
+@pytest.mark.asyncio
+async def test_dry_run_reports_without_deleting(client, db_session):
+    owner = await _user(client, "reaper4")
+    gid = (await _group(client, owner, "still-here")).json()["group_id"]
+    g = await db_session.get(Group, gid)
+    g.created_at = datetime.now(timezone.utc) - timedelta(days=30)
+    await db_session.commit()
+
+    gone = await prune_unhosted_groups(db_session, dry_run=True)
+    assert [name for _, name in gone] == ["still-here"]
+    assert await db_session.get(Group, gid) is not None, "dry run deleted a group"
+
+
+@pytest.mark.asyncio
+async def test_collecting_a_group_takes_its_memberships_with_it(client, db_session):
+    """Nothing cascades in the schema, and an orphan row keeps the group in /mine."""
+    owner = await _user(client, "reaper5")
+    await _user(client, "tagalong")
+    gid = (await _group(client, owner, "doomed")).json()["group_id"]
+    await client.post(f"/v1/groups/{gid}/members/tagalong", json={}, headers=owner)
+
+    g = await db_session.get(Group, gid)
+    g.created_at = datetime.now(timezone.utc) - timedelta(days=9)
+    await db_session.commit()
+
+    await prune_unhosted_groups(db_session)
+
+    rows = (await db_session.execute(
+        select(GroupMember).where(GroupMember.group_id == gid))).scalars().all()
+    assert rows == []
+
+
+# ── What a public group may be ────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_public_group_cannot_be_invite_only(client):
+    """It would be listed to everyone and admit nobody.
+
+    Admission by request was the alternative and was dropped: between strangers
+    the only channel is the hub, so a one-time code would travel through the
+    party it exists to exclude.
+    """
+    owner = await _user(client, "policy1")
+    r = await _group(client, owner, "contradiction", "public", join_policy="invite")
+    assert r.status_code == 422, r.text
+    assert "private" in r.json()["detail"]
+
+
+@pytest.mark.asyncio
+async def test_a_public_group_is_open(client):
+    owner = await _user(client, "policy2")
+    r = await _group(client, owner, "welcoming", "public", join_policy="open")
+    assert r.status_code == 201, r.text
+
+
+@pytest.mark.asyncio
+async def test_a_private_group_is_invite_only_by_default(client, db_session):
+    owner = await _user(client, "policy3")
+    gid = (await _group(client, owner, "closed")).json()["group_id"]
+    assert (await db_session.get(Group, gid)).join_policy == "invite"
+
+
+# ── The stamp itself ──────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_node_registration_stamps_hosted_at(client, db_session):
+    """The tests above set `hosted_at` directly; this one exercises the writer.
+
+    `_mark_hosted` is called from the node WebSocket handler with the group list
+    the hub derived from its own tables — a node can narrow that set but never
+    widen it (finding C2), so being announced here is evidence, not a claim.
+    """
+    from meshbay_hub.api.revocation import _mark_hosted
+
+    owner = await _user(client, "stamped")
+    gid = (await _group(client, owner, "about-to-be-hosted")).json()["group_id"]
+    assert (await db_session.get(Group, gid)).hosted_at is None
+
+    await _mark_hosted([gid])
+
+    await db_session.refresh(await db_session.get(Group, gid))
+    assert (await db_session.get(Group, gid)).hosted_at is not None
+
+
+@pytest.mark.asyncio
+async def test_the_stamp_is_not_moved_by_a_later_reconnection(client, db_session):
+    """It records that a node once existed, not when one last showed up."""
+    from meshbay_hub.api.revocation import _mark_hosted
+
+    owner = await _user(client, "stamped2")
+    gid = (await _group(client, owner, "steady")).json()["group_id"]
+    first = datetime.now(timezone.utc) - timedelta(days=30)
+    await _mark_hosted([gid])
+    g = await db_session.get(Group, gid)
+    g.hosted_at = first
+    await db_session.commit()
+
+    await _mark_hosted([gid])
+
+    g = await db_session.get(Group, gid)
+    await db_session.refresh(g)
+    # SQLite hands back a naive datetime where PostgreSQL keeps the offset, so
+    # the comparison is made on common ground rather than on the driver.
+    stored = g.hosted_at.replace(tzinfo=timezone.utc) if g.hosted_at.tzinfo is None \
+        else g.hosted_at
+    assert abs((stored - first).total_seconds()) < 1, "the stamp moved"
+
+
+@pytest.mark.asyncio
+async def test_marking_hosted_survives_an_empty_list(client):
+    """A node that hosts nothing must not take the socket down."""
+    from meshbay_hub.api.revocation import _mark_hosted
+    await _mark_hosted([])
diff --git a/packages/meshbay-hub/tests/test_group_leave_and_quota.py b/packages/meshbay-hub/tests/test_group_leave_and_quota.py
new file mode 100644
index 0000000..dfa18b8
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_group_leave_and_quota.py
@@ -0,0 +1,258 @@
+"""
+Leaving a group of your own accord, and the cap on public groups.
+
+Sits beside `test_group_membership.py`, which covers the owner removing someone
+else. The two are deliberately different endpoints rather than one with an
+exception for the self case, so they are tested apart.
+
+The refusals are the interesting half: the owner who cannot walk out and leave
+the group unmanageable, and the eleventh public group. What leaving does *not*
+do is asserted too — the hub can only drop a membership row. The node keeps the
+identity it pinned, the keypair bundle and the uploaded files until its operator
+removes them, and whoever left still holds the group key they were served.
+"""
+
+import base64
+import hashlib
+from datetime import datetime, timezone
+
+import pytest
+from sqlalchemy import select
+
+from meshbay_hub.api.groups import MAX_PUBLIC_GROUPS
+from meshbay_hub.db.models import Group, GroupMember, User
+
+
+def _auth_key(password: str, username: str) -> str:
+    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
+    return base64.b64encode(
+        hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
+
+
+async def _user(client, username, password="a-long-enough-passphrase"):
+    await client.post("/v1/users/register", json={
+        "username": username, "email": f"{username}@example.com",
+        "auth_key": _auth_key(password, username)})
+    r = await client.post("/v1/users/login", json={
+        "username": username, "auth_key": _auth_key(password, username)})
+    return {"Authorization": f"Bearer {r.json()['access_token']}"}
+
+
+async def _group(client, owner, name, visibility="private"):
+    # A public group must be open to join — invite-only is refused, because a
+    # group everyone can find and nobody can enter is a dead end.
+    body = {"name": name, "visibility": visibility}
+    if visibility == "public":
+        body["join_policy"] = "open"
+    r = await client.post("/v1/groups", json=body, headers=owner)
+    assert r.status_code == 201, r.text
+    return r.json()["group_id"]
+
+
+# ── Leaving ───────────────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_member_can_leave(client, db_session):
+    owner = await _user(client, "owner1")
+    member = await _user(client, "member1")
+    gid = await _group(client, owner, "readers")
+    await client.post(f"/v1/groups/{gid}/members/member1", json={}, headers=owner)
+
+    # Marked hosted, or the member would not see the group in the first place
+    # and the assertion below would hold whether or not leaving worked.
+    g = await db_session.get(Group, gid)
+    g.hosted_at = datetime.now(timezone.utc)
+    await db_session.commit()
+
+    before = await client.get("/v1/groups/mine", headers=member)
+    assert [g["id"] for g in before.json()["groups"]] == [gid], "precondition"
+
+    r = await client.post(f"/v1/groups/{gid}/leave", headers=member)
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "left"
+
+    mine = await client.get("/v1/groups/mine", headers=member)
+    assert [g["id"] for g in mine.json()["groups"]] == []
+
+
+@pytest.mark.asyncio
+async def test_leaving_removes_only_that_membership_row(client, db_session):
+    """The group and everyone else in it are untouched — this is not a deletion."""
+    owner = await _user(client, "owner2")
+    member = await _user(client, "member2")
+    gid = await _group(client, owner, "still-here")
+    await client.post(f"/v1/groups/{gid}/members/member2", json={}, headers=owner)
+
+    await client.post(f"/v1/groups/{gid}/leave", headers=member)
+
+    assert await db_session.get(Group, gid) is not None
+    rows = (await db_session.execute(
+        select(GroupMember).where(GroupMember.group_id == gid))).scalars().all()
+    assert len(rows) == 1, "the owner should still be a member"
+
+    mine = await client.get("/v1/groups/mine", headers=owner)
+    assert [g["id"] for g in mine.json()["groups"]] == [gid]
+
+
+@pytest.mark.asyncio
+async def test_leaving_does_not_touch_the_account_or_its_other_groups(client, db_session):
+    owner = await _user(client, "owner3")
+    member = await _user(client, "member3")
+    elsewhere = await _user(client, "owner3b")
+    gid = await _group(client, owner, "leaving")
+    other = await _group(client, elsewhere, "staying")
+    await client.post(f"/v1/groups/{gid}/members/member3", json={}, headers=owner)
+    await client.post(f"/v1/groups/{other}/members/member3", json={}, headers=elsewhere)
+
+    await client.post(f"/v1/groups/{gid}/leave", headers=member)
+
+    user = (await db_session.execute(
+        select(User).where(User.username == "member3"))).scalar_one()
+    assert user.status == "active"
+    assert await db_session.get(GroupMember, (other, user.id)) is not None
+
+
+@pytest.mark.asyncio
+async def test_the_owner_cannot_leave_their_own_group(client):
+    """It would leave the group with nobody able to admit, edit or delete it."""
+    owner = await _user(client, "owner4")
+    gid = await _group(client, owner, "orphan-risk")
+
+    r = await client.post(f"/v1/groups/{gid}/leave", headers=owner)
+    assert r.status_code == 409, r.text
+    assert "own this group" in r.json()["detail"]
+
+    mine = await client.get("/v1/groups/mine", headers=owner)
+    assert [g["id"] for g in mine.json()["groups"]] == [gid]
+
+
+@pytest.mark.asyncio
+async def test_leaving_twice_is_refused(client):
+    owner = await _user(client, "owner5")
+    member = await _user(client, "member5")
+    gid = await _group(client, owner, "once")
+    await client.post(f"/v1/groups/{gid}/members/member5", json={}, headers=owner)
+
+    assert (await client.post(f"/v1/groups/{gid}/leave",
+                              headers=member)).status_code == 200
+    assert (await client.post(f"/v1/groups/{gid}/leave",
+                              headers=member)).status_code == 404
+
+
+@pytest.mark.asyncio
+async def test_leaving_a_group_you_were_never_in_is_refused(client):
+    owner = await _user(client, "owner6")
+    stranger = await _user(client, "stranger6")
+    gid = await _group(client, owner, "not-yours")
+
+    r = await client.post(f"/v1/groups/{gid}/leave", headers=stranger)
+    assert r.status_code == 404
+
+
+# ── Public group cap ──────────────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_public_groups_are_capped(client):
+    owner = await _user(client, "prolific")
+    for i in range(MAX_PUBLIC_GROUPS):
+        await _group(client, owner, f"public-{i}", visibility="public")
+
+    r = await client.post("/v1/groups",
+                          json={"name": "one-too-many", "visibility": "public",
+                                "join_policy": "open"},
+                          headers=owner)
+    assert r.status_code == 409, r.text
+    detail = r.json()["detail"]
+    assert str(MAX_PUBLIC_GROUPS) in detail
+    assert "private" in detail, "the message should name the way out"
+
+
+@pytest.mark.asyncio
+async def test_private_groups_are_not_capped(client):
+    """Private groups cost other people nothing — they are invisible to non-members."""
+    owner = await _user(client, "hoarder")
+    for i in range(MAX_PUBLIC_GROUPS + 5):
+        await _group(client, owner, f"private-{i}")
+
+    mine = await client.get("/v1/groups/mine", headers=owner)
+    assert len(mine.json()["groups"]) == MAX_PUBLIC_GROUPS + 5
+
+
+@pytest.mark.asyncio
+async def test_a_suspended_public_group_does_not_hold_a_slot(client, db_session):
+    """Moderation already dealt with the owner; the slot should not punish twice."""
+    owner = await _user(client, "moderated")
+    ids = [await _group(client, owner, f"pub-{i}", visibility="public")
+           for i in range(MAX_PUBLIC_GROUPS)]
+
+    suspended = await db_session.get(Group, ids[0])
+    suspended.status = "suspended"
+    await db_session.commit()
+
+    r = await client.post("/v1/groups",
+                          json={"name": "replacement", "visibility": "public",
+                                "join_policy": "open"},
+                          headers=owner)
+    assert r.status_code == 201, r.text
+
+
+@pytest.mark.asyncio
+async def test_the_cap_is_per_owner(client):
+    """Being a member of someone else's public groups costs nothing."""
+    a = await _user(client, "ownera")
+    await _user(client, "ownerb")
+    b = await _user(client, "ownerb2")
+    for i in range(MAX_PUBLIC_GROUPS):
+        gid = await _group(client, a, f"a-pub-{i}", visibility="public")
+        await client.post(f"/v1/groups/{gid}/members/ownerb2", json={}, headers=a)
+
+    r = await client.post("/v1/groups",
+                          json={"name": "b-first", "visibility": "public",
+                                "join_policy": "open"},
+                          headers=b)
+    assert r.status_code == 201, r.text
+
+
+@pytest.mark.asyncio
+async def test_hub_staff_are_exempt(client, db_session):
+    """The cap is anti-spam, not a rule about running an instance."""
+    owner = await _user(client, "instanceadmin")
+    user = (await db_session.execute(
+        select(User).where(User.username == "instanceadmin"))).scalar_one()
+    user.role = "admin"
+    await db_session.commit()
+
+    for i in range(MAX_PUBLIC_GROUPS + 2):
+        r = await client.post("/v1/groups",
+                              json={"name": f"admin-pub-{i}", "visibility": "public",
+                                    "join_policy": "open"},
+                              headers=owner)
+        assert r.status_code == 201, r.text
+
+
+# ── Presence in the group list ────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_the_group_list_reports_node_presence(client):
+    """`node_online` rides on the request the sidebar already makes.
+
+    It reflects the hub's signaling registry, so it says a node is connected
+    *to the hub* — not that this browser can reach it, and not something a
+    dishonest hub could not fake. The client downgrades it on a connection it
+    tried and failed, which is the evidence that concerns the reader.
+    """
+    owner = await _user(client, "watcher")
+    gid = await _group(client, owner, "quiet")
+
+    mine = await client.get("/v1/groups/mine", headers=owner)
+    entry = next(g for g in mine.json()["groups"] if g["id"] == gid)
+    assert entry["node_online"] is False
+
+    from meshbay_hub.api import revocation
+    revocation._node_groups["node-x"] = [gid]
+    try:
+        mine = await client.get("/v1/groups/mine", headers=owner)
+        entry = next(g for g in mine.json()["groups"] if g["id"] == gid)
+        assert entry["node_online"] is True
+    finally:
+        revocation._node_groups.pop("node-x", None)
diff --git a/packages/meshbay-hub/tests/test_groups_self_service.py b/packages/meshbay-hub/tests/test_groups_self_service.py
index a60209d..3202658 100644
--- a/packages/meshbay-hub/tests/test_groups_self_service.py
+++ b/packages/meshbay-hub/tests/test_groups_self_service.py
@@ -40,6 +40,16 @@ async def _create_group(client, token, name="test-group", visibility="public",
     return r.json()["group_id"]
 
 
+
+async def _mark_hosted(db_session, *group_ids):
+    """Pretend a node announced these groups, as /v1/nodes/ws would."""
+    from datetime import datetime, timezone
+    from meshbay_hub.db.models import Group
+    for gid in group_ids:
+        (await db_session.get(Group, gid)).hosted_at = datetime.now(timezone.utc)
+    await db_session.commit()
+
+
 @pytest.mark.asyncio
 async def test_create_group(client):
     await _register(client, "alice", email="a@x.com")
@@ -73,8 +83,11 @@ async def test_join_open_group(client):
 async def test_join_invite_group_rejected(client):
     await _register(client, "alice", email="a@x.com")
     alice_token = await _login(client, "alice")
+    # Private: invite-only is refused on a public group now, since a group
+    # everyone can find and nobody can enter is a dead end. What is under test
+    # here — /join refusing a group that is not open — is unchanged.
     gid = await _create_group(client, alice_token, "invite-group",
-                              join_policy="invite")
+                              visibility="private", join_policy="invite")
 
     await _register(client, "bob", email="b@x.com")
     bob_token = await _login(client, "bob")
@@ -132,11 +145,14 @@ async def test_group_members_non_member_denied(client):
 
 
 @pytest.mark.asyncio
-async def test_group_search(client):
+async def test_group_search(client, db_session):
     await _register(client, "alice", email="a@x.com")
     token = await _login(client, "alice")
-    await _create_group(client, token, "alpha-team")
-    await _create_group(client, token, "beta-team")
+    a = await _create_group(client, token, "alpha-team")
+    b = await _create_group(client, token, "beta-team")
+    # The directory shows groups a node has announced. Marked here so this test
+    # exercises the search filter rather than the hosting one.
+    await _mark_hosted(db_session, a, b)
 
     r = await client.get("/v1/groups?q=alpha")
     assert r.status_code == 200
diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py
index 5a2cf86..5314b93 100644
--- a/packages/meshbay-hub/tests/test_hub_api.py
+++ b/packages/meshbay-hub/tests/test_hub_api.py
@@ -365,7 +365,7 @@ async def test_jwt_contains_groups_claim(client):
 # ── My groups (9.6) ─────────────────────────────────────────────────────────
 
 @pytest.mark.asyncio
-async def test_my_groups(client):
+async def test_my_groups(client, db_session):
     """GET /v1/groups/mine returns groups the user belongs to."""
     pk_ed_a, pk_x_a, _ = _gen_user_keys()
     pk_ed_b, pk_x_b, _ = _gen_user_keys()
@@ -396,6 +396,14 @@ async def test_my_groups(client):
                       json={},
                       headers={"Authorization": f"Bearer {alice_token}"})
 
+    # A group no node has announced is shown to its owner only — a member would
+    # otherwise see a name they cannot open. Stamped here so the rest of this
+    # test is about membership, which is what it was written for.
+    from datetime import datetime, timezone
+    from meshbay_hub.db.models import Group
+    (await db_session.get(Group, group_id)).hosted_at = datetime.now(timezone.utc)
+    await db_session.commit()
+
     # Re-login to get fresh token with group claims
     bob_token = (await client.post("/v1/users/login",
                  json={"username": "mg_bob", "password": "bobpass99"})).json()["access_token"]