scripts/dev_server.py

"""Local dev backend for the test environment (see testenv/README.md).

Runs the REAL bucket-sync FastAPI app against a **filesystem-persistent
FakeHub**: every bucket lives under ``--root`` as ``{org}/{bucket}/...``, so
state survives restarts and the dashboard (in ``LOCAL_BUCKET_DIR`` mode) reads
the very same files the backend writes. No HF org, tokens, or Spaces — the
whole stack runs offline.

Identity is scriptable, mirroring tests/conftest.py: ANY bearer token resolves
to ``--user`` (an org admin, so organizer-gated features like broadcasts work
too), and agent bucket-ownership proofs work because the fake honors the same
``hf://buckets/{org}/{slug}-{agent}/...`` layout on disk.

    python scripts/dev_server.py --root ../.testenv/buckets --seed
"""
from __future__ import annotations

import argparse
import logging
import os
import sys
from pathlib import Path

_BACKEND = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(_BACKEND))          # app.*
sys.path.insert(0, str(_BACKEND / "tests"))  # fakes (the canonical hub fake)

# Dev-friendly limits unless the caller pins their own — the production
# default (5 raw msgs/min) gets in the way of hammering a test stack.
# Must be set before app.config reads the env.
os.environ.setdefault("RAW_MESSAGE_PER_MINUTE", "60")
os.environ.setdefault("RAW_MESSAGE_PER_HOUR", "1000")

from app.audit import AuditLogger                      # noqa: E402
from app.config import Settings                        # noqa: E402
from app.dedup import PromotionLRU                     # noqa: E402
from app.deps import (                                 # noqa: E402
    get_audit,
    get_bucket_write_limiter,
    get_dedup,
    get_hub,
    get_notifier,
    get_org_roles,
    get_raw_message_limiter,
    get_read_model,
    get_registration_limiter,
    get_settings_dep,
    get_verification_status,
    get_verifier,
)
from app.main import app as fastapi_app                # noqa: E402
from app.notify import Notifier                        # noqa: E402
from app.org_roles import OrgRoles                     # noqa: E402
from app.rate_limit import CompoundLimiter, TokenBucket  # noqa: E402
from app.read_model import ReadModel                   # noqa: E402
from app.verification import VerificationStatusStore   # noqa: E402
from app.verifier import Verifier                      # noqa: E402
from fakes import FakeHub, FakeJobRunner, seed_agent   # noqa: E402


log = logging.getLogger("dev-server")


class PersistentFakeHub(FakeHub):
    """FakeHub that mirrors every bucket to a directory tree.

    Layout: ``{root}/{org}/{bucket}/{path...}`` — the central bucket therefore
    lives at ``{root}/{ORG}/{SLUG}-main-bucket/``, which is exactly what the
    dashboard's ``LOCAL_BUCKET_DIR`` should point at. All reads stay in-memory
    (loaded once at boot); every write/delete goes through to disk.
    """

    def __init__(self, settings: Settings, root: Path):
        super().__init__(settings)
        self._root = root
        self._load()

    # ── disk mirroring ────────────────────────────────────────────
    def _file(self, bucket: str, path: str) -> Path:
        return self._root / bucket / path

    def _persist(self, bucket: str, path: str) -> None:
        data = self.buckets.get(bucket, {}).get(path)
        if data is None:
            return
        f = self._file(bucket, path)
        f.parent.mkdir(parents=True, exist_ok=True)
        f.write_bytes(data)

    def _load(self) -> None:
        if not self._root.is_dir():
            return
        n = 0
        for org_dir in self._root.iterdir():
            if not org_dir.is_dir():
                continue
            for bucket_dir in org_dir.iterdir():
                if not bucket_dir.is_dir():
                    continue
                bucket = f"{org_dir.name}/{bucket_dir.name}"
                files = self.buckets.setdefault(bucket, {})
                for f in bucket_dir.rglob("*"):
                    if f.is_file():
                        files[str(f.relative_to(bucket_dir))] = f.read_bytes()
                        n += 1
        if n:
            log.info("loaded %d files from %s", n, self._root)

    # ── write-through overrides ───────────────────────────────────
    def seed(self, path: str, text: str, bucket: str | None = None) -> None:
        super().seed(path, text, bucket)
        self._persist(bucket or self._settings.central_bucket, path)

    def write_text_central(self, path: str, text: str) -> None:
        super().write_text_central(path, text)
        self._persist(self._settings.central_bucket, path)

    def write_bytes_central(self, path: str, data: bytes) -> None:
        super().write_bytes_central(path, data)
        self._persist(self._settings.central_bucket, path)

    def write_many_central(self, items: list[tuple[bytes, str]]) -> None:
        super().write_many_central(items)
        for _, p in items:
            self._persist(self._settings.central_bucket, p)

    def delete_central(self, path: str) -> None:
        super().delete_central(path)
        f = self._file(self._settings.central_bucket, path)
        if f.is_file():
            f.unlink()

    def write_bytes_to_bucket(self, bucket: str, path: str, data: bytes) -> None:
        super().write_bytes_to_bucket(bucket, path, data)
        self._persist(bucket, path)

    def append_jsonl_audit(self, path: str, line: str) -> None:
        super().append_jsonl_audit(path, line)
        self._persist(self._settings.audit_bucket, path)

    def write_bytes_audit(self, path: str, data: bytes) -> None:
        super().write_bytes_audit(path, data)
        self._persist(self._settings.audit_bucket, path)

    def copy_file_to_central(self, src_bucket: str, src_xet_hash: str, dest_path: str) -> None:
        super().copy_file_to_central(src_bucket, src_xet_hash, dest_path)
        self._persist(self._settings.central_bucket, dest_path)

    # ── live disk fallbacks ───────────────────────────────────────
    # Files dropped into {root}/{org}/{bucket}/ AFTER boot (the "agent writes
    # to their scratch bucket, then promotes" workflow) are picked up without
    # a restart: reads fall back to disk and cache in.

    def read_bytes(self, uri) -> bytes:
        try:
            return super().read_bytes(uri)
        except FileNotFoundError:
            from app.naming import SourceURI, parse_source_uri
            parsed = uri if isinstance(uri, SourceURI) else parse_source_uri(uri)
            if parsed is not None:
                f = self._file(f"{parsed.org}/{parsed.bucket}", parsed.path)
                if f.is_file():
                    data = f.read_bytes()
                    self.buckets.setdefault(f"{parsed.org}/{parsed.bucket}", {})[parsed.path] = data
                    return data
            raise

    def bucket_exists(self, bucket: str) -> bool:
        return super().bucket_exists(bucket) or (self._root / bucket).is_dir()

    def copy_tree_to_central(self, src_bucket: str, src_prefix: str, dest_prefix: str):
        out = super().copy_tree_to_central(src_bucket, src_prefix, dest_prefix)
        for _src, dest, _size in out:
            self._persist(self._settings.central_bucket, dest)
        return out


def seed_world(hub: PersistentFakeHub, settings: Settings) -> None:
    """A small, believable starting state: two registered agents with scratch
    buckets (handshakes in place, so every source-URI flow works out of the
    box) and a couple of board messages. Idempotent — seeding an already
    seeded world just rewrites the same files."""
    for agent, hf_user in (("byte-bandit", "bb-hf"), ("delta-coder", "dc-hf")):
        seed_agent(hub, agent, hf_user=hf_user)
        scratch = settings.agent_bucket(agent)
        hub.seed(".bucket-sync-handshake", hf_user, bucket=scratch)
        hub.seed("subscribe.md", "following", bucket=scratch)
    hub.seed(
        "message_board/20260707-090000-000_byte-bandit.md",
        "---\nagent: byte-bandit\ntimestamp: 2026-07-07 09:00 UTC\nvia: raw\ntype: agent\n---\n"
        "joining; planning an arithmetic-coder baseline\n",
    )
    hub.seed(
        "message_board/20260707-091500-000_delta-coder.md",
        "---\nagent: delta-coder\ntimestamp: 2026-07-07 09:15 UTC\nvia: raw\ntype: agent\n---\n"
        "@byte-bandit interested — comparing notes on context models\n",
    )
    log.info("seeded 2 agents + 2 board messages")


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--root", default=str(_BACKEND.parent / ".testenv" / "buckets"),
                    help="directory that plays the role of HF bucket storage")
    ap.add_argument("--org", default="local-org")
    ap.add_argument("--slug", default="collab")
    ap.add_argument("--user", default="tester",
                    help="every bearer token resolves to this HF user (an org admin)")
    ap.add_argument("--host", default="127.0.0.1")
    ap.add_argument("--port", type=int, default=8100)
    ap.add_argument("--seed", action="store_true", help="seed agents + board messages")
    args = ap.parse_args()

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")

    settings = Settings(
        HF_TOKEN="dev-admin-token",
        ORG=args.org,
        COLLAB_SLUG=args.slug,
        AUDIT_BUCKET=f"{args.org}-private/{args.slug}-audit",
    )
    root = Path(args.root).resolve()
    root.mkdir(parents=True, exist_ok=True)
    hub = PersistentFakeHub(settings, root)
    hub.whoami_user = args.user
    hub.whoami_email = f"{args.user}@example.com"
    hub.whoami_orgs = {args.org}
    # The dev user is an org admin: organizer-gated paths (broadcast, /v1/me
    # is_organizer) behave like production for an organizer session.
    hub.org_roles = {args.user: "admin"}
    hub.org_roles_by_email = {f"{args.user}@example.com": (args.user, "admin")}

    if args.seed:
        seed_world(hub, settings)

    read_model = ReadModel(hub, settings)
    dedup = PromotionLRU(settings.dedup_lru_size)
    verification = VerificationStatusStore(hub, runs_prefix=settings.verification_runs_prefix)
    # Every singleton in app/deps.py reads the env-backed settings, so each one
    # this Settings() must reach needs an override below — the notifier included,
    # or /v1/healthz and every wait= route 500s on a Settings ValidationError.
    notifier = Notifier(
        max_waiters_per_owner=settings.longpoll_max_waiters_per_owner,
        max_waiters_total=settings.longpoll_max_waiters_total,
        wake_spread_s=settings.longpoll_wake_spread_s,
        wake_spread_threshold=settings.longpoll_wake_spread_threshold,
    )
    # The verifier posts verdict messages, so it holds the notifier too — without
    # it a verdict would land silently and never wake a parked watcher.
    verifier = Verifier(settings, hub, read_model, verification, FakeJobRunner(),
                        spawn=lambda _name, fn: fn(), notifier=notifier)

    def compound(burst: int, sustained: int) -> CompoundLimiter:
        return CompoundLimiter(
            TokenBucket(capacity=burst, refill_per_minute=burst),
            TokenBucket(capacity=sustained, refill_per_minute=sustained),
        )

    fastapi_app.dependency_overrides.update({
        get_settings_dep: lambda: settings,
        get_hub: lambda: hub,
        get_read_model: lambda: read_model,
        get_notifier: lambda: notifier,
        get_org_roles: lambda: OrgRoles(hub, settings),
        get_audit: lambda: AuditLogger(hub),
        get_dedup: lambda: dedup,
        get_verification_status: lambda: verification,
        get_verifier: lambda: verifier,
        # Real (env-tunable) production limiter shapes — the point of the
        # test environment is realism, just with dev-friendly defaults.
        get_bucket_write_limiter: lambda: compound(
            settings.bucket_write_burst, settings.bucket_write_per_minute),
        get_raw_message_limiter: lambda: compound(
            settings.raw_message_per_minute, settings.raw_message_per_hour),
        get_registration_limiter: lambda: TokenBucket(
            capacity=settings.registration_per_minute,
            refill_per_minute=settings.registration_per_minute),
    })

    log.info("central bucket on disk: %s", root / settings.central_bucket)
    log.info("bearer identity: %s (admin of %s)", args.user, args.org)

    import uvicorn
    uvicorn.run(fastapi_app, host=args.host, port=args.port, log_level="info")


if __name__ == "__main__":
    main()