fix(cfnetwork): preserve shared sessions and delegate-queue completions

Ignore both invalidation methods on the shared session; deliver task completion through the owning session delegate queue. Extend the standalone probe to exercise singleton reuse and a suspended custom queue. Require stable old-client process identities and complete window data before optional recovery, and continue launch after shutdown failures.

tests/test_hidden_riot_recovery.py

@@ -21,6 +21,7 @@ class RecoveryTests(unittest.TestCase):
         self.root = Path(self.temp.name)
         self.proc = self.root / 'proc'
         self.proc.mkdir()
+        (self.proc / "uptime").write_text("1000 0\n")
         self.home = self.root / 'home'
         self.home_patch = patch.object(Path, 'home', return_value=self.home)
         self.home_patch.start()
@@ -29,11 +30,13 @@ class RecoveryTests(unittest.TestCase):
         self.env_patch.start()
         self.addCleanup(self.env_patch.stop)
 
-    def process(self, pid, executable, prefix=None):
+    def process(self, pid, executable, prefix=None, age=120):
         folder = self.proc / str(pid)
         folder.mkdir()
         (folder / 'cmdline').write_bytes(executable.encode() + b'\0')
         (folder / 'environ').write_bytes(b'' if prefix is None else b'DPREFIX=' + os.fsencode(prefix) + b'\0')
+        fields = ['S'] + ['0'] * 18 + [str(int((1000 - age) * os.sysconf('SC_CLK_TCK')))]
+        (folder / 'stat').write_text(str(pid) + ' (Riot Client (worker)) ' + ' '.join(fields))
         return folder
 
     def riot(self):
@@ -59,7 +62,7 @@ class RecoveryTests(unittest.TestCase):
     def test_visible_client_is_preserved(self):
         self.riot()
         for identity in ('Riot Client', 'League of Legends'):
-            result, calls = self.run_recovery([{'class': identity, 'mapped': True, 'hidden': False}])
+            result, calls = self.run_recovery([{'class': identity, 'title': identity, 'mapped': True, 'hidden': False}])
             self.assertFalse(result)
             self.assertEqual(len(calls), 1)
 
@@ -96,7 +99,8 @@ class RecoveryTests(unittest.TestCase):
 
     def test_malformed_compositor_response_is_preserved(self):
         self.riot()
-        for raw in ('invalid', '{}', '[null]'):
+        for raw in ('invalid', '{}', '[null]', '[{}]', '[{"class":"editor"}]',
+                    '[{"class":"Riot","title":"Riot","mapped":"true","hidden":false}]'):
             result, calls = self.run_recovery(raw=raw)
             self.assertFalse(result)
             self.assertEqual(len(calls), 1)
@@ -111,6 +115,45 @@ class RecoveryTests(unittest.TestCase):
         with patch.object(Path, 'read_bytes', read):
             self.assertEqual(self.run_recovery(), (False, []))
 
+    def test_young_client_prevents_recovery_even_if_an_old_client_exists(self):
+        self.riot()
+        self.process(125, '/Applications/Riot Client.app/Contents/MacOS/RiotClientServices', age=5)
+        self.assertEqual(self.run_recovery(), (False, []))
+
+    def test_missing_or_invalid_age_information_prevents_recovery(self):
+        proc = self.riot()
+        for value in ('garbage', '123 (worker) S'):
+            (proc / 'stat').write_text(value)
+            self.assertEqual(self.run_recovery(), (False, []))
+        (proc / 'stat').unlink()
+        self.assertEqual(self.run_recovery(), (False, []))
+
+    def test_shutdown_failures_are_reported_without_aborting_launch(self):
+        self.riot()
+        for error in (FileNotFoundError(), subprocess.TimeoutExpired('darling', 25),
+                      subprocess.CalledProcessError(1, 'darling')):
+            def command(args, **kwargs):
+                if args[0] == 'hyprctl':
+                    return subprocess.CompletedProcess(args, 0, stdout='[]')
+                raise error
+            with patch.object(recovery.subprocess, 'run', side_effect=command):
+                self.assertFalse(recovery.recover(self.proc))
+
+    def test_client_started_during_window_inspection_is_preserved(self):
+        self.riot()
+        def command(args, **kwargs):
+            self.assertEqual(args[0], 'hyprctl')
+            self.process(125, '/Applications/Riot Client.app/Contents/MacOS/RiotClientServices', age=1)
+            return subprocess.CompletedProcess(args, 0, stdout='[]')
+        with patch.object(recovery.subprocess, 'run', side_effect=command):
+            self.assertFalse(recovery.recover(self.proc))
+
+    def test_valid_unrelated_window_does_not_prevent_recovery(self):
+        self.riot()
+        result, calls = self.run_recovery([{'class':'editor', 'title':'file', 'mapped':True, 'hidden':False}])
+        self.assertTrue(result)
+        self.assertEqual(calls[-1], ['darling', 'shutdown'])
+
 
 if __name__ == '__main__':
     unittest.main()

tests/test_urlsession_runner.py

@@ -32,6 +32,8 @@ else:
     raise AssertionError('reserved error endpoint unexpectedly accepted a connection')
 print('LIFECYCLE_PASS: fixture')
 print('URLSESSION_PASS: fixture')
+print('SHARED_SESSION_PASS: fixture')
+print('QUEUE_PASS: fixture')
 ''')
             executable.chmod(0o755)
             probe = root / 'probe'