Fix log-stream ordering, server lifecycle, style selector, shape validation and 3D/node panel defects

WebATM-integrated/tests/test_log_streamer.py

@@ -79,6 +79,83 @@ def test_history_is_bounded_to_max_history():
     assert [item["line"] for item in streamer.history()] == ["l2", "l3", "l4"]
 
 
+class ReentrantSocketIO(FakeSocketIO):
+    """Runs a newly scheduled flush task *during* an emit.
+
+    FakeSocketIO drains tasks strictly one after another, which hides the real
+    threading-mode behaviour where a second flush task can emit while the first
+    is still working through its chunks. This stand-in models that by firing
+    ``on_first_emit`` mid-flush and giving any task it schedules a turn there
+    and then.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.on_first_emit = None
+        self._reentered = False
+
+    def emit(self, event, payload):
+        super().emit(event, payload)
+        if self.on_first_emit and not self._reentered:
+            self._reentered = True
+            hook, self.on_first_emit = self.on_first_emit, None
+            hook()  # a log line arrives mid-flush...
+            self.run_all()  # ...and its flush task gets a turn
+
+
+def test_a_line_arriving_mid_flush_does_not_overtake_the_batch_being_emitted():
+    """Only one flush task may be live: a second one emitting concurrently
+    would interleave its newer lines among the older chunks, delivering the
+    stream out of order despite seq being assigned in order."""
+    sio = ReentrantSocketIO()
+    streamer = LogStreamer(sio, batch_max=2)
+
+    for i in range(5):
+        streamer.feed_line(f"l{i}")
+    sio.on_first_emit = lambda: streamer.feed_line("late")
+    sio.run_all()
+
+    seqs = [item["seq"] for _, payload in sio.emitted for item in payload["lines"]]
+    assert seqs == sorted(seqs), f"lines delivered out of order: {seqs}"
+    assert _lines(sio) == ["l0", "l1", "l2", "l3", "l4", "late"]
+
+
+def test_a_line_arriving_as_the_flusher_winds_down_is_still_delivered():
+    """The scheduled flag is cleared under the same lock feed_line appends
+    under, so a line can never be left pending with no flush task coming."""
+    sio = FakeSocketIO()
+    streamer = LogStreamer(sio)
+
+    streamer.feed_line("first")
+    sio.run_all()
+    streamer.feed_line("second")
+    sio.run_all()
+
+    assert _lines(sio) == ["first", "second"]
+
+
+def test_stream_recovers_after_an_emit_raises():
+    """A raising emit must not leave the flush flag stuck True -- feed_line only
+    schedules while it is False, so the stream would go silent for good."""
+    sio = FakeSocketIO()
+    streamer = LogStreamer(sio)
+
+    def boom(event, payload):
+        raise RuntimeError("socket write failed")
+
+    sio.emit = boom
+    streamer.feed_line("during-outage")
+    with pytest.raises(RuntimeError):
+        sio.run_all()
+
+    # The next line must schedule a fresh flush and get through.
+    sio.emit = lambda event, payload: sio.emitted.append((event, payload))
+    streamer.feed_line("after-outage")
+    sio.run_all()
+
+    assert _lines(sio) == ["after-outage"]
+
+
 def test_feed_line_recovers_after_a_failed_schedule():
     """A failed start_background_task must not leave _flush_scheduled stuck
     True, which would silence the stream forever."""

WebATM-integrated/tests/test_process_manager.py

@@ -18,6 +18,8 @@
 import threading
 import time
 
+import pytest
+import webatm_integrated.process_manager as pm
 from webatm_integrated.process_manager import BlueSkyProcessManager
 
 
@@ -147,6 +149,57 @@ def on_line(line: str) -> None:
         manager.kill()
 
 
+def test_a_failing_signal_does_not_strand_the_state_at_stopping(monkeypatch):
+    """If killpg raises anything but ProcessLookupError, stop() must still leave
+    a usable state: a stranded "stopping" makes every later start() wait 15s and
+    then refuse, with no control surface able to clear it."""
+    manager = BlueSkyProcessManager(
+        cmd=[sys.executable, "-c", "import time; time.sleep(60)"]
+    )
+    try:
+        assert manager.start()["success"] is True
+
+        def denied(pgid, sig):
+            raise PermissionError("operation not permitted")
+
+        monkeypatch.setattr(pm.os, "killpg", denied)
+        with pytest.raises(PermissionError):
+            manager.stop()
+
+        # The process really is still up, so that is what status must report --
+        # and a retry must be able to proceed rather than hit the stop-wait.
+        assert manager.status()["status"] == "running"
+        monkeypatch.undo()
+        assert manager.stop()["success"] is True
+        assert manager.status()["running"] is False
+    finally:
+        manager.kill()
+
+
+def test_kill_on_a_stopped_server_does_not_claim_it_killed_something():
+    """The UI renders this message verbatim, so Kill on an already-stopped
+    server must say so rather than report a kill that never happened."""
+    manager = BlueSkyProcessManager(cmd=[sys.executable, "-c", "pass"])
+
+    result = manager.kill()
+
+    assert result["success"] is True
+    assert result["status"] == "stopped"
+    assert result["message"] == "BlueSky server is not running"
+
+
+def test_kill_reports_a_kill_when_a_process_was_actually_running():
+    manager = BlueSkyProcessManager(
+        cmd=[sys.executable, "-c", "import time; time.sleep(60)"]
+    )
+    assert manager.start()["success"] is True
+
+    result = manager.kill()
+
+    assert result["success"] is True
+    assert result["message"] == "BlueSky server killed"
+
+
 def test_restart_propagates_stop_failure_instead_of_claiming_success():
     """restart() must not report "restarted" while the old tree is still alive."""
     manager = BlueSkyProcessManager(cmd=[sys.executable, "-c", "pass"])

WebATM-integrated/webatm_integrated/log_streamer.py

@@ -66,15 +66,32 @@ def feed_line(self, line: str) -> None:
                     raise
 
     def _flush_after_delay(self) -> None:
-        # Cooperative sleep via SocketIO so it matches the active async mode.
-        self._sio.sleep(self._batch_ms / 1000.0)
-        with self._lock:
-            batch = self._pending
-            self._pending = []
-            self._flush_scheduled = False
-        for start in range(0, len(batch), self._batch_max):
-            chunk = batch[start : start + self._batch_max]
-            self._sio.emit(EVENT, {"lines": chunk})
+        # Drain until empty rather than flushing once, so only one flush task is
+        # ever live: clearing the flag before emitting would let a second task
+        # emit concurrently, interleaving newer lines among the older chunks.
+        try:
+            while True:
+                # Cooperative sleep via SocketIO to match the active async mode.
+                self._sio.sleep(self._batch_ms / 1000.0)
+                with self._lock:
+                    batch = self._pending
+                    self._pending = []
+                    if not batch:
+                        # Cleared under the lock feed_line appends under, so a
+                        # line arriving now always gets a flush task scheduled
+                        # for it -- it can never be stranded unflushed.
+                        self._flush_scheduled = False
+                        return
+                for start in range(0, len(batch), self._batch_max):
+                    chunk = batch[start : start + self._batch_max]
+                    self._sio.emit(EVENT, {"lines": chunk})
+        except BaseException:
+            # A raising emit/sleep must not leave the flag stuck True: feed_line
+            # only schedules while it is False, so the stream would go silent
+            # for good. Clearing it lets the next line start a fresh flush.
+            with self._lock:
+                self._flush_scheduled = False
+            raise
 
     def history(self) -> list[dict]:
         """Return a snapshot of buffered lines for late-joining clients.

WebATM-integrated/webatm_integrated/process_manager.py

@@ -32,6 +32,14 @@ def _default_spawn(target: Callable, *args) -> threading.Thread:
     return thread
 
 
+def _signal_group(pgid: int, sig: int) -> None:
+    """Signal a process group, tolerating one that has already exited."""
+    try:
+        os.killpg(pgid, sig)
+    except ProcessLookupError:
+        pass
+
+
 class BlueSkyProcessManager:
     """Thread-safe lifecycle manager for the ``bluesky --headless`` process tree.
 
@@ -189,59 +197,77 @@ def stop(self, sig: int = signal.SIGTERM, escalate_after: float = 5.0) -> dict:
                     "message": "BlueSky server is not running",
                 }
             self._state = "stopping"
-            try:
-                pgid = os.getpgid(proc.pid)
-            except ProcessLookupError:
-                self._state = "stopped"
-                return {
-                    "success": True,
-                    "status": "stopped",
-                    "message": "BlueSky server already exited",
-                }
 
-        # Signal the whole group (server + all node children) outside the lock.
         try:
-            os.killpg(pgid, sig)
+            return self._terminate(proc, sig, escalate_after)
+        except BaseException:
+            # Never leave the state stranded at "stopping": start() waits out a
+            # shutdown in that state, so it would block for 15s and then refuse
+            # to start, with no control surface able to clear it.
+            self._settle(proc)
+            raise
+
+    def _terminate(
+        self, proc: subprocess.Popen, sig: int, escalate_after: float
+    ) -> dict:
+        """Signal ``proc``'s group, escalating to SIGKILL, and settle the state."""
+        try:
+            pgid = os.getpgid(proc.pid)
         except ProcessLookupError:
-            pass
+            self._settle(proc)
+            return {
+                "success": True,
+                "status": "stopped",
+                "message": "BlueSky server already exited",
+            }
 
+        # Signal the whole group (server + all node children) outside the lock.
+        _signal_group(pgid, sig)
         try:
             proc.wait(timeout=escalate_after)
         except subprocess.TimeoutExpired:
-            try:
-                os.killpg(pgid, signal.SIGKILL)
-            except ProcessLookupError:
-                pass
+            _signal_group(pgid, signal.SIGKILL)
             try:
                 proc.wait(timeout=5)
             except subprocess.TimeoutExpired:
                 logger.error("BlueSky process group %s survived SIGKILL", pgid)
-                with self._lock:
-                    if self._proc is proc:
-                        self._state = "running"
+                self._settle(proc)
                 return {
                     "success": False,
                     "status": "error",
                     "message": "BlueSky server did not exit after SIGKILL",
                 }
 
-        with self._lock:
-            if self._proc is proc:
-                self._state = "stopped"
+        self._settle(proc)
         return {
             "success": True,
             "status": "stopped",
             "message": "BlueSky server stopped",
         }
 
+    def _settle(self, proc: subprocess.Popen) -> None:
+        """Re-derive the state from whether ``proc`` is actually still alive.
+
+        Skipped if a restart already installed a different process, whose own
+        state must not be clobbered by this one's teardown.
+        """
+        with self._lock:
+            if self._proc is proc:
+                self._state = "running" if proc.poll() is None else "stopped"
+
     def kill(self) -> dict:
         """Force-kill the whole process group immediately (no graceful wait).
 
         Returns:
             dict: Result with ``success``, ``status`` and ``message``.
         """
+        with self._lock:
+            proc = self._proc
+            was_running = proc is not None and proc.poll() is None
         result = self.stop(sig=signal.SIGKILL, escalate_after=2.0)
-        if result.get("success") and result.get("status") == "stopped":
+        # Only claim a kill if there was actually a live process to kill --
+        # otherwise keep stop()'s "is not running", which the UI shows verbatim.
+        if was_running and result.get("success") and result.get("status") == "stopped":
             result["message"] = "BlueSky server killed"
         return result
 

WebATM/static/css/style.css

@@ -957,20 +957,6 @@ html.wa-open-threeD-controls #threeD-controls {
     min-width: 80px;
 }
 
-/* Aircraft model dropdown - ensure proper right alignment */
-#aircraft-model-container {
-    display: flex !important;
-    justify-content: space-between !important;
-    align-items: center !important;
-}
-
-/* Aircraft 3D scale input - ensure proper right alignment */
-#aircraft-3d-scale-container {
-    display: flex !important;
-    justify-content: space-between !important;
-    align-items: center !important;
-}
-
 /* Aircraft appearance controls group */
 .aircraft-appearance-controls {
     margin-top: 8px;

WebATM/templates/index.html

@@ -1229,34 +1229,6 @@ <h3>Upload a Plugin</h3>
 
 
 
-    <!-- Add Node Demo Restriction Modal -->
-    <div id="add-node-demo-modal" class="modal" style="display: none;">
-        <div class="modal-content">
-            <div class="modal-header">
-                <h3>Add Node</h3>
-                <button class="modal-close" id="add-node-demo-close">&times;</button>
-            </div>
-            <div class="modal-body">
-                <div class="setting-group">
-                    <div class="demo-message">
-                        <span class="demo-badge">DEMO MODE</span>
-                        <p><strong>This feature is not available in Demo mode</strong></p>
-                        <p>In the full version of WebATM for BlueSky, you can:</p>
-                        <ul style="margin: 10px 0; padding-left: 20px; color: var(--text-secondary);">
-                            <li>Add multiple simulation nodes for distributed computing</li>
-                            <li>Manage node configurations and resources</li>
-                            <li>Monitor individual node performance</li>
-                        </ul>
-                        <p style="color: var(--accent-primary);"><em>This demo is limited to a single simulation node.</em></p>
-                    </div>
-                </div>
-            </div>
-            <div class="modal-footer">
-                <button id="add-node-demo-ok" class="btn-primary">OK</button>
-            </div>
-        </div>
-    </div>
-
     <!-- Scripts -->
     <!-- maplibre-gl and socket.io-client are bundled by webpack via npm imports;
          no CDN script tags needed (keeps the app functional without internet). -->