Spaces:
Sleeping
Sleeping
| """Per-session run state — one RunSession per browser session, never shared. | |
| Several people use the Space at the same time, each in their own Gradio session. | |
| Nothing about a run may leak from one session into another: not the cooperative | |
| control flags (pause / cancel), and above all not the flagged episodes published | |
| by the check / sync / trajectory review gates — those name someone else's dataset. | |
| So every flag that used to be a module-level global lives here, on a RunSession | |
| looked up by the caller's `gr.Request.session_hash`. Handlers annotate a | |
| `request: gr.Request` parameter (Gradio injects it — it is NOT a UI input) and | |
| call `session.get(request)`; the worker thread captures the object itself, so a | |
| run keeps working on its own flags even after its page is gone. | |
| """ | |
| import threading | |
| import time | |
| import gradio as gr | |
| # A session is dropped once its page has been silent for this long. The review | |
| # gr.Timer ticks twice a second while a page is open, so anything idle for hours | |
| # is a closed tab (Gradio's unload event isn't relied on — a killed tab or a lost | |
| # connection never fires it). | |
| _TTL_S = 6 * 3600 | |
| _sessions: dict[str, "RunSession"] = {} | |
| _lock = threading.Lock() | |
| class RunSession: | |
| """One user's run: cooperative flags + the review gate's published state. | |
| stop — abandon the run (Cancel confirmed) | |
| pause — hold the worker at the next safe checkpoint (Pause) | |
| cancel_pending — Cancel was clicked, awaiting confirmation | |
| review* — the gate the worker blocks on while the user triages the | |
| flagged episodes (see review.py) | |
| """ | |
| def __init__(self) -> None: | |
| self.stop = threading.Event() | |
| self.pause = threading.Event() | |
| self.cancel_pending = threading.Event() | |
| # A run has two sequential review gates — pre-SLAM (kind "input": | |
| # completeness + sync) and post-SLAM (kind "trajectory") — so the same | |
| # fields are reused twice; `review_kind` tells the panel which gate it is | |
| # currently drawing. | |
| self.review = threading.Event() # worker is awaiting the user's decision | |
| self.review_done = threading.Event() # Continue clicked → release the worker | |
| self.review_items: list[dict] = [] # flagged episodes published for the panel | |
| self.review_drop: list[str] = [] # episode names the user chose to drop | |
| self.review_kind = "trajectory" | |
| self.seen = time.monotonic() | |
| # ---- review gate ---------------------------------------------------- | |
| def review_state(self) -> dict: | |
| """Snapshot for the reactive @gr.render; `open` drives whether it shows.""" | |
| return {"open": self.review.is_set(), | |
| "kind": self.review_kind, | |
| "items": list(self.review_items), | |
| "dropped": list(self.review_drop)} | |
| def reset_review(self) -> None: | |
| """Clear every review flag (called by the controller's run reset).""" | |
| self.review.clear() | |
| self.review_done.clear() | |
| self.review_drop.clear() | |
| self.review_kind = "trajectory" | |
| # ---- control flags -------------------------------------------------- | |
| def start_run(self) -> None: | |
| """Fresh run: clear every flag left over from the previous one.""" | |
| self.stop.clear() | |
| self.pause.clear() | |
| self.cancel_pending.clear() | |
| self.reset_review() | |
| def wait_if_paused(self) -> None: | |
| """Block while paused (the worker's checkpoint gate). Returns at once if a | |
| stop/cancel is requested, so pausing then cancelling never deadlocks.""" | |
| while self.pause.is_set() and not self.stop.is_set(): | |
| time.sleep(0.2) | |
| def get(request: gr.Request | None) -> RunSession: | |
| """The RunSession for this browser session, created on first use. | |
| `request` is None only if Gradio couldn't inject one (it always can for event | |
| handlers); a shared fallback session is used then rather than crashing — it is | |
| the one case where two callers could meet, and it can't be reached from the UI. | |
| """ | |
| key = getattr(request, "session_hash", None) or "_fallback" | |
| now = time.monotonic() | |
| with _lock: | |
| for stale in [k for k, s in _sessions.items() if now - s.seen > _TTL_S]: | |
| del _sessions[stale] | |
| sess = _sessions.get(key) | |
| if sess is None: | |
| sess = _sessions[key] = RunSession() | |
| sess.seen = now | |
| return sess | |