Bundled KDE Connect: no copy left behind by a stop or a local read error

- A start turned off while copying removes the copy instead of starting an
  agent that would be killed before it could tidy up.
- A local read error while copying removes the partial copy too.
- The retry race test holds the new start until the retry has decided, so
  it fails every time without the fix (checked 3/3), not by luck.

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

tests/test_input.py

@@ -172,6 +172,24 @@ class Bundled(unittest.TestCase):
         with self.assertRaises(self.server.Failure):
             self.deliver(frame_has=False, damaged=True)
 
+    def test_a_failed_copy_is_removed(self):
+        calls = []
+
+        def ssh(remote, stdin=None, timeout=30, text=True):
+            calls.append(remote)
+            if "cat >" in remote and len([c for c in calls if "cat >" in c]) == 2:
+                raise self.server.Failure("Broken pipe")
+            return b"" if not text else ""
+        folder, packages = self.fake_bundle()
+        old = self.server.ssh, self.server.KDECONNECT, self.server.LOCAL
+        self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = ssh, folder, False
+        try:
+            with self.assertRaises(self.server.Failure):
+                self.server.InputAgent(packages=packages).deliver(lambda m: None)
+        finally:
+            self.server.ssh, self.server.KDECONNECT, self.server.LOCAL = old
+        self.assertTrue(calls[-1].startswith("rm -rf .local/share/frame-control/kdeconnect/incoming/"), calls[-1])
+
     def lifecycle(self, agent_lines, deliver=None):
         """An InputAgent whose ssh and agent are fakes. Returns it, the folders each launch
         used, and the fake agents. A fake agent reports its lines, then keeps running
@@ -263,33 +281,63 @@ class Bundled(unittest.TestCase):
         agent.start()  # while the first launch is still copying
         self.assertTrue(entered.wait(5), "the second start didn't launch")
         gate.set()
-        self.wait_for(lambda: len(procs) == 2 and agent.status == {"state": "ready"})
-        # The stopped launch's agent was ended; the new one is the one in use.
-        self.wait_for(lambda: sum(p.ended.is_set() for p in procs) == 1)
-        self.assertFalse(agent.proc.ended.is_set())
+        self.wait_for(lambda: agent.status == {"state": "ready"})
+        time.sleep(0.1)
+        # The stopped launch started no agent; the new one is the one in use.
+        self.assertEqual(len(procs), 1)
+        self.assertIs(agent.proc, procs[0])
         agent.stop()
         self.wait_for(lambda: all(p.ended.is_set() for p in procs))
 
     def test_retry_and_a_new_start_never_run_two_agents(self):
         # A start() landing just as an agent that asked for the packages exits: exactly one
-        # of it and the retry launches, and stop() ends everything.
+        # of it and the retry launches, and stop() ends everything. The new start is held
+        # in its copy until the retry has decided, so neither can see the other's agent.
+        first, decided = [], threading.Event()
         agent, launches, procs = self.lifecycle([[b'{"state": "need-packages"}\n'], [b'{"state": "ready"}\n']])
-        watch, raced = agent._watch, []
+        watch, launch, deliver = agent._watch, agent._launch, agent.deliver
 
         def racing_watch(proc, errors, retry=False):
             wanted = watch(proc, errors, retry)
-            if wanted and not raced:
-                raced.append(True)
+            if wanted and not first:
+                first.append(threading.current_thread())
                 agent.start()  # lands between the agent exiting and the retry
             return wanted
-        agent._watch = racing_watch
+
+        def held_deliver(report, force=False):
+            if force:
+                decided.set()  # the retry went ahead
+            elif first and threading.current_thread() is not first[0]:
+                decided.wait(5)  # the new start waits until the retry has decided
+            return deliver(report, force)
+
+        def first_launch(generation, force=False):
+            try:
+                launch(generation, force)
+            finally:
+                if first and threading.current_thread() is first[0]:
+                    decided.set()  # the retry declined
+        agent._watch, agent._launch, agent.deliver = racing_watch, first_launch, held_deliver
         agent.start()
         self.wait_for(lambda: agent.status == {"state": "ready"})
-        time.sleep(0.2)
+        time.sleep(0.1)
         self.assertEqual(sum(not p.ended.is_set() for p in procs), 1, launches)
         agent.stop()
         self.wait_for(lambda: all(p.ended.is_set() for p in procs))
 
+    def test_stopped_during_the_copy_removes_it(self):
+        gate, entered = threading.Event(), threading.Event()
+        agent, launches, procs = self.lifecycle([], deliver=lambda: (entered.set(), gate.wait(5)))
+        removed = []
+        agent.discard = removed.append
+        agent.deliver = lambda report, force=False: (entered.set(), gate.wait(5), "~/incoming/x")[2]
+        agent.start()
+        self.assertTrue(entered.wait(5))
+        agent.stop()
+        gate.set()
+        self.wait_for(lambda: removed == ["incoming/x"])
+        self.assertEqual(procs, [])  # no agent started for it
+
     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

@@ -648,14 +648,19 @@ class InputAgent:
                 quoted = shlex.quote(name)
                 ssh(f"cd {folder} && cat > {quoted}.part && mv {quoted}.part {quoted}",
                     stdin=path.read_bytes(), text=False, timeout=600)
-        except Failure:
-            try:
-                ssh(f"rm -rf {folder}", timeout=20)  # a partial copy is no use to anyone
-            except Failure:
-                pass  # the agent tidies it later
+        except (Failure, OSError):
+            self.discard(folder)  # a partial copy is no use to anyone
             raise
         return f"~/{folder}"
 
+    def discard(self, folder):
+        """Remove a copy no agent will take over (best effort; agents tidy up old ones too)."""
+        if folder and not LOCAL:
+            try:
+                ssh(f"rm -rf {folder}", timeout=20)
+            except Failure:
+                pass
+
     def start(self):
         with self.lock:
             if self.launching == self.generation or (self.proc and self.proc.poll() is None):
@@ -674,6 +679,11 @@ class InputAgent:
         try:
             ensure_master()
             folder = self.deliver(report, force)
+            with self.lock:
+                stopped = self.generation != generation
+            if stopped:  # turned off while copying: no agent will take the copy over
+                self.discard(folder.removeprefix("~/"))
+                raise Failure("stopped")
             proc = subprocess.Popen([*SSH, FRAME, self.command(folder)], stdin=subprocess.PIPE,
                                     stdout=subprocess.PIPE, stderr=errors)
         except (Failure, OSError) as e: