diff --git a/epiphany-core/src/heartbeat_state.rs b/epiphany-core/src/heartbeat_state.rs index cbf4371d..99a42a8e 100644 --- a/epiphany-core/src/heartbeat_state.rs +++ b/epiphany-core/src/heartbeat_state.rs @@ -235,15 +235,19 @@ pub fn tick_heartbeat_store( complete_pending_turn(&mut state, index)?; resident_ack_to_consume = Some((resident_store.to_path_buf(), ack.ack_id)); } + let resident_pressure_pending = + if let Some(resident_store) = options.resident_self_store.as_deref() { + crate::pending_resident_self_pressure(resident_store)? + } else { + false + }; let mut effective_options = options.clone(); - if let Some(resident_store) = options.resident_self_store.as_deref() - && crate::pending_resident_self_pressure(resident_store)? - { + if resident_pressure_pending { effective_options.coordinator_action = Some("resident-self-pressure".to_string()); effective_options.target_role = Some("coordinator".to_string()); effective_options.defer_completion = true; } - let result = tick_once(&mut state, &effective_options)?; + let result = tick_once(&mut state, &effective_options, resident_pressure_pending)?; commit_heartbeat_state_transaction(store_path, expected_state, &state)?; if result["event"]["selectedRole"] == "coordinator" && effective_options.defer_completion @@ -661,7 +665,7 @@ pub fn pump_heartbeat_store( agent_store: options.agent_store.clone(), resident_self_store: options.resident_self_store.clone(), }; - match tick_once(&mut state, &tick_options) { + match tick_once(&mut state, &tick_options, false) { Ok(result) => { write_json_artifact( artifact_dir.join(format!("{}.initiative.json", tick_options.schedule_id)), @@ -1142,14 +1146,19 @@ pub fn recover_stale_heartbeat_store( fn tick_once( state: &mut EpiphanyHeartbeatStateEntry, options: &HeartbeatTickOptions, + force_work_role: bool, ) -> Result { let rate = state.target_heartbeat_rate.max(0.001); let work_role = work_role_for_action( options.coordinator_action.as_deref(), options.target_role.as_deref(), ); - let (selected_index, selection_kind, selection_reason) = - select_participant(state, work_role.as_deref(), options.urgency)?; + let (selected_index, selection_kind, selection_reason) = select_participant( + state, + work_role.as_deref(), + options.urgency, + force_work_role, + )?; let selected = state.participants[selected_index].clone(); let action = action_for_selection( state, @@ -1570,6 +1579,7 @@ fn select_participant( state: &EpiphanyHeartbeatStateEntry, work_role: Option<&str>, urgency: f64, + force_work_role: bool, ) -> Result<(usize, &'static str, String)> { let active: Vec = state .participants @@ -1595,6 +1605,15 @@ fn select_participant( .unwrap_or("unknown") )); } + if force_work_role && candidate.status == "active" { + return Ok(( + index, + "admitted_work", + format!( + "Admitted {work_role} work selected its owning role before idle initiative." + ), + )); + } let reaction_readiness = candidate.reaction_bias * urgency - candidate.current_load; if candidate.status == "active" && reaction_readiness >= candidate.interrupt_threshold { return Ok(( @@ -4257,6 +4276,64 @@ mod tests { Ok(()) } + #[test] + fn resident_pressure_selects_self_before_idle_lane_initiative() -> Result<()> { + let temp = tempfile::tempdir()?; + let heartbeat_store = temp.path().join("heartbeat.cc"); + let resident_store = temp.path().join("resident.cc"); + let artifact_dir = temp.path().join("artifacts"); + initialize_heartbeat_store(&heartbeat_store, 1.0)?; + let mut heartbeat = load_heartbeat_state_entry(&heartbeat_store)?.unwrap(); + heartbeat + .participants + .iter_mut() + .find(|participant| participant.role_id == "coordinator") + .unwrap() + .next_ready_at = 100.0; + write_heartbeat_state_entry(&heartbeat_store, &heartbeat)?; + crate::enqueue_resident_self_pressure( + &resident_store, + &crate::ResidentSelfPressure { + schema_version: crate::RESIDENT_SELF_PRESSURE_SCHEMA_VERSION.into(), + pressure_id: "wake-health".into(), + kind: "operator-objective".into(), + provenance_ref: "test://wake-health".into(), + objective: "Perform a bounded health check.".into(), + created_at_millis: 1, + status: "pending".into(), + consumed_by_grant_id: None, + private_state_exposed: false, + }, + )?; + + let pulse = pulse_resident_self_heartbeat( + &heartbeat_store, + &resident_store, + &artifact_dir, + false, + "wake-health", + "test://wake-health", + None, + )?; + + assert_eq!(pulse.status, "granted"); + assert!(pulse.grant_id.is_some()); + let heartbeat = load_heartbeat_state_entry(&heartbeat_store)?.unwrap(); + assert_eq!( + heartbeat + .participants + .iter() + .find(|participant| participant.role_id == "coordinator") + .and_then(|participant| participant.pending_turn.as_ref()) + .map(|turn| turn.action_id.as_str()), + Some("heartbeat.coordinator.work") + ); + assert!(heartbeat.participants.iter().all(|participant| { + participant.role_id == "coordinator" || participant.pending_turn.is_none() + })); + Ok(()) + } + #[test] fn high_heat_cannot_requeue_running_participant() -> Result<()> { let temp = tempfile::tempdir()?;