| """Server-owned web runtime surviving browser reconnects. |
| |
| The PyBullet world remains the source of truth. This object persists only |
| session/UI metadata, conversation state, task lifecycle, and a bounded event |
| journal so a reloaded page can catch up without resetting the simulation. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import asyncio |
| from collections import deque |
| import threading |
| import time |
| import uuid |
|
|
| from llm_panda.conversation import ConversationManager |
| from llm_panda.execution.cancellation import CancellationToken |
|
|
|
|
| class WebRuntime: |
| def __init__(self, max_events: int = 750): |
| self.runtime_id = uuid.uuid4().hex |
| self.conversation_manager = ConversationManager( |
| session_id=self.runtime_id |
| ) |
| self._lock = threading.RLock() |
| self._subscribers = set() |
| self._events = deque(maxlen=max_events) |
| self._seq = 0 |
| self._generation = 0 |
| self._active = None |
| self._pending = None |
| self.execution_state = "idle" |
| self.latest_plan = None |
| self.latest_result = None |
| self.latest_state = None |
| self.latest_init = None |
| self.current_step = None |
|
|
| async def subscribe(self, websocket): |
| with self._lock: |
| self._subscribers.add(websocket) |
|
|
| async def unsubscribe(self, websocket): |
| with self._lock: |
| self._subscribers.discard(websocket) |
|
|
| def submit(self, prompt: str, use_mock: bool = False): |
| with self._lock: |
| self._generation += 1 |
| command = { |
| "task_id": uuid.uuid4().hex, |
| "generation": self._generation, |
| "prompt": prompt, |
| "use_mock": bool(use_mock), |
| "created_at": time.time(), |
| } |
| if self._active is not None: |
| self._active["token"].request("superseded_by_new_command") |
| self.execution_state = "interrupt_requested" |
| else: |
| self.execution_state = "planning" |
| |
| self._pending = command |
| return dict(command) |
|
|
| def take_pending(self): |
| with self._lock: |
| command = self._pending |
| self._pending = None |
| if command is None: |
| return None |
| token = CancellationToken() |
| command = {**command, "token": token} |
| self._active = command |
| self.execution_state = "planning" |
| self.current_step = None |
| self.latest_result = None |
| return command |
|
|
| def request_interrupt(self, reason="user_cancelled"): |
| with self._lock: |
| if self._active is None: |
| return False |
| self._active["token"].request(reason) |
| self.execution_state = "interrupt_requested" |
| return True |
|
|
| def cancel_all(self, reason="user_cancelled"): |
| """Cancel active work and discard commands that have not started.""" |
| with self._lock: |
| self._pending = None |
| if self._active is None: |
| self.execution_state = "idle" |
| return False |
| self._active["token"].request(reason) |
| self.execution_state = "interrupt_requested" |
| return True |
|
|
| def set_status(self, state, generation=None, current_step=None): |
| with self._lock: |
| if ( |
| generation is not None |
| and self._active is not None |
| and generation != self._active["generation"] |
| ): |
| return False |
| self.execution_state = state |
| self.current_step = current_step |
| return True |
|
|
| def complete(self, generation, state="completed"): |
| with self._lock: |
| if ( |
| self._active is None |
| or generation != self._active["generation"] |
| ): |
| return False |
| self._active = None |
| self.execution_state = ( |
| "planning" if self._pending is not None else state |
| ) |
| self.current_step = None |
| return True |
|
|
| @property |
| def active_token(self): |
| with self._lock: |
| return self._active["token"] if self._active else None |
|
|
| @property |
| def active_generation(self): |
| with self._lock: |
| return self._active["generation"] if self._active else None |
|
|
| @property |
| def has_pending(self): |
| with self._lock: |
| return self._pending is not None |
|
|
| def _prepare_event(self, payload, journal=True): |
| event = dict(payload) |
| with self._lock: |
| event_generation = event.get("generation") |
| is_current = ( |
| event_generation is None |
| or event_generation >= self._generation |
| ) |
| if event.get("type") == "state": |
| self.latest_state = event |
| elif event.get("type") == "plan" and is_current: |
| self.latest_plan = event |
| elif event.get("type") == "result" and is_current: |
| self.latest_result = event |
| if journal: |
| self._seq += 1 |
| event["seq"] = self._seq |
| self._events.append(event) |
| return event |
|
|
| async def publish(self, payload, journal=True): |
| event = self._prepare_event(payload, journal=journal) |
| with self._lock: |
| subscribers = list(self._subscribers) |
| stale = [] |
| for websocket in subscribers: |
| try: |
| await websocket.send_json(event) |
| except Exception: |
| stale.append(websocket) |
| if stale: |
| with self._lock: |
| for websocket in stale: |
| self._subscribers.discard(websocket) |
| return event |
|
|
| def publish_threadsafe(self, payload, loop, journal=True): |
| return asyncio.run_coroutine_threadsafe( |
| self.publish(payload, journal=journal), |
| loop, |
| ) |
|
|
| def events_after(self, seq): |
| with self._lock: |
| return [ |
| dict(event) |
| for event in self._events |
| if event.get("seq", 0) > int(seq or 0) |
| ] |
|
|
| def clear_task_history(self): |
| """Clear UI/task history after an explicit reset, preserving sequence.""" |
| with self._lock: |
| self._events.clear() |
| self.latest_plan = None |
| self.latest_result = None |
| self.current_step = None |
| self.execution_state = "idle" |
|
|
| def snapshot(self): |
| with self._lock: |
| pending = self.conversation_manager.pending |
| return { |
| "type": "runtime_snapshot", |
| "runtime_id": self.runtime_id, |
| "execution_state": self.execution_state, |
| "active_task_id": ( |
| self._active["task_id"] if self._active else None |
| ), |
| "active_prompt": ( |
| self._active["prompt"] if self._active else None |
| ), |
| "generation": ( |
| self._active["generation"] |
| if self._active |
| else self._generation |
| ), |
| "current_step": self.current_step, |
| "plan": self.latest_plan, |
| "result": self.latest_result, |
| "state": self.latest_state, |
| "events": [dict(event) for event in self._events], |
| "last_event_seq": self._seq, |
| "conversation_pending": pending is not None, |
| } |
|
|
