fix(engine): close concurrent preempt race in target-state ownership transfer (#1994)

When two components race to own the same target state, the prior fix
relied on tokio scheduling to serialize their `pre_commit` ->
`sink_apply` -> `commit` sequences. That holds for LMDB (microsecond
ops) but breaks under PG latency, where the loser's `delete` lands
after the winner's `upsert` and the target state is lost.

Add a `pending_process_token: Option<u128>` to `TargetStateInfoItem`,
written by `pre_commit` whenever it queues a sink action and cleared
by `commit_in_txn`'s retention pass. A detection sub-pass at the top
of `pre_commit` peeks the token on every preempt-source item: live
(same process token) -> return `PendingRetry` with the unconsumed
declared map so `submit()` can re-invoke after a backoff; dead
(crashed prior process) -> force `prev_may_be_missing=true` on
reconcile.

The detection sub-pass also caches old-owner `tracking_info` bytes
into a per-call `HashMap<StablePath, Vec<u8>>` shared with the Phase 1
preempt branch -- each owner is read once, modified in place across
multiple preempts, and emitted as a single `DeferredWrite` at the end
of Phase 1.

`set_provider_generation` (OnceLock-backed, can't run twice) is
deferred until after the last possible PendingRetry exit. On
`sink_apply` / `commit_in_txn` failure, `rollback_pending_tokens`
re-reads tracking_info and clears every token matching this process's
token; retried indefinitely with logged backoff until success. The
process-exits-mid-rollback case is covered by the dead-token recovery
path on next startup.

`contained_target_state_paths` wrapped in `Arc` to avoid full HashSet
rehash on every retry iteration.

See `specs/target_state_ownership_transfer/concurrent_preempt_race_fix.md`.

Author: georgeh0

rust/core/src/engine/environment.rs

@@ -26,6 +26,12 @@ struct EnvironmentInner<Prof: EngineProfile> {
     /// See `specs/memo_validation/plan.md` → "Extension: state validation
     /// for tracked context values".
     context_initial_states: RwLock<HashMap<Fingerprint, Vec<Prof::FunctionData>>>,
+    /// Per-process liveness token. Used by `pre_commit` to distinguish a
+    /// pending tracking-info entry written by this process (live → caller
+    /// backs off and retries) from one left behind by a crashed prior
+    /// process (dead → take the recovery path).
+    /// See `specs/target_state_ownership_transfer/concurrent_preempt_race_fix.md`.
+    process_token: u128,
 }
 
 #[derive(Clone)]
@@ -47,10 +53,16 @@ impl<Prof: EngineProfile> Environment<Prof> {
             host_runtime_ctx,
             logic_set: RwLock::new(HashSet::new()),
             context_initial_states: RwLock::new(HashMap::new()),
+            process_token: uuid::Uuid::new_v4().as_u128(),
         });
         Ok(Self { inner: state })
     }
 
+    /// Liveness token for the current process. See `EnvironmentInner::process_token`.
+    pub fn process_token(&self) -> u128 {
+        self.inner.process_token
+    }
+
     pub fn storage(&self) -> &Storage {
         &self.inner.storage
     }