fix(engine): bind the targeting under the lock that protects it

CHANGELOG-INTERNAL.md

@@ -42,6 +42,33 @@ a `### BREAKING` section placed FIRST in that version, and each such line is pre
 - Help text and the flag tables in both READMEs now state that the flag is valid on its own.
 - Version bump deliberately NOT taken (convention 34): the owner closes it in `VERSION.txt`.
 
+### Changed: start() binds the targeting under the lock that protects it
+
+`_start_locked` read `self._targeting` bare and then acted on that reference three times
+(`set_table`, a synchronous `refresh`, `retarget`), while every other access to the field - in
+`set_target` and `target_for` - is under `_target_lock`. A concurrent `set_target()` landing in the
+middle would leave the resolver pointed at an ORPHAN while the core tested against the object that
+replaced it, which is the same class of mismatch `set_target()` itself was fixed for.
+
+- **This is consistency of access, NOT a fixed symptom, and the commit says so.** The race was not
+  reproduced: 25 start/stop cycles against a concurrent applier in `test_concurrency_chaos.py` stayed
+  green, and the window is narrow today because the CLI applies settings before start, so only a GUI
+  "Apply" landing inside start could hit it. Recorded this way deliberately, so a later session does
+  not read it as "we fixed a race we had seen".
+- Lock order checked before widening the hold, not after: `_stop_lock` -> `_target_lock` ->
+  `ProcessTargeting._lock` -> `PortTable._lock`, and nothing walks it the other way - the resolver
+  never calls back into the engine, and `retarget()` is only reached with `_target_lock` already
+  held. In this spot there is not even contention: `_resolver.start()` comes after the block and
+  `stop()` joins the resolver, so the synchronous `refresh()` has no contender. Cost is one cold
+  resolve (~36 ms elevated, per `portmap.info`) at session start.
+- Rejected: locking only the READ. It narrows the window without closing it, and the point is that
+  start binds ONE coherent object rather than that a mismatch becomes less likely.
+- New test: `tests/test_target_resolver.py::test_start_binds_the_targeting_under_the_lock_that_protects_it`
+  - a mechanical guard (an instrumented `_target_lock` that records acquisitions), not a timing test,
+  because a timing test for this window would be flaky and prove nothing either way. Mutation-checked:
+  reverting to the bare read turns it red. Its docstring states what it does not claim - that the
+  race was real.
+
 ### Tests: the concurrency chaos suite now includes the SocketWatcher
 
 The suite's own charter is "many threads hammering one engine" and it names the failure it exists to

beantester/engine.py

@@ -592,18 +592,39 @@ def _start_locked(self, filt, divert, duration, socket_source=None):
             # otherwise None and the poller stands. A failure to open the SOCKET
             # handle degrades, not kills.
             self._start_socketwatch(real_windivert, socket_source)
-            targeting = self._targeting
-            if targeting is not None:
-                # Point targeting at the live map when we have one, at the poller
-                # otherwise (2c). Targeting may have been built before start() (the
-                # GUI applies it first), against the poller, so this is where it is
-                # (re)bound to whatever this session actually has.
-                targeting.set_table(self._targeting_table())
-                with crashlog.quiet("engine.target"):
-                    targeting.refresh()
-                # Reconcile: whichever path installed the target (target_for alone, or
-                # set_target), the session starts with the resolver pointed at it.
-                self._resolver.retarget(targeting)
+            # Held across the whole binding, not just the read. ``_targeting`` is
+            # shared with every other thread that can install a target (a GUI
+            # "Apply", a scenario step), and this block reads it and then acts on
+            # that reference three times. Bare, a concurrent set_target() landing in
+            # the middle would leave the resolver pointed at an ORPHAN while the core
+            # tested against the object that replaced it - the same class of mismatch
+            # set_target() was fixed for (see its docstring). Consistency of access,
+            # not an observed symptom: the race was NOT reproduced (25 start/stop
+            # cycles against a concurrent applier in test_concurrency_chaos.py stayed
+            # green), and it is narrow today because the CLI applies before start.
+            #
+            # Safe to hold here. Lock order is _stop_lock -> _target_lock ->
+            # ProcessTargeting._lock -> PortTable._lock, and nothing walks it the
+            # other way: the resolver never calls back into the engine, and
+            # retarget() is only ever reached with _target_lock already held. In this
+            # spot there is not even contention - _resolver.start() is below, and
+            # stop() joins the resolver - so the synchronous refresh() has no
+            # contender; it costs a cold resolve (~36 ms elevated, see portmap.info)
+            # once, at session start.
+            with self._target_lock:
+                targeting = self._targeting
+                if targeting is not None:
+                    # Point targeting at the live map when we have one, at the poller
+                    # otherwise (2c). Targeting may have been built before start()
+                    # (the GUI applies it first), against the poller, so this is
+                    # where it is (re)bound to whatever this session actually has.
+                    targeting.set_table(self._targeting_table())
+                    with crashlog.quiet("engine.target"):
+                        targeting.refresh()
+                    # Reconcile: whichever path installed the target (target_for
+                    # alone, or set_target), the session starts with the resolver
+                    # pointed at it.
+                    self._resolver.retarget(targeting)
             # UNCONDITIONALLY, target or no target: the resolver's life is the SESSION's.
             # Starting it only when a target already exists meant that narrowing down
             # mid-run - press START, watch, then type a process - left nobody keeping

tests/test_target_resolver.py

@@ -468,3 +468,45 @@ def test_repeated_start_stop_cycles_do_not_stack_resolver_threads():
     time.sleep(0.2)
     leaked = {t.name for t in threading.enumerate()} - before
     check("five start/stop cycles leave no thread behind", not leaked, f"({leaked})")
+
+
+def test_start_binds_the_targeting_under_the_lock_that_protects_it():
+    """``_targeting`` is shared with every thread that can install a target, so the
+    start path must not read it bare and then act on that reference three times.
+
+    A mechanical guard rather than an attempt to reproduce the race: the window is
+    narrow (the CLI applies before start, so only a GUI "Apply" landing inside start
+    could hit it) and a timing test for it would be flaky and prove nothing either
+    way. What this pins down is that the access is CONSISTENT with every other one -
+    if a concurrent set_target() ever does land in the middle, the resolver cannot end
+    up pointed at an orphan while the core tests against its replacement.
+
+    This does NOT claim the race was real; it was not reproduced. It claims the field
+    is no longer touched without its lock.
+    """
+    targeting, _ = _targeting()
+    engine = BeanEngine()
+    engine.set_target(True, targeting)
+
+    real_lock = engine._target_lock
+    taken = []
+
+    class _Watched:
+        """Passes straight through to the real lock, recording every acquisition."""
+
+        def __enter__(self):
+            taken.append(True)
+            return real_lock.__enter__()
+
+        def __exit__(self, *exc):
+            return real_lock.__exit__(*exc)
+
+    engine._target_lock = _Watched()
+    try:
+        engine.start("true", divert=SyntheticDivert(seed=7))
+        engine.stop()
+    finally:
+        engine._target_lock = real_lock
+
+    check("start() took the targeting lock before binding the target", taken,
+          f"({taken})")