app.py

"""Grabette fleet — operator dashboard + command broker (Docker HF Space).

A free-tier-friendly Docker Space. It is self-contained (no `grabette` import) so
it deploys standalone to HF. Responsibilities:

  * Operator UI + login via HF Spaces native OAuth (`hf_oauth: true`).
  * Command broker: an in-memory, per-owner device registry + command queue.
  * Device auth: devices call with `Authorization: Bearer <hf_token>`; we resolve
    the owner via `whoami` (cached) and group devices by HF identity.

Transport is short-polling (devices GET /api/devices/poll every couple seconds).
That polling traffic doubles as the keep-alive that stops a free Space sleeping
(sleep is timed from the last request). State is in-memory, so a restart drops
it — devices simply re-register on their next poll; durable data lives in the
device's own HF datasets, not here.

Designed to be duplicated per user: one Space = one owner = one fleet.
"""

from __future__ import annotations

import asyncio
import logging
import os
import time
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Optional

from fastapi import Depends, FastAPI, Header, HTTPException, Request
from fastapi.responses import HTMLResponse, RedirectResponse

logger = logging.getLogger("grabette-fleet")
from huggingface_hub import attach_huggingface_oauth, parse_huggingface_oauth, whoami
from pydantic import BaseModel

# A device is "online" if the fleet heard from it within this window. Liveness
# no longer rides on the command poll (long-polling holds that open ~LONG_POLL_S,
# and muxing runs off the poll loop) but on a dedicated lightweight HEARTBEAT the
# device sends every DEVICE_HEARTBEAT_S. So this only needs to cover a missed
# heartbeat + jitter (~3× the interval) — small, for fast disconnect detection.
DEVICE_HEARTBEAT_S = 5.0   # must match the device's relay heartbeat cadence
ONLINE_WINDOW = 15.0       # ~3 missed heartbeats → offline within ~15s
# Only discard a device's queued commands once it's been gone THIS long — far
# beyond ONLINE_WINDOW and any transient blip. Clearing on the mere online
# window (15s) would nuke an in-flight start/stop the instant a device flapped,
# breaking group sync; a device that's briefly "offline" but still holding a
# long-poll receives its queued command the moment it's enqueued regardless.
STALE_QUEUE_S = 120.0
WHOAMI_TTL = 300.0
# Long-polling: the poll endpoint holds the connection open up to this many
# seconds when the device's queue is empty, returning the instant a command is
# enqueued (see _enqueue). This cuts command-delivery latency from ~1 poll
# interval to a network round-trip — notably a fanned-out group STOP reaches a
# peer in ~ms, so grouped episodes end within ~ms of each other instead of ~1s.
# MUST stay below BOTH the HF Space reverse-proxy idle timeout AND ONLINE_WINDOW
# (a held poll only refreshes last_seen when it returns, so a device mid-hold
# must not age out as offline). Set to 0 to DISABLE → classic short-polling: the
# poll returns immediately and the client throttles to its poll_interval. The
# relay client auto-detects which mode the server is in, so flipping this needs
# no device change — the escape hatch if the Space's proxy/sleep misbehaves.
LONG_POLL_S = 25.0
# LeRobot dataset generation: the fleet gathers the selected tasks' episodes,
# has each device push its OWN streams to a shared raw dataset (by role), then
# triggers a processing Space that converts raw → LeRobot (mono or bimanual,
# per the device set). Overridable so the test Space can be targeted.
LEROBOT_SPACE_URL = os.environ.get(
    "GRABETTE_y.LEROBOT_SPACE_URL", "https://pollen-robotics-grabette-slam-duplicate.hf.space"
).rstrip("/")
# Lead time before a group's synchronized start actually fires. The device
# warms its hardware DURING this lead, then waits out the shared T0 on its own
# NTP-disciplined clock. The lead must cover (poll delivery + hardware warmup)
# so warmup finishes BEFORE T0 — otherwise the variable warmup leaks into the
# start and the devices drift apart. Warmup is only long when the OAK-D is cold
# (multi-second cold boot); once it has recorded recently it stays warm (device
# keepalive), so we use two leads:
# Must cover (poll delivery to the peer ~2.5s + OAK-D cold boot). Measured: a
# peer overran a 12s lead by ~0.7s (cold boot ≈10.2s, but it only had ~9.5s
# after poll delivery), so 15s gives ~2.3s margin over the observed cold boot.
GROUP_START_LEAD_COLD_S = 15.0   # OAK-D likely asleep → cover poll delivery + cold boot
# Even when warm, the lead must exceed the WORST-CASE command delivery to a
# peer: a device polls only every ~2.5s AND its relay is blocked while it muxes
# the previous episode's stop (several seconds), so a too-short warm lead makes
# the peer receive its start_capture after T0 (→ started late, best-effort — it
# no longer drops the episode, but it's desynced). 3s covers the common case;
# the real cure for back-to-back on-time starts is non-blocking relay muxing.
GROUP_START_LEAD_WARM_S = 1.0
# Fleet can't see the OAK-D power state directly; it infers "warm" from the time
# since the session's last recording stop. This window MUST stay safely below
# the device's OAK-D keepalive (GRABETTE oakd_keepalive_s, default 30s) so that
# whenever fleet says "warm" the OAK-D is DEFINITELY still powered — a false
# "warm" would desync the start, a false "cold" only costs an unnecessarily long
# lead. Fleet's stop timestamp is the dispatch time, earlier than the device's
# actual keepalive countdown start, which makes the estimate extra conservative.
OAK_WARM_WINDOW_S = 25.0
# Lead for a lead-based SYNCHRONIZED stop (shared future T_stop fanned to every
# member so they end together). CURRENTLY UNUSED: stops are dispatched
# immediately (see _dispatch_episode_stop) so a button press feels instant and
# peers trail only by the ~1s poll delivery. Kept here (with the device-side
# CaptureScheduler.schedule_stop path) so lead-based sync can be re-enabled
# without re-plumbing — e.g. if delivery latency ever grew. To switch back:
# have _dispatch_episode_stop send args={"stop_at_utc": now + GROUP_STOP_LEAD_S}.
GROUP_STOP_LEAD_S = 3.0


# --- device identity (Bearer token -> HF username, cached) -------------------
_whoami_cache: dict[str, tuple[str, float]] = {}
# Derived namespace list per owner (username + orgs) for the dataset owner
# dropdown — NOT the raw token. Device tokens list orgs reliably (unlike the
# short-lived operator OAuth token), so we compute the list from a device token
# when one passes through, cache the RESULT, and discard the token. In a shared
# Space this avoids holding every user's write-capable token in memory.
# owner -> (expiry_epoch, [namespaces]).
_namespaces_cache: dict[str, tuple[float, list[str]]] = {}
_NAMESPACES_TTL = 600.0


def _cached_user(token: str) -> Optional[str]:
    """Fast path: no I/O. Returns the cached username if still fresh."""
    hit = _whoami_cache.get(token)
    if hit and hit[1] > time.time():
        return hit[0]
    return None


def _whoami_blocking(token: str) -> str:
    """Slow path: hits the HF API. Synchronous — the huggingface_hub client
    has no async variant, so this must always be run off the event loop
    thread (see verify_user) or a single cache-miss stalls every other
    request the whole broker is serving (all owners' polling, dispatch, and
    the synchronized-start scheduling), since this process runs a single
    worker with all state in memory.
    """
    try:
        info = whoami(token=token)
    except Exception as e:  # noqa: BLE001
        raise ValueError(f"Invalid HF token: {e}") from e
    name = (info or {}).get("name", "")
    if not name:
        raise ValueError("Could not resolve HF username")
    _whoami_cache[token] = (name, time.time() + WHOAMI_TTL)
    return name


async def verify_user(token: str) -> str:
    cached = _cached_user(token)
    if cached is not None:
        return cached
    return await asyncio.to_thread(_whoami_blocking, token)


async def device_auth(authorization: Optional[str] = Header(None)) -> tuple[str, str]:
    """FastAPI dep: resolve (owner, token) from a device's Bearer HF token.

    Returns the raw token too — handlers use it transiently within the request
    (e.g. dispatching a device's own Space call). It is deliberately NOT retained
    anywhere on the fleet: in a shared Space that would concentrate every user's
    write-capable HF token in one process. Owner identity is resolved via a cached
    whoami; the operator's namespace list is derived + cached separately (see
    _touch_namespaces) so no raw token needs to be kept.
    """
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(401, "Missing 'Authorization: Bearer <hf_token>'")
    token = authorization[len("Bearer ") :].strip()
    try:
        owner = await verify_user(token)
    except ValueError as e:
        raise HTTPException(401, str(e)) from e
    return owner, token


async def device_owner(authorization: Optional[str] = Header(None)) -> str:
    """FastAPI dep: resolve just the device's owner from its Bearer HF token."""
    owner, _token = await device_auth(authorization)
    return owner


def _cache_namespaces_blocking(owner: str, token: str) -> None:
    """Blocking: resolve the owner's pushable namespaces (username + orgs) from a
    device token and cache the RESULT — never the token itself."""
    names = [owner]
    try:
        info = whoami(token=token)
        names += [o["name"] for o in (info.get("orgs") or []) if o.get("name")]
    except Exception:
        logger.debug("namespaces whoami failed for %s", owner, exc_info=True)
    _namespaces_cache[owner] = (time.time() + _NAMESPACES_TTL, list(dict.fromkeys(names)))


def _touch_namespaces(owner: str, token: Optional[str]) -> None:
    """TTL-gated, fire-and-forget: refresh the owner's namespace list from a
    passing device token, so the dataset-owner dropdown works WITHOUT the fleet
    keeping the token. Marks the cache fresh up front to avoid a refresh stampede."""
    if not token:
        return
    ent = _namespaces_cache.get(owner)
    if ent and ent[0] > time.time():
        return  # still fresh
    _namespaces_cache[owner] = (time.time() + _NAMESPACES_TTL, ent[1] if ent else [owner])
    t = asyncio.create_task(asyncio.to_thread(_cache_namespaces_blocking, owner, token))
    _bg_tasks.add(t)
    t.add_done_callback(_bg_tasks.discard)


def operator_auth(request: Request) -> tuple[str, Optional[str]]:
    """Resolve (owner, oauth_access_token) for the logged-in operator, or 401.

    The OAuth token identifies the operator (their namespace). The fleet keeps
    no durable state of its own — the task/episode view is aggregated live from
    connected devices' reports (see _reported_tasks)."""
    info = parse_huggingface_oauth(request)
    if info is None:
        raise HTTPException(401, "Not logged in")
    owner = info.user_info.preferred_username or info.user_info.name
    return owner, getattr(info, "access_token", None)


def operator_name(request: Request) -> str:
    """Resolve the logged-in operator's HF username, or 401."""
    return operator_auth(request)[0]


# --- in-memory fleet state ---------------------------------------------------
@dataclass
class Command:
    id: str
    type: str
    args: dict[str, Any]
    status: str = "pending"  # pending | sent | done
    result: Optional[dict[str, Any]] = None
    created_at: float = field(default_factory=time.time)
    done_at: Optional[float] = None


@dataclass
class Device:
    device_id: str
    name: str
    capabilities: list[str]
    hand: str = ""  # "left" or "right", reported by the device on register
    ip: str = ""  # device LAN IPv4, reported by the device on register
    battery: Optional[float] = None  # last-reported battery %, sent via heartbeat
    # Device self-reported activity, sent on the heartbeat:
    # "" (device not updated / unknown) | idle | capturing | uploading | processing.
    # Empty falls back to fleet-side inference (see _device_activity).
    reported_status: str = ""
    # This device's recorded tasks, reported on register (see TaskManager.report_tasks).
    # The device is the durable source of truth; the fleet aggregates these across
    # connected devices (phase 2). Stored but not yet consumed in phase 1.
    tasks: list[dict] = field(default_factory=list)
    # Bumped each register (when tasks are (re)reported). Feeds the aggregation
    # cache signature so it recomputes only when a device's report changed.
    report_rev: int = 0
    last_seen: float = field(default_factory=time.time)
    queue: list[Command] = field(default_factory=list)
    history: list[Command] = field(default_factory=list)
    pending_delete: bool = False
    # Set whenever a command is enqueued (see _enqueue) to release a long-poll
    # holding on this device. Excluded from repr/eq — it's live runtime state,
    # never serialized (devices aren't persisted). asyncio.Event() binds to a
    # loop lazily (on first await), so constructing it off-loop here is fine.
    wakeup: asyncio.Event = field(default_factory=asyncio.Event, repr=False, compare=False)

    @property
    def online(self) -> bool:
        return (time.time() - self.last_seen) < ONLINE_WINDOW


FLEET: dict[str, dict[str, Device]] = {}  # owner -> device_id -> Device


def _fleet_of(owner: str) -> dict[str, Device]:
    return FLEET.setdefault(owner, {})


def kind_of(dev: Device) -> str:
    """Mirrors the frontend's kindOf(): classify a device as grabette/gripette/casquette."""
    s = f"{dev.name} {dev.device_id} {' '.join(dev.capabilities)}".lower()
    if "gripette" in s:
        return "gripette"
    if "casquette" in s:
        return "casquette"
    return "grabette"


def _device_slot(dev: Device) -> Optional[str]:
    """The task-signature role this device fills: 'left'/'right' for a handed
    grabette, 'casquette' for a casquette, else None (not recordable)."""
    k = kind_of(dev)
    if k == "casquette":
        return "casquette"
    if k == "grabette" and dev.hand in ("left", "right"):
        return dev.hand
    return None


# --- tasks (registry, source of truth for task names sent to devices) --------
# A task is a *type of action*. It carries only the device *roles* it expects
# (its "signature"), never concrete device ids — those live on the group. The
# task NAME is the stable join key devices resolve locally via
# get_or_create_task, so it must come from this single registry.
VALID_SLOTS = ("left", "right", "casquette")


@dataclass
class Task:
    id: str
    name: str
    description: str = ""
    # Expected device roles, a subset of VALID_SLOTS (e.g. ["left","right"] for
    # a bimanual task, ["right"] for a single right-hand one). Empty = no
    # constraint. Used to validate the group assigned to the task and to tell
    # the dataset builder which roles produce data.
    device_signature: list[str] = field(default_factory=list)
    created_at: float = field(default_factory=time.time)


TASKS: dict[str, dict[str, Task]] = {}  # owner -> task_id -> Task


def _tasks_of(owner: str) -> dict[str, Task]:
    return TASKS.setdefault(owner, {})


# --- task/episode aggregation from device reports ---------------------------
# The devices are the durable source of truth for tasks and for who recorded each
# episode. On connect each device reports its tasks (see TaskManager.report_tasks);
# the fleet merges the reports from currently-connected devices to reconstruct the
# task/episode view — so ANY operator sees the existing tasks and can generate a
# dataset from them, regardless of which HF account did the original acquisition.

# Task names to hide briefly after an edit(rename)/delete, until the devices have
# processed the command and re-reported without them. Bridges the window where a
# stale report would otherwise resurrect a just-deleted/renamed task. owner ->
# {name: expiry_epoch}.
TASK_SUPPRESSED: dict[str, dict[str, float]] = {}
_SUPPRESS_S = 20.0
# Grace after a session stops before its episodes can be considered "orphaned":
# covers the window where one member has registered + re-reported the episode but
# the peer hasn't yet, so an unfinished recording isn't mistaken for a lost pair.
ORPHAN_GRACE_S = 30.0
# Orphaned-episode reconciliation is EVENT-DRIVEN: recomputed when a device
# registers (connects / re-reports its tasks) — the only moments orphans change —
# and cached here, so the operator UI just reads the cache instead of forcing a
# recompute on every poll. owner -> list of orphan-episode dicts.
ORPHANS_PENDING: dict[str, list[dict]] = {}


def _suppress_task_name(owner: str, name: str) -> None:
    TASK_SUPPRESSED.setdefault(owner, {})[name] = time.time() + _SUPPRESS_S


def _suppressed_names(owner: str) -> set[str]:
    now = time.time()
    supp = TASK_SUPPRESSED.get(owner)
    if not supp:
        return set()
    # Prune expired so a name can come back if it's ever legitimately re-created.
    for n in [n for n, exp in supp.items() if exp <= now]:
        del supp[n]
    return set(supp)


def _devices_reporting_task(owner: str, name: str) -> list["Device"]:
    """Online devices whose latest report includes a task with this name."""
    out = []
    for dev in _fleet_of(owner).values():
        if dev.online and any((t.get("name") == name) for t in (dev.tasks or [])):
            out.append(dev)
    return out

def _task_episode_items(t: dict):
    """Yield (episode_id, members) for a reported task, supporting the compact
    'groups' format (episodes grouped by shared membership) and the legacy
    per-episode 'episodes' format (older device firmware)."""
    if "groups" in t:
        for grp in t.get("groups", []):
            members = grp.get("members") or {}
            for eid in grp.get("episode_ids", []):
                if eid:
                    yield eid, members
    else:
        for ep in t.get("episodes", []):
            eid = ep.get("episode_id")
            if eid:
                yield eid, (ep.get("members") or {})


# Aggregation is memoized per owner: recomputing it (O(all reported episodes))
# on every dashboard poll doesn't scale. A cheap signature over the owner's
# devices (online status + a per-device report revision + suppressed names)
# tells us when anything that feeds the aggregation actually changed; otherwise
# we return the cached result. owner -> (signature, aggregation).
_REPORTED_CACHE: dict[str, tuple] = {}


def _reported_sig(owner: str):
    devs = _fleet_of(owner)
    return (tuple(sorted((d.device_id, d.online, d.report_rev) for d in devs.values())),
            tuple(sorted(_suppressed_names(owner))))


def _compute_reported_tasks(owner: str) -> dict[str, dict]:
    agg: dict[str, dict] = {}
    suppressed = _suppressed_names(owner)  # just-edited/deleted names, hidden briefly
    for dev in _fleet_of(owner).values():
        if not dev.online:
            continue
        for t in dev.tasks or []:
            name = t.get("name")
            if not name or name in suppressed:
                continue
            entry = agg.setdefault(name, {"description": "", "device_signature": [], "episodes": {}})
            if t.get("description") and not entry["description"]:
                entry["description"] = t["description"]
            if t.get("device_signature") and not entry["device_signature"]:
                entry["device_signature"] = list(t["device_signature"])
            for eid, ep_members in _task_episode_items(t):
                members = entry["episodes"].setdefault(eid, {})
                for role, who in ep_members.items():
                    members.setdefault(role, who)
    return agg


def _reported_tasks(owner: str) -> dict[str, dict]:
    """Merge online devices' reported tasks, keyed by task name (memoized). The
    SAME episode is reported by every member device (same episode_id + members),
    so we dedup by episode_id and union members. Returns {name: {"description",
    "device_signature", "episodes": {episode_id: {role: {device_id, name}}}}}.
    The result is cached and must be treated as READ-ONLY by callers."""
    sig = _reported_sig(owner)
    cached = _REPORTED_CACHE.get(owner)
    if cached is not None and cached[0] == sig:
        return cached[1]
    agg = _compute_reported_tasks(owner)
    _REPORTED_CACHE[owner] = (sig, agg)
    return agg


def _episode_started_at(episode_id: str) -> float:
    """Best-effort epoch seconds from an episode id (UTC 'YYYYmmdd_HHMMSS')."""
    try:
        return datetime.strptime(episode_id, "%Y%m%d_%H%M%S").replace(tzinfo=timezone.utc).timestamp()
    except ValueError:
        return 0.0


def _episodes_from_entry(entry: Optional[dict]) -> list[dict]:
    """Normalise a _reported_tasks entry's episodes into a sorted list carrying
    both members (role → {device_id, name}) and a plain roles map (role → id)."""
    if not entry:
        return []
    out = []
    for eid, members in entry["episodes"].items():
        out.append({
            "episode_id": eid,
            "members": members,
            "roles": {role: who.get("device_id") for role, who in members.items()},
            "started_at": _episode_started_at(eid),
        })
    out.sort(key=lambda e: e["episode_id"])
    return out


def _online_episode_reporters(owner: str) -> dict[str, dict]:
    """From CONNECTED devices, map episode_id → {role: {device_id, name}} built
    from each reporting device's slot. Lets the fleet reconstruct who recorded an
    episode saved before the device persisted its members — the reporting device
    IS a member, and its role is its slot. One pass over the reports."""
    out: dict[str, dict] = {}
    for dev in _fleet_of(owner).values():
        if not dev.online:
            continue
        role = _device_slot(dev)
        if not role:
            continue
        who = {"device_id": dev.device_id, "name": dev.name}
        for t in dev.tasks or []:
            for eid, _members in _task_episode_items(t):
                if eid:
                    out.setdefault(eid, {}).setdefault(role, who)
    return out


def _reconcile_tasks(owner: str) -> None:
    """Ensure a Task exists locally for every task reported by connected devices,
    so a freshly-connected operator's registry is populated from the devices (the
    source of truth). Purely in-memory: the fleet persists nothing, so this is
    rebuilt from the reports on every restart."""
    tasks = _tasks_of(owner)
    by_name = {t.name: t for t in tasks.values()}
    for name, entry in _reported_tasks(owner).items():
        t = by_name.get(name)
        if t is None:
            tid = uuid.uuid4().hex[:8]
            tasks[tid] = Task(id=tid, name=name, description=entry["description"],
                              device_signature=list(entry["device_signature"]))
        else:
            # Backfill from the authoritative device report where we're missing it.
            if not t.device_signature and entry["device_signature"]:
                t.device_signature = list(entry["device_signature"])
            if not t.description and entry["description"]:
                t.description = entry["description"]


def _orphan_episodes(owner: str) -> list[dict]:
    """Episodes that lost a pair — detected purely from the live reports, no
    tombstone. An episode is orphaned when one of its member devices is ONLINE
    but no longer reports it (it deleted the episode / task), while another online
    member still holds it. This is what surfaces after a multi-device task was
    deleted with a peer offline: on reconnect, the peer's copies are seen as
    orphaned (their now-online partner deleted them) and cleanup is proposed.

    Episodes whose missing member is OFFLINE are NOT flagged — we can't tell if
    that device still has them. Just-deleted (suppressed) task names are skipped
    so a normal both-online delete doesn't flash transient orphans mid-teardown.
    Returns one entry per orphaned episode."""
    fleet = _fleet_of(owner)
    suppressed = _suppressed_names(owner)
    online = {d.device_id: d for d in fleet.values() if d.online}
    # Episodes still "in flux" — being recorded now, or just stopped — must NOT be
    # flagged: one member registers + re-reports before the other, so for a few
    # seconds it exists on one device and not (yet) the peer, which looks orphaned
    # but is just an unfinished recording. Skip any episode in an open session, or
    # in a session stopped within the grace window (covers the re-report lag).
    now = time.time()
    in_flux: set[str] = set()
    for s in _sessions_of(owner).values():
        if s.status == "open" or (s.last_stop_at is not None and now - s.last_stop_at < ORPHAN_GRACE_S):
            for ep in s.episodes:
                eid = ep.get("episode_id")
                if eid:
                    in_flux.add(eid)
    dev_eps: dict[str, set[str]] = {}          # device_id -> episode ids it reports
    ep_info: dict[str, dict] = {}              # episode_id -> {members, task}
    for dev in online.values():
        eids: set[str] = set()
        for t in dev.tasks or []:
            name = t.get("name")
            if not name or name in suppressed:
                continue
            for eid, ep_members in _task_episode_items(t):
                eids.add(eid)
                ep_info.setdefault(eid, {"members": ep_members, "task": name})
        dev_eps[dev.device_id] = eids

    def _name(mid: str, fallback: str) -> str:
        return online[mid].name if mid in online else (fallback or mid)

    out = []
    for eid, info in ep_info.items():
        if eid in in_flux:
            continue  # being recorded / just stopped — not a real orphan
        member_ids = {w.get("device_id"): w.get("name") for w in info["members"].values() if w.get("device_id")}
        deleters = [(m, _name(m, n)) for m, n in member_ids.items()
                    if m in online and eid not in dev_eps.get(m, set())]
        holders = [(m, _name(m, n)) for m, n in member_ids.items()
                   if m in online and eid in dev_eps.get(m, set())]
        if deleters and holders:  # a peer deleted it, but someone online still has it
            out.append({
                "episode_id": eid, "task": info["task"], "started_at": _episode_started_at(eid),
                "holders": [{"device_id": m, "name": n} for m, n in holders],
                "deleted_by": [{"device_id": m, "name": n} for m, n in deleters],
            })
    return out


def _task_name(owner: str, task_id: str) -> str:
    t = _tasks_of(owner).get(task_id)
    return t.name if t else ""


@dataclass
class Group:
    id: str
    name: str
    left: str = ""  # device_id of the left-hand grabette in this group
    right: str = ""  # device_id of the right-hand grabette in this group
    casquette: str = ""  # device_id of the casquette in this group
    created_at: float = field(default_factory=time.time)
    # (No task here — the task is chosen when a session is launched, not stored
    # on the group. See Session.)


GROUPS: dict[str, dict[str, Group]] = {}  # owner -> group_id -> Group


def _groups_of(owner: str) -> dict[str, Group]:
    return GROUPS.setdefault(owner, {})


# --- sessions (fleet-only: the recording-run + upload manifest) --------------
# A session is a run of one or more episodes recorded together for ONE task.
# It carries its own role→device membership (a "group of one" is just a
# single-entry map — no Group object is materialised) and the per-episode
# manifest: which physical device held which role for each episode. That
# manifest is what lets the dataset builder regroup an episode's data even
# when the device filling a role changes between episodes.
@dataclass
class Session:
    id: str
    task_id: str
    members: dict[str, str]  # role -> device_id, captured at launch
    status: str = "open"     # open | closed
    # Optimistic recording indicator: True between an episode start and its
    # stop. Maintained on every start/stop path (operator + physical button).
    # Not a hardware truth (real per-device state is reconciled in phase 3) —
    # enough to drive the "● recording" indicator in the UI.
    recording: bool = False
    # Epoch time of the last episode stop dispatched for this session. Drives
    # the warm/cold lead decision: within OAK_WARM_WINDOW_S the OAK-D is still
    # powered (device keepalive), so the next episode can use the short lead.
    # None until the first episode stops → first episode uses the cold lead.
    last_stop_at: Optional[float] = None
    # "stopping" phase, reconciled against reality: set True when an episode stop
    # is dispatched, and cleared when every dispatched stop_capture command has
    # reported its result (i.e. the devices actually finished tearing down + mux)
    # — so the UI's "stopping" indicator ends when the devices really stop, not
    # after a fixed guess. pending_stop_cmds holds the command ids still awaited.
    stopping: bool = False
    pending_stop_cmds: set[str] = field(default_factory=set)
    # Each entry: {"episode_id": str, "roles": {role: device_id}, "started_at": float}
    episodes: list[dict[str, Any]] = field(default_factory=list)
    started_at: float = field(default_factory=time.time)


SESSIONS: dict[str, dict[str, Session]] = {}  # owner -> session_id -> Session


@dataclass
class DatasetJob:
    """A LeRobot dataset build. Transient runtime state (not persisted): the
    fleet dispatches per-device uploads, waits for them, then (step 3) triggers
    the processing Space and tracks it to completion."""
    id: str
    task_ids: list[str]
    roles: list[str]              # the shared device signature of the selected tasks
    raw_repo: str                 # {owner}/grabette-raw-<job> — deleted after processing
    target_repo: str              # {owner}/<name> — the resulting LeRobot dataset
    status: str = "uploading"     # uploading | raw_ready | processing | done | error
    message: str = ""
    # 0..1 fraction for the determinate upload phase; None once processing starts
    # (the Space conversion runs opaquely behind one blocking device call, so the
    # UI shows an indeterminate bar there). 1.0 when done.
    progress: Optional[float] = 0.0
    result_url: Optional[str] = None
    error: Optional[str] = None
    upload_cmds: dict[str, str] = field(default_factory=dict)  # device_id -> command id
    created_at: float = field(default_factory=time.time)


DATASET_JOBS: dict[str, dict[str, DatasetJob]] = {}  # owner -> job_id -> job


def _dataset_jobs_of(owner: str) -> dict[str, DatasetJob]:
    return DATASET_JOBS.setdefault(owner, {})


def _sessions_of(owner: str) -> dict[str, Session]:
    return SESSIONS.setdefault(owner, {})


def _open_session_for_device(owner: str, device_id: str) -> Optional[Session]:
    for s in _sessions_of(owner).values():
        if s.status == "open" and device_id in s.members.values():
            return s
    return None


# --- device activity: what a device is doing, for the operator UI + the recording
# gates. Prefer what the fleet can INFER on its own from the work it dispatched (an
# in-flight upload/process command in the device's queue); then the device's own
# self-reported status (which also covers dashboard-initiated work the fleet never
# sees); then capture inferred from open-session membership. Pure in-memory reads,
# no extra I/O, so it stays cheap at fleet scale.
_RECORDING_BLOCKERS = ("uploading", "processing")  # dataset work that must not overlap a recording


def _device_activity(owner: str, dev: Device) -> str:
    """One of: idle | capturing | uploading | processing."""
    # Dataset work the fleet ITSELF dispatched — authoritative, and the device may
    # not self-report it (relay commands don't create a local job). A command is
    # removed from the queue on result, so anything left here is live.
    types = {c.type for c in dev.queue}
    if "process_dataset" in types:
        return "processing"
    if "upload_episodes" in types:
        return "uploading"
    # Device self-reported activity — covers local dashboard work (SLAM push /
    # episode upload) and live capture that the fleet can't see on its own.
    st = (dev.reported_status or "").strip().lower()
    if st in ("capturing", "uploading", "processing"):
        return st
    # Capture inferred from open-session membership (fallback if not reported).
    if _open_session_for_device(owner, dev.device_id) is not None:
        return "capturing"
    return "idle"


def _busy_recording_blockers(owner: str, device_ids) -> list[str]:
    """Subset of device_ids currently tied up by dataset work (upload/convert),
    which must be free before they can (re)start a recording."""
    fleet = _fleet_of(owner)
    return [did for did in device_ids
            if (d := fleet.get(did)) is not None
            and _device_activity(owner, d) in _RECORDING_BLOCKERS]


def _remove_from_groups(owner: str, device_id: str) -> None:
    for g in _groups_of(owner).values():
        if g.left == device_id:
            g.left = ""
        if g.right == device_id:
            g.right = ""
        if g.casquette == device_id:
            g.casquette = ""


def _validate_task_signature(owner: str, task_id: str, slots: set[str]) -> None:
    """The devices launched for a task must fill exactly the task's required
    roles — exact match (not just superset) so the dataset builder knows
    precisely which roles produce data. No-op when the task has no signature."""
    t = _tasks_of(owner).get(task_id)
    if t is None:
        raise HTTPException(404, f"Task {task_id} not found")
    if t.device_signature and set(t.device_signature) != slots:
        raise HTTPException(400, {
            "message": "selected devices don't match the task's required devices",
            "task": t.name,
            "required": sorted(t.device_signature),
            "got": sorted(slots),
        })


# --- persistence: none -------------------------------------------------------
# The fleet keeps NO durable state of its own. Devices are the source of truth:
# each re-registers on connect and reports its tasks + per-episode membership
# (see TaskManager.report_tasks), which the fleet aggregates in memory (see
# _reported_tasks / _reconcile_tasks). This means ANY operator sees the existing
# tasks and can build a dataset from them, regardless of which HF account did the
# acquisition — and nothing is ever written to an HF namespace on the fleet's
# behalf. Groups and open sessions are runtime-only and reset on a Space restart.

# Strong refs to fire-and-forget background tasks (e.g. dataset jobs). asyncio
# only weakly references tasks, so without this the GC can cancel one mid-run.
_bg_tasks: set[asyncio.Task] = set()


def _enqueue(dev: "Device", cmd: Command) -> None:
    """Queue a command for a device AND wake any long-poll holding on it, so the
    command is delivered on the next network round-trip instead of the next poll
    interval. Centralized on purpose: a bare dev.queue.append() would silently
    strand the command for up to LONG_POLL_S until the hold times out."""
    dev.queue.append(cmd)
    dev.wakeup.set()


async def _operator_loaded(request: Request) -> str:
    """Resolve the authenticated operator (namespace). Kept as the single
    dependency every operator-facing handler uses; there is no snapshot to load
    anymore — the task/episode view is aggregated live from device reports."""
    owner, _token = operator_auth(request)
    return owner


# --- app + OAuth -------------------------------------------------------------
app = FastAPI(title="Grabette fleet")
attach_huggingface_oauth(app)  # adds /oauth/huggingface/{login,logout,callback}


@app.get("/healthz")
async def healthz() -> dict[str, str]:
    """Public, auth-free liveness ping. A device hits this to WAKE the (free-tier,
    sleep-when-idle) Space and confirm it's up before starting OAuth — so the
    OAuth callback, which HF routes through this Space, lands on a running relay
    even when no operator has a fleet dashboard tab open to keep it warm."""
    return {"status": "ok"}


# === device-facing API (Bearer auth) ========================================
class RegisterReq(BaseModel):
    device_id: str
    name: str = ""
    capabilities: list[str] = []
    hand: str = ""  # "left" or "right" (from GRABETTE_HAND on the device)
    ip: str = ""  # device LAN IPv4, best-effort
    tasks: list[dict] = []  # this device's recorded tasks (source of truth for aggregation)


class ResultReq(BaseModel):
    device_id: str
    command_id: str
    result: dict[str, Any] = {}


@app.post("/api/devices/register")
async def register(req: RegisterReq, auth: tuple[str, str] = Depends(device_auth)) -> dict[str, Any]:
    owner, token = auth
    _touch_namespaces(owner, token)  # refresh the owner's org list (token not kept)
    fleet = _fleet_of(owner)
    dev = fleet.get(req.device_id)
    if dev is None:
        dev = Device(req.device_id, req.name or req.device_id, req.capabilities,
                     hand=req.hand, ip=req.ip, tasks=req.tasks)
        fleet[req.device_id] = dev
    else:
        dev.name = req.name or dev.name
        dev.capabilities = req.capabilities or dev.capabilities
        dev.hand = req.hand or dev.hand
        dev.ip = req.ip or dev.ip
        # Reported every register: the device is authoritative, so replace wholesale.
        dev.tasks = req.tasks
    dev.report_rev += 1  # invalidates the aggregation cache for this owner
    dev.last_seen = time.time()
    # A device just (re)reported its tasks — the moment orphaned episodes can
    # appear/change. Recompute the reconciliation snapshot now (event-driven)
    # rather than on every operator poll.
    ORPHANS_PENDING[owner] = _orphan_episodes(owner)
    return {"status": "ok", "pending": len(dev.queue)}


@app.post("/api/devices/heartbeat")
async def heartbeat(device_id: str, battery: Optional[float] = None,
                    status: Optional[str] = None,
                    auth: tuple[str, str] = Depends(device_auth)) -> dict[str, str]:
    """Lightweight liveness ping, sent every DEVICE_HEARTBEAT_S independently of
    the (long-held) command poll — this is what keeps last_seen fresh and lets
    the fleet detect a disconnect within ONLINE_WINDOW. Deliberately does no HF
    I/O so it stays cheap at a few-second cadence. Also carries the device's
    battery % and its self-reported activity, so the fleet can show device state
    without polling."""
    owner, _token = auth
    dev = _fleet_of(owner).get(device_id)
    if dev is None:
        raise HTTPException(404, "Device not registered")
    dev.last_seen = time.time()
    if battery is not None:
        dev.battery = battery
    if status is not None:
        dev.reported_status = status
    return {"status": "ok"}


@app.get("/api/devices/poll")
async def poll(device_id: str, auth: tuple[str, str] = Depends(device_auth)) -> dict[str, Any]:
    owner, token = auth
    dev = _fleet_of(owner).get(device_id)
    if dev is None:
        raise HTTPException(404, "Device not registered")
    _touch_namespaces(owner, token)  # keeps the org list fresh (TTL-gated; token not kept)
    dev.last_seen = time.time()
    # Long-poll: hold the connection open until a command is enqueued (which
    # sets dev.wakeup) or LONG_POLL_S elapses, so a command is delivered on the
    # next round-trip instead of the next poll interval. Clear the event FIRST,
    # then re-check the queue: an enqueue landing in that gap sets the event, so
    # the wait() returns immediately rather than stranding the command. When
    # LONG_POLL_S<=0 this whole block is skipped → short-poll (return at once).
    if LONG_POLL_S > 0:
        dev.wakeup.clear()
        if not any(c.status == "pending" for c in dev.queue):
            try:
                await asyncio.wait_for(dev.wakeup.wait(), timeout=LONG_POLL_S)
            except asyncio.TimeoutError:
                pass
            # NB: do NOT refresh last_seen here — a held poll that times out on a
            # DEAD device would otherwise look like fresh contact and delay
            # offline detection. last_seen reflects the poll's ARRIVAL (above);
            # liveness is kept fresh by the separate lightweight heartbeat.
    pending = [c for c in dev.queue if c.status == "pending"]
    for c in pending:
        c.status = "sent"
    return {"commands": [{"id": c.id, "type": c.type, "args": c.args} for c in pending]}


@app.post("/api/devices/result")
async def result(req: ResultReq, auth: tuple[str, str] = Depends(device_auth)) -> dict[str, str]:
    owner, _token = auth
    dev = _fleet_of(owner).get(req.device_id)
    if dev is None:
        raise HTTPException(404, "Device not registered")
    for c in dev.queue:
        if c.id == req.command_id:
            c.status = "done"
            c.result = req.result
            c.done_at = time.time()
            dev.queue.remove(c)
            dev.history.insert(0, c)
            del dev.history[20:]
            if c.type == "logout" and dev.pending_delete:
                fleet = _fleet_of(owner)
                fleet.pop(dev.device_id, None)
                _remove_from_groups(owner, dev.device_id)
            break
    # A stop_capture result means that device actually finished tearing down its
    # capture. End the session's "stopping" phase once EVERY dispatched stop has
    # reported — so the UI leaves "stopping" exactly when the devices really stop.
    for s in _sessions_of(owner).values():
        if req.command_id in s.pending_stop_cmds:
            s.pending_stop_cmds.discard(req.command_id)
            if not s.pending_stop_cmds:
                s.stopping = False
            break
    return {"status": "ok"}


# === operator-facing API (session OAuth) =====================================
class DispatchReq(BaseModel):
    device_id: str
    type: str
    args: dict[str, Any] = {}


@app.get("/api/fleet/me")
async def me(request: Request) -> dict[str, Any]:
    info = parse_huggingface_oauth(request)
    if info is None:
        return {"logged_in": False}
    return {"logged_in": True, "username": info.user_info.preferred_username or info.user_info.name}


@app.get("/api/fleet/namespaces")
async def namespaces(request: Request) -> dict[str, Any]:
    """The namespaces the operator can push a dataset to: their own username +
    the orgs they belong to. Used to populate the dataset owner dropdown.

    Served from the cache built when a device token passed through (device tokens
    list orgs reliably; see _touch_namespaces) — so the fleet keeps no raw token.
    Cold start (no device seen yet for this owner): best-effort via the operator's
    OAuth token, which lists orgs poorly, falling back to just the username."""
    owner, oauth_token = operator_auth(request)
    ent = _namespaces_cache.get(owner)
    if ent and ent[1]:
        return {"namespaces": ent[1], "default": owner}
    names = [owner]
    if oauth_token:
        try:
            info = await asyncio.to_thread(whoami, oauth_token)
            names += [o["name"] for o in (info.get("orgs") or []) if o.get("name")]
        except Exception:
            logger.warning("whoami for namespaces failed for %s", owner, exc_info=True)
    seen: set[str] = set()
    uniq = [n for n in names if not (n in seen or seen.add(n))]
    return {"namespaces": uniq, "default": owner}


@app.get("/api/fleet/devices")
async def list_devices(request: Request) -> dict[str, Any]:
    owner = operator_name(request)
    devices = list(_fleet_of(owner).values())
    now = time.time()
    for d in devices:
        # Only drop stale commands for a LONG-gone device — never on the mere
        # 15s online window (would nuke an in-flight group start/stop).
        if now - d.last_seen > STALE_QUEUE_S:
            d.queue.clear()
    return {
        "owner": owner,
        "devices": [
            {
                "device_id": d.device_id,
                "name": d.name,
                "online": d.online,
                "capabilities": d.capabilities,
                "hand": d.hand,
                "ip": d.ip,
                "battery": d.battery,
                "activity": _device_activity(owner, d),
                "pending": len(d.queue),
                "history": [
                    {"type": c.type, "args": c.args, "result": c.result, "ts": c.done_at or c.created_at}
                    for c in d.history[:5]
                ],
            }
            for d in devices
        ],
    }


@app.delete("/api/fleet/devices/{device_id}")
async def remove_device(device_id: str, request: Request) -> dict[str, str]:
    owner = await _operator_loaded(request)
    fleet = _fleet_of(owner)
    if device_id not in fleet:
        raise HTTPException(404, "Device not found")
    del fleet[device_id]
    _remove_from_groups(owner, device_id)
    return {"status": "ok"}


@app.post("/api/fleet/devices/{device_id}/remove")
async def remove_device_graceful(device_id: str, request: Request) -> dict[str, str]:
    """Remove a device: dispatch logout if online (then delete on result), delete immediately if offline."""
    owner = await _operator_loaded(request)
    fleet = _fleet_of(owner)
    dev = fleet.get(device_id)
    if dev is None:
        raise HTTPException(404, "Device not found")
    if dev.online:
        dev.pending_delete = True
        cmd = Command(id=uuid.uuid4().hex[:12], type="logout", args={})
        _enqueue(dev, cmd)
    else:
        fleet.pop(device_id, None)
        _remove_from_groups(owner, device_id)
    return {"status": "ok"}


# === tasks (operator-facing, session OAuth) ==================================
class TaskReq(BaseModel):
    name: str = ""
    description: str = ""
    device_signature: list[str] = []


def _task_dict(t: Task) -> dict[str, Any]:
    return {"id": t.id, "name": t.name, "description": t.description,
            "device_signature": t.device_signature}


@app.get("/api/fleet/tasks")
async def list_tasks(request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    _reconcile_tasks(owner)  # surface tasks reported by connected devices
    agg = _reported_tasks(owner)
    reporters = _online_episode_reporters(owner)  # for repairing incomplete episodes
    out = []
    for t in _tasks_of(owner).values():
        d = _task_dict(t)
        eps = _episodes_from_entry(agg.get(t.name))
        sig = set(t.device_signature or [])
        for ep in eps:
            # Roles we actually know a device for. Incomplete = the task's expected
            # roles aren't all covered (or none is), e.g. episodes recorded before
            # the device stored its members. Fillable = a connected device that
            # reports this episode could supply a role we're missing.
            present = {r for r, w in ep["members"].items() if w.get("device_id")}
            incomplete = (bool(sig) and not sig.issubset(present)) or not present
            recon = reporters.get(ep["episode_id"], {})
            ep["incomplete"] = incomplete
            ep["fillable"] = incomplete and any(r not in present for r in recon)
        d["episodes"] = eps
        d["episode_count"] = len(eps)
        out.append(d)
    return {"tasks": out}


@app.post("/api/fleet/tasks/{task_id}/fill-devices")
async def fill_task_devices(task_id: str, request: Request) -> dict[str, Any]:
    """Repair ALL of a task's incomplete episodes at once (recorded before the
    device persisted their members): reconstruct role → device from the CONNECTED
    devices that still report each episode, then persist onto them so the fleet's
    view — and the devices themselves — become the durable source of truth.
    One batched command per device (not per episode). Partial by design: fills
    only the roles whose devices are online now, so a re-run completes the rest."""
    owner = await _operator_loaded(request)
    task = _tasks_of(owner).get(task_id)
    if task is None:
        raise HTTPException(404, "Task not found")
    eps = _episodes_from_entry(_reported_tasks(owner).get(task.name))
    sig = set(task.device_signature or [])
    # One pass: episode_id → online devices that still report it.
    ep_devs: dict[str, list[Device]] = {}
    for dev in _fleet_of(owner).values():
        if not dev.online:
            continue
        seen: set[str] = set()
        for t in dev.tasks or []:
            for eid, _m in _task_episode_items(t):
                if eid and eid not in seen:
                    seen.add(eid)
                    ep_devs.setdefault(eid, []).append(dev)
    # Build one batch of (episode_id, reconstructed members) per target device.
    batches: dict[str, dict] = {}  # device_id → {"dev": Device, "entries": [...]}
    filled: set[str] = set()
    for ep in eps:
        eid = ep["episode_id"]
        present = {r for r, w in ep["members"].items() if w.get("device_id")}
        incomplete = (bool(sig) and not sig.issubset(present)) or not present
        if not incomplete:
            continue
        devs = ep_devs.get(eid, [])
        recon: dict[str, dict] = {}
        for dev in devs:
            role = _device_slot(dev)
            if role:
                recon.setdefault(role, {"device_id": dev.device_id, "name": dev.name})
        if not any(r not in present for r in recon):  # nothing new we can add now
            continue
        filled.add(eid)
        for dev in devs:
            batches.setdefault(dev.device_id, {"dev": dev, "entries": []})["entries"].append(
                {"episode_id": eid, "members": recon})
    if not batches:
        raise HTTPException(409, {"message": "no incomplete episode can be repaired right now — "
                                             "connect the grabettes that recorded them, then retry"})
    for b in batches.values():
        cmd = Command(id=uuid.uuid4().hex[:12], type="set_episode_members",
                      args={"task_name": task.name, "episodes": b["entries"],
                            "device_signature": list(task.device_signature)})
        _enqueue(b["dev"], cmd)
    return {"status": "ok", "episodes": len(filled),
            "devices": [b["dev"].device_id for b in batches.values()]}


def _clean_signature(sig: list[str]) -> list[str]:
    bad = [s for s in sig if s not in VALID_SLOTS]
    if bad:
        raise HTTPException(400, f"Invalid device_signature entries: {bad}; allowed: {list(VALID_SLOTS)}")
    # de-dup, canonical order
    cleaned = [s for s in VALID_SLOTS if s in sig]
    if not cleaned:
        raise HTTPException(400, "A task must require at least one device")
    return cleaned


@app.post("/api/fleet/tasks")
async def create_task(req: TaskReq, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    if not req.name.strip():
        raise HTTPException(400, "Task needs a name")
    tasks = _tasks_of(owner)
    if any(t.name == req.name.strip() for t in tasks.values()):
        raise HTTPException(400, "A task with this name already exists")
    tid = uuid.uuid4().hex[:8]
    tasks[tid] = Task(id=tid, name=req.name.strip(), description=req.description.strip(),
                      device_signature=_clean_signature(req.device_signature))
    return {"status": "ok", "id": tid, "task": _task_dict(tasks[tid])}


@app.put("/api/fleet/tasks/{task_id}")
async def update_task(task_id: str, req: TaskReq, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    _reconcile_tasks(owner)
    t = _tasks_of(owner).get(task_id)
    if t is None:
        raise HTTPException(404, "Task not found")
    old_name = t.name
    new_name = req.name.strip() or old_name
    new_desc = req.description.strip()
    new_sig = _clean_signature(req.device_signature)

    # The device signature is locked ONCE it's known and episodes exist — the
    # recordings were made with that device set, so changing it would misdescribe
    # them. But an old task with episodes and NO recorded signature can be set
    # once (to backfill it); after that it's known → locked.
    has_episodes = bool(_episodes_from_entry(_reported_tasks(owner).get(old_name)))
    sig_locked = has_episodes and bool(t.device_signature)
    if sig_locked and sorted(new_sig) != sorted(t.device_signature):
        raise HTTPException(409, {"message": "cannot change the required devices of a task "
                                             "that already has recorded episodes"})
    if new_name != old_name and any(o.name == new_name for o in _tasks_of(owner).values()):
        raise HTTPException(400, "A task with this name already exists")

    sig_changed = (not sig_locked) and sorted(new_sig) != sorted(t.device_signature)

    # Apply on the fleet, then propagate to every device that recorded this task
    # (the source of truth) so the change survives the next report reconcile.
    t.name = new_name
    t.description = new_desc
    if not sig_locked:
        t.device_signature = new_sig
    for dev in _devices_reporting_task(owner, old_name):
        args = {"name": old_name,
                "new_name": new_name if new_name != old_name else None,
                "description": new_desc}
        if sig_changed:  # backfilling an old task's signature → devices must store it
            args["device_signature"] = new_sig
        _enqueue(dev, Command(id=uuid.uuid4().hex[:12], type="edit_task", args=args))
    if new_name != old_name:
        _suppress_task_name(owner, old_name)  # hide stale old-name reports until re-report
    return {"status": "ok", "task": _task_dict(t)}


@app.delete("/api/fleet/tasks/{task_id}")
async def delete_task(task_id: str, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    _reconcile_tasks(owner)
    tasks = _tasks_of(owner)
    t = tasks.get(task_id)
    if t is None:
        raise HTTPException(404, "Task not found")
    if any(s.status == "open" and s.task_id == task_id for s in _sessions_of(owner).values()):
        raise HTTPException(409, "Cannot delete a task with an open session")
    name = t.name
    # Delete on every ONLINE device that has this task — the task AND its recorded
    # episodes go (the device is the source of truth; removing here alone would let
    # the next report resurrect it). Offline members keep their copy for now; they
    # are reconciled when they reconnect (see orphan detection, phase 3).
    dispatched = []
    for dev in _devices_reporting_task(owner, name):
        _enqueue(dev, Command(id=uuid.uuid4().hex[:12], type="delete_task", args={"name": name}))
        dispatched.append(dev.device_id)
    del tasks[task_id]
    _suppress_task_name(owner, name)  # hide it until the devices re-report without it
    return {"status": "ok", "dispatched": dispatched}


class OrphanCleanupReq(BaseModel):
    episode_ids: list[str] = []


@app.get("/api/fleet/orphans")
async def list_orphans(request: Request) -> dict[str, Any]:
    """Orphaned episodes (a peer deleted them but a connected device still holds
    them), grouped by task for the reconciliation banner. Returns the cached
    snapshot computed on the last device register — not a per-poll recompute."""
    owner = await _operator_loaded(request)
    reported = _reported_tasks(owner)
    groups: dict[str, dict] = {}
    for o in ORPHANS_PENDING.get(owner, []):
        g = groups.setdefault(o["task"], {"task": o["task"], "episode_ids": [],
                                          "holders": {}, "deleted_by": {}})
        g["episode_ids"].append(o["episode_id"])
        for h in o["holders"]:
            g["holders"][h["device_id"]] = h["name"]
        for d in o["deleted_by"]:
            g["deleted_by"][d["device_id"]] = d["name"]
    out = []
    for g in groups.values():
        # If EVERY remaining episode of the task is orphaned, the whole task was
        # deleted on the peer(s) — say so, rather than just "episodes lost pairs".
        entry = reported.get(g["task"])
        total = len(entry["episodes"]) if entry else len(g["episode_ids"])
        out.append({"task": g["task"], "episode_ids": g["episode_ids"],
                    "count": len(g["episode_ids"]),
                    "whole_task": len(g["episode_ids"]) >= total,
                    "holders": list(g["holders"].values()),
                    "deleted_by": list(g["deleted_by"].values())})
    return {"groups": out}


@app.post("/api/fleet/orphans/cleanup")
async def cleanup_orphans(req: OrphanCleanupReq, request: Request) -> dict[str, Any]:
    """Delete the given orphaned episodes on the devices still holding them.
    Recomputes orphans so only genuinely-orphaned episodes are touched."""
    owner = await _operator_loaded(request)
    # Recompute live here (accuracy matters for a destructive action).
    orphans = {o["episode_id"]: o for o in _orphan_episodes(owner)}
    fleet = _fleet_of(owner)
    dispatched = 0
    cleaned = set()
    for eid in req.episode_ids:
        o = orphans.get(eid)
        if not o:
            continue
        for h in o["holders"]:
            dev = fleet.get(h["device_id"])
            if dev is not None and dev.online:
                _enqueue(dev, Command(id=uuid.uuid4().hex[:12], type="delete_episode",
                                      args={"episode_id": eid}))
                dispatched += 1
                cleaned.add(eid)
    # Drop the cleaned episodes from the cached snapshot so the banner clears at
    # once; the holders' re-register will confirm on the next report.
    if cleaned and owner in ORPHANS_PENDING:
        ORPHANS_PENDING[owner] = [o for o in ORPHANS_PENDING[owner] if o["episode_id"] not in cleaned]
    return {"status": "ok", "dispatched": dispatched}


# === device groups (operator-facing, session OAuth) ==========================
class GroupReq(BaseModel):
    name: str = ""
    left: str = ""
    right: str = ""
    casquette: str = ""


def _validate_group_members(owner: str, req: GroupReq, group_id: Optional[str] = None) -> None:
    fleet = _fleet_of(owner)
    groups = _groups_of(owner)
    slots = {"left": req.left, "right": req.right, "casquette": req.casquette}
    if not any(slots.values()):
        raise HTTPException(400, "Group needs at least one device")
    for slot, device_id in slots.items():
        if not device_id:
            continue
        dev = fleet.get(device_id)
        if dev is None:
            raise HTTPException(404, f"Device {device_id} not found")
        if slot in ("left", "right"):
            if kind_of(dev) != "grabette" or dev.hand != slot:
                raise HTTPException(400, f"Device {device_id} is not a {slot}-hand grabette")
        elif kind_of(dev) != "casquette":
            raise HTTPException(400, f"Device {device_id} is not a casquette")
        for gid, g in groups.items():
            if gid == group_id:
                continue
            if device_id in (g.left, g.right, g.casquette):
                raise HTTPException(400, f"Device {device_id} is already in another group")


@app.get("/api/fleet/groups")
async def list_groups(request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    groups = list(_groups_of(owner).values())
    return {
        "groups": [
            {"id": g.id, "name": g.name, "left": g.left, "right": g.right, "casquette": g.casquette}
            for g in groups
        ]
    }


@app.post("/api/fleet/groups")
async def create_group(req: GroupReq, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    _validate_group_members(owner, req)
    groups = _groups_of(owner)
    gid = uuid.uuid4().hex[:8]
    groups[gid] = Group(
        id=gid, name=req.name.strip() or f"Group {len(groups) + 1}", left=req.left, right=req.right, casquette=req.casquette,
    )
    return {"status": "ok", "id": gid}


@app.put("/api/fleet/groups/{group_id}")
async def update_group(group_id: str, req: GroupReq, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    groups = _groups_of(owner)
    g = groups.get(group_id)
    if g is None:
        raise HTTPException(404, "Group not found")
    _validate_group_members(owner, req, group_id=group_id)
    g.name = req.name.strip() or g.name
    g.left, g.right, g.casquette = req.left, req.right, req.casquette
    return {"status": "ok"}


@app.delete("/api/fleet/groups/{group_id}")
async def delete_group(group_id: str, request: Request) -> dict[str, str]:
    owner = await _operator_loaded(request)
    groups = _groups_of(owner)
    if group_id not in groups:
        raise HTTPException(404, "Group not found")
    del groups[group_id]
    return {"status": "ok"}


def _group_members(g: Group) -> list[str]:
    return [d for d in (g.left, g.right, g.casquette) if d]


def _lead_for(session: Session) -> float:
    """Pick the start lead for the next episode: short if the OAK-D is still
    warm (a recent episode stopped within OAK_WARM_WINDOW_S), long otherwise
    (first episode of the session, or the OAK-D has since gone to sleep) so the
    cold boot finishes before T0. Conservative: unknown/old → cold."""
    if session.last_stop_at is not None and (time.time() - session.last_stop_at) < OAK_WARM_WINDOW_S:
        return GROUP_START_LEAD_WARM_S
    return GROUP_START_LEAD_COLD_S


def _named_members(owner: str, members: dict[str, str]) -> dict[str, dict[str, str]]:
    """role → {device_id, name}. Sent to devices so each records who its peers
    were (by stable device_id + display name), letting a device name an offline
    peer later — the device is the durable source of truth for episode membership."""
    fleet = _fleet_of(owner)
    return {
        r: {"device_id": d, "name": (fleet[d].name if d in fleet else d)}
        for r, d in members.items()
    }


def _schedule_episode_start(
    owner: str, task_name: str, members: dict[str, str], lead_s: float,
    exclude_device_id: Optional[str] = None, signature: Optional[list[str]] = None,
) -> str:
    """Enqueue a synchronized start_capture to every member device (except
    exclude_device_id, which self-schedules from the returned T0). Raises 409
    if any member is offline. Returns the shared start_at_utc (ISO).

    Each command carries the episode's full membership (role → device_id+name)
    and the task's device signature so every device persists who recorded with
    it — the fleet no longer needs its own HF-backed record of this."""
    fleet = _fleet_of(owner)
    device_ids = list(members.values())
    offline = [d for d in device_ids if d not in fleet or not fleet[d].online]
    if offline:
        raise HTTPException(409, {"message": "one or more devices are offline", "offline": offline})

    named = _named_members(owner, members)
    target_iso = (datetime.now(timezone.utc) + timedelta(seconds=lead_s)).isoformat()
    for device_id in device_ids:
        if device_id == exclude_device_id:
            continue
        cmd = Command(
            id=uuid.uuid4().hex[:12], type="start_capture",
            args={"task_name": task_name, "start_at_utc": target_iso,
                  "members": named, "signature": signature or []},
        )
        _enqueue(fleet[device_id], cmd)
    return target_iso


def _dispatch_episode_stop(owner: str, members: dict[str, str], exclude_device_id: Optional[str] = None) -> list[str]:
    """Fan out an IMMEDIATE stop to every member device except exclude_device_id
    (the acting device, which has already stopped locally). Each peer stops as
    soon as it receives the command, i.e. within ~1 poll interval — the pressed
    device is instant and peers trail only by the short delivery latency, no
    scheduled lead. Chaining fast stays safe because the next episode's start
    still uses a lead (GROUP_START_LEAD_*) that the peer's stop+mux fits inside.

    Returns the dispatched stop command ids so the caller can wait for their
    results (see Session.pending_stop_cmds) and end the "stopping" phase exactly
    when the devices report done, rather than after a fixed guess.

    (A synchronized, lead-based stop — shared future T_stop via GROUP_STOP_LEAD_S
    + the device-side CaptureScheduler.schedule_stop path — remains wired up but
    unused here; it'd matter only if delivery latency grew. With long-polling
    delivery drops to a network round-trip, making even this spread negligible.)"""
    fleet = _fleet_of(owner)
    ids: list[str] = []
    for device_id in members.values():
        if device_id == exclude_device_id:
            continue
        dev = fleet.get(device_id)
        if dev is not None:
            cmd = Command(id=uuid.uuid4().hex[:12], type="stop_capture", args={})
            _enqueue(dev, cmd)
            ids.append(cmd.id)
    return ids


def _begin_episode_stop(s: "Session", stop_cmd_ids: list[str]) -> None:
    """Mark a session as no longer recording and enter the 'stopping' phase,
    awaiting the dispatched stops' results (see Session.stopping). Only claims
    'stopping' when there are commands to wait on — a stop with no tracked device
    (e.g. a solo device that stopped locally) goes straight to idle."""
    s.recording = False
    s.last_stop_at = time.time()  # also drives the warm/cold lead for the next episode
    s.pending_stop_cmds = set(stop_cmd_ids)
    s.stopping = bool(stop_cmd_ids)


def _episode_id_for_target(target_iso: str) -> str:
    """Mirror the device's episode_id_for(T0): same UTC second → same id, so
    the manifest entry matches the folder every device actually creates."""
    return datetime.fromisoformat(target_iso).astimezone(timezone.utc).strftime("%Y%m%d_%H%M%S")


# === sessions (operator-facing, session OAuth) ===============================
# The recording unit. Launch = pick a task + a target (a group, or a single
# device treated as a group of one), validate the target's roles match the
# task's required devices, then record one or more synchronized episodes into
# the session's manifest, and close it.
class SessionLaunchReq(BaseModel):
    task_id: str
    # A device id per role, picked from the fleet at launch time. The device
    # selection IS the grouping — no Group object is created (the session owns
    # its members map).
    left: str = ""
    right: str = ""
    casquette: str = ""


def _resolve_members(owner: str, req: SessionLaunchReq) -> dict[str, str]:
    """Build the session's role→device map from the per-role device ids,
    checking each device actually fills the role it was placed in."""
    fleet = _fleet_of(owner)
    members: dict[str, str] = {}
    for role, device_id in (("left", req.left), ("right", req.right), ("casquette", req.casquette)):
        if not device_id:
            continue
        dev = fleet.get(device_id)
        if dev is None:
            raise HTTPException(404, f"Device {device_id} not found")
        if _device_slot(dev) != role:
            raise HTTPException(400, f"Device {device_id} cannot fill the {role} role")
        if not dev.online:
            # A session can't be launched with an offline device — it couldn't
            # receive the synchronized start and would fail at the first episode.
            raise HTTPException(409, {"message": f"{dev.name} is offline", "offline": [device_id]})
        members[role] = device_id
    if not members:
        raise HTTPException(400, "Select at least one device")
    return members


# Safety cap: if a device never reports its stop result (e.g. it dropped offline
# mid-stop), don't leave the "stopping" phase stuck — expose it as ended after this.
STOP_PHASE_MAX_S = 30.0


def _session_dict(owner: str, s: Session) -> dict[str, Any]:
    fleet = _fleet_of(owner)
    def dname(d):
        dev = fleet.get(d)
        return dev.name if dev else d
    # "stopping" ends when every dispatched stop reported (s.stopping cleared);
    # the time cap is only a fallback for a device that never reports.
    stopping = s.stopping and s.last_stop_at is not None and (time.time() - s.last_stop_at) < STOP_PHASE_MAX_S
    return {
        "id": s.id, "status": s.status, "task_id": s.task_id,
        "task_name": _task_name(owner, s.task_id),
        "recording": s.recording,
        "stopping": stopping,  # accurate "stopping" phase — ends when the devices really stop
        "started_at": s.started_at,
        "members": {r: {"device_id": d, "name": dname(d)} for r, d in s.members.items()},
        "episode_count": len(s.episodes),
        "episodes": s.episodes,
    }


@app.get("/api/fleet/sessions")
async def list_sessions(request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    return {"sessions": [_session_dict(owner, s) for s in _sessions_of(owner).values()]}


@app.post("/api/fleet/sessions")
async def launch_session(req: SessionLaunchReq, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    task = _tasks_of(owner).get(req.task_id)
    if task is None:
        raise HTTPException(404, "Task not found")
    members = _resolve_members(owner, req)
    # The target must provide EXACTLY the roles the task requires.
    _validate_task_signature(owner, req.task_id, set(members.keys()))
    # A single open session at a time (fleet-wide). Launching is done from the
    # task list; the running one must be closed before another can start.
    if any(s.status == "open" for s in _sessions_of(owner).values()):
        raise HTTPException(409, {"message": "a session is already running; close it first"})
    # A device tied up by a dataset upload/conversion can't also record.
    busy = _busy_recording_blockers(owner, members.values())
    if busy:
        fleet = _fleet_of(owner)
        names = ", ".join(fleet[d].name for d in busy if d in fleet) or "a device"
        raise HTTPException(409, {"message": f"{names} is busy processing a dataset — wait for it to finish",
                                  "devices": busy})
    sid = uuid.uuid4().hex[:8]
    _sessions_of(owner)[sid] = Session(id=sid, task_id=req.task_id, members=members)
    # No warm-up on launch: keeping the OAK-D on until the first episode would
    # drain the battery if the operator waits. Instead the first episode uses
    # the COLD lead (long enough for the scheduler to warm the OAK-D during the
    # lead), and after that the device keepalive keeps it warm between episodes.
    return {"status": "ok", "id": sid, "session": _session_dict(owner, _sessions_of(owner)[sid])}


def _get_open_session(owner: str, session_id: str) -> Session:
    s = _sessions_of(owner).get(session_id)
    if s is None:
        raise HTTPException(404, "Session not found")
    if s.status != "open":
        raise HTTPException(409, "Session is closed")
    return s


@app.post("/api/fleet/sessions/{session_id}/episode/start")
async def session_episode_start(session_id: str, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    s = _get_open_session(owner, session_id)
    # A member pulled into a dataset upload/conversion can't start a new episode.
    busy = _busy_recording_blockers(owner, s.members.values())
    if busy:
        fleet = _fleet_of(owner)
        names = ", ".join(fleet[d].name for d in busy if d in fleet) or "a device"
        raise HTTPException(409, {"message": f"{names} is busy processing a dataset — wait for it to finish",
                                  "devices": busy})
    _task = _tasks_of(owner).get(s.task_id)
    target_iso = _schedule_episode_start(
        owner, _task_name(owner, s.task_id), s.members, _lead_for(s),
        signature=_task.device_signature if _task else [],
    )
    episode_id = _episode_id_for_target(target_iso)
    # Record the manifest entry up front from the deterministic id; per-device
    # success is reconciled later via /api/devices/result (phase 3). start_at_utc
    # is the shared T0 — the UI shows "initializing" until then, "recording" after.
    s.episodes.append({"episode_id": episode_id, "roles": dict(s.members),
                       "started_at": datetime.now(timezone.utc).isoformat(),
                       "start_at_utc": target_iso})
    s.recording = True
    return {"status": "scheduled", "scheduled_start_utc": target_iso, "episode_id": episode_id,
            "devices": list(s.members.values())}


@app.post("/api/fleet/sessions/{session_id}/episode/stop")
async def session_episode_stop(session_id: str, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    s = _get_open_session(owner, session_id)
    ids = _dispatch_episode_stop(owner, s.members)
    _begin_episode_stop(s, ids)
    return {"status": "ok", "devices": list(s.members.values())}


@app.post("/api/fleet/sessions/{session_id}/episode/delete-last")
async def session_delete_last_episode(session_id: str, request: Request) -> dict[str, Any]:
    """Drop the session's most recent episode from the manifest and tell each
    device that recorded it to delete its local files. Refused while recording
    (the last manifest entry is then the in-progress episode)."""
    owner = await _operator_loaded(request)
    s = _get_open_session(owner, session_id)
    if s.recording:
        raise HTTPException(409, {"message": "stop the current episode before deleting"})
    if not s.episodes:
        raise HTTPException(409, {"message": "no episode to delete"})
    ep = s.episodes.pop()  # most recent
    episode_id = ep["episode_id"]
    fleet = _fleet_of(owner)
    devices = list((ep.get("roles") or {}).values())
    for dev_id in devices:
        dev = fleet.get(dev_id)
        if dev is not None:
            _enqueue(dev, Command(id=uuid.uuid4().hex[:12], type="delete_episode",
                                  args={"episode_id": episode_id}))
    return {"status": "ok", "episode_id": episode_id, "devices": devices}


@app.post("/api/fleet/sessions/{session_id}/close")
async def close_session(session_id: str, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    s = _sessions_of(owner).get(session_id)
    if s is None:
        raise HTTPException(404, "Session not found")
    # Best-effort: make sure nothing is left recording on the members.
    _dispatch_episode_stop(owner, s.members)
    if not s.episodes:
        # A session that recorded nothing isn't worth keeping — drop it so it
        # never clutters the task's history.
        del _sessions_of(owner)[session_id]
        return {"status": "ok", "discarded": True}
    s.recording = False
    s.last_stop_at = time.time()
    s.status = "closed"
    return {"status": "ok"}


@app.delete("/api/fleet/sessions/{session_id}")
async def delete_session(session_id: str, request: Request) -> dict[str, str]:
    owner = await _operator_loaded(request)
    if session_id not in _sessions_of(owner):
        raise HTTPException(404, "Session not found")
    del _sessions_of(owner)[session_id]
    return {"status": "ok"}


# === LeRobot dataset generation (operator OAuth) ============================
# Selected tasks (all sharing one device signature) → each involved device
# pushes its own streams to a shared raw dataset (by role) → a processing Space
# converts raw → LeRobot. The fleet only orchestrates; devices upload with their
# own write tokens (the fleet can't reach them, and needs no token for uploads).
class DatasetReq(BaseModel):
    task_ids: list[str]
    name: str = ""  # target dataset name; bare name → {owner}/{name}
    private: bool = False  # create the resulting LeRobot dataset as private
    # If True, include only episodes whose EVERY role-device is currently online
    # (skip episodes that reference an offline device) instead of requiring the
    # whole set of devices to be up.
    only_available: bool = False
    # "Use only" advanced option: a subset of roles (e.g. ["left"] or
    # ["left","casquette"]). When set, ANY task whose signature is a superset is
    # eligible, and only these roles' devices are uploaded → build a dataset from
    # just this sub-combination. Empty → use each selected task's full signature.
    roles: list[str] = []


def _command_status(dev: Device, cmd_id: str) -> Optional[Command]:
    """Find a dispatched command by id, whether still queued or completed
    (results move a command from queue to history)."""
    for c in list(dev.queue) + list(dev.history):
        if c.id == cmd_id:
            return c
    return None


def _resolve_dataset_plan(owner: str, task_ids: list[str], only_available: bool = False,
                          roles_override: Optional[list[str]] = None) -> tuple[list[str], dict[str, dict]]:
    """Validate the selection and build the per-device upload plan from the
    device reports. Returns (roles, plan) where plan maps device_id ->
    {"role": role, "episode_ids": set[str]}.

    roles_override ("Use only"): a subset of roles. When given, every selected
    task must merely CONTAIN these roles (superset), and only these roles' devices
    are uploaded — so you can build, say, a left-only dataset from bimanual tasks.
    When absent, the selected tasks must share one signature and all its roles are
    used.

    only_available: skip episodes that reference a device which is currently
    offline — the resulting dataset covers only what can be fully uploaded now."""
    _reconcile_tasks(owner)
    tasks = _tasks_of(owner)
    sel = [tasks[t] for t in task_ids if t in tasks]
    if not sel:
        raise HTTPException(404, "no matching tasks")
    override = [r for r in (roles_override or []) if r in VALID_SLOTS]
    if override:
        oset = set(override)
        bad = [t.name for t in sel if not oset.issubset(set(t.device_signature))]
        if bad:
            raise HTTPException(409, {"message": "some selected tasks don't have the chosen devices",
                                      "tasks": bad})
        roles = sorted(oset)
    else:
        sigs = {tuple(sorted(t.device_signature)) for t in sel}
        if len(sigs) != 1:
            raise HTTPException(409, {"message": "selected tasks must share the same device signature"})
        roles = sorted(sel[0].device_signature)
    agg = _reported_tasks(owner)
    fleet = _fleet_of(owner)

    def _online(d):
        return d in fleet and fleet[d].online

    # Episodes come from the devices' reports (the source of truth), matched by
    # task name. members' device_id per role tells us which device to ask to
    # upload that role's data for each episode. With an override, restrict to
    # just those roles so only the requested devices upload.
    role_set = set(roles)
    plan: dict[str, dict] = {}
    for t in sel:
        for ep in _episodes_from_entry(agg.get(t.name)):
            roles_map = {r: d for r, d in ep["roles"].items() if r in role_set}
            if not roles_map:
                continue  # this episode lacks the requested roles
            if only_available and not all(_online(d) for d in roles_map.values()):
                continue  # a needed device is offline → episode unusable
            for role, dev_id in roles_map.items():
                if not dev_id:
                    continue
                p = plan.setdefault(dev_id, {"role": role, "episode_ids": set()})
                p["episode_ids"].add(ep["episode_id"])
    if not plan:
        msg = ("no episodes have all their devices online"
               if only_available else "selected tasks have no recorded episodes")
        raise HTTPException(409, {"message": msg})
    return roles, plan


async def _run_dataset_job(owner: str, job: DatasetJob, private: bool,
                           processor_device_id: str) -> None:
    """Background: wait for every device's upload to complete, then have ONE
    device trigger the processing Space (raw → LeRobot, mono/bimanual per the
    device set) with ITS OWN long-lived token and report the result back.

    The fleet never handles an HF token for the Space call: the device→Space
    channel is the same one the SLAM flow already uses, so no token is cached on
    the fleet nor forwarded through it (avoids widening the token's blast radius
    and the short-lived-OAuth-token expiry problem). The Space downloads the raw,
    builds + pushes the dataset, and deletes the raw itself."""
    fleet = _fleet_of(owner)
    deadline = time.time() + 1800.0  # 30-min cap for the whole upload phase
    try:
        pending = dict(job.upload_cmds)  # device_id -> command id
        n = len(pending)
        while pending:
            if time.time() > deadline:
                raise RuntimeError("upload timed out")
            await asyncio.sleep(2.0)
            for dev_id, cmd_id in list(pending.items()):
                dev = fleet.get(dev_id)
                c = _command_status(dev, cmd_id) if dev else None
                if c is not None and c.status == "done":
                    res = c.result or {}
                    if res.get("status") != "ok":
                        raise RuntimeError(f"{(dev.name if dev else dev_id)}: {res.get('message', 'upload failed')}")
                    pending.pop(dev_id)
                    job.progress = (n - len(pending)) / n if n else 1.0
                    job.message = f"Uploaded {n - len(pending)}/{n} device(s)… (this can take several minutes)"
                    continue
                # Not done yet — bail fast if the device dropped offline (it went
                # dark before finishing, so its upload will never complete). An
                # actively-uploading device keeps polling → stays online, so this
                # only fires on a real disconnect (within ~ONLINE_WINDOW).
                if dev is None or not dev.online:
                    who = dev.name if dev else dev_id
                    raise RuntimeError(f"device '{who}' went offline before finishing its upload — reconnect it and retry")

        # Raw dataset complete → one device runs the processing (calls the Space
        # with its own token; the command completes when the Space is done).
        job.status = "processing"
        job.progress = None  # opaque Space conversion → indeterminate bar
        job.message = "Converting to LeRobot… (this can take several minutes)"
        dev = fleet.get(processor_device_id)
        if dev is None or not dev.online:
            raise RuntimeError("no online device available to run processing")
        proc = Command(id=uuid.uuid4().hex[:12], type="process_dataset",
                       args={"space_url": LEROBOT_SPACE_URL, "source_repo": job.raw_repo,
                             "target_repo": job.target_repo, "roles": job.roles,
                             "private": private, "task": job.target_repo.split("/")[-1]})
        _enqueue(dev, proc)
        proc_deadline = time.time() + 3600.0  # 60-min cap for processing
        while True:
            if time.time() > proc_deadline:
                raise RuntimeError("processing timed out")
            await asyncio.sleep(3.0)
            c = _command_status(dev, proc.id)
            if c is None or c.status != "done":
                # Bail fast if the processing device dropped offline mid-run
                # (it keeps polling while working, so this is a real disconnect).
                d = fleet.get(processor_device_id)
                if d is None or not d.online:
                    who = d.name if d else processor_device_id
                    raise RuntimeError(f"device '{who}' went offline during processing — reconnect it and retry")
                continue
            res = c.result or {}
            if res.get("status") != "ok":
                raise RuntimeError(res.get("message", "processing failed"))
            job.result_url = res.get("result_url") or f"https://huggingface.co/datasets/{job.target_repo}"
            job.status, job.message, job.progress = "done", "Dataset ready.", 1.0
            return
    except Exception as e:  # noqa: BLE001
        job.status, job.error = "error", str(e)
        job.message = f"Failed: {e}"
        logger.warning("dataset job %s failed: %s", job.id, e, exc_info=True)


@app.post("/api/fleet/lerobot-dataset")
async def create_lerobot_dataset(req: DatasetReq, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    roles, plan = _resolve_dataset_plan(owner, req.task_ids, req.only_available, req.roles or None)
    fleet = _fleet_of(owner)
    # With only_available the plan already excludes offline-device episodes; this
    # guard then bites only in the default mode (require the whole set up).
    offline = [d for d in plan if d not in fleet or not fleet[d].online]
    if offline:
        raise HTTPException(409, {"message": "some recording devices are offline", "devices": offline})
    # One member device runs the processing (calls the Space with its own token).
    processor = next(iter(plan))
    job_id = uuid.uuid4().hex[:8]
    name = (req.name or f"grabette-dataset-{job_id}").strip()
    target_repo = name if "/" in name else f"{owner}/{name}"
    # Intermediate raw dataset named after the target (…-raw) instead of random,
    # so it's easy to spot and correlate. Same namespace as the target.
    raw_repo = f"{target_repo}-raw"
    job = DatasetJob(id=job_id, task_ids=list(req.task_ids), roles=roles,
                     raw_repo=raw_repo, target_repo=target_repo,
                     message=f"Uploading episodes from {len(plan)} device(s)… (this can take several minutes)")
    _dataset_jobs_of(owner)[job_id] = job
    for dev_id, p in plan.items():
        cmd = Command(id=uuid.uuid4().hex[:12], type="upload_episodes",
                      args={"raw_repo": raw_repo, "role": p["role"],
                            "episode_ids": sorted(p["episode_ids"]),
                            "private": req.private})
        _enqueue(fleet[dev_id], cmd)
        job.upload_cmds[dev_id] = cmd.id
    t = asyncio.create_task(_run_dataset_job(owner, job, req.private, processor))
    _bg_tasks.add(t)
    t.add_done_callback(_bg_tasks.discard)
    return {"status": "ok", "job_id": job_id, "raw_repo": raw_repo,
            "target_repo": target_repo, "roles": roles, "devices": list(plan.keys())}


@app.get("/api/fleet/lerobot-dataset/{job_id}")
async def lerobot_dataset_status(job_id: str, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    job = _dataset_jobs_of(owner).get(job_id)
    if job is None:
        raise HTTPException(404, "job not found")
    return {"id": job.id, "status": job.status, "message": job.message,
            "progress": job.progress,
            "raw_repo": job.raw_repo, "target_repo": job.target_repo,
            "result_url": job.result_url, "error": job.error}


# === device-facing sync API (Bearer auth) — physical-button episode start/stop
# A button press records an episode into the OPEN session containing this
# device (found by membership). It behaves exactly like the operator starting
# an episode from the dashboard: same task, same synchronized T0 across the
# session's members. No open session → solo (the device records locally). ======
@app.post("/api/devices/{device_id}/sync/start")
async def device_sync_start(device_id: str, auth: tuple[str, str] = Depends(device_auth)) -> dict[str, Any]:
    owner, _token = auth
    if device_id not in _fleet_of(owner):
        raise HTTPException(404, "Device not registered")
    s = _open_session_for_device(owner, device_id)
    if s is None:
        return {"status": "solo"}
    task_name = _task_name(owner, s.task_id)
    _task = _tasks_of(owner).get(s.task_id)
    signature = _task.device_signature if _task else []
    target_iso = _schedule_episode_start(
        owner, task_name, s.members, _lead_for(s),
        exclude_device_id=device_id, signature=signature,
    )
    episode_id = _episode_id_for_target(target_iso)
    s.episodes.append({"episode_id": episode_id, "roles": dict(s.members),
                       "started_at": datetime.now(timezone.utc).isoformat(),
                       "start_at_utc": target_iso})
    s.recording = True
    peers = [d for d in s.members.values() if d != device_id]
    # Full membership + signature so the pressing device persists who it recorded
    # with (it self-schedules from scheduled_start_utc, so it's excluded above).
    return {"status": "scheduled", "scheduled_start_utc": target_iso, "task_name": task_name,
            "peers": peers, "members": _named_members(owner, s.members), "signature": signature}


@app.post("/api/devices/{device_id}/sync/stop")
async def device_sync_stop(device_id: str, auth: tuple[str, str] = Depends(device_auth)) -> dict[str, Any]:
    owner, _token = auth
    if device_id not in _fleet_of(owner):
        raise HTTPException(404, "Device not registered")
    s = _open_session_for_device(owner, device_id)
    if s is None:
        return {"status": "solo"}
    ids = _dispatch_episode_stop(owner, s.members, exclude_device_id=device_id)
    _begin_episode_stop(s, ids)
    return {"status": "ok", "peers": [d for d in s.members.values() if d != device_id]}


@app.post("/api/fleet/dispatch")
async def dispatch(req: DispatchReq, request: Request) -> dict[str, Any]:
    owner = await _operator_loaded(request)
    dev = _fleet_of(owner).get(req.device_id)
    if dev is None:
        raise HTTPException(404, "Device not found in your fleet")
    cmd = Command(id=uuid.uuid4().hex[:12], type=req.type, args=dict(req.args))
    _enqueue(dev, cmd)
    return {"status": "queued", "command_id": cmd.id}


# === OAuth relay for grabette devices ========================================
@app.get("/oauth/grabette/callback", response_model=None)
async def grabette_oauth_relay(
    code: str | None = None,
    state: str | None = None,
    error: str | None = None,
    error_description: str | None = None,
) -> RedirectResponse | HTMLResponse:
    """Relay the HF OAuth callback to the originating grabette on the local network.

    The grabette encodes its mDNS hostname into the OAuth state as
    ``{hostname}|{session_id}``. This endpoint splits that apart and issues a
    302 redirect so the user's browser (on the same LAN as the grabette) reaches
    ``http://{hostname}.local:8000/api/hf-auth/oauth/callback``.

    Only this Space URL needs to be registered as a redirect_uri in the HF
    OAuth app — one entry covers every grabette regardless of hostname.
    """
    if not state or "|" not in state:
        return HTMLResponse("Missing or invalid state parameter.", status_code=400)

    hostname, session_id = state.split("|", 1)
    from urllib.parse import urlencode
    base = f"http://{hostname}.local:8000/api/hf-auth/oauth/callback"

    if error:
        # Forward the error to the grabette so it can mark the session as failed
        # and stop the polling loop on the frontend.
        params: dict = {"error": error, "state": session_id}
        if error_description:
            params["error_description"] = error_description
        return RedirectResponse(f"{base}?{urlencode(params)}", status_code=302)

    if not code:
        return HTMLResponse("Missing code.", status_code=400)

    return RedirectResponse(
        f"{base}?{urlencode({'code': code, 'state': session_id})}",
        status_code=302,
    )


# === UI ======================================================================
@app.get("/")
async def index() -> HTMLResponse:
    return HTMLResponse(_INDEX)


_INDEX = """<!DOCTYPE html><html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>Grabette fleet</title><style>
 *{box-sizing:border-box}
 body{font-family:-apple-system,system-ui,sans-serif;background:linear-gradient(135deg,#1a1a2e,#16213e);
  color:#fff;min-height:100vh;margin:0;padding:2rem;display:flex;justify-content:center}
 .wrap{width:100%;max-width:640px}h1{font-size:1.3rem;margin:0 0 .3rem}
 h2{font-size:1rem;margin:0;display:flex;align-items:center;gap:.5rem}
 .intro{color:#c3cbe0;font-size:.88rem;line-height:1.5;margin:0 0 1.4rem}
 .card{background:rgba(255,255,255,.06);padding:1.2rem;border-radius:14px;margin-bottom:1rem}
 .card.groups{background:linear-gradient(135deg,rgba(139,92,246,.18),rgba(59,130,246,.10));
  border:1px solid rgba(167,139,250,.4);margin-bottom:1.4rem}
 .card.groups h2{color:#c4b5fd}
 .card.groups .count{background:rgba(167,139,250,.22);color:#e9d5ff}
 .card-head{display:flex;align-items:center;justify-content:space-between;margin-bottom:.9rem}
 .count{background:rgba(255,255,255,.12);color:#e2e8f0;font-size:.72rem;font-weight:600;
  padding:.12rem .55rem;border-radius:999px}
 a.btn,button{padding:.55rem 1rem;border:0;border-radius:8px;cursor:pointer;font-weight:600;text-decoration:none;display:inline-block}
 .primary{background:#ffcc4d;color:#1a1a2e}.logout{background:#ef4444;color:#fff}
 .who-row{display:flex;align-items:center;justify-content:space-between;gap:1rem;flex-wrap:wrap}
 .logout-icon{background:#16213e;color:#fff;border:1px solid #2d4a7a;padding:.3rem .5rem;line-height:1}
 .del-icon{background:#16213e;color:#ef4444;border:1px solid #ef4444;padding:.3rem .5rem;line-height:1}
 .rec-icon{background:#16213e;color:#10b981;border:1px solid #2d4a7a;padding:.3rem .5rem;line-height:1}
 .stop-icon{background:#16213e;color:#ef4444;border:1px solid #2d4a7a;padding:.3rem .5rem;line-height:1}
 .dash-icon{background:#16213e;color:#7dd3fc;border:1px solid #2d4a7a;padding:.3rem .5rem;line-height:1}
 button:disabled{opacity:.4;cursor:not-allowed}
 .muted{color:#a0aec0;font-size:.82rem}
 table{width:100%;border-collapse:collapse;font-size:.85rem;table-layout:fixed}
 td,th{text-align:left;padding:.45rem .35rem;border-bottom:1px solid #334;vertical-align:top}
 .c-dot{width:1.2rem}.c-act{width:30%}.c-tools{width:6.6rem;text-align:right}
 .name b{word-break:break-word}
 .result{font-size:.8rem;color:#cbd5e0;
  background:rgba(0,0,0,.25);border-radius:6px;padding:.5rem .6rem;margin-top:.4rem}
 .result.empty{background:none;padding:0}
 .rlabel{font-size:.74rem;font-weight:600;color:#8b98ad;margin-bottom:.35rem;display:flex;align-items:center;gap:.5rem}
 .rclose{margin-left:auto;background:none;border:0;color:#8b98ad;font-size:1.05rem;line-height:1;cursor:pointer;padding:0 .15rem;font-weight:400}
 .rclose:hover{color:#fff}
 .kv{display:grid;grid-template-columns:auto 1fr;gap:.2rem .7rem}
 .kv .k{color:#8b98ad}.kv .v{overflow-wrap:anywhere;word-break:break-word}
 .pill{display:inline-block;padding:.03rem .5rem;border-radius:999px;font-size:.72rem;font-weight:600}
 .pill.rec{background:#ef4444;color:#fff}.pill.idle{background:rgba(255,255,255,.14);color:#cbd5e0}
 .pill.init{background:rgba(245,158,11,.22);color:#fcd34d}
 .pill.stopping{background:rgba(239,68,68,.22);color:#fca5a5}
 .pill.hand{background:rgba(125,211,252,.16);color:#7dd3fc;margin-left:.4rem;vertical-align:middle}
 .rec-dur{font-variant-numeric:tabular-nums;margin-left:.15rem}
 .batt{display:inline-block;margin-left:.4rem;padding:.03rem .45rem;border-radius:999px;font-size:.7rem;font-weight:700;vertical-align:middle}
 .batt::before{content:"🔋 "}
 .batt.batt-ok{background:rgba(16,185,129,.2);color:#a7f3d0}
 .batt.batt-mid{background:rgba(234,179,8,.2);color:#fde68a}
 .batt.batt-low{background:rgba(239,68,68,.22);color:#fca5a5}
 /* device activity badge (capturing / uploading / converting) */
 .act-badge{display:inline-block;margin-left:.4rem;padding:.03rem .5rem;border-radius:999px;font-size:.7rem;font-weight:700;vertical-align:middle;white-space:nowrap}
 .act-badge.act-cap{background:rgba(239,68,68,.22);color:#fca5a5}
 .act-badge.act-up{background:rgba(59,130,246,.2);color:#93c5fd}
 .act-badge.act-proc{background:rgba(139,92,246,.22);color:#d6c9fb}
 .err{color:#fca5a5}
 .raw-d{margin-top:.45rem}.raw-d summary{cursor:pointer;color:#8b98ad;font-size:.72rem}
 .raw-d pre{white-space:pre-wrap;overflow-wrap:anywhere;word-break:break-word;margin:.3rem 0 0;font-size:.72rem;
  max-height:14rem;overflow:auto;background:rgba(0,0,0,.25);border-radius:5px;padding:.4rem .5rem}
 .acts{display:flex;flex-wrap:wrap;gap:.3rem}.acts button{padding:.22rem .7rem;font-size:.8rem}
 .acts button.act-on{box-shadow:inset 0 0 0 2px #1a1a2e}
 .tools{display:flex;gap:.3rem;justify-content:flex-end}
 .dot{display:inline-block;width:.6rem;height:.6rem;border-radius:50%;margin-top:.3rem}
 .on{background:#10b981}.off{background:#ef4444}
 fieldset:disabled{opacity:.45}fieldset{border:0;padding:0;margin:0}code{color:#ffcc4d}
 .c-plus{width:2.4rem}
 .plus-btn{width:1.8rem;height:1.8rem;border-radius:50%;padding:0;font-size:1.1rem;font-weight:700;
  line-height:1;display:inline-flex;align-items:center;justify-content:center}
 .plus-btn.blue{background:#3b82f6;color:#fff}
 .plus-btn.green{background:#10b981;color:#fff}
 .plus-btn.grey{background:#3a4358;color:#6b7280;cursor:not-allowed}
 .pill.group{color:#fff}
 .task-row{padding:.5rem .4rem;border-radius:8px;cursor:pointer}
 .task-row:hover{background:rgba(255,255,255,.06)}
 /* Top line: name takes the full width, action buttons pinned right. */
 .task-top{display:flex;align-items:center;gap:.5rem}
 .task-top .tname{font-weight:600;flex:1;min-width:0;overflow-wrap:anywhere}
 .task-acts{display:flex;gap:.3rem;flex-shrink:0;margin-left:auto}
 /* Second line: "Required devices" tags + description wrap below the name. */
 .task-sub{color:#a0aec0;font-size:.78rem;margin-top:.3rem;display:flex;flex-wrap:wrap;align-items:center;gap:.3rem}
 .task-sub .pill.hand{margin:0}
 #task-editor,.task-edit-panel,.task-del-panel{display:flex;flex-direction:column;gap:.6rem;align-items:stretch}
 .subpanel{margin-top:.9rem;padding:.9rem 1rem;border-radius:10px;border:1px solid rgba(16,185,129,.35);
  background:rgba(0,0,0,.18)}
 .subpanel-title{font-weight:600;font-size:.85rem;color:#6ee7b7}
 .subpanel.task-del-panel{border-color:rgba(239,68,68,.4)}
 .subpanel-title.danger{color:#fca5a5}
 .del-warn{font-size:.85rem;color:#c3cbe0}
 button.del-confirm{background:#ef4444;color:#fff}
 #orphan-bar{margin-bottom:1.4rem}
 .orphan-acc{border:1px solid rgba(245,158,11,.4);border-radius:10px;background:rgba(245,158,11,.08)}
 .orphan-acc-head{display:flex;justify-content:space-between;align-items:center;gap:.5rem;
  padding:.65rem .9rem;cursor:pointer;color:#fcd34d;font-weight:600;font-size:.9rem;user-select:none}
 .orphan-acc-head:hover{background:rgba(245,158,11,.06)}
 .orphan-caret{color:#fcd34d}
 .orphan-acc-body{padding:0 .8rem .8rem;display:flex;flex-direction:column;gap:.7rem}
 .orphan-item{padding:.8rem .9rem;border-radius:10px;border:1px solid rgba(245,158,11,.4);
  background:rgba(245,158,11,.1);display:flex;flex-direction:column;gap:.4rem;align-items:flex-start}
 .orphan-head{font-size:.92rem;color:#fcd34d}
 .orphan-detail{font-size:.82rem;color:#c3cbe0}
 .warn-ico{width:15px;height:15px;flex-shrink:0}
 .orphan-acc-head .warn-ico,.hist-toggle.warn .warn-ico{width:14px;height:14px;vertical-align:-2px}
 .task-warn{display:inline-flex;align-items:center;margin-left:.4rem;color:#fbbf24;vertical-align:middle}
 .task-warn .warn-ico{width:15px;height:15px}
 .hist-toggle.warn{color:#fcd34d}
 .hist-toggle.warn:hover{color:#fde68a}
 select.role-select{width:100%;padding:.5rem .6rem;border-radius:8px;border:1px solid #334;
  background:#fff;color:#111;font:inherit}
 select.role-select option{background:#fff;color:#111}
 .role-pick{display:flex;align-items:center;gap:.6rem}
 .role-pick .sig-label{min-width:5.5rem}
 .fleet-group{margin-top:1rem}.fleet-group:first-of-type{margin-top:.3rem}
 .fleet-group h3{font-size:.9rem;margin:0 0 .4rem;color:#c3cbe0;display:flex;align-items:center;gap:.5rem}
 .subcount{background:rgba(255,255,255,.12);color:#e2e8f0;font-size:.7rem;font-weight:600;padding:.1rem .5rem;border-radius:999px}
 .session-panel{margin-top:.9rem;padding:1rem;border-radius:12px;border:1px solid rgba(239,68,68,.4);
  background:rgba(0,0,0,.22)}
 .sp-head{display:flex;align-items:center;justify-content:space-between;gap:.6rem;margin-bottom:.4rem}
 .sp-task{font-weight:700;font-size:1.05rem}
 /* Big recording timer, centered just above the record button. */
 .sp-timer-wrap{display:flex;justify-content:center;min-height:1.9rem;margin:.6rem 0 .2rem}
 .sp-timer{font-variant-numeric:tabular-nums;font-size:1.9rem;font-weight:800;line-height:1;color:#fff}
 @keyframes sp-pulse{50%{opacity:.35}}
 /* Phone-camera-style record toggle: red circle (start) ↔ red square (stop). */
 .sp-rec{display:flex;justify-content:center;margin:0 0 1rem}
 .rec-toggle{width:72px;height:72px;border-radius:999px;border:4px solid rgba(255,255,255,.55);background:transparent;display:inline-flex;align-items:center;justify-content:center;cursor:pointer;padding:0}
 .rec-toggle:hover{border-color:rgba(255,255,255,.85)}
 .rec-toggle .rt-inner{background:#ef4444;transition:width .15s,height .15s,border-radius .15s}
 .rec-toggle.idle .rt-inner{width:52px;height:52px;border-radius:999px}
 .rec-toggle.recording .rt-inner{width:28px;height:28px;border-radius:8px}
 .rec-toggle.busy{cursor:default;border-color:rgba(255,255,255,.35)}
 .rec-toggle.busy .rt-inner{width:34px;height:34px;border-radius:999px;opacity:.6;animation:sp-pulse 1s infinite}
 /* record circle blocked because a member is busy with a dataset job */
 .rec-toggle.blocked{cursor:not-allowed;opacity:.4;border-color:rgba(255,255,255,.3)}
 .sp-busy-note{margin-top:.6rem;text-align:center;font-size:.8rem;color:#d6c9fb}
 .sp-meta{display:flex;align-items:baseline;gap:.6rem;margin-top:.35rem;font-size:.9rem}
 .sp-label{color:#8b98ad;font-size:.78rem;font-weight:600;min-width:5rem}
 .sp-devs{display:flex;flex-wrap:wrap;gap:.8rem}
 .sp-dev{display:inline-flex;align-items:center;gap:.35rem}
 .sp-actions{display:flex;gap:.5rem;margin-top:.8rem;flex-wrap:wrap}
 .sp-actions button{display:inline-flex;align-items:center;gap:.4rem;padding:.4rem .8rem;font-size:.85rem}
 .sp-actions .del-ep{background:#16213e;color:#ef4444;border:1px solid #ef4444}
 .sp-actions .del-ep:disabled{opacity:.45;cursor:not-allowed}
 .sp-close{display:block;width:100%;margin-top:1rem;padding:.6rem;font-size:.9rem;border:0;border-radius:8px;cursor:pointer;font-weight:600}
 /* ===== Session panel — two zones: recording box vs management box ===== */
 .del-ep{background:#16213e;color:#ef4444;border:1px solid #ef4444;padding:.45rem .8rem;border-radius:8px;font-size:.85rem;cursor:pointer}
 .del-ep:disabled{opacity:.45;cursor:not-allowed}
 .close-btn{background:#ffcc4d;color:#1a1a2e;border:0;padding:.55rem;border-radius:8px;font-weight:700;font-size:.9rem;cursor:pointer;width:100%;margin-top:.8rem}
 .var-c{padding:0;overflow:hidden}
 .sp-rec-zone{padding:1rem;background:rgba(239,68,68,.08)}
 .sp-manage-zone{padding:1rem;background:rgba(255,255,255,.04);border-top:1px solid rgba(255,255,255,.12)}
 /* Delete last episode: bottom-right, inside the recording zone */
 .sp-rec-del{display:flex;justify-content:flex-end;margin-top:.8rem}
 /* Recorded-episode count: one large number */
 .sp-section-label{font-size:.72rem;font-weight:700;color:#8b98ad;letter-spacing:.05em;text-transform:uppercase;margin:.2rem 0 .3rem}
 .ep-big{display:flex;align-items:baseline;gap:.5rem;margin:.15rem 0 .6rem}
 .ep-big-num{font-size:2.1rem;font-weight:800;line-height:1;font-variant-numeric:tabular-nums;
  background:linear-gradient(135deg,#10b981,#3b82f6);-webkit-background-clip:text;background-clip:text;color:transparent}
 .ep-big-lbl{font-size:.82rem;color:#8b98ad}
 .task-block{border-bottom:1px solid #223;}
 .task-detail{padding:.2rem .4rem .7rem 1.6rem}
 .task-launch{display:flex;flex-direction:column;gap:.5rem;margin:.2rem 0 .6rem}
 .launch-hint{font-size:.78rem;align-self:center}
 .section-note{font-size:.8rem;line-height:1.35;margin:.1rem 0 .8rem}
 .task-eps{margin-top:.35rem}
 .ep-count{display:inline-block;font-size:.72rem;color:#a7f3d0;background:rgba(16,185,129,.18);padding:.08rem .55rem;border-radius:999px;white-space:nowrap}
 .ghost{background:rgba(255,255,255,.08);color:#c3cbe0;border:1px solid rgba(255,255,255,.18);padding:.3rem .7rem;font-size:.8rem;border-radius:8px;cursor:pointer}
 .ghost:hover{background:rgba(255,255,255,.14)}
 .task-row.select{cursor:pointer}
 .task-row.select .task-top{gap:.6rem}
 .task-row.disabled{opacity:.4;cursor:not-allowed}
 .task-check{width:16px;height:16px;accent-color:#10b981;pointer-events:none;flex-shrink:0}
 /* Vertical stack: content recap, then the labelled name field, then Private,
    then the button — each on its own line. */
 #dataset-bar{display:none;flex-direction:column;align-items:flex-start;gap:.6rem;
  margin-top:.8rem;padding:.7rem .8rem;border-radius:10px;background:linear-gradient(135deg,rgba(139,92,246,.14),rgba(59,130,246,.10));border:1px solid rgba(139,92,246,.32)}
 #top-sticky{position:sticky;top:0;z-index:30;background:#181e36;padding:.7rem 0;margin-bottom:.9rem;
  display:flex;flex-direction:column;gap:.7rem;box-shadow:0 6px 12px -8px rgba(0,0,0,.7)}
 #top-sticky .card{margin-bottom:0}
 .fleet-head{cursor:pointer;user-select:none;margin-bottom:0}
 .fleet-head h2{display:inline}
 .fleet-caret{color:#8b98ad;font-size:.9rem}
 .fleet-recap{display:inline-flex;flex-wrap:wrap;gap:.3rem;align-items:center}
 .fleet-chip{font-size:.68rem;font-weight:600;color:#c3cbe0;background:rgba(255,255,255,.1);padding:.06rem .45rem;border-radius:999px;white-space:nowrap}
 .fleet-chip.zero{color:#6b7280;background:rgba(255,255,255,.05)}
 .fleet-chip.busy{color:#d6c9fb;background:rgba(139,92,246,.22)}
 #fleet-body{margin-top:.9rem;max-height:42vh;overflow:auto}
 .page-tabs{display:flex;gap:4px;padding:4px;border-radius:999px;
  background:rgba(255,255,255,.05);border:1px solid rgba(255,255,255,.1)}
 .page-tabs .seg-btn{flex:1;justify-content:center}
 .seg-btn{display:inline-flex;align-items:center;justify-content:center;gap:.5rem;border:0;background:transparent;
  color:#aeb8cc;font:inherit;font-size:1rem;font-weight:700;padding:.7rem 1.1rem;border-radius:999px;cursor:pointer;
  white-space:nowrap;transition:background .12s,color .12s,box-shadow .12s}
 .seg-btn svg{width:16px;height:16px}
 .seg-btn:hover{color:#e2e8f0}
 #seg-record.active{background:linear-gradient(135deg,rgba(16,185,129,.5),rgba(59,130,246,.42));color:#fff;box-shadow:0 1px 8px rgba(16,185,129,.3)}
 #seg-dataset.active{background:linear-gradient(135deg,rgba(139,92,246,.5),rgba(59,130,246,.42));color:#fff;box-shadow:0 1px 8px rgba(139,92,246,.3)}
 .mode-dataset .step-num{background:linear-gradient(135deg,#8b5cf6,#3b82f6);color:#fff}
 .mode-dataset .ep-count{background:rgba(139,92,246,.2);color:#d6c9fb}
 #ds-gen{background:linear-gradient(135deg,#8b5cf6,#3b82f6);color:#fff}
 .ds-info{font-size:.82rem;color:#c3cbe0}
 .ds-step{display:flex;align-items:center;gap:.55rem;font-weight:700;font-size:1.05rem;color:#e2e8f0}
 #ds-step1{margin:.2rem 0 .6rem}
 #ds-advanced{margin:-.2rem 0 .7rem}
 .adv-toggle{font-size:.8rem;color:#8b98ad;cursor:pointer;user-select:none;padding:.2rem 0}
 .adv-toggle:hover{color:#c3cbe0}
 .adv-body{display:flex;align-items:center;flex-wrap:wrap;gap:.4rem;padding:.3rem 0 .1rem;font-size:.82rem}
 .use-only{border:1px solid rgba(255,255,255,.18);background:rgba(255,255,255,.05);color:#c3cbe0;
  font:inherit;font-size:.8rem;font-weight:600;padding:.22rem .6rem;border-radius:7px;cursor:pointer}
 .use-only:hover{background:rgba(255,255,255,.12)}
 .use-only.active{background:rgba(16,185,129,.22);color:#a7f3d0;border-color:rgba(16,185,129,.5)}
 .adv-hint{flex-basis:100%;font-size:.75rem;margin-top:.15rem}
 .step-num{display:inline-flex;align-items:center;justify-content:center;width:1.7rem;height:1.7rem;flex-shrink:0;
  border-radius:999px;background:linear-gradient(135deg,#10b981,#3b82f6);color:#fff;font-size:.85rem;font-weight:700}
 .ds-field{display:flex;flex-direction:column;gap:.3rem;width:100%}
 .ds-label{font-size:.75rem;font-weight:600;color:#8b98ad}
 .ds-req{color:#fca5a5}
 .ds-devlabel{font-size:.75rem}
 .ds-devlist{display:flex;flex-wrap:wrap;gap:.3rem .8rem;margin-top:.2rem}
 .ds-dev{display:inline-flex;align-items:center;gap:.35rem;font-size:.82rem;color:#c3cbe0}
 .ds-dot{width:.55rem;height:.55rem;border-radius:999px;flex-shrink:0}
 .ds-dot.on{background:#10b981}
 .ds-dot.off{background:#ef4444}
 .ds-avail{display:inline-flex;align-items:center;gap:.3rem;font-size:.8rem;color:#c3cbe0;cursor:pointer}
 .ds-avail input{width:auto;min-width:0;margin:0}
 .ds-target{display:flex;align-items:center;gap:.3rem;flex-wrap:wrap}
 .ds-target select,.ds-target #ds-name{padding:.35rem .5rem;border-radius:8px;border:1px solid #334;font:inherit;font-size:.82rem}
 .ds-target select{background:#fff;color:#111}
 .ds-target #ds-name{background:rgba(255,255,255,.05);color:#fff;min-width:8rem}
 .ds-slash{color:#8b98ad}
 /* Keep the checkbox tight to its label (don't inherit the text-input width). */
 .ds-private{display:inline-flex;align-items:center;gap:.25rem;font-size:.8rem;color:#c3cbe0;cursor:pointer;white-space:nowrap}
 .ds-private input{width:auto;min-width:0;margin:0}
 .ds-job{width:100%;font-size:.8rem;color:#c3cbe0}
 .ds-job.err{color:#fca5a5}
 .ds-job.ok{color:#a7f3d0}
 .ds-progress{width:100%;height:6px;border-radius:999px;background:rgba(255,255,255,.12);overflow:hidden}
 .ds-progress-fill{height:100%;width:0;border-radius:999px;background:#10b981;transition:width .3s ease}
 .ds-progress.indet .ds-progress-fill{width:40%;background:linear-gradient(90deg,rgba(16,185,129,0),#10b981,rgba(16,185,129,0));animation:ds-indet 1.2s linear infinite}
 @keyframes ds-indet{0%{transform:translateX(-120%)}100%{transform:translateX(320%)}}
 .ds-job a{color:#7dd3fc}
 .hist-toggle{font-size:.8rem;color:#8b98ad;cursor:pointer;user-select:none;padding:.25rem 0}
 .hist-toggle:hover{color:#c3cbe0}
 .hist-row{font-size:.82rem;padding:.25rem 0;color:#cbd5e0}
 .hist-head{display:flex;align-items:center;justify-content:space-between;gap:.6rem;flex-wrap:wrap}
 .fill-task{background:#16213e;color:#7dd3fc;border:1px solid #2d4a7a;padding:.2rem .6rem;border-radius:7px;font-size:.76rem;cursor:pointer;white-space:nowrap}
 .fill-ep-hint{font-size:.75rem;color:#8b98ad;margin-left:.3rem;white-space:nowrap}
 .act-edit{background:#16213e;color:#7dd3fc;border:1px solid #2d4a7a;padding:.3rem .5rem;line-height:1}
 #task-name-input,#task-desc-input,.edit-input{width:100%;padding:.5rem .6rem;border-radius:8px;border:1px solid #334;
  background:rgba(255,255,255,.05);color:#fff;font:inherit}
 .sig-row{display:flex;align-items:center;gap:.7rem;flex-wrap:wrap}
 .sig-label{color:#c3cbe0;font-size:.85rem;font-weight:600}
 .sig-label.task-label{font-size:1.05rem;font-weight:700;color:#e2e8f0}
 .sig-checks{display:flex;gap:.8rem;align-items:center;color:#c3cbe0;font-size:.85rem}
 .sig-checks label{display:flex;align-items:center;gap:.3rem;cursor:pointer}
 .editor-actions{display:flex;gap:.5rem}
 .pill.task{background:rgba(255,204,77,.16);color:#ffcc4d}
 .card.tasks{margin-bottom:1.4rem}
 .card.sessions{background:linear-gradient(135deg,rgba(239,68,68,.14),rgba(139,92,246,.10));
  border:1px solid rgba(239,68,68,.32);margin-bottom:1.4rem}
 .card.sessions h2{color:#fca5a5}
 .card.sessions .count{background:rgba(239,68,68,.22);color:#fecaca}
 .pill.open{background:#10b981;color:#fff}
 .pill.closed{background:rgba(255,255,255,.14);color:#cbd5e0}
 button.validate{background:linear-gradient(135deg,#10b981,#3b82f6);color:#fff}
 button.cancel{background:rgba(255,255,255,.12);color:#e2e8f0}
 /* Phone: reclaim the wide desktop padding and let side-by-side rows stack. */
 @media (max-width:600px){
  body{padding:.7rem}
  .card{padding:.9rem}
  .role-pick{flex-wrap:wrap}
  .role-pick select{flex:1;min-width:0}
  .sig-row{align-items:flex-start}
 }
</style></head><body><div class="wrap">
<h1>Grabette fleet <span class="muted">— operator dashboard</span></h1>
<p class="intro">This dashboard lists every Grabette, Gripette and Casquette that has been
 detected and is connected to your HuggingFace account. Each device is grouped by type below;
 sign in with HuggingFace to see and control the devices that report to your fleet.</p>
<div class="card"><h2>HuggingFace login</h2><div id="who" style="margin-top:.7rem">Checking…</div></div>
<fieldset id="gated" disabled>
 <div id="top-sticky">
  <div class="page-tabs" id="page-tabs" role="tablist" aria-label="Mode">
   <button class="seg-btn active" id="seg-record" role="tab" onclick="setSelectMode(false)"><svg viewBox="0 0 24 24" fill="currentColor"><circle cx="12" cy="12" r="7"/></svg>Record</button>
   <button class="seg-btn" id="seg-dataset" role="tab" onclick="setSelectMode(true)"><svg viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M12 2 2 7l10 5 10-5-10-5Z"/><path d="m2 17 10 5 10-5"/><path d="m2 12 10 5 10-5"/></svg>Build dataset</button>
  </div>
  <div class="card fleet-acc" id="fleet-card">
   <div class="card-head fleet-head" onclick="toggleFleet()">
    <div style="display:flex;align-items:center;gap:.5rem;flex-wrap:wrap"><h2>Fleet</h2><span class="fleet-recap" id="fleet-recap"></span></div>
    <span class="fleet-caret" id="fleet-caret">▸</span>
   </div>
   <div id="fleet-body" style="display:none">
    <div class="fleet-group" data-kind="grabette">
     <h3>Grabettes <span class="subcount" id="count-grabette">0</span></h3>
     <table><tbody id="tb-grabette"></tbody></table>
     <div class="muted empty" id="empty-grabette">No Grabettes detected yet.</div>
    </div>
    <div class="fleet-group" data-kind="gripette">
     <h3>Gripettes <span class="subcount" id="count-gripette">0</span></h3>
     <table><tbody id="tb-gripette"></tbody></table>
     <div class="muted empty" id="empty-gripette">No Gripettes detected yet.</div>
    </div>
    <div class="fleet-group" data-kind="casquette">
     <h3>Casquettes <span class="subcount" id="count-casquette">0</span></h3>
     <table><tbody id="tb-casquette"></tbody></table>
     <div class="muted empty" id="empty-casquette">No Casquettes detected yet.</div>
    </div>
   </div>
  </div>
 </div>
 <div id="orphan-bar" style="display:none"></div>
 <div class="card sessions" id="sessions-card" style="display:none">
  <div class="card-head"><div style="display:flex;align-items:center;gap:.5rem"><h2><svg width="15" height="15" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><circle cx="12" cy="12" r="10"/><polyline points="12 6 12 12 16 14"/></svg>Session</h2></div></div>
  <div id="session-list"></div>
 </div>
 <div class="card tasks mode-record" id="tasks-card">
  <div class="card-head"><div style="display:flex;align-items:center;gap:.5rem"><h2><svg width="15" height="15" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M9 11l3 3L22 4"/><path d="M21 12v7a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2V5a2 2 0 0 1 2-2h11"/></svg>Tasks</h2><span class="count" id="count-tasks">0</span></div></div>
  <p class="section-note muted">These tasks come from the grabettes connected to this account — those online now, plus any connected earlier in this session. Episode counts reflect only the grabettes currently online.</p>
  <div id="ds-step1" class="ds-step" style="display:none"></div>
  <div id="ds-advanced" style="display:none"></div>
  <div id="task-list" class="muted empty">No tasks yet.</div>
  <div id="dataset-bar"></div>
  <button class="primary" id="btn-create-task" onclick="startCreateTask()" style="margin-top:.9rem">Create task</button>
  <div id="task-editor" class="subpanel" style="display:none">
   <div class="subpanel-title" id="task-editor-title">New task</div>
   <input id="task-name-input" type="text" placeholder="Task name">
   <input id="task-desc-input" type="text" placeholder="Description (optional)">
   <div class="sig-row">
    <span class="sig-label">Required devices</span>
    <span class="sig-checks">
     <label><input type="checkbox" id="task-sig-left">lgrabette</label>
     <label><input type="checkbox" id="task-sig-right">rgrabette</label>
     <label><input type="checkbox" id="task-sig-casquette">casquette</label>
    </span>
   </div>
   <div class="editor-actions">
    <button class="validate" onclick="validateTask()">Validate task</button>
    <button class="cancel" onclick="cancelTaskEdit()">Cancel</button>
   </div>
  </div>
 </div>
</fieldset></div><script>
const $=id=>document.getElementById(id);let loggedIn=false;
const KINDS=['grabette','gripette','casquette'];
let DEVICES=[];let TASKS=[];let SESSIONS=[];let ORPHANS=[];
let taskEditId=null;let taskDeleteId='';
let launchRolePick={left:'',right:'',casquette:''};
let expandedTaskId='';let historyOpenId='';let issuesOpenId='';let orphansOpen=false;let groupsOpenId='';let fleetOpen=false;
// Dataset selection mode: pick several tasks (same device signature) to build a
// LeRobot dataset. The first pick locks the signature; incompatible tasks are
// disabled. Selection persists across the 3s refresh (module-level state).
let selectMode=false;let datasetSel=new Set();let datasetJob=null;let datasetPollTimer=null;
let datasetName='';let datasetOwner='';let datasetNamespaces=null;let datasetPrivate=false;let datasetOnlyAvailable=false;
// "Use only" advanced option: a role subset (empty = each task's full signature).
let datasetUseOnly=[];let datasetAdvOpen=false;
const USE_ONLY_OPTS=[{label:'L',roles:['left']},{label:'R',roles:['right']},{label:'LC',roles:['left','casquette']},{label:'RC',roles:['right','casquette']}];
const ROLE_LABEL={left:'lgrabette',right:'rgrabette',casquette:'casquette'};
function deviceName(device_id){const d=DEVICES.find(x=>x.device_id===device_id);return d?d.name:device_id;}
function deviceBattery(device_id){const d=DEVICES.find(x=>x.device_id===device_id);return d?d.battery:null;}
function batteryPill(b){
 if(b===null||b===undefined)return '';
 const p=Math.round(b);const cls=p<=15?'batt-low':p<=40?'batt-mid':'batt-ok';
 return `<span class="batt ${cls}" title="Battery ${p}%">${p}%</span>`;}
function fmtDur(ms){const s=Math.max(0,Math.floor(ms/1000));return Math.floor(s/60)+':'+String(s%60).padStart(2,'0');}
function tickRecDur(){document.querySelectorAll('.rec-dur').forEach(el=>{const st=+el.dataset.start;if(st)el.textContent=fmtDur(Date.now()-st);});}
function slotOf(d){const k=kindOf(d);if(k==='casquette')return 'casquette';if(k==='grabette'&&(d.hand==='left'||d.hand==='right'))return d.hand;return null;}
// device activity ('idle'|'capturing'|'uploading'|'processing') reported by
// the fleet (self-reported by the device, or inferred from dispatched work).
function deviceActivity(id){const d=DEVICES.find(x=>x.device_id===id);return (d&&d.activity)||'idle';}
// Busy with dataset work → cannot (re)start a recording (mirrors the server gate).
function deviceBusyForRec(id){const a=deviceActivity(id);return a==='uploading'||a==='processing';}
const ACT_LABEL={capturing:'● Recording',uploading:'↑ Uploading',processing:'⚙ Converting'};
function activityBadge(id){
 const a=deviceActivity(id);
 if(a==='idle')return '';
 const cls={capturing:'act-cap',uploading:'act-up',processing:'act-proc'}[a]||'';
 return `<span class="act-badge ${cls}" title="Device is ${a}">${ACT_LABEL[a]||a}</span>`;}
function errText(j){const d=j&&j.detail;if(!d)return '';if(typeof d==='string')return d;return d.message||JSON.stringify(d);}
// ── Tasks ──
function startCreateTask(){taskEditId=null;$('task-editor-title').textContent='New task';$('task-name-input').value='';$('task-desc-input').value='';
 ['left','right','casquette'].forEach(s=>$('task-sig-'+s).checked=false);
 $('task-editor').style.display='flex';$('btn-create-task').style.display='none';}
// Edit happens INLINE, in the task's own block (see taskEditorHtml in
// renderTaskList) — not in the bottom editor — so it's visible even far down a
// long list. Just flag the task and re-render.
function editTask(id){taskEditId=(taskEditId===id)?'':id;expandedTaskId='';renderTaskList();}
// Inline editor for one task. The required devices are LOCKED once they're KNOWN
// and the task has episodes (changing them would misdescribe the recordings). An
// old task with episodes but no recorded signature can be set once to backfill it.
function sigLocked(t){return (t.episode_count||0)>0 && (t.device_signature||[]).length>0;}
function taskEditorHtml(t){
 const locked=sigLocked(t);
 const backfill=(t.episode_count||0)>0 && !locked;  // old task: set the devices once
 const chk=s=>`<label><input type="checkbox" class="edit-sig" value="${s}" ${(t.device_signature||[]).includes(s)?'checked':''} ${locked?'disabled':''}>${ROLE_LABEL[s]}</label>`;
 let note='';
 if(locked)note=`<div class="muted" style="font-size:.8rem">Locked — this task already has ${t.episode_count} recorded episode(s), so its required devices can't change.</div>`;
 else if(backfill)note=`<div class="muted" style="font-size:.8rem">This task has recordings but no devices set — choose them now to fix it. This can only be set once.</div>`;
 return `<div class="subpanel task-edit-panel">
   <div class="subpanel-title">Edit task</div>
   <input id="edit-name" class="edit-input" type="text" placeholder="Task name" value="${esc(t.name)}">
   <input id="edit-desc" class="edit-input" type="text" placeholder="Description (optional)" value="${esc(t.description||'')}">
   <div class="sig-row"><span class="sig-label">Required devices</span>
    <span class="sig-checks">${chk('left')}${chk('right')}${chk('casquette')}</span></div>
   ${note}
   <div class="editor-actions">
    <button class="validate" onclick="validateTask()">Save</button>
    <button class="cancel" onclick="editTask('${t.id}')">Cancel</button>
   </div>
  </div>`;}
function cancelTaskEdit(){taskEditId=null;$('task-editor').style.display='none';$('btn-create-task').style.display='inline-block';}
async function validateTask(){
 let name,description,device_signature;
 if(taskEditId){  // inline editor
  name=$('edit-name').value.trim();description=$('edit-desc').value.trim();
  device_signature=[...document.querySelectorAll('.edit-sig:checked')].map(c=>c.value);
 }else{           // bottom "create task" editor
  name=$('task-name-input').value.trim();description=$('task-desc-input').value.trim();
  device_signature=['left','right','casquette'].filter(s=>$('task-sig-'+s).checked);
 }
 if(!name){alert('Task needs a name');return;}
 if(!device_signature.length){alert('Select at least one required device');return;}
 const body={name,description,device_signature};
 const url=taskEditId?`/api/fleet/tasks/${taskEditId}`:'/api/fleet/tasks';
 const method=taskEditId?'PUT':'POST';
 const r=await fetch(url,{method,headers:{'Content-Type':'application/json'},body:JSON.stringify(body)});
 if(!r.ok){const j=await r.json().catch(()=>({}));alert(errText(j)||'Failed to save task');return;}
 taskEditId=null;$('task-editor').style.display='none';$('btn-create-task').style.display='inline-block';
 await refresh();}
// Delete asks for confirmation in an inline panel (taskDeleteHtml) that spells
// out the consequences (episodes + which devices lose their recordings), rather
// than a bare confirm(). Clicking delete again cancels.
function deleteTask(id){taskDeleteId=(taskDeleteId===id)?'':id;taskEditId='';expandedTaskId='';renderTaskList();}
function taskDeleteHtml(t){
 const eps=t.episodes||[];
 const devs=new Map();  // device_id -> {name, online} across all this task's episodes
 for(const ep of eps)for(const m of Object.values(ep.members||{}))if(m&&m.device_id)devs.set(m.device_id,{name:m.name,online:deviceOnline(m.device_id)});
 const list=[...devs.values()].map(d=>`<span class="ds-dev"><span class="ds-dot ${d.online?'on':'off'}"></span>${esc(d.name)}${d.online?'':' <span class="muted">offline</span>'}</span>`).join('');
 const offline=[...devs.values()].filter(d=>!d.online).length;
 return `<div class="subpanel task-del-panel">
   <div class="subpanel-title danger">Delete “${esc(t.name)}”?</div>
   <div class="del-warn">This permanently deletes the task and its <b>${eps.length}</b> recorded episode(s) on:</div>
   <div class="ds-devlist">${list||'<span class="muted">no recording devices</span>'}</div>
   ${offline?`<div class="del-warn muted">${offline} device(s) offline — they keep their copy until they reconnect, then you'll be prompted to finish the cleanup.</div>`:''}
   <div class="editor-actions">
    <button class="del-confirm" onclick='confirmDeleteTask("${t.id}")'>Delete permanently</button>
    <button class="cancel" onclick='deleteTask("${t.id}")'>Cancel</button>
   </div>
  </div>`;}
async function confirmDeleteTask(id){
 const r=await fetch(`/api/fleet/tasks/${id}`,{method:'DELETE'});
 if(!r.ok){const j=await r.json().catch(()=>({}));alert(errText(j)||'Failed to delete task');return;}
 taskDeleteId='';await refresh();}
function sigBadges(sig){return (sig||[]).map(s=>`<span class="pill hand" title="${s} role">${s[0].toUpperCase()}</span>`).join(' ');}
function warnIcon(){return '<svg class="warn-ico" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M10.29 3.86 1.82 18a2 2 0 0 0 1.71 3h16.94a2 2 0 0 0 1.71-3L13.71 3.86a2 2 0 0 0-3.42 0z"/><line x1="12" y1="9" x2="12" y2="13"/><line x1="12" y1="17" x2="12.01" y2="17"/></svg>';}
function toggleIssues(id){issuesOpenId=issuesOpenId===id?'':id;renderTaskList();}
function toggleGroups(id){groupsOpenId=groupsOpenId===id?'':id;renderTaskList();}
function toggleFleet(){fleetOpen=!fleetOpen;$('fleet-body').style.display=fleetOpen?'':'none';$('fleet-caret').textContent=fleetOpen?'▾':'▸';}
// Header recap (accordion collapsed): count of each device type ONLINE now.
function renderFleetRecap(){
 let lg=0,rg=0,grip=0,casq=0,busy=0;
 for(const d of DEVICES){
  if(!d.online)continue;
  if(d.activity==='uploading'||d.activity==='processing')busy++;
  const k=kindOf(d);
  if(k==='grabette'){if(d.hand==='left')lg++;else if(d.hand==='right')rg++;}
  else if(k==='gripette')grip++;
  else if(k==='casquette')casq++;
 }
 const chip=(label,n,title)=>`<span class="fleet-chip${n?'':' zero'}" title="${title}">${label} ${n}</span>`;
 // Busy chip is visible even with the Fleet accordion collapsed (e.g. from Record).
 const busyChip=busy?`<span class="fleet-chip busy" title="${busy} device(s) busy with a dataset upload/conversion">⚙ ${busy} busy</span>`:'';
 $('fleet-recap').innerHTML=chip('L',lg,'lgrabettes online')+chip('R',rg,'rgrabettes online')+chip('C',casq,'casquettes online')+chip('Grip',grip,'gripettes online')+busyChip;}
// Distinct device combinations used to record a task, with episode count each.
// Built from the connected devices' reported episodes (their members map).
function taskCombos(t){
 const m=new Map();
 for(const ep of (t.episodes||[])){
  const entries=Object.entries(ep.members||{}).sort((a,b)=>a[0].localeCompare(b[0]));
  if(!entries.length)continue;
  const key=entries.map(([r,mm])=>r+':'+(mm.device_id||'?')).join('|');
  let g=m.get(key);
  if(!g){g={roles:entries.map(([r,mm])=>({role:r,name:mm.name,online:deviceOnline(mm.device_id)})),count:0};m.set(key,g);}
  g.count++;
 }
 return [...m.values()];}
function comboRowHtml(g){
 const devs=g.roles.map(r=>`<span class="pill hand" title="${r.role}">${r.role[0].toUpperCase()}</span> ${esc(r.name)}${r.online?'':' <span class="muted">offline</span>'}`).join(' · ');
 return `<div class="hist-row">${devs} — <b>${g.count}</b> episode${g.count>1?'s':''}</div>`;}
function toggleTaskDetail(id){
 if(expandedTaskId===id){expandedTaskId='';}
 else{expandedTaskId=id;launchRolePick={left:'',right:'',casquette:''};historyOpenId='';}
 renderTaskList();}
function toggleHistory(id){historyOpenId=historyOpenId===id?'':id;renderTaskList();}
function fmtTs(sec){try{return new Date(sec*1000).toLocaleString();}catch(e){return '';}}
// ── Dataset selection ──
function sigKey(sig){return (sig||[]).slice().sort().join('+');}
function datasetLockedSig(){for(const t of TASKS){if(datasetSel.has(t.id))return sigKey(t.device_signature);}return '';}
function useOnlySet(){return datasetUseOnly.length>0;}
function sameRoles(a,b){return a.length===b.length&&a.every(r=>b.includes(r));}
function taskHasRoles(t,roles){const sig=new Set(t.device_signature||[]);return roles.every(r=>sig.has(r));}
// A task can be picked for the dataset if: (with "use only") its signature is a
// superset of the chosen roles; else (default) it matches the first pick's exact
// signature. This is what lets a bimanual task be used for a left-only dataset.
function datasetCompatible(t){
 if(useOnlySet())return taskHasRoles(t,datasetUseOnly);
 const locked=datasetLockedSig();
 return !locked||sigKey(t.device_signature)===locked;}
// Roles actually used to build the dataset: the "use only" subset, or null=all.
function activeRoles(){return useOnlySet()?datasetUseOnly:null;}
function setUseOnly(i){
 const roles=USE_ONLY_OPTS[i].roles;
 datasetUseOnly=sameRoles(datasetUseOnly,roles)?[]:roles.slice();  // click active = clear
 // Drop any now-incompatible selection under the new constraint.
 for(const id of [...datasetSel]){const t=TASKS.find(x=>x.id===id);if(!t||!datasetCompatible(t))datasetSel.delete(id);}
 renderTaskList();}
function renderAdvanced(){
 const el=$('ds-advanced');
 if(!selectMode||!TASKS.length){el.style.display='none';return;}
 el.style.display='block';
 const opts=USE_ONLY_OPTS.map((o,i)=>`<button class="use-only${sameRoles(datasetUseOnly,o.roles)?' active':''}" onclick="setUseOnly(${i})">${o.label}</button>`).join('');
 el.innerHTML=`<div class="adv-toggle" onclick="datasetAdvOpen=!datasetAdvOpen;renderAdvanced()">${datasetAdvOpen?'▾':'▸'} Advanced options</div>`+
  (datasetAdvOpen?`<div class="adv-body"><span class="muted">Use only:</span> ${opts}`+
   `<div class="adv-hint muted">Includes any task that has at least these devices, and uploads only these to build the dataset.</div></div>`:'');}
// Switch the task list between the two segmented modes. Idempotent: re-clicking
// the active segment is a no-op (so a dataset in progress isn't reset).
function setSelectMode(on){
 on=!!on;
 if(on===selectMode)return;
 selectMode=on;datasetSel.clear();expandedTaskId='';datasetJob=null;
 // Reset any open task editor/delete/create panel so it doesn't linger across modes.
 taskEditId=null;taskDeleteId='';$('task-editor').style.display='none';
 if(datasetPollTimer){clearTimeout(datasetPollTimer);datasetPollTimer=null;}
 if(selectMode){datasetName='';datasetPrivate=false;datasetOnlyAvailable=false;datasetUseOnly=[];datasetAdvOpen=false;if(datasetNamespaces===null)fetchNamespaces();}
 $('seg-record').classList.toggle('active',!selectMode);
 $('seg-dataset').classList.toggle('active',selectMode);
 // Drive the per-mode accent colour of the numbers/counts/encart below.
 $('tasks-card').classList.toggle('mode-dataset',selectMode);
 $('tasks-card').classList.toggle('mode-record',!selectMode);
 $('btn-create-task').style.display=selectMode?'none':'inline-block';
 renderTaskList();}  // renderTaskList sets the step-1 header text for the mode
async function fetchNamespaces(){
 try{
  const r=await fetch('/api/fleet/namespaces');
  if(r.ok){const j=await r.json();datasetNamespaces=j.namespaces||[];
   if(!datasetOwner)datasetOwner=j.default||datasetNamespaces[0]||'';
   renderDatasetBar();}
 }catch(e){}
}
function toggleDatasetPick(id){
 const t=TASKS.find(x=>x.id===id);if(!t)return;
 if(datasetSel.has(id)){datasetSel.delete(id);}
 else{if(!datasetCompatible(t))return;datasetSel.add(id);}
 renderTaskList();}
// ── Dataset device involvement (computed live from the tasks' reported episodes + DEVICES) ──
function deviceOnline(id){const d=DEVICES.find(x=>x.device_id===id);return !!(d&&d.online);}
function datasetEpisodes(){ // every episode across the selected tasks (from device reports)
 const eps=[];
 for(const t of TASKS){if(!datasetSel.has(t.id))continue;for(const ep of (t.episodes||[]))eps.push(ep);}
 return eps;}
function datasetDevices(){ // Map device_id -> role, union over selected episodes (only the active roles)
 const m=new Map();const only=activeRoles();
 for(const ep of datasetEpisodes())for(const [role,dev] of Object.entries(ep.roles||{})){
  if(only&&!only.includes(role))continue;
  m.set(dev,role);}
 return m;}
function datasetEpisodeCounts(){ // total, and how many have ALL their (active-role) devices online
 let total=0,avail=0;const only=activeRoles();
 for(const ep of datasetEpisodes()){
  const d=Object.entries(ep.roles||{}).filter(([r])=>!only||only.includes(r)).map(([,dev])=>dev);
  if(!d.length)continue;
  total++;if(d.every(deviceOnline))avail++;}
 return {total,avail};}
function datasetDevicesHtml(){
 const m=datasetDevices();
 if(!m.size)return '';
 const items=[...m.entries()].map(([id,role])=>{
  const on=deviceOnline(id);
  return `<span class="ds-dev"><span class="ds-dot ${on?'on':'off'}"></span>${esc(deviceName(id))}`+
   ` <span class="pill hand">${role[0].toUpperCase()}</span>${on?'':' <span class="muted">offline</span>'}</span>`;
 }).join('');
 return `<span class="ds-devlabel muted">Devices needed:</span><span class="ds-devlist">${items}</span>`;}
function renderDatasetBar(){
 const bar=$('dataset-bar');
 if(!selectMode){bar.style.display='none';bar.innerHTML='';return;}
 bar.style.display='flex';
 const sel=TASKS.filter(t=>datasetSel.has(t.id));
 const counts=datasetEpisodeCounts();
 const epText=datasetOnlyAvailable?`${counts.avail} / ${counts.total} episodes`:`${counts.total} episodes`;
 // With "use only" the dataset's roles are the chosen subset, not the task sig.
 const sig=useOnlySet()?datasetUseOnly:(sel.length?sel[0].device_signature:[]);
 const infoHtml=`${sel.length} task(s) · ${sigBadges(sig)||'<span class="muted">no device set</span>'} · ${epText}`;
 let jobCls='',jobHtml='';
 if(datasetJob){
  const m=esc(datasetJob.message||datasetJob.status||'');
  if(datasetJob.status==='error'){jobCls='err';jobHtml='✗ '+m;}
  else if(datasetJob.status==='done'){jobCls='ok';jobHtml='✓ '+m+(datasetJob.result_url?` — <a href="${datasetJob.result_url}" target="_blank" rel="noopener">open</a>`:'');}
  else{jobHtml='… '+m;}
 }
 // Build the controls once (so the name <input> keeps focus/value across the 3s
 // refresh re-renders); afterwards only update the dynamic text/state.
 if(!$('ds-name')){
  bar.innerHTML=
   `<div class="ds-step"><span class="step-num">2</span> Name your destination repository</div>`+
   `<span class="ds-info" id="ds-info">${infoHtml}</span>`+
   `<div id="ds-devices"></div>`+
   `<label class="ds-avail"><input type="checkbox" id="ds-avail" ${datasetOnlyAvailable?'checked':''} onchange="datasetOnlyAvailable=this.checked;renderDatasetBar()"> Use only available devices</label>`+
   `<div class="ds-field"><label class="ds-label" for="ds-name">Dataset name <span class="ds-req">*</span></label>`+
   `<span class="ds-target"><select id="ds-owner" onchange="datasetOwner=this.value"></select>`+
   `<span class="ds-slash">/</span>`+
   `<input id="ds-name" placeholder="dataset-name" value="${esc(datasetName)}" oninput="datasetName=this.value;$('ds-gen').disabled=!dsCanGen()"></span></div>`+
   `<label class="ds-private"><input type="checkbox" id="ds-private" ${datasetPrivate?'checked':''} onchange="datasetPrivate=this.checked"> Private</label>`+
   `<button class="validate" id="ds-gen" onclick="generateDataset()">Generate LeRobot dataset</button>`+
   `<div class="ds-progress" id="ds-progress" style="display:none"><div class="ds-progress-fill"></div></div>`+
   `<div class="ds-job" id="ds-job"></div>`;
 }else{
  $('ds-info').innerHTML=infoHtml;
 }
 $('ds-devices').innerHTML=datasetDevicesHtml();
 // Owner options: (re)fill when the namespace list changes; keep the selection.
 const ownerSel=$('ds-owner');
 const opts=(datasetNamespaces&&datasetNamespaces.length)?datasetNamespaces:(datasetOwner?[datasetOwner]:[]);
 if(ownerSel.options.length!==opts.length)
  ownerSel.innerHTML=opts.map(n=>`<option value="${esc(n)}">${esc(n)}</option>`).join('');
 if(datasetOwner)ownerSel.value=datasetOwner;
 $('ds-gen').disabled=!dsCanGen();
 const jobEl=$('ds-job');jobEl.className='ds-job'+(jobCls?' '+jobCls:'');jobEl.innerHTML=jobHtml;
 // Progress bar: determinate during upload (fraction of devices done), animated
 // (indeterminate) during the opaque Space conversion, full & green when done.
 const prog=$('ds-progress'),fill=prog.querySelector('.ds-progress-fill');
 if(!datasetJob||datasetJob.status==='error'){prog.style.display='none';}
 else{
  prog.style.display='block';
  const done=datasetJob.status==='done';
  // Same animated (indeterminate) bar for both work phases (upload + convert);
  // full green only when done. Each phase can take several minutes.
  const indet=!done;
  prog.className='ds-progress'+(indet?' indet':'')+(done?' done':'');
  fill.style.width=indet?'':'100%';
 }}
// Generate requires: ≥1 task, a name, no running job, at least one usable
// episode, and — in default mode — every needed device online (else the backend
// refuses). With "use only available devices", offline devices are tolerated
// (their episodes are dropped) as long as ≥1 episode remains.
function dsCanGen(){
 const busy=datasetJob&&(datasetJob.status==='uploading'||datasetJob.status==='processing');
 if(busy||datasetSel.size===0||!(datasetName||'').trim())return false;
 const c=datasetEpisodeCounts();
 if(datasetOnlyAvailable)return c.avail>0;
 if(c.total===0)return false;
 for(const id of datasetDevices().keys())if(!deviceOnline(id))return false;
 return true;}
async function generateDataset(){
 const ids=[...datasetSel];
 if(!ids.length)return;
 const name=(datasetName||'').trim();
 if(!name){alert('Enter a dataset name');return;}
 const target=datasetOwner?datasetOwner+'/'+name:name;
 const r=await fetch('/api/fleet/lerobot-dataset',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({task_ids:ids,name:target,private:datasetPrivate,only_available:datasetOnlyAvailable,roles:datasetUseOnly})});
 const j=await r.json().catch(()=>({}));
 if(!r.ok){alert(errText(j)||'Failed to start dataset generation');return;}
 datasetJob={status:'uploading',message:'Starting… (this can take several minutes)'};renderDatasetBar();
 pollDatasetJob(j.job_id);}
function pollDatasetJob(jobId){
 if(datasetPollTimer)clearTimeout(datasetPollTimer);
 const step=async()=>{
  try{
   const r=await fetch('/api/fleet/lerobot-dataset/'+jobId);
   if(r.ok){datasetJob=await r.json();renderDatasetBar();
    if(datasetJob.status==='done'||datasetJob.status==='error'||datasetJob.status==='raw_ready')return;}
  }catch(e){}
  datasetPollTimer=setTimeout(step,2000);
 };
 step();}
// Per-role device dropdowns for a task's required devices. Selected values are
// baked in as `selected` attributes so the 3s refresh re-render keeps the picks.
function pickRole(role,val){launchRolePick[role]=val;renderTaskList();}
function launchRolesHtml(t){
 return (t.device_signature||[]).map(role=>{
  const opts=DEVICES.filter(d=>slotOf(d)===role).map(d=>{
   const sel=launchRolePick[role]===d.device_id?' selected':'';
   // Offline devices can't be picked — a session needs every device online.
   // Nor can a device busy with a dataset upload/conversion.
   const busy=deviceBusyForRec(d.device_id);
   const dis=(!d.online||busy)?' disabled':'';
   const suffix=!d.online?' (offline)':busy?` (${deviceActivity(d.device_id)==='processing'?'converting':'uploading'})`:'';
   return `<option value="${d.device_id}"${sel}${dis}>${esc(d.name)}${suffix}</option>`;
  }).join('');
  const none=launchRolePick[role]?'':' selected';
  return `<label class="role-pick"><span class="sig-label">${ROLE_LABEL[role]}</span>`+
   `<select class="role-select" onchange="pickRole('${role}',this.value)">`+
   `<option value=""${none}>— select —</option>${opts}</select></label>`;
 }).join('');}
// Expanded task detail: the session launcher (role dropdowns + Launch button),
// then a collapsed-by-default session history.
function taskDetailHtml(t){
 const running=SESSIONS.some(s=>s.status==='open');
 // Episodes recorded for this task, newest first — sourced from the devices'
 // reports so they show (and name their peers, even offline ones) for ANY
 // operator, regardless of which account did the acquisition.
 const eps=(t.episodes||[]).slice().sort((a,b)=>b.episode_id.localeCompare(a.episode_id));
 const roles=t.device_signature||[];
 const missing=roles.filter(role=>!launchRolePick[role]);
 const offlinePick=roles.filter(role=>launchRolePick[role]&&!deviceOnline(launchRolePick[role]));
 const busyPick=roles.filter(role=>launchRolePick[role]&&deviceBusyForRec(launchRolePick[role]));
 let launchBtn;
 if(running){
  launchBtn=`<button class="validate" disabled>Launch session</button><span class="muted launch-hint">A session is already running — close it first.</span>`;
 }else if(missing.length){
  launchBtn=`<button class="validate" disabled>Launch session</button><span class="muted launch-hint">Select a device for every role — required to launch.</span>`;
 }else if(offlinePick.length){
  launchBtn=`<button class="validate" disabled>Launch session</button><span class="muted launch-hint">A selected device is offline — every device must be online.</span>`;
 }else if(busyPick.length){
  launchBtn=`<button class="validate" disabled>Launch session</button><span class="muted launch-hint">A selected device is busy processing a dataset — wait for it to finish.</span>`;
 }else{
  launchBtn=`<button class="validate" onclick='launchSession("${t.id}")'>Launch session</button>`;
 }
 const histOpen=historyOpenId===t.id;
 const epRows=eps.length?eps.map(ep=>{
  const names=Object.entries(ep.members||{}).map(([role,m])=>`<span class="pill hand" title="${role}">${role[0].toUpperCase()}</span> ${esc(m.name)}${deviceOnline(m.device_id)?'':' <span class="muted">offline</span>'}`).join(' ')||'<span class="muted">no devices recorded</span>';
  // Episodes saved before the device persisted their members show up incomplete;
  // repair is bulk, at the task level (see the "Fill devices" button below).
  const mark=ep.incomplete?` <span class="fill-ep-hint" title="Recorded before members were saved — use “Fill devices” to repair">${warnIcon()} incomplete</span>`:'';
  return `<div class="hist-row"><span class="muted">${fmtTs(ep.started_at)}</span> · ${names}${mark}</div>`;
 }).join(''):'<div class="muted" style="font-size:.82rem">No episode recorded yet.</div>';
 // Task-level repair: count incomplete episodes and how many can be filled now.
 const incN=eps.filter(e=>e.incomplete).length;
 const fillN=eps.filter(e=>e.fillable).length;
 const fillCtrl=fillN
  ?`<button class="fill-task" title="Backfill who recorded these episodes onto the connected grabettes" onclick='fillTaskDevices("${t.id}")'>Fill devices (${fillN})</button>`
  :(incN?`<span class="fill-ep-hint" title="Connect the grabettes that recorded these episodes to repair them">${warnIcon()} ${incN} incomplete — connect their grabettes</span>`:'');
 // Pairing-issues accordion — only for a task that currently has orphans. Same
 // style as the global banner, nested under the Episodes accordion.
 const g=orphanFor(t.name);
 const issuesOpen=issuesOpenId===t.id;
 const issues=g?`<div class="hist-toggle warn" onclick='toggleIssues("${t.id}")'>${issuesOpen?'▾':'▸'} ${warnIcon()} Pairing issues (${g.count})</div>${issuesOpen?orphanCardHtml(g):''}`:'';
 // Device groups: which device combinations recorded this task, and how many
 // episodes each. From connected devices' reports (same scope as Episodes).
 const combos=taskCombos(t);
 const groupsOpen=groupsOpenId===t.id;
 const groups=combos.length?`<div class="hist-toggle" onclick='toggleGroups("${t.id}")'>${groupsOpen?'▾':'▸'} Device groups (${combos.length})</div>${groupsOpen?combos.map(comboRowHtml).join(''):''}`:'';
 const connNote='<span class="muted" style="font-weight:400"> · from connected devices</span>';
 return `<div class="task-detail">
   <div class="ds-step" style="margin:.2rem 0 .6rem"><span class="step-num">2</span> Select the devices to use</div>
   <div class="task-launch">${launchRolesHtml(t)}<div class="editor-actions">${launchBtn}</div></div>
   ${groups}
   <div class="hist-head"><div class="hist-toggle" onclick='toggleHistory("${t.id}")'>${histOpen?'▾':'▸'} Episodes (${eps.length})${connNote}</div>${fillCtrl}</div>
   ${histOpen?epRows:''}
   ${issues}
  </div>`;}
function taskEpsHtml(total){return `<div class="task-eps"><span class="ep-count" title="Total episodes recorded for this task">${total} ${total===1?'episode':'episodes'}</span></div>`;}
function taskSubHtml(t){return `<div class="task-sub"><span class="muted">Required devices:</span> ${sigBadges(t.device_signature)}${t.description?` · ${esc(t.description)}`:''}</div>`;}
function renderTaskList(){
 // Don't clobber an inline edit-in-progress on the periodic refresh — rebuilding
 // innerHTML would wipe what the user is typing. Skip while its editor is live.
 if(taskEditId && $('edit-name'))return;
 const el=$('task-list');$('count-tasks').textContent=TASKS.length;
 const step1=$('ds-step1');
 if(!TASKS.length){el.className='muted empty';el.textContent='No tasks yet.';step1.style.display='none';renderAdvanced();renderDatasetBar();return;}
 el.className='';
 // Step 1 header, styled the same in both modes but worded for the task at hand.
 step1.style.display='flex';
 step1.innerHTML='<span class="step-num">1</span> '+(selectMode?'Select tasks to include':'Select the task to record');
 renderAdvanced();
 if(selectMode){
  el.innerHTML=TASKS.map(t=>{
   const checked=datasetSel.has(t.id);
   const disabled=!checked&&!datasetCompatible(t);
   return `<div class="task-block">
    <div class="task-row select${disabled?' disabled':''}" ${disabled?'':`onclick='toggleDatasetPick("${t.id}")'`}>
     <div class="task-top">
      <input type="checkbox" class="task-check" ${checked?'checked':''} ${disabled?'disabled':''}>
      <span class="tname">${esc(t.name)}</span>
     </div>
     ${taskEpsHtml(t.episode_count||0)}
     ${taskSubHtml(t)}
    </div>
   </div>`;
  }).join('');
  renderDatasetBar();return;
 }
 el.innerHTML=TASKS.map(t=>{
  const editing=taskEditId===t.id;
  const deleting=taskDeleteId===t.id;
  const detail=editing?taskEditorHtml(t):(deleting?taskDeleteHtml(t):(expandedTaskId===t.id?taskDetailHtml(t):''));
  const hasIssues=!!orphanFor(t.name);
  return `<div class="task-block">
    <div class="task-row" ${(editing||deleting)?'':`onclick='toggleTaskDetail("${t.id}")'`}>
     <div class="task-top">
      <span class="tname">${esc(t.name)}${hasIssues?`<span class="task-warn" title="Episode pairing issues">${warnIcon()}</span>`:''}</span>
      <div class="task-acts">
       <button class="act-edit" title="Edit task" onclick='event.stopPropagation();editTask("${t.id}")'>
        <svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M12 20h9"/><path d="M16.5 3.5a2.1 2.1 0 0 1 3 3L7 19l-4 1 1-4Z"/></svg>
       </button>
       <button class="del-icon" title="Delete task" onclick='event.stopPropagation();deleteTask("${t.id}")'>
        <svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg>
       </button>
      </div>
     </div>
     ${taskEpsHtml(t.episode_count||0)}
     ${taskSubHtml(t)}
    </div>${detail}
   </div>`;
 }).join('');
 renderDatasetBar();}
// ── Sessions ──
// The launcher now lives inside each task's expanded detail (taskDetailHtml):
// pick a device per required role, then Launch. A single session runs at a time.
async function launchSession(taskId){
 const t=TASKS.find(x=>x.id===taskId);if(!t)return;
 const body={task_id:taskId};
 for(const role of (t.device_signature||[])){
  const id=launchRolePick[role];
  if(!id){alert('Select a device for '+ROLE_LABEL[role]);return;}
  if(!deviceOnline(id)){alert(deviceName(id)+' is offline — a session needs every device online.');return;}
  body[role]=id;
 }
 const r=await fetch('/api/fleet/sessions',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify(body)});
 if(!r.ok){const j=await r.json().catch(()=>({}));alert(errText(j)||'Failed to launch session');return;}
 expandedTaskId='';launchRolePick={left:'',right:'',casquette:''};await refresh();}
// Single record toggle (phone-camera style): start when idle, stop when
// recording/initializing; ignored mid-stop.
function episodeToggle(id){
 const s=SESSIONS.find(x=>x.id===id);if(!s)return;
 const p=sessionPhase(s);
 if(p==='idle')episodeStart(id);
 else if(p==='recording'||p==='initializing')episodeStop(id);}
async function episodeStart(id){
 const r=await fetch(`/api/fleet/sessions/${id}/episode/start`,{method:'POST'});
 const j=await r.json().catch(()=>({}));
 if(!r.ok){alert(errText(j)||'Failed to start episode');return;}
 await refresh();}  // the "Initializing…" state shows the pre-T0 wait — no alert needed
async function episodeStop(id){await fetch(`/api/fleet/sessions/${id}/episode/stop`,{method:'POST'});await refresh();}
async function deleteLastEpisode(id){
 if(!confirm('Delete the last episode of this session on ALL its devices? This cannot be undone.'))return;
 const r=await fetch(`/api/fleet/sessions/${id}/episode/delete-last`,{method:'POST'});
 if(!r.ok){const j=await r.json().catch(()=>({}));alert(errText(j)||'Failed to delete episode');return;}
 await refresh();}
async function fillTaskDevices(taskId){
 const r=await fetch(`/api/fleet/tasks/${encodeURIComponent(taskId)}/fill-devices`,{method:'POST'});
 if(!r.ok){const j=await r.json().catch(()=>({}));alert(errText(j)||'Failed to fill devices');return;}
 await refresh();}
async function closeSession(id){if(!confirm('Close this session?'))return;await fetch(`/api/fleet/sessions/${id}/close`,{method:'POST'});await refresh();}
async function deleteSession(id){if(!confirm('Delete this session and its manifest?'))return;await fetch(`/api/fleet/sessions/${id}`,{method:'DELETE'});await refresh();}
// The session's capture phase, mirroring the grabette LEDs (off/blink/solid/
// fast-blink): idle → nothing running; initializing → episode dispatched, still
// before the shared T0 (devices warming/waiting); recording → past T0; stopping
// → devices tearing down + muxing. initializing↔recording is derived from T0;
// stopping is reported by the fleet and ends when the devices actually finish
// (s.stopping is cleared as each stop_capture reports its result).
function sessionPhase(s){
 if(s.recording){
  const ep=(s.episodes&&s.episodes.length)?s.episodes[s.episodes.length-1]:null;
  const t0=ep&&ep.start_at_utc?Date.parse(ep.start_at_utc):0;
  return (t0&&Date.now()<t0)?'initializing':'recording';
 }
 return s.stopping?'stopping':'idle';}
function phasePill(s){
 // State pill for the panel header — the recording duration lives above the
 // record button, not in the pill.
 const p=sessionPhase(s);
 if(p==='recording')return '<span class="pill rec">● Recording</span>';
 if(p==='initializing')return '<span class="pill init">◌ Initializing…</span>';
 if(p==='stopping')return '<span class="pill stopping">◍ Stopping…</span>';
 return '<span class="pill idle">○ Idle</span>';}
// Episodes that lost their pair (a peer deleted the task while this device was
// offline; on reconnect its copies are orphaned). Cleanup deletes only the ones
// whose deleting peer is back online — those still paired with an offline device
// are left untouched.
function orphanFor(name){return ORPHANS.find(g=>g.task===name)||null;}  // group for a task, or null
function orphanCardHtml(g){
 const i=ORPHANS.indexOf(g);
 const dels=g.deleted_by.map(esc).join(', ')||'a peer';
 const hold=g.holders.map(esc).join(', ');
 const head=g.whole_task
  ?`<b>${esc(g.task)}</b> — task deleted on <b>${dels}</b>; ${g.count} episode(s) still here`
  :`<b>${esc(g.task)}</b> — ${g.count} episode(s) lost their pair`;
 const detail=g.whole_task
  ?`This task was deleted (with its recordings) on <b>${dels}</b>, but ${g.count} orphaned episode(s) remain on <b>${esc(hold)}</b>.`
  :`Deleted on <b>${dels}</b>, still present on <b>${esc(hold)}</b>.`;
 const btn=g.whole_task
  ?`Delete the task and its ${g.count} episode(s) on ${esc(hold)}`
  :`Delete the ${g.count} orphaned episode(s) on ${esc(hold)}`;
 return `<div class="orphan-item">
    <div class="orphan-head">${head}</div>
    <div class="orphan-detail">${detail}</div>
    <button class="del-confirm" onclick="cleanupOrphans(${i})">${btn}</button>
   </div>`;}
// Collapsed accordion under the tabs summarising all pairing issues.
function renderOrphans(){
 const bar=$('orphan-bar');
 if(!ORPHANS.length){bar.style.display='none';bar.innerHTML='';orphansOpen=false;return;}
 bar.style.display='block';
 const n=ORPHANS.length;
 bar.innerHTML=`<div class="orphan-acc">
   <div class="orphan-acc-head" onclick="orphansOpen=!orphansOpen;renderOrphans()">
    <span>${warnIcon()} ${n} task${n>1?'s':''} with episode pairing issues</span>
    <span class="orphan-caret">${orphansOpen?'▾':'▸'}</span>
   </div>
   ${orphansOpen?`<div class="orphan-acc-body">${ORPHANS.map(orphanCardHtml).join('')}</div>`:''}
  </div>`;}
async function cleanupOrphans(i){
 const g=ORPHANS[i];if(!g)return;
 if(!confirm(`Permanently delete ${g.episode_ids.length} orphaned episode(s) of “${g.task}”? Their pair was already removed, so they can't complete a dataset.`))return;
 const r=await fetch('/api/fleet/orphans/cleanup',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({episode_ids:g.episode_ids})});
 if(!r.ok){alert('Failed to clean up orphaned episodes');return;}
 await refresh();}
function renderSessionList(){
 // A single running session REPLACES the task list: when one is open, show the
 // session card and hide the mode tabs + Tasks card (the Fleet accordion in the
 // sticky bar stays); otherwise restore them and hide the (empty) session card.
 const open=SESSIONS.filter(s=>s.status==='open');
 const el=$('session-list'),card=$('sessions-card'),tasksCard=$('tasks-card'),tabs=$('page-tabs');
 if(!open.length){
  card.style.display='none';
  tasksCard.style.display='';
  if(tabs)tabs.style.display='';
  return;
 }
 card.style.display='';
 tasksCard.style.display='none';
 if(tabs)tabs.style.display='none';
 el.className='';
 el.innerHTML=open.map(s=>{
  const devs=Object.entries(s.members).map(([role,m])=>{
   const url=`http://${encodeURIComponent(m.name)}.local:8000`;
   return `<span class="sp-dev"><span class="pill hand" title="${role}">${role[0].toUpperCase()}</span> ${esc(m.name)}${deviceOnline(m.device_id)?batteryPill(deviceBattery(m.device_id)):''}${deviceBusyForRec(m.device_id)?activityBadge(m.device_id):''}`+
    ` <a class="dash-icon" target="_blank" rel="noopener" title="Open ${esc(m.name)} dashboard" href="${url}"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M18 13v6a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2V8a2 2 0 0 1 2-2h6"/><polyline points="15 3 21 3 21 9"/><line x1="10" y1="14" x2="21" y2="3"/></svg></a></span>`;
  }).join('');
  const p=sessionPhase(s);
  const ep=(s.episodes&&s.episodes.length)?s.episodes[s.episodes.length-1]:null;
  const t0=ep&&ep.start_at_utc?Date.parse(ep.start_at_utc):0;
  // Count only FINISHED episodes: the in-progress one (recording) isn't counted yet.
  const done=Math.max(0,(s.episode_count||0)-(s.recording?1:0));
  const M={
   idle:{lbl:'Idle',cls:'idle',btn:'idle',dis:'',title:'Start episode'},
   initializing:{lbl:'Initializing…',cls:'init',btn:'busy',dis:'',title:'Cancel'},
   recording:{lbl:'Recording',cls:'rec',btn:'recording',dis:'',title:'Stop episode'},
   stopping:{lbl:'Stopping…',cls:'stopping',btn:'busy',dis:'disabled',title:'Stopping…'},
  }[p];
  const timer=(p==='recording'&&t0)?`<span class="sp-timer rec-dur" data-start="${t0}">${fmtDur(Date.now()-t0)}</span>`:'';
  const task=esc(s.task_name||'(task deleted)');
  // A member busy with a dataset upload/conversion blocks STARTING a new episode
  // (mirrors the server gate); an in-progress recording is left alone.
  const busyMembers=Object.values(s.members).map(m=>m.device_id).filter(deviceBusyForRec);
  const recBlocked=(p==='idle')&&busyMembers.length>0;
  const circDis=M.dis||(recBlocked?'disabled':'');
  const circTitle=recBlocked?'A device is busy processing a dataset — wait for it to finish':M.title;
  const circle=`<div class="sp-rec"><button class="rec-toggle ${M.btn}${recBlocked?' blocked':''}" ${circDis} title="${circTitle}" onclick='episodeToggle("${s.id}")'><span class="rt-inner"></span></button></div>`;
  const busyNote=recBlocked?`<div class="sp-busy-note">${busyMembers.map(deviceName).join(', ')} busy processing a dataset — recording paused until it finishes.</div>`:'';
  const del=`<button class="del-ep" ${(p!=='idle'||!done)?'disabled':''} title="Delete the last episode on all its devices" onclick='deleteLastEpisode("${s.id}")'>Delete last episode</button>`;
  const close=`<button class="close-btn" onclick='closeSession("${s.id}")'>Close session</button>`;
  const epWord=`episode${done>1?'s':''}`;
  return `
   <div class="session-panel var-c">
    <div class="sp-rec-zone">
     <div class="sp-head"><span class="sp-task">${task}</span>${phasePill(s)}</div>
     <div class="sp-timer-wrap">${timer}</div>
     ${circle}
     ${busyNote}
     <div class="sp-rec-del">${del}</div>
    </div>
    <div class="sp-manage-zone">
     <div class="sp-section-label">Episodes</div>
     <div class="ep-big"><span class="ep-big-num">${done}</span><span class="ep-big-lbl">${epWord} recorded this session</span></div>
     <div class="sp-meta"><span class="sp-label">Devices</span><span class="sp-devs">${devs}</span></div>
     ${close}
    </div>
   </div>`;
 }).join('');
 tickRecDur();}
const openRaw=new Set();
function rawToggle(id,open){open?openRaw.add(id):openRaw.delete(id);}
const showState=new Set();
function runState(id){showState.add(id);dispatch(id,'get_state',{});}
function hideState(id){showState.delete(id);refresh();}
function kindOf(d){
 const s=`${d.name||''} ${d.device_id||''} ${(d.capabilities||[]).join(' ')}`.toLowerCase();
 if(s.includes('gripette'))return 'gripette';
 if(s.includes('casquette'))return 'casquette';
 return 'grabette';}
function btn(id,type,args,online){const dis=online?'':'disabled';return `<button class="primary" ${dis} onclick='dispatch("${id}","${type}",${JSON.stringify(args||{})})'>${type}</button>`;}
async function dispatch(device_id,type,args){
 await fetch('/api/fleet/dispatch',{method:'POST',headers:{'Content-Type':'application/json'},
  body:JSON.stringify({device_id,type,args})});refresh();}
async function removeDevice(device_id){
 await fetch(`/api/fleet/devices/${device_id}/remove`,{method:'POST'});refresh();}
async function checkLogin(){
 // Never leave #who stuck on "Checking…": if the request fails (the free-tier
 // Space is waking up, a redeploy is in flight, or a transient network error),
 // show a retrying message and let the 3s tick recover instead of throwing.
 try{
  const r=await fetch('/api/fleet/me');
  if(!r.ok)throw new Error('HTTP '+r.status);
  const m=await r.json();loggedIn=m.logged_in;
  if(loggedIn){$('who').innerHTML=`<div class="who-row"><span>Signed in as <b>${m.username}</b></span><a class="btn logout" href="/oauth/huggingface/logout">Logout</a></div>`;}
  else{$('who').innerHTML=`<a class="btn primary" href="/oauth/huggingface/login">Sign in with HuggingFace</a>`;}
 }catch(e){
  loggedIn=false;
  $('who').innerHTML=`<span class="muted">Connecting to the Space… <span style="opacity:.7">(it may be waking up — retrying)</span></span>`;
 }
 $('gated').disabled=!loggedIn;}
function esc(s){return String(s).replace(/[&<>]/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;'}[c]));}
function cap(s){return String(s).replace(/(^|_)([a-z])/g,(m,p,c)=>p+c.toUpperCase());}
function kv(k,v){return `<div class="k">${esc(k)}</div><div class="v">${v}</div>`;}
function fmtState(st){
 const rows=[];
 rows.push(kv('Daemon',`${esc(st.state||'?')}${st.backend?` · <span class="muted">${esc(st.backend)}</span>`:''}`));
 const sens=st.sensor||{};const cap=sens.capture||{};
 if(cap.is_capturing){
  const bits=[];
  if(cap.duration_seconds!=null)bits.push(`${Math.round(cap.duration_seconds)}s`);
  if(cap.frame_count!=null)bits.push(`${cap.frame_count} frames`);
  rows.push(kv('Capture',`<span class="pill rec">● Recording</span> ${esc(bits.join(' · '))}`));
  if(cap.session_id)rows.push(kv('Session',esc(cap.session_id)));
 }else if(cap.is_starting){
  rows.push(kv('Capture',`<span class="pill idle">○ Starting…</span>`));
 }else{
  rows.push(kv('Capture',`<span class="pill idle">○ Idle</span>`));
 }
 if(sens.angle){const deg=r=>(Number(r)*180/Math.PI).toFixed(1);
  rows.push(kv('Angle',`prox ${deg(sens.angle.proximal)}° · dist ${deg(sens.angle.distal)}°`));}
 if(st.error)rows.push(kv('Error',`<span class="err">${esc(st.error)}</span>`));
 return `<div class="kv">${rows.join('')}</div>`;}
function formatResult(type,res){
 if(res==null)return '<span class="muted">no result</span>';
 if(res.status==='error')return `<div class="err">⚠ ${esc(res.message||'error')}</div>`;
 if(type==='get_state'&&res.state)return fmtState(res.state);
 if(type==='start_capture')return `<div>▶ Capture started${res.episode_id?` · <span class="muted">${esc(res.episode_id)}</span>`:''}</div>`;
 if(type==='stop_capture'){const r=res.result||{};const bits=[];
  if(r.frame_count!=null)bits.push(`${r.frame_count} frames`);
  if(r.duration_seconds!=null)bits.push(`${Math.round(r.duration_seconds)}s`);
  return `<div>■ Capture stopped${bits.length?` · <span class="muted">${esc(bits.join(' · '))}</span>`:''}</div>`;}
 if(type==='logout')return '<div>Logged out</div>';
 return `<pre>${esc(JSON.stringify(res,null,1))}</pre>`;}
function rowHtml(d){
 const last=d.history[0];
 const caps=d.capabilities||[];
 const hasLogout=caps.includes('logout');
 const logoutDis=d.online?'':'disabled';
 const logoutBtn=hasLogout
  ?`<button class="logout-icon" title="Logout" ${logoutDis} onclick='dispatch("${d.device_id}","logout",{})'><svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M9 21H5a2 2 0 0 1-2-2V5a2 2 0 0 1 2-2h4"/><polyline points="16 17 21 12 16 7"/><line x1="21" y1="12" x2="9" y2="12"/></svg></button>`
  :'';
 const deleteBtn=`<button class="del-icon" title="Remove device" onclick='removeDevice("${d.device_id}")'><svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg></button>`;
 const dashUrl=`http://${encodeURIComponent(d.name)}.local:8000`;
 const dashBtn=`<a class="dash-icon" target="_blank" rel="noopener" title="Open device dashboard (${dashUrl})" href="${dashUrl}"><svg width="16" height="16" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round"><path d="M18 13v6a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2V8a2 2 0 0 1 2-2h6"/><polyline points="15 3 21 3 21 9"/><line x1="10" y1="14" x2="21" y2="3"/></svg></a>`;
 const handBadge=d.hand?`<span class="pill hand" title="${esc(d.hand)} hand">${d.hand==='left'?'L':d.hand==='right'?'R':esc(d.hand)}</span>`:'';
 const showRes=last&&!(last.type==='get_state'&&!showState.has(d.device_id));
 const rtime=last&&last.ts?new Date(last.ts*1000).toLocaleTimeString():'';
 const closeBtn=last&&last.type==='get_state'?`<button class="rclose" title="Close" onclick='hideState("${d.device_id}")'>×</button>`:'';
 const resultHtml=showRes
  ?`<div class="result"><div class="rlabel">${esc(cap(last.type))}${rtime?` · ${esc(rtime)}`:''}${closeBtn}</div>${formatResult(last.type,last.result)}`
   +`<details class="raw-d"${openRaw.has(d.device_id)?' open':''} ontoggle='rawToggle("${d.device_id}",this.open)'><summary>raw</summary><pre>${esc(JSON.stringify(last.result,null,1))}</pre></details></div>`
  :'<div class="result empty muted">No command run yet.</div>';
 return `<tr>
   <td class="c-dot"><span class="dot ${d.online?'on':'off'}" title="${d.online?'online':'offline'}"></span></td>
   <td class="name"><b>${d.name}</b>${handBadge}${d.online?batteryPill(d.battery):''}${d.online?activityBadge(d.device_id):''}<br><span class="muted">${d.device_id}${d.ip?` · ${esc(d.ip)}`:''}${d.pending?` · ${d.pending} pending`:''}</span>${resultHtml}</td>
   <td class="c-tools"><div class="tools">${dashBtn}${logoutBtn}${deleteBtn}</div></td></tr>`;}
async function refresh(){
 if(!loggedIn)return;
 const [dr,tr,sr,or_]=await Promise.all([fetch('/api/fleet/devices'),fetch('/api/fleet/tasks'),fetch('/api/fleet/sessions'),fetch('/api/fleet/orphans')]);
 if(dr.ok){DEVICES=(await dr.json()).devices;}
 if(tr.ok){TASKS=(await tr.json()).tasks;}
 if(sr.ok){SESSIONS=(await sr.json()).sessions;}
 if(or_.ok){ORPHANS=(await or_.json()).groups;}
 // Fleet tables (Devices must be applied before the launcher/roles render).
 const kinds={grabette:[],gripette:[],casquette:[]};
 for(const d of DEVICES)kinds[kindOf(d)].push(d);
 for(const k of KINDS){
  const list=kinds[k];
  $('tb-'+k).innerHTML=list.map(rowHtml).join('');
  $('count-'+k).textContent=list.length;
  $('empty-'+k).style.display=list.length?'none':'block';}
 renderFleetRecap();
 // Tasks (with inline launcher) + the running session.
 renderTaskList();
 renderSessionList();
 renderOrphans();}
async function tick(){
 try{await checkLogin();}catch(e){}
 try{await refresh();}catch(e){}
}
tick();setInterval(tick,3000);
// 1s UI tick. While a session is open, poll session state at 1s (not the 3s
// full refresh) so phase changes show promptly: initializing→recording (at T0)
// and, crucially, stopping→idle the moment the fleet marks the devices' stop
// complete — instead of lagging up to 3s. Devices/tasks stay on the 3s refresh.
async function uiTick(){
 if(SESSIONS.some(s=>s.status==='open')){
  try{const r=await fetch('/api/fleet/sessions');if(r.ok)SESSIONS=(await r.json()).sessions;}catch(e){}
  renderSessionList();
 }
 tickRecDur();
}
setInterval(uiTick,1000);
</script></body></html>"""