chore: reset NIGHTLY_CHANGELOG.md after nightly-0.9.30-nightly12 release

A testing-<sha> build has no Gitea releases (the testing workflow builds only
a Docker image), so the in-app update check must not fall through to the
stable path - which would wrongly offer a stable "update". handle_api_update_check
now short-circuits for a "testing" version with docker_only:true and nothing
to update (no Gitea round-trip at all), and handle_api_update_apply refuses
self-update on the testing channel the same way it already does for nightly.

Two new tests cover both. (Mirrors the same fix already on the testing
branch, applied here to the pre-refactor monolithic module.)

config_loader.py

@@ -7,8 +7,11 @@ import os
 import sys
 import pathlib
 import configparser
+import logging
 from typing import Optional
 
+log = logging.getLogger("kobrax.config")
+
 _BASE = pathlib.Path(sys.executable).parent if getattr(sys, "frozen", False) else pathlib.Path(__file__).parent
 
 CONFIG_SECTION_CONNECTION = "connection"
@@ -116,7 +119,6 @@ def migrate_env_to_config(env_path: pathlib.Path, config_path: pathlib.Path):
             k, _, v = line.partition("=")
             env_vals[k.strip()] = v.strip()
 
-    config_path.parent.mkdir(parents=True, exist_ok=True)
     cfg = configparser.ConfigParser(interpolation=None)
     cfg[CONFIG_SECTION_CONNECTION] = {
         "printer_ip": env_vals.get("PRINTER_IP", ""),
@@ -136,10 +138,20 @@ def migrate_env_to_config(env_path: pathlib.Path, config_path: pathlib.Path):
     cfg[CONFIG_SECTION_BRIDGE] = {
         "poll_interval": "3",
     }
-    with open(config_path, "w", encoding="utf-8") as f:
-        f.write("# KX-Bridge Konfigurationsdatei\n")
-        f.write("# Automatically migrated from .env\n\n")
-        cfg.write(f)
+    # This runs at module import time (see the "Laden" section below) - an
+    # uncaught mkdir/write failure (e.g. a read-only filesystem) would crash
+    # the whole bridge at startup with a raw traceback. Log a clear diagnostic
+    # before re-raising, so the actual cause (permissions, disk full) is
+    # visible instead of a bare stack trace pointing into configparser.
+    try:
+        config_path.parent.mkdir(parents=True, exist_ok=True)
+        with open(config_path, "w", encoding="utf-8") as f:
+            f.write("# KX-Bridge Konfigurationsdatei\n")
+            f.write("# Automatically migrated from .env\n\n")
+            cfg.write(f)
+    except OSError as e:
+        log.error("Failed to write migrated config.ini to %s: %s", config_path, e)
+        raise
 
 
 def find_config_path() -> pathlib.Path:
@@ -445,9 +457,25 @@ def get(key: str, default: str = "") -> str:
     return os.environ.get(key, default)
 
 
+def _safe_int(value: str, default: int) -> int:
+    """Falls back to `default` instead of raising on a non-numeric value.
+
+    All of these run at module import time - an uncaught ValueError here
+    (e.g. from a hand-edited config.ini with a typo like `mqtt_port = 98833x`)
+    would crash the entire bridge before it even starts, with a raw traceback
+    instead of a clear diagnostic. list_printers() already guards this same
+    class of input the same way; this applies it to the module-level
+    shortcuts too."""
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        log.warning("config: expected a number, got %r - using default %r", value, default)
+        return default
+
+
 # Frequently used shortcuts
 PRINTER_IP       = get("PRINTER_IP",       "")
-MQTT_PORT        = int(get("MQTT_PORT",    "9883"))
+MQTT_PORT        = _safe_int(get("MQTT_PORT", "9883"), 9883)
 USERNAME         = get("MQTT_USERNAME",    "")
 PASSWORD         = get("MQTT_PASSWORD",    "")
 MODE_ID          = get("MODE_ID",          "")
@@ -456,14 +484,14 @@ POWER_ON_URL     = get("POWER_ON_URL",     "")
 POWER_OFF_URL    = get("POWER_OFF_URL",    "")
 POWER_STATUS_URL = get("POWER_STATUS_URL", "")
 DEFAULT_AMS_SLOT = get("DEFAULT_AMS_SLOT", "auto")
-AUTO_LEVELING           = int(get("AUTO_LEVELING",          "1"))
-VIBRATION_COMPENSATION  = int(get("VIBRATION_COMPENSATION", "0"))
-CAMERA_ON_PRINT         = int(get("CAMERA_ON_PRINT",        "0"))
-WEB_UPLOAD_WARNING = int(get("WEB_UPLOAD_WARNING", "1"))
-DELETE_PRINTER_FILE_AFTER_PRINT = int(get("DELETE_PRINTER_FILE_AFTER_PRINT", "0"))
-PRINT_START_DIALOG = int(get("PRINT_START_DIALOG", get("FILE_READY_DIALOG", "1")))
+AUTO_LEVELING           = _safe_int(get("AUTO_LEVELING",          "1"), 1)
+VIBRATION_COMPENSATION  = _safe_int(get("VIBRATION_COMPENSATION", "0"), 0)
+CAMERA_ON_PRINT         = _safe_int(get("CAMERA_ON_PRINT",        "0"), 0)
+WEB_UPLOAD_WARNING = _safe_int(get("WEB_UPLOAD_WARNING", "1"), 1)
+DELETE_PRINTER_FILE_AFTER_PRINT = _safe_int(get("DELETE_PRINTER_FILE_AFTER_PRINT", "0"), 0)
+PRINT_START_DIALOG = _safe_int(get("PRINT_START_DIALOG", get("FILE_READY_DIALOG", "1")), 1)
 SPOOLMAN_SERVER    = get("SPOOLMAN_SERVER",    "")
-SPOOLMAN_SYNC_RATE = int(get("SPOOLMAN_SYNC_RATE", "0"))
+SPOOLMAN_SYNC_RATE = _safe_int(get("SPOOLMAN_SYNC_RATE", "0"), 0)
 BRIDGE_HOST_IP     = get("BRIDGE_HOST_IP",     "")
-POLL_INTERVAL      = int(get("POLL_INTERVAL",  "3"))
-VERBOSE_HTTP_LOG   = int(get("VERBOSE_HTTP_LOG", "0"))
+POLL_INTERVAL      = _safe_int(get("POLL_INTERVAL",  "3"), 3)
+VERBOSE_HTTP_LOG   = _safe_int(get("VERBOSE_HTTP_LOG", "0"), 0)

kobrax_client.py

@@ -178,6 +178,14 @@ class KobraXClient:
         self._pending_msgid: dict[str, dict] = {}
         # Pending requests by msg_type/report topic suffix
         self._pending_report: dict[str, dict] = {}
+        # Guards _pending_msgid/_pending_report against concurrent mutation:
+        # the reader thread resolves entries in _dispatch() while publish()
+        # (called from the poll loop and, via run_in_executor, HTTP handler
+        # threads) registers/cleans them up - without this, two concurrent
+        # publish() calls for the same msg_type can race on the
+        # check-then-set for a report_key slot, and _dispatch() could observe
+        # a dict mid-mutation.
+        self._pending_lock = threading.Lock()
 
         # Optional callbacks: topic_suffix → callable(payload_dict)
         self.callbacks: dict[str, callable] = {}
@@ -446,34 +454,46 @@ class KobraXClient:
     def _drain(self):
         buf = self._buf
         idx = 0
-        while idx < len(buf):
-            ptype = buf[idx] & 0xF0
-            i = idx + 1
-            mul = 1
-            rem = 0
-            while i < len(buf):
-                b = buf[i]
-                rem += (b & 0x7F) * mul
-                mul *= 128
-                i += 1
-                if not (b & 0x80):
+        try:
+            while idx < len(buf):
+                ptype = buf[idx] & 0xF0
+                i = idx + 1
+                mul = 1
+                rem = 0
+                while i < len(buf):
+                    b = buf[i]
+                    rem += (b & 0x7F) * mul
+                    mul *= 128
+                    i += 1
+                    if not (b & 0x80):
+                        break
+                if i + rem > len(buf):
                     break
-            if i + rem > len(buf):
-                break
-            pkt = buf[i:i + rem]
-            idx = i + rem
+                pkt = buf[i:i + rem]
+                idx = i + rem
 
-            if ptype == 0x30:
-                topic, raw_payload = _parse_publish(pkt)
-                if topic is None:
-                    continue
-                try:
-                    payload = json.loads(raw_payload)
-                except Exception:
-                    payload = {"_raw": raw_payload.decode("utf-8", errors="replace")}
-                self._dispatch(topic, payload)
-
-        self._buf = buf[idx:]
+                if ptype == 0x30:
+                    topic, raw_payload = _parse_publish(pkt)
+                    if topic is None:
+                        continue
+                    try:
+                        payload = json.loads(raw_payload)
+                    except Exception:
+                        payload = {"_raw": raw_payload.decode("utf-8", errors="replace")}
+                    try:
+                        self._dispatch(topic, payload)
+                    except Exception as e:
+                        # A single malformed/unexpected message (e.g. valid JSON
+                        # that isn't an object, like a bare number or list) must
+                        # not be reprocessed forever: without this, an exception
+                        # here would skip the buffer-advance below, leaving the
+                        # same bad packet at the front of self._buf so every
+                        # future _drain() call crashes on it again - each one
+                        # forcing a reconnect via the reader loop's exception
+                        # handler, an endless self-inflicted reconnect loop.
+                        log.warning("dispatch error for %s: %s", topic, e)
+        finally:
+            self._buf = buf[idx:]
 
     def _dedup_hash(self, suffix: str, payload: dict) -> str:
         """Hash payload ignoring volatile per-tick fields for dedup check."""
@@ -484,6 +504,9 @@ class KobraXClient:
         return hashlib.md5(json.dumps(stable, sort_keys=True).encode(), usedforsecurity=False).hexdigest()
 
     def _dispatch(self, topic: str, payload: dict):
+        if not isinstance(payload, dict):
+            log.warning("dispatch: non-dict payload on %s: %r", topic, payload)
+            return
         suffix = "/".join(topic.split("/")[-2:])
 
         if self._raw_log:
@@ -511,18 +534,29 @@ class KobraXClient:
                 log.info("RX %-25s  state=%-12s  data=%s",
                          suffix, state, json.dumps(payload.get("data"), ensure_ascii=False))
 
-        # Resolve by report topic suffix (e.g. "info/report")
-        if suffix in self._pending_report:
-            entry = self._pending_report[suffix]
-            entry["result"] = payload
-            entry["event"].set()
+        msgid = payload.get("msgid")
+        with self._pending_lock:
+            report_entry = self._pending_report.get(suffix)
+            msgid_entry  = self._pending_msgid.get(msgid) if msgid else None
+
+        # Resolve by report topic suffix (e.g. "info/report"). If the payload
+        # carries a msgid that doesn't match what this waiter is actually
+        # expecting, it's a stale/late reply for a different, already-timed-out
+        # request that happens to share the same report_key - don't deliver it
+        # to the wrong caller.
+        if report_entry is not None:
+            entry_msgid = report_entry.get("msgid")
+            if not entry_msgid or not msgid or entry_msgid == msgid:
+                report_entry["result"] = payload
+                report_entry["event"].set()
+            else:
+                log.debug("dispatch: msgid mismatch for %s report (waiting=%s, got=%s) - ignoring stale reply",
+                           suffix, entry_msgid, msgid)
 
         # Resolve by msgid (for generic response ACK)
-        msgid = payload.get("msgid")
-        if msgid and msgid in self._pending_msgid:
-            entry = self._pending_msgid[msgid]
-            entry["result"] = payload
-            entry["event"].set()
+        if msgid_entry is not None:
+            msgid_entry["result"] = payload
+            msgid_entry["event"].set()
 
         # User callbacks by topic suffix (last two path components)
         if suffix in self.callbacks:
@@ -558,13 +592,18 @@ class KobraXClient:
         # Also register by report topic as fallback for responses without msgid.
         report_key = f"{msg_type}/report"
         event  = threading.Event()
-        entry  = {"event": event, "result": None}
-        self._pending_msgid[msgid] = entry
-        # Only register report-key waiter if nobody else is waiting on it
+        # entry carries its own msgid so _dispatch()'s report-suffix path can
+        # confirm a reply actually belongs to THIS request before delivering
+        # it - without that, a late reply for an already-timed-out request A
+        # could be handed to a newer request B waiting on the same report_key.
+        entry  = {"event": event, "result": None, "msgid": msgid}
         report_registered = False
-        if report_key not in self._pending_report:
-            self._pending_report[report_key] = entry
-            report_registered = True
+        with self._pending_lock:
+            self._pending_msgid[msgid] = entry
+            # Only register report-key waiter if nobody else is waiting on it
+            if report_key not in self._pending_report:
+                self._pending_report[report_key] = entry
+                report_registered = True
 
         topic = self._pub_topic(msg_type)
         # Status poll TX (query/getInfo) is pure noise (every few seconds) ->
@@ -578,9 +617,10 @@ class KobraXClient:
                 self._sock.sendall(_build_publish(topic, payload))
         except Exception as e:
             log.error("send error: %s, reconnecting…", e)
-            self._pending_msgid.pop(msgid, None)
-            if report_registered:
-                self._pending_report.pop(report_key, None)
+            with self._pending_lock:
+                self._pending_msgid.pop(msgid, None)
+                if report_registered:
+                    self._pending_report.pop(report_key, None)
             # Non-blocking: never hang the poll-loop thread inside publish()
             # while a reconnect is running / during backoff (see _reconnect
             # docstring) - it must return so kobra_state can flip to "offline".
@@ -590,22 +630,25 @@ class KobraXClient:
             try:
                 with self._lock:
                     self._sock.sendall(_build_publish(topic, payload))
-                self._pending_msgid[msgid] = entry
-                if report_registered:
-                    self._pending_report[report_key] = entry
+                with self._pending_lock:
+                    self._pending_msgid[msgid] = entry
+                    if report_registered:
+                        self._pending_report[report_key] = entry
             except Exception:
                 return None
 
         if timeout <= 0:
-            self._pending_msgid.pop(msgid, None)
-            if report_registered:
-                self._pending_report.pop(report_key, None)
+            with self._pending_lock:
+                self._pending_msgid.pop(msgid, None)
+                if report_registered:
+                    self._pending_report.pop(report_key, None)
             return None
 
         received = event.wait(timeout)
-        self._pending_msgid.pop(msgid, None)
-        if report_registered:
-            self._pending_report.pop(report_key, None)
+        with self._pending_lock:
+            self._pending_msgid.pop(msgid, None)
+            if report_registered:
+                self._pending_report.pop(report_key, None)
         if not received:
             return None
         return entry["result"]
@@ -710,7 +753,9 @@ class KobraXClient:
                 raise RuntimeError("Could not get info/report for upload URL")
             upload_url = info["data"]["urls"]["fileUploadurl"]
         # parse token from URL query string
-        token = upload_url.split("?s=")[1] if "?s=" in upload_url else ""
+        if "?s=" not in upload_url:
+            raise RuntimeError(f"Upload: no session token ('?s=') in upload URL: {upload_url!r}")
+        token = upload_url.split("?s=")[1]
 
         with open(filepath, "rb") as f:
             file_data = f.read()
@@ -760,19 +805,25 @@ class KobraXClient:
         # (the printer processes the file before replying).
         _ai  = socket.getaddrinfo(self.host, 18910, socket.AF_INET, socket.SOCK_STREAM)
         sock = socket.create_connection(_ai[0][4], timeout=10)
-        sock.settimeout(None)  # blocking during send
-        sock.sendall(headers + body)
-        sock.settimeout(180)
-        response = b""
         try:
-            while True:
-                chunk = sock.recv(65536)
-                if not chunk:
-                    break
-                response += chunk
-        except socket.timeout:
-            pass
-        sock.close()
+            sock.settimeout(None)  # blocking during send
+            sock.sendall(headers + body)
+            sock.settimeout(180)
+            response = b""
+            try:
+                while True:
+                    chunk = sock.recv(65536)
+                    if not chunk:
+                        break
+                    response += chunk
+            except socket.timeout:
+                pass
+        finally:
+            # Without this, a sendall()/recv() failure other than
+            # socket.timeout (e.g. ConnectionResetError/BrokenPipeError if
+            # the printer drops the connection mid-upload) skipped
+            # sock.close() entirely, leaking the fd on every failed attempt.
+            sock.close()
 
         # parse HTTP response body
         if b"\r\n\r\n" in response:

kobrax_moonraker_bridge.py

@@ -713,7 +713,7 @@ class CameraCache:
                 await asyncio.sleep(2.0)
                 continue
             try:
-                self._proc_jpeg = await asyncio.create_subprocess_exec(
+                proc = await asyncio.create_subprocess_exec(
                     _find_ffmpeg(), "-loglevel", "warning",
                     *self._input_args(url), "-i", url,
                     "-vf", "fps=2",
@@ -722,6 +722,7 @@ class CameraCache:
                     stdout=asyncio.subprocess.PIPE,
                     stderr=asyncio.subprocess.PIPE,
                 )
+                self._proc_jpeg = proc
             except Exception as e:
                 log.warning(f"CameraCache: ffmpeg-jpeg start failed: {e}")
                 await asyncio.sleep(3.0)
@@ -731,7 +732,7 @@ class CameraCache:
             rc = None
             try:
                 while True:
-                    chunk = await self._proc_jpeg.stdout.read(self.TS_CHUNK)
+                    chunk = await proc.stdout.read(self.TS_CHUNK)
                     if not chunk:
                         break
                     buf += chunk
@@ -751,26 +752,31 @@ class CameraCache:
             except Exception as e:
                 log.debug(f"CameraCache: jpeg-loop unterbrochen: {e}")
             finally:
-                # Kill + wait - otherwise the child process lingers as a zombie and
-                # asyncio reports "Unknown child pid ..." on the next reaper tick.
-                if self._proc_jpeg is not None:
+                # NOTE: cleanup operates on the local `proc` reference, not on
+                # self._proc_jpeg - see _run_mjpeg_loop's identical comment.
+                # If this task got cancelled (e.g. by reset()), a new task may
+                # already have started and assigned its own process to
+                # self._proc_jpeg by the time we reach here; killing that
+                # shared attribute instead of our own local proc would kill
+                # the WRONG (newer) process and leak this one as an orphan.
+                try:
+                    proc.kill()
+                except Exception:
+                    pass
+                try:
+                    await proc.wait()
+                except Exception:
+                    pass
+                rc = proc.returncode
+                if rc:
                     try:
-                        self._proc_jpeg.kill()
+                        err = await proc.stderr.read(500)
+                        if err:
+                            log.warning(f"CameraCache: ffmpeg-jpeg stderr: {err.decode(errors='replace').strip()}")
                     except Exception:
                         pass
-                    try:
-                        await self._proc_jpeg.wait()
-                    except Exception:
-                        pass
-                    rc = self._proc_jpeg.returncode
-                    if rc:
-                        try:
-                            err = await self._proc_jpeg.stderr.read(500)
-                            if err:
-                                log.warning(f"CameraCache: ffmpeg-jpeg stderr: {err.decode(errors='replace').strip()}")
-                        except Exception:
-                            pass
-                self._proc_jpeg = None
+                if self._proc_jpeg is proc:
+                    self._proc_jpeg = None
             if rc:
                 self._fail_count_jpeg += 1
                 delay = min(2.0 * (2 ** self._fail_count_jpeg), 300.0)
@@ -788,7 +794,7 @@ class CameraCache:
                 await asyncio.sleep(2.0)
                 continue
             try:
-                self._proc_h264 = await asyncio.create_subprocess_exec(
+                proc = await asyncio.create_subprocess_exec(
                     _find_ffmpeg(), "-loglevel", "warning",
                     *self._input_args(url), "-i", url,
                     "-c:v", "copy", "-an",
@@ -796,6 +802,7 @@ class CameraCache:
                     stdout=asyncio.subprocess.PIPE,
                     stderr=asyncio.subprocess.PIPE,
                 )
+                self._proc_h264 = proc
             except Exception as e:
                 log.warning(f"CameraCache: ffmpeg-h264 start failed: {e}")
                 await asyncio.sleep(3.0)
@@ -804,7 +811,7 @@ class CameraCache:
             rc = None
             try:
                 while True:
-                    chunk = await self._proc_h264.stdout.read(self.TS_CHUNK)
+                    chunk = await proc.stdout.read(self.TS_CHUNK)
                     if not chunk:
                         break
                     # Fanout: non-blocking per subscriber; slow clients
@@ -822,24 +829,31 @@ class CameraCache:
             except Exception as e:
                 log.debug(f"CameraCache: h264-loop unterbrochen: {e}")
             finally:
-                if self._proc_h264 is not None:
+                # NOTE: cleanup operates on the local `proc` reference, not on
+                # self._proc_h264 - see _run_mjpeg_loop's identical comment.
+                # If this task got cancelled (e.g. by reset()), a new task may
+                # already have started and assigned its own process to
+                # self._proc_h264 by the time we reach here; killing that
+                # shared attribute instead of our own local proc would kill
+                # the WRONG (newer) process and leak this one as an orphan.
+                try:
+                    proc.kill()
+                except Exception:
+                    pass
+                try:
+                    await proc.wait()
+                except Exception:
+                    pass
+                rc = proc.returncode
+                if rc:
                     try:
-                        self._proc_h264.kill()
+                        err = await proc.stderr.read(500)
+                        if err:
+                            log.warning(f"CameraCache: ffmpeg-h264 stderr: {err.decode(errors='replace').strip()}")
                     except Exception:
                         pass
-                    try:
-                        await self._proc_h264.wait()
-                    except Exception:
-                        pass
-                    rc = self._proc_h264.returncode
-                    if rc:
-                        try:
-                            err = await self._proc_h264.stderr.read(500)
-                            if err:
-                                log.warning(f"CameraCache: ffmpeg-h264 stderr: {err.decode(errors='replace').strip()}")
-                        except Exception:
-                            pass
-                self._proc_h264 = None
+                if self._proc_h264 is proc:
+                    self._proc_h264 = None
             if rc:
                 self._fail_count_h264 += 1
                 delay = min(2.0 * (2 ** self._fail_count_h264), 300.0)
@@ -5093,7 +5107,10 @@ class KobraXBridge:
 
     async def handle_api_settings_post(self, request):
         import configparser
-        data = await request.json()
+        try:
+            data = await request.json()
+        except Exception:
+            return self._json_cors({"error": "invalid json"}, status=400)
         config_path = self._find_config_path()
         config_path.parent.mkdir(parents=True, exist_ok=True)
 
@@ -5434,6 +5451,21 @@ class KobraXBridge:
 
     async def handle_api_update_check(self, request):
         current = self._read_version()
+        # Testing channel (testing-<sha>) has no Gitea releases at all - it's
+        # a Docker-only channel. Report that directly instead of falling
+        # through to the stable path (which would wrongly offer a stable
+        # "update"). The :testing image is rolling, so there's nothing to
+        # compare a version against.
+        if "testing" in current:
+            return web.json_response({
+                "current":          current,
+                "latest":           current,
+                "update_available": False,
+                "tag":              current,
+                "docker_only":      True,
+                "changelog":        "Testing channel - updates are delivered via Docker: "
+                                    "docker compose pull && docker compose up -d",
+            })
         is_nightly = "nightly" in current
         is_dev = "-dev+" in current
         if is_nightly:
@@ -5513,11 +5545,16 @@ class KobraXBridge:
     ]
 
     async def handle_api_update_apply(self, request):
-        data = await request.json()
+        try:
+            data = await request.json()
+        except Exception:
+            return web.json_response({"error": "invalid json"}, status=400)
         new_tag = data.get("tag", "")
-        if "nightly" in self._read_version():
+        _cur = self._read_version()
+        if "nightly" in _cur or "testing" in _cur:
+            channel = "testing" if "testing" in _cur else "nightly"
             return web.json_response(
-                {"error": "nightly updates are delivered via Docker: "
+                {"error": f"{channel} updates are delivered via Docker: "
                           "docker compose pull && docker compose up -d"}, status=400)
         if getattr(sys, "frozen", False):
             return web.json_response(

orca_filaments.py

@@ -9,8 +9,11 @@ them as a normalized list with (id, name, vendor, type, color).
 from __future__ import annotations
 
 import json
+import logging
 import re
 
+log = logging.getLogger("kobrax.filaments")
+
 
 def first_str(value, default: str = "") -> str:
     """Orca profiles store some fields as ['value']. Returns the first
@@ -69,7 +72,17 @@ def parse_profile(data: dict, by_name: dict | None = None,
     if system_index:
         for p in system_index:
             if isinstance(p, dict) and p.get("name"):
-                sys_by_name[p["name"]] = p
+                pname = p["name"]
+                if pname in sys_by_name and sys_by_name[pname] is not p:
+                    # clean_name() deliberately collapses variant-suffixed
+                    # names (e.g. "...@base" vs "...@Anycubic Kobra X 0.4
+                    # nozzle") onto the same cleaned name - expected, but the
+                    # last-write-wins overwrite here was previously silent,
+                    # making an unexpected inherits-parent resolution hard to
+                    # debug.
+                    log.debug("orca_filaments: duplicate system profile name %r - "
+                              "overwriting %r with %r", pname, sys_by_name[pname].get("id"), p.get("id"))
+                sys_by_name[pname] = p
 
     def _resolve(key: str, depth: int = 5):
         cur_list = [data]
@@ -119,7 +132,7 @@ def parse_profile(data: dict, by_name: dict | None = None,
     if not fid or not isinstance(fid, str):
         return None
 
-    name_raw = data.get("name", fid)
+    name_raw = first_str(data.get("name"), fid)
     name = clean_name(name_raw)
     vendor = first_str(_resolve_full("filament_vendor")) or (path_vendor or "Generic")
     ftype  = first_str(_resolve_full("filament_type"), "")

tests/test_camera_process_race.py

@@ -0,0 +1,83 @@
+"""Camera ffmpeg process-handle race (code review finding).
+
+_run_jpeg_loop()/_run_h264_loop() used to operate on the shared instance
+attribute (self._proc_jpeg / self._proc_h264) in their cleanup, instead of a
+local reference to the process they themselves started - the same bug
+_run_mjpeg_loop() already had fixed with a documented local-`proc` pattern.
+
+If a loop's task is cancelled (e.g. via CameraCache.reset() after a stream
+URL rotation) while a new task has already started and assigned its own
+process to the shared attribute, the cancelled task's cleanup would kill
+and null out the NEWER process instead of its own - leaking its own actual
+ffmpeg child as an orphan that nothing ever cleans up again.
+"""
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from kobrax_moonraker_bridge import CameraCache
+
+
+def _fake_proc(name):
+    """A minimal stand-in for asyncio.subprocess.Process good enough to
+    drive _run_jpeg_loop()'s read/kill/wait/returncode usage."""
+    proc = MagicMock(name=name)
+    proc.returncode = 0
+    proc.kill = MagicMock()
+    proc.wait = AsyncMock()
+    proc.stdout = MagicMock()
+    proc.stderr = MagicMock()
+    proc.stderr.read = AsyncMock(return_value=b"")
+    return proc
+
+
+@pytest.mark.asyncio
+async def test_jpeg_loop_cleanup_does_not_kill_a_newer_process():
+    cache = CameraCache()
+    cache._url = "http://printer/live/streamtoken"
+
+    old_proc = _fake_proc("old")
+    new_proc = _fake_proc("new")
+
+    # old_proc's stdout.read blocks forever until cancelled - simulating the
+    # loop being stuck reading from a stale connection, same as the real bug.
+    stuck = asyncio.Event()
+
+    async def old_stdout_read(_n):
+        await stuck.wait()
+        return b""
+
+    old_proc.stdout.read = old_stdout_read
+
+    create_calls = []
+
+    async def fake_create_subprocess_exec(*args, **kwargs):
+        create_calls.append(1)
+        return old_proc if len(create_calls) == 1 else new_proc
+
+    with patch("kobrax_moonraker_bridge.asyncio.create_subprocess_exec",
+               side_effect=fake_create_subprocess_exec), \
+         patch("kobrax_moonraker_bridge._find_ffmpeg", return_value="ffmpeg"):
+        task = asyncio.create_task(cache._run_jpeg_loop())
+        # Let the loop start and assign old_proc to the shared attribute.
+        await asyncio.sleep(0.05)
+        assert cache._proc_jpeg is old_proc
+
+        # Simulate a second loop iteration's process already having been
+        # assigned to the shared attribute before the cancelled task's
+        # cleanup runs - the exact race window from the bug report.
+        cache._proc_jpeg = new_proc
+
+        task.cancel()
+        try:
+            await asyncio.wait_for(task, timeout=2.0)
+        except (asyncio.CancelledError, asyncio.TimeoutError):
+            pass
+
+    # The cancelled task must have killed/waited on ITS OWN process (old_proc),
+    # not the newer one that had already taken over the shared attribute.
+    old_proc.kill.assert_called_once()
+    new_proc.kill.assert_not_called()
+    # And it must not have clobbered the newer process's slot.
+    assert cache._proc_jpeg is new_proc

tests/test_client_robustness.py

@@ -0,0 +1,189 @@
+"""Robustness fixes found during a targeted code review of kobrax_client.py:
+
+1. _drain()/_dispatch() must not get stuck reprocessing the same malformed
+   packet forever when the printer sends valid JSON that isn't an object
+   (e.g. a bare number or list) - previously an exception from _dispatch()
+   escaped before self._buf was advanced, so the same bytes sat at the
+   front of the buffer and crashed every subsequent _drain() call, each one
+   forcing a reconnect via the reader loop's exception handler.
+2. publish()'s pending-dict registration/cleanup must be safe against
+   concurrent callers for the same msg_type, and a stale/late reply must
+   not be delivered to a newer, unrelated caller waiting on the same
+   report-topic suffix.
+3. upload_gcode() must raise a clear error for a malformed upload URL
+   (missing "?s=") instead of silently sending an unauthenticated request,
+   and must not leak the upload socket on a send/recv failure.
+"""
+import argparse
+import json
+import socket
+import tempfile
+import threading
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from kobrax_client import KobraXClient, _build_publish
+
+
+def _client(**overrides):
+    kwargs = dict(
+        host="192.168.1.100", username="u", password="p",
+        mode_id="20030", device_id="abc123", port=9883,
+        client_id="test",
+    )
+    kwargs.update(overrides)
+    return KobraXClient(**kwargs)
+
+
+def _frame(topic: str, payload) -> bytes:
+    """Builds a raw MQTT PUBLISH frame carrying `payload` as JSON, the same
+    wire format _drain() parses."""
+    return _build_publish(topic, json.dumps(payload))
+
+
+# --- Fix 1: malformed (non-dict) JSON payload must not wedge the buffer ---
+
+def test_drain_advances_buffer_past_non_dict_payload():
+    c = _client()
+    # A bare JSON number is valid JSON but not a dict - json.loads succeeds,
+    # but _dispatch()'s dict-oriented logic would previously crash on it.
+    bad = _frame("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report", 42)
+    good = _frame("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report", {"state": "done"})
+    c._buf = bad + good
+
+    c._drain()  # must not raise
+
+    # Both packets must have been consumed - the malformed one logged/skipped,
+    # the buffer advanced past it so the following valid packet is also
+    # processed, not left stuck behind it.
+    assert c._buf == b""
+
+
+def test_drain_does_not_reprocess_bad_packet_on_repeated_calls():
+    """Regression guard for the original bug: before the fix, a crash in
+    _dispatch() left self._buf untouched, so the bad packet stayed at the
+    front and every _drain() call re-crashed on it."""
+    c = _client()
+    bad = _frame("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report", [1, 2, 3])
+    c._buf = bad
+
+    c._drain()
+    assert c._buf == b""  # consumed, not stuck
+
+    # A second call on the now-empty buffer must be a no-op, not a re-crash.
+    c._drain()
+    assert c._buf == b""
+
+
+def test_dispatch_rejects_non_dict_payload_directly():
+    c = _client()
+    c._dispatch("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report", "not a dict")
+    # No exception, no registered pending entries get corrupted.
+    assert c._pending_report == {}
+
+
+# --- Fix 2: pending-dict thread-safety + msgid correlation ---
+
+def test_publish_concurrent_same_msg_type_both_get_delivered():
+    """Two overlapping publish() calls for the same msg_type must each
+    receive their OWN reply, not have one silently miss out because the
+    other already claimed the shared report_key slot."""
+    c = _client()
+    c._running = True
+    c._sock = MagicMock()
+    c._ensure_reader = lambda: None  # no real reader thread needed for this test
+
+    results = {}
+
+    def call(label):
+        results[label] = c.publish("info", "query", timeout=2.0)
+
+    t1 = threading.Thread(target=call, args=("a",))
+    t2 = threading.Thread(target=call, args=("b",))
+    t1.start()
+    time.sleep(0.02)
+    t2.start()
+    time.sleep(0.05)
+
+    # Simulate the printer replying to whichever msgid-bearing requests are
+    # currently pending, by msgid (the reliable path both requests can use).
+    with c._pending_lock:
+        pending = dict(c._pending_msgid)
+    for msgid, entry in pending.items():
+        c._dispatch("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report",
+                    {"msgid": msgid, "state": "done", "data": {"ok": True}})
+
+    t1.join(timeout=3)
+    t2.join(timeout=3)
+
+    assert results["a"] is not None
+    assert results["b"] is not None
+
+
+def test_dispatch_ignores_stale_reply_with_mismatched_msgid():
+    """A late reply carrying a msgid that doesn't match what the current
+    report_key waiter is expecting must not be delivered to it - it belongs
+    to an earlier, already-resolved/timed-out request."""
+    c = _client()
+    c._running = True
+    c._sock = MagicMock()
+
+    event = threading.Event()
+    entry = {"event": event, "result": None, "msgid": "the-real-one"}
+    with c._pending_lock:
+        c._pending_report["info/report"] = entry
+
+    # A stale reply for a different msgid must not resolve this waiter.
+    c._dispatch("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report",
+                {"msgid": "some-other-stale-id", "state": "done"})
+    assert not event.is_set()
+    assert entry["result"] is None
+
+    # The matching reply must resolve it.
+    c._dispatch("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report",
+                {"msgid": "the-real-one", "state": "done"})
+    assert event.is_set()
+    assert entry["result"]["msgid"] == "the-real-one"
+
+
+def test_dispatch_delivers_reply_without_msgid_as_before():
+    """Regression guard: plenty of printer reports carry no msgid at all
+    (e.g. spontaneous status pushes) - those must still resolve a waiter
+    registered without one (entry["msgid"] is falsy)."""
+    c = _client()
+    event = threading.Event()
+    entry = {"event": event, "result": None, "msgid": None}
+    with c._pending_lock:
+        c._pending_report["info/report"] = entry
+
+    c._dispatch("anycubic/anycubicCloud/v1/printer/public/20030/abc123/info/report",
+                {"state": "done"})
+    assert event.is_set()
+
+
+# --- Fix 3: upload_gcode() URL validation + socket cleanup ---
+
+def test_upload_gcode_raises_clear_error_on_missing_session_token(tmp_path):
+    c = _client()
+    f = tmp_path / "test.gcode"
+    f.write_text("; gcode")
+
+    with pytest.raises(RuntimeError, match="session token"):
+        c.upload_gcode(str(f), upload_url="http://192.168.1.100:18910/gcode_upload")
+
+
+def test_upload_gcode_closes_socket_on_send_failure(tmp_path):
+    c = _client()
+    f = tmp_path / "test.gcode"
+    f.write_text("; gcode")
+
+    fake_sock = MagicMock()
+    fake_sock.sendall.side_effect = ConnectionResetError("printer went away")
+
+    with patch("socket.create_connection", return_value=fake_sock):
+        with pytest.raises(ConnectionResetError):
+            c.upload_gcode(str(f), upload_url="http://192.168.1.100:18910/gcode_upload?s=tok123")
+
+    fake_sock.close.assert_called_once()

tests/test_config_loader_robustness.py

@@ -0,0 +1,66 @@
+"""config_loader.py robustness fixes found during code review:
+
+1. _safe_int() must fall back to a default instead of raising - the
+   module-level numeric shortcuts (MQTT_PORT, POLL_INTERVAL, etc.) run this
+   at import time, so an uncaught ValueError there previously crashed the
+   entire bridge on startup if config.ini had a typo'd numeric value
+   (e.g. "mqtt_port = 98833x"), with a raw traceback instead of a clear
+   diagnostic. list_printers() already guarded this same class of input the
+   same way; this applies it to the module-level shortcuts too.
+2. migrate_env_to_config() must log a clear error (not a bare traceback)
+   if writing the migrated config.ini fails (e.g. permission error).
+"""
+import subprocess
+import sys
+import textwrap
+
+import config_loader
+
+
+def test_safe_int_returns_value_for_valid_numeric_string():
+    assert config_loader._safe_int("42", 0) == 42
+
+
+def test_safe_int_falls_back_to_default_on_garbage():
+    assert config_loader._safe_int("98833x", 9883) == 9883
+
+
+def test_safe_int_falls_back_to_default_on_empty_string():
+    assert config_loader._safe_int("", 3) == 3
+
+
+def test_safe_int_falls_back_to_default_on_none():
+    assert config_loader._safe_int(None, 5) == 5
+
+
+def test_config_loader_import_survives_malformed_config_ini(tmp_path):
+    """End-to-end regression guard: importing config_loader with a
+    config.ini containing a non-numeric mqtt_port must not raise - it must
+    fall back to the default instead. Run in a subprocess since
+    config_loader executes its migration/loading logic at import time and
+    Python caches modules, so a plain re-import in this test process
+    wouldn't actually re-exercise the import-time code path."""
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+    (config_dir / "config.ini").write_text(
+        "[connection]\n"
+        "printer_ip = 192.168.1.50\n"
+        "mqtt_port = 98833x\n"  # malformed - not a number
+    )
+
+    script = textwrap.dedent(f"""
+        import sys
+        sys.path.insert(0, {str(tmp_path.parent.parent)!r})
+        import config_loader
+        config_loader._BASE = __import__("pathlib").Path({str(tmp_path)!r})
+        config_loader._find_config_file = lambda: {str(config_dir / "config.ini")!r} \
+            and __import__("pathlib").Path({str(config_dir / "config.ini")!r})
+        import importlib
+        importlib.reload(config_loader)
+        assert config_loader.MQTT_PORT == 9883, config_loader.MQTT_PORT
+        print("OK")
+    """)
+
+    result = subprocess.run([sys.executable, "-c", script], capture_output=True, text=True, timeout=10)
+    assert result.returncode == 0, f"stdout={result.stdout!r} stderr={result.stderr!r}"
+    assert "OK" in result.stdout

tests/test_orca_filaments_parser.py

@@ -0,0 +1,182 @@
+"""orca_filaments.py parser robustness (code review finding).
+
+No dedicated test file existed for parse_profile()/parse_profile_bytes()/
+clean_name() before this - existing tests only used pre-parsed profile dicts
+as fixtures, never exercised the actual parsing logic.
+"""
+import json
+import logging
+
+from orca_filaments import clean_name, first_str, parse_profile, parse_profile_bytes
+
+
+def test_clean_name_strips_base_suffix():
+    assert clean_name("PolyTerra PLA @base") == "PolyTerra PLA"
+
+
+def test_clean_name_strips_printer_and_nozzle_suffix():
+    assert clean_name("Anycubic PLA @Anycubic Kobra X 0.4 nozzle") == "Anycubic PLA"
+
+
+def test_clean_name_strips_bare_nozzle_suffix():
+    assert clean_name("Anker Generic PLA 0.4 nozzle") == "Anker Generic PLA"
+
+
+def test_clean_name_returns_raw_when_stripping_leaves_nothing():
+    """An all-suffix name has nothing left after stripping - falls back to
+    the original raw string rather than returning an empty string."""
+    assert clean_name("@base") == "@base"
+
+
+def test_first_str_unwraps_single_element_list():
+    assert first_str(["PLA"]) == "PLA"
+
+
+def test_first_str_passes_through_plain_string():
+    assert first_str("PLA") == "PLA"
+
+
+def test_first_str_returns_default_for_empty_list():
+    assert first_str([], "fallback") == "fallback"
+
+
+def test_first_str_returns_default_for_other_types():
+    assert first_str(42, "fallback") == "fallback"
+    assert first_str(None, "fallback") == "fallback"
+
+
+def test_parse_profile_rejects_non_dict():
+    assert parse_profile([1, 2, 3]) is None
+    assert parse_profile("not a dict") is None
+    assert parse_profile(None) is None
+
+
+def test_parse_profile_rejects_stub_without_id_or_parent():
+    data = {"type": "filament"}  # no inherits, no filament_id
+    assert parse_profile(data) is None
+
+
+def test_parse_profile_rejects_instantiation_false():
+    data = {"type": "filament", "filament_id": "GFL01", "instantiation": "false"}
+    assert parse_profile(data) is None
+
+
+def test_parse_profile_minimal_valid_profile():
+    data = {
+        "type": "filament",
+        "filament_id": "GFL01",
+        "name": "Generic PLA",
+        "filament_vendor": ["Generic"],
+        "filament_type": ["PLA"],
+        "default_filament_colour": ["#FFFFFF"],
+    }
+    result = parse_profile(data)
+    assert result == {
+        "id": "GFL01",
+        "name": "Generic PLA",
+        "vendor": "Generic",
+        "type": "PLA",
+        "color": "#FFFFFF",
+    }
+
+
+def test_parse_profile_name_as_list_does_not_crash():
+    """Regression guard: `name` wasn't previously routed through first_str()
+    like the other fields are, unlike filament_vendor/filament_type/
+    default_filament_colour just below it - a list value here used to raise
+    TypeError inside clean_name()'s re.sub()."""
+    data = {
+        "type": "filament",
+        "filament_id": "GFL02",
+        "name": ["Geeetech PLA Basic"],
+        "filament_vendor": ["Geeetech"],
+        "filament_type": ["PLA"],
+    }
+    result = parse_profile(data)
+    assert result is not None
+    assert result["name"] == "Geeetech PLA Basic"
+
+
+def test_parse_profile_missing_name_falls_back_to_filament_id():
+    data = {"type": "filament", "filament_id": "GFL03", "filament_vendor": ["Generic"]}
+    result = parse_profile(data)
+    assert result["name"] == "GFL03"
+
+
+def test_parse_profile_missing_optional_fields_default_to_empty_string():
+    data = {"type": "filament", "filament_id": "GFL04", "name": "Mystery Filament"}
+    result = parse_profile(data)
+    assert result["type"] == ""
+    assert result["color"] == ""
+    assert result["vendor"] == "Generic"  # no path_vendor given either
+
+
+def test_parse_profile_inherits_via_by_name():
+    parent = {"type": "filament", "filament_id": "GFL05", "filament_vendor": ["Geeetech"], "filament_type": ["PLA"]}
+    child = {"type": "filament", "inherits": "Geeetech PLA @base", "name": "Geeetech PLA Basic"}
+    by_name = {"Geeetech PLA @base": [parent]}
+    result = parse_profile(child, by_name=by_name)
+    assert result is not None
+    assert result["id"] == "GFL05"
+    assert result["vendor"] == "Geeetech"
+    assert result["type"] == "PLA"
+
+
+def test_parse_profile_inherits_via_system_index():
+    system_index = [{
+        "id": "GFL06", "name": "Geeetech PLA", "vendor": "Geeetech", "type": "PLA", "color": "",
+    }]
+    user_profile = {"type": "filament", "inherits": "Geeetech PLA @base", "name": "My Geeetech Override"}
+    result = parse_profile(user_profile, system_index=system_index)
+    assert result is not None
+    assert result["id"] == "GFL06"
+    assert result["vendor"] == "Geeetech"
+
+
+def test_parse_profile_inherits_cycle_does_not_infinite_loop():
+    """A inherits B, B inherits A - _resolve()'s hard depth=5 bound must
+    terminate this rather than recursing forever."""
+    a = {"type": "filament", "inherits": "B"}
+    b = {"type": "filament", "inherits": "A"}
+    by_name = {"A": [a], "B": [b]}
+    # Neither profile has a filament_id anywhere in the cycle - must return
+    # None (not hang, not crash) after exhausting the depth limit.
+    result = parse_profile(a, by_name=by_name)
+    assert result is None
+
+
+def test_parse_profile_duplicate_system_names_logs_and_uses_last(caplog):
+    """clean_name() deliberately collapses variant-suffixed names onto the
+    same cleaned name - sys_by_name's last-write-wins overwrite on collision
+    is expected, but must now be observable via a debug log instead of
+    silent."""
+    system_index = [
+        {"id": "GFL07", "name": "PolyTerra PLA", "vendor": "Polymaker", "type": "PLA", "color": ""},
+        {"id": "GFL08", "name": "PolyTerra PLA", "vendor": "Polymaker", "type": "PLA", "color": ""},
+    ]
+    user_profile = {"type": "filament", "inherits": "PolyTerra PLA @base", "name": "Override"}
+    with caplog.at_level(logging.DEBUG, logger="kobrax.filaments"):
+        result = parse_profile(user_profile, system_index=system_index)
+    assert result is not None
+    assert result["id"] == "GFL08"  # last one wins, as before
+    assert any("duplicate system profile name" in r.message for r in caplog.records)
+
+
+def test_parse_profile_bytes_valid_json():
+    blob = json.dumps({
+        "type": "filament", "filament_id": "GFL09", "name": "Test PLA",
+        "filament_vendor": ["Test"], "filament_type": ["PLA"],
+    }).encode("utf-8")
+    result = parse_profile_bytes(blob)
+    assert result is not None
+    assert result["id"] == "GFL09"
+
+
+def test_parse_profile_bytes_malformed_json_returns_none():
+    assert parse_profile_bytes(b"{not valid json") is None
+
+
+def test_parse_profile_bytes_non_dict_json_returns_none():
+    assert parse_profile_bytes(b"[1, 2, 3]") is None
+    assert parse_profile_bytes(b'"just a string"') is None
+    assert parse_profile_bytes(b"42") is None

tests/test_settings.py

@@ -39,6 +39,15 @@ async def test_settings_get_returns_configured_values(client_configured):
     assert data["device_id"]  == "abc123deadbeef"
 
 
+@pytest.mark.asyncio
+async def test_settings_post_invalid_json_returns_400(client):
+    """A malformed/non-JSON body must be a clean 400, not an unhandled 500
+    with a raw JSONDecodeError traceback (code review finding)."""
+    c, _ = client
+    resp = await c.post("/api/settings", data=b"not json", headers={"Content-Type": "application/json"})
+    assert resp.status == 400
+
+
 @pytest.mark.asyncio
 async def test_settings_post_writes_config_ini(client):
     """POST /api/settings schreibt Werte in config.ini (Migration von .env, v0.9.x)."""

tests/test_update_check.py

@@ -12,6 +12,49 @@ from unittest.mock import AsyncMock, MagicMock, patch
 import pytest
 
 
+@pytest.mark.asyncio
+async def test_update_apply_invalid_json_returns_400(client):
+    """A malformed/non-JSON body must be a clean 400, not an unhandled 500
+    with a raw JSONDecodeError traceback (code review finding)."""
+    c, _ = client
+    resp = await c.post("/api/update/apply", data=b"not json", headers={"Content-Type": "application/json"})
+    assert resp.status == 400
+
+
+@pytest.mark.asyncio
+async def test_update_check_testing_channel_is_docker_only(client):
+    """A testing-<sha> build has no Gitea releases at all - the check must
+    report a docker-only channel with nothing to update, NOT fall through to
+    the stable path and wrongly offer a stable "update". Must not even call
+    the Gitea API."""
+    c, bridge = client
+    bridge._read_version = lambda: "testing-2e4dbf0"
+
+    with patch("aiohttp.ClientSession.get") as mock_get:
+        resp = await c.get("/api/update/check")
+        data = await resp.json()
+
+    assert resp.status == 200
+    assert data["update_available"] is False
+    assert data["docker_only"] is True
+    assert data["current"] == "testing-2e4dbf0"
+    mock_get.assert_not_called()  # no Gitea round-trip for the testing channel
+
+
+@pytest.mark.asyncio
+async def test_update_apply_testing_channel_blocked(client):
+    """Self-update must be refused on the testing channel, same as nightly -
+    testing images are delivered via Docker only."""
+    c, bridge = client
+    bridge._read_version = lambda: "testing-2e4dbf0"
+
+    resp = await c.post("/api/update/apply", json={"tag": "whatever"})
+    data = await resp.json()
+    assert resp.status == 400
+    assert "testing" in data["error"]
+    assert "docker" in data["error"].lower()
+
+
 def _fake_releases_response(payload):
     resp = MagicMock()
     resp.status = 200