fix(lifecycle): preserve ownership, cancellation, and cleanup outcomes (#4238)

* fix(cleanup): preserve owner recovery instructions

Keep cleanup remediation through combined errors and direct or managed one-shot analysis while retaining primary diagnostics.

Refs: #4227

* fix(hopper): recover late startup and retained cleanup

Authenticate initial and cleanup handshakes through the same identity owner. Expose startup and cleanup activity, retain health after target retirement, and expire abandoned unauthenticated listeners without timing out authenticated work.

Refs: #4227

* fix(providers): retain composite lifecycle facts across retries

Preserve observed child health with incomplete peer coverage, retry only unresolved clients, and retain all child cleanup recovery actions.

Refs: #4233

* fix(process): honor cancellation through capture cleanup

Check the request signal after cleanup and before final publication, retaining completed observations under typed cancellation.

Refs: #4005

* fix(native): preserve failed LLDB launch ownership and exit proof

Publish failed-launch process identity and observed exit separately from object validity. Retire only the exact owned process and retain uncertainty when exit cannot be established.

Refs: #4232

* fix(native): honor cancellation after trace finalization

Check cancellation after runtime-root cleanup and retain collected trace observations as partial cancellation evidence.

Refs: #4009

* fix(cli): preserve analysis results when provider cleanup fails

Reuse the existing one-shot finalizer across thirteen command families to retain completed Evidence and primary diagnostics on close failure. Cover success, denied cleanup, and combined primary/cleanup failure through the public CLI.

Refs: #3935

* docs(lifecycle): define completion invariants and regression scenarios

Document distinct ownership/readiness/exit facts, late cancellation and atomic publication boundaries, partial child cleanup, and diagnostic retention. Link representative regressions and route contributor guidance through AGENTS.md.

Refs: #4227

* fix(session): retain run identity through pending cleanup

* fix(cli): preserve original cleanup resource diagnostics

* fix(native): retain completed facts through finalization failure

* fix(providers): report retained cleanup paths at their owners

bridge/hopper_bridge.py

@@ -1057,6 +1057,10 @@ def _serve_connection(connection):
                 ):
                     raise PermissionError("Invalid bridge capability")
                 authenticated = True
+                # The connection grace period only bounds unauthenticated idle
+                # clients. Native analysis and a connected owner's idle time
+                # remain unbounded after capability validation.
+                connection.settimeout(None)
                 _write_message(file, {"id": request_id, "event": {
                     "type": "progress",
                     "phase": "hopper_bridge",
@@ -1148,13 +1152,17 @@ def _run():
     server.bind(REA_SOCKET)
     os.chmod(REA_SOCKET, 0o600)
     server.listen(1)
+    # A bridge can start after its caller's startup/cleanup deadline. Bound
+    # the first accept as well as reconnects so an orphan cannot pin Hopper's
+    # dedicated Python thread and prevent later documents from starting REA.
+    server.settimeout(REA_RECONNECT_TIMEOUT_SECONDS)
     try:
         while True:
             try:
                 connection, _ = server.accept()
             except socket.timeout:
                 break
-            connection.settimeout(None)
+            connection.settimeout(REA_RECONNECT_TIMEOUT_SECONDS)
             if _serve_connection(connection):
                 break
             # Match REA's cleanup handshake budget without pinning Hopper's

bridge/native/rea_lldb_tracer.py

@@ -555,6 +555,38 @@ def _failed_launch_description(process):
     return None
 
 
+def _retire_failed_launch(process, config, listener, stdout, stderr):
+    """Reconcile an SBProcess returned alongside a failed Launch error."""
+    if not process.IsValid():
+        stdout.close()
+        stderr.close()
+        return False
+
+    pid = process.GetProcessID()
+    if pid > 0:
+        # Even an already-exited process needs its PID published so the host
+        # can distinguish a completed launch failure from an unknown target.
+        with open(config["pid_path"], "w") as handle:
+            handle.write(str(pid))
+
+    if process.GetState() != lldb.eStateExited:
+        stdout.mark_launched()
+        stderr.mark_launched()
+        if pid > 0:
+            # Hold a surviving target until the host has captured its start
+            # identity. If that proof is unavailable, Kill still addresses
+            # this exact SBProcess created by the current launch request.
+            _wait_for_identity_ack(config)
+        process.Kill()
+        _wait_for_exit(listener, process, 5)
+
+    exit_observed = process.GetState() == lldb.eStateExited
+    expected_exit = exit_observed
+    stdout.close(expected_exit=expected_exit)
+    stderr.close(expected_exit=expected_exit)
+    return exit_observed
+
+
 def trace(debugger, config):
     global _TRACE_CONTEXT
     debugger.SetAsync(True)
@@ -630,8 +662,9 @@ def trace(debugger, config):
             )
         process = target.Launch(_launch_info(target, config, listener), error)
         if not error.Success() or not process.IsValid():
-            stdout.close()
-            stderr.close()
+            exit_observed = _retire_failed_launch(
+                process, config, listener, stdout, stderr
+            )
             return {
                 "status": "launch-error",
                 "error": error.GetCString(),
@@ -640,15 +673,26 @@ def trace(debugger, config):
                     if process.IsValid()
                     else {}
                 ),
+                **(
+                    {"target_process_exit_observed": True}
+                    if exit_observed
+                    else {}
+                ),
             }
         launch_failure = _failed_launch_description(process)
         if launch_failure is not None:
-            stdout.close(expected_exit=True)
-            stderr.close(expected_exit=True)
+            exit_observed = _retire_failed_launch(
+                process, config, listener, stdout, stderr
+            )
             return {
                 "status": "launch-error",
                 "error": launch_failure,
                 "target_process_created": True,
+                **(
+                    {"target_process_exit_observed": True}
+                    if exit_observed
+                    else {}
+                ),
             }
         stdout.mark_launched()
         stderr.mark_launched()

bridge/native/test_rea_lldb_tracer.py

@@ -15,6 +15,8 @@ fake_lldb = sys.modules.setdefault("lldb", types.ModuleType("lldb"))
 fake_lldb.LLDB_INVALID_ADDRESS = -1
 fake_lldb.eLaunchFlagDebug = 2
 fake_lldb.eLaunchFlagStopAtEntry = 4
+fake_lldb.eStateExited = 10
+fake_lldb.SBEvent = lambda: object()
 spec = importlib.util.spec_from_file_location(
     "rea_lldb_tracer", os.path.join(os.path.dirname(__file__), "rea_lldb_tracer.py")
 )
@@ -23,6 +25,136 @@ spec.loader.exec_module(tracer)
 
 
 class NativeTraceBudgetTests(unittest.TestCase):
+    def test_failed_launch_does_not_infer_exit_from_a_failed_kill(self):
+        class Process:
+            def IsValid(self):
+                return True
+
+            def GetProcessID(self):
+                return 12345
+
+            def GetState(self):
+                return 6
+
+            def Kill(self):
+                return None
+
+        class Sink:
+            close_exit = None
+
+            def mark_launched(self):
+                pass
+
+            def close(self, expected_exit=False):
+                self.close_exit = expected_exit
+
+        with tempfile.TemporaryDirectory() as directory:
+            stdout, stderr = Sink(), Sink()
+            with patch.object(tracer, "_wait_for_identity_ack", return_value="unavailable"), patch.object(
+                tracer, "_wait_for_exit", return_value=None
+            ):
+                exited = tracer._retire_failed_launch(
+                    Process(), {"pid_path": os.path.join(directory, "pid")}, object(), stdout, stderr
+                )
+        self.assertFalse(exited)
+        self.assertFalse(stdout.close_exit)
+        self.assertFalse(stderr.close_exit)
+
+    def test_failed_launch_publishes_pid_and_retires_live_process(self):
+        class Process:
+            state = 6
+            killed = False
+
+            def IsValid(self):
+                return True
+
+            def GetProcessID(self):
+                return 12345
+
+            def GetState(self):
+                return self.state
+
+            def Kill(self):
+                self.killed = True
+                self.state = tracer.lldb.eStateExited
+
+        class Sink:
+            launched = False
+            close_exit = None
+
+            def mark_launched(self):
+                self.launched = True
+
+            def close(self, expected_exit=False):
+                self.close_exit = expected_exit
+
+        with tempfile.TemporaryDirectory() as directory:
+            pid_path = os.path.join(directory, "pid")
+            process = Process()
+            stdout, stderr = Sink(), Sink()
+
+            def acknowledge(config):
+                with open(config["pid_path"], "r") as pid_file:
+                    self.assertEqual(pid_file.read(), "12345")
+                return "readable"
+
+            with patch.object(tracer, "_wait_for_identity_ack", acknowledge):
+                exited = tracer._retire_failed_launch(
+                    process,
+                    {"pid_path": pid_path},
+                    object(),
+                    stdout,
+                    stderr,
+                )
+
+        self.assertTrue(exited)
+        self.assertTrue(process.killed)
+        self.assertTrue(stdout.launched)
+        self.assertTrue(stderr.launched)
+        self.assertTrue(stdout.close_exit)
+        self.assertTrue(stderr.close_exit)
+
+    def test_failed_launch_records_when_valid_process_is_already_exited(self):
+        class Process:
+            def IsValid(self):
+                return True
+
+            def GetProcessID(self):
+                return 12345
+
+            def GetState(self):
+                return tracer.lldb.eStateExited
+
+            def Kill(self):
+                raise AssertionError("an exited process must not be killed")
+
+        class Sink:
+            close_exit = None
+
+            def mark_launched(self):
+                raise AssertionError("an exited process has no output to drain")
+
+            def close(self, expected_exit=False):
+                self.close_exit = expected_exit
+
+        with tempfile.TemporaryDirectory() as directory:
+            pid_path = os.path.join(directory, "pid")
+            stdout, stderr = Sink(), Sink()
+            with patch.object(
+                tracer,
+                "_wait_for_identity_ack",
+                side_effect=AssertionError("exit is already observed"),
+            ):
+                exited = tracer._retire_failed_launch(
+                    Process(), {"pid_path": pid_path}, object(), stdout, stderr
+                )
+            with open(pid_path, "r") as pid_file:
+                self.assertEqual(pid_file.read(), "12345")
+
+        self.assertTrue(exited)
+        self.assertTrue(stdout.close_exit)
+        self.assertTrue(stderr.close_exit)
+
     def test_pre_target_integrity_failure_reports_no_process_created(self):
         original_debugger = getattr(tracer.lldb, "SBDebugger", None)