Stabilize resident coordinator thread ownership

epiphany-core/src/bin/epiphany-swarm.rs

@@ -12,7 +12,8 @@ use epiphany_core::{
     import_bifrost_persona_feedback_deliveries, ingest_resident_self_domain_pressure,
     load_epiphany_cultmesh_swarm_brake, load_resident_self_state, observe_process_instance,
     prepare_resident_self_launch, publish_resident_provider_readiness, resident_self_child_claim,
-    resident_self_local_provider_status, terminate_process_instance,
+    resident_coordinator_thread_id, resident_self_local_provider_status,
+    terminate_process_instance,
     validate_persona_feedback_store_separation, validate_resident_self_store_separation,
 };
 use serde_json::json;
@@ -185,7 +186,7 @@ fn cycle(
             return Ok(ResidentSelfOutcome::Braked);
         }
         let launch = CoordinatorLaunch {
-            turn_id: format!("resident-self-turn-{}", prepared.grant.grant_id),
+            turn_id: resident_coordinator_thread_id(&args.policy.release_runtime_id)?,
             wake: epiphany_core::ResidentSelfWake::Explicit {
                 objective: prepared.grant.objective.clone(),
             },
@@ -282,7 +283,7 @@ fn cycle(
         return Ok(ResidentSelfOutcome::Sleeping);
     };
     let launch = CoordinatorLaunch {
-        turn_id: format!("resident-self-turn-{}", prepared.grant.grant_id),
+        turn_id: resident_coordinator_thread_id(&args.policy.release_runtime_id)?,
         wake: epiphany_core::ResidentSelfWake::Explicit {
             objective: prepared.grant.objective.clone(),
         },

epiphany-core/src/resident_self.rs

@@ -533,6 +533,25 @@ pub fn coordinator_argv(
     argv
 }
 
+pub fn resident_coordinator_thread_id(runtime_id: &str) -> Result<String> {
+    let runtime_id = runtime_id.trim();
+    if runtime_id.is_empty() {
+        return Err(anyhow!("resident coordinator thread requires runtime identity"));
+    }
+    Ok(format!("resident-self-thread-{runtime_id}"))
+}
+
+fn prepared_coordinator_thread_id(argv: &[String]) -> Result<String> {
+    let index = argv
+        .iter()
+        .position(|value| value == "--thread-id")
+        .ok_or_else(|| anyhow!("prepared coordinator argv lost thread identity"))?;
+    argv.get(index + 1)
+        .filter(|value| !value.trim().is_empty())
+        .cloned()
+        .ok_or_else(|| anyhow!("prepared coordinator argv has empty thread identity"))
+}
+
 #[cfg(test)]
 fn reconcile_resident_self(
     state: &mut ResidentSelfState,
@@ -1033,7 +1052,7 @@ pub fn prepare_resident_self_launch(
     let Some(mut grant) = pending_resident_self_grant(path)? else {
         return Ok(None);
     };
-    let turn_id = format!("resident-self-turn-{}", grant.grant_id);
+    let turn_id = resident_coordinator_thread_id(&policy.release_runtime_id)?;
     let wake = ResidentSelfWake::Explicit {
         objective: grant.objective.clone(),
     };
@@ -1196,7 +1215,7 @@ pub fn acknowledge_resident_self_launch(
         ));
     }
     let lease = ResidentSelfTurnLease {
-        turn_id: format!("resident-self-turn-{}", prepared.grant.grant_id),
+        turn_id: prepared_coordinator_thread_id(&prepared.argv)?,
         wake: ResidentSelfWake::Explicit {
             objective: prepared.grant.objective.clone(),
         },