Bundled KDE Connect: one safety net for every launch failure

Any unexpected error while launching clears the launch and reports it, so
the pad can always be started again; the temporary stderr file is made
inside the handled path and a failure reading it is tolerated.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

tests/test_input.py

@@ -379,6 +379,21 @@ class Bundled(unittest.TestCase):
         agent.start()
         self.wait_for(lambda: agent.status == {"state": "ready"})
 
+    def test_any_unexpected_error_still_reports_and_allows_a_retry(self):
+        agent, launches, procs = self.lifecycle([])
+        old = self.server.tempfile.TemporaryFile
+        self.server.tempfile.TemporaryFile = lambda: (_ for _ in ()).throw(OSError(24, "Too many open files"))
+        try:
+            agent.start()
+            self.wait_for(lambda: agent.status.get("state") == "error")
+        finally:
+            self.server.tempfile.TemporaryFile = old
+        self.assertIsNone(agent.launching)
+        agent.deliver = lambda report, force=False: (_ for _ in ()).throw(ValueError("something odd"))
+        agent.start()
+        self.wait_for(lambda: "something odd" in agent.status.get("message", ""))
+        self.assertIsNone(agent.launching)
+
     def test_copies_go_to_a_folder_of_their_own(self):
         first, _, _ = self.deliver(frame_has=False)
         second, _, _ = self.deliver(frame_has=False)

ui/server.py

@@ -672,12 +672,23 @@ class InputAgent:
         threading.Thread(target=self._launch, args=(generation,), daemon=True).start()
 
     def _launch(self, generation, force=False):
+        try:
+            self._launch_once(generation, force)
+        except Exception as e:  # whatever went wrong, never leave it stuck "starting"
+            with self.lock:
+                if self.launching == generation:
+                    self.launching = None
+                if self.generation == generation and self.status.get("state") in ("starting", "installing"):
+                    self.status = {"state": "error", "message": f"Couldn't start the keyboard and trackpad: {e}"}
+
+    def _launch_once(self, generation, force):
         def report(message):
             with self.lock:
                 if self.generation == generation:
                     self.status = {"state": "installing", "message": message}
-        errors, folder = tempfile.TemporaryFile(), ""
+        folder = ""
         try:
+            errors = tempfile.TemporaryFile()
             ensure_master()
             folder = self.deliver(report, force)
             with self.lock:
@@ -740,8 +751,11 @@ class InputAgent:
                         self.status = status
         proc.wait()
         _live_tunnels.discard(proc)
-        errors.seek(0)
-        detail = strip_ansi(errors.read().decode(errors="replace")).strip()
+        try:
+            errors.seek(0)
+            detail = strip_ansi(errors.read().decode(errors="replace")).strip()
+        except OSError:
+            detail = ""
         with self.lock:
             if self.proc is proc and self.status.get("state") != "error" and not (wanted and retry):
                 message = detail.splitlines()[-1] if detail else "The connection to the Frame ended"