pub fn refresh_control_ownership(zone: &mut Zone, _house_power_enabled: bool) {
    let now = Utc::now();
    let (owner, source, resume_at, reason) = if zone.device_manual_override {
        let source = match zone.control_source.as_str() {
            "home_assistant_direct" | "web_direct" | "external" => zone.control_source.clone(),
            _ => "external".into(),
        };
        ("direct_manual", source, zone.device_manual_override_until, "Direct/manual device control has priority".to_string())
    } else if zone.local_thermostat_power.is_some() {
        let source = match zone.control_source.as_str() {
            "home_assistant_thermostat" | "web_thermostat" => zone.control_source.clone(),
            _ => "local_thermostat".into(),
        };
        let resume_at = zone.temporary_quick_thermostat.as_ref()
            .and_then(temporary_quick_thermostat_next_deadline)
            .or(zone.local_thermostat_resume_at.clone());
        let reason = if zone.local_thermostat_power == Some(false) {
            "Local thermostat is explicitly off".into()
        } else if zone.temporary_quick_thermostat.is_some() {
            "Temporary Quick Thermostat owns the zone".into()
        } else {
            "Local thermostat owns the zone".into()
        };
        ("local_thermostat", source, resume_at, reason)
    } else {
        let source = if zone.control_source.starts_with("group:")
            || matches!(zone.control_source.as_str(), "automation.device" | "house_power")
        {
            zone.control_source.clone()
        } else {
            "automation".into()
        };
        ("automation", source, zone.manual_override_until, "Automatic thermostat/schedule control".to_string())
    };
    if zone.control_owner != owner || zone.control_source != source {
        zone.control_since = Some(now);
    } else if zone.control_since.is_none() {
        zone.control_since = Some(now);
    }
    zone.control_owner = owner.into();
    zone.control_source = source;
    zone.control_resume_at = resume_at;
    zone.control_reason = reason;
}

fn normalized_direct_source(source: &str) -> &'static str {
    if source.contains("home_assistant") { "home_assistant_direct" }
    else if source == "device.manual_control" { "web_direct" }
    else { "external" }
}

pub fn reset_device_manual_override(zone: &mut Zone) -> bool {
    let now = Utc::now();
    let changed = zone.device_manual_override
        || zone.device_manual_override_since.is_some()
        || zone.device_manual_override_until.is_some()
        || !zone.device_manual_override_fields.is_empty()
        || zone.device_manual_override_baseline.is_some();
    zone.device_manual_override = false;
    zone.device_manual_override_since = None;
    zone.device_manual_override_until = None;
    zone.device_manual_override_fields.clear();
    zone.device_manual_override_baseline = None;
    if let Some(session) = zone.temporary_quick_thermostat.as_mut() {
        let pause = session.paused_at.take()
            .map(|paused_at| now.signed_duration_since(paused_at))
            .filter(|pause| *pause > chrono::Duration::zero());
        if session.activated_at.is_some() {
            if let Some(pause) = pause {
                if matches!(session.finish_kind.as_str(), "duration" | "until") {
                    session.expires_at = session.expires_at.map(|at| at + pause);
                }
                if matches!(session.finish_kind.as_str(), "temperature_reached" | "temperature_stable") {
                    session.safety_expires_at = session.safety_expires_at.map(|at| at + pause);
                }
            }
            session.state = "active".into();
        } else {
            // A due session blocked by manual ownership has not started its work clock.
            // Preserve the requested remaining `until` window by excluding manual wait time.
            if session.finish_kind == "until" {
                if let Some(pause) = pause {
                    session.expires_at = session.expires_at.map(|at| at + pause);
                }
            }
            session.state = "scheduled".into();
        }
        session.condition_started_at = None;
        session.condition_last_observed_at = None;
    }
    if zone.control_owner == "direct_manual" {
        zone.control_owner = "automation".into();
        zone.control_source = "automation".into();
        zone.control_since = Some(now);
        zone.control_resume_at = None;
        zone.control_reason = "Manual takeover cleared; automation may resume".into();
    }
    changed
}

#[cfg(test)]
fn manual_override_matches_baseline(zone: &Zone, device: &Device) -> bool {
    let Some(baseline) = zone.device_manual_override_baseline.as_ref() else { return false; };
    if zone.device_manual_override_fields.is_empty() { return false; }
    // If the unit was OFF before takeover, returning it to OFF is operationally the same
    // controller state even if the remote retained a different mode/target internally.
    // Those dormant values will be set explicitly if automation later powers the unit.
    if !baseline.power { return !device.power; }
    zone.device_manual_override_fields.iter().all(|field| match field.as_str() {
        "power" => device.power == baseline.power,
        "mode" => device.mode == baseline.mode,
        "target_temperature" => device.target_temperature.round() == baseline.target_temperature.round(),
        "fan_speed" => device.fan_speed == baseline.fan_speed,
        "quiet" => device.quiet == baseline.quiet,
        "sleep" => device.sleep == baseline.sleep,
        _ => false,
    })
}

fn persist_manual_override_clear(state: &AppState, zone: &mut Zone, source: &str, restored: bool) -> Result<bool, AppError> {
    if !reset_device_manual_override(zone) { return Ok(false); }
    let now = Utc::now();
    let local_resume_rearmed = restored && rearm_local_thermostat_resume(zone, now.clone());
    zone.updated_at = now;
    state.db.save_zone(zone)?;
    state.broadcast("zone.updated", serde_json::to_value(&*zone)?);
    let (kind, message) = if restored {
        ("zone.device_manual_override_restored", format!("Manual device control returned {} to its previous state", zone.name))
    } else {
        ("zone.device_manual_override_cleared", format!("Manual device control ended for {}", zone.name))
    };
    state.log("info", kind, &message, json!({
        "zone_id": zone.id, "device_id": zone.device_id, "source": source,
        "local_thermostat_resume_rearmed": local_resume_rearmed,
        "local_thermostat_resume_at": zone.local_thermostat_resume_at,
    }));
    state.wake_zone_control();
    Ok(true)
}

fn set_device_manual_override(state: &AppState, zone: &mut Zone, fields: Vec<String>, source: &str, baseline: &Device) -> Result<(), AppError> {
    if fields.is_empty() { return Ok(()); }
    let now = Utc::now();
    if !zone.device_manual_override {
        zone.device_manual_override_since = Some(now);
        zone.device_manual_override_baseline = Some(baseline.into());
        zone.control_since = Some(now);
    }
    zone.device_manual_override = true;
    if let Some(session) = zone.temporary_quick_thermostat.as_mut() {
        // A future scheduled session has no ownership yet. Do not mark it paused until its
        // requested start actually becomes due while manual control is still present.
        if session.activated_at.is_some() || session.started_at <= now {
            if session.paused_at.is_none() { session.paused_at = Some(now); }
            session.state = "paused_manual".into();
            session.condition_started_at = None;
            session.condition_last_observed_at = None;
        }
    }
    zone.control_owner = "direct_manual".into();
    zone.control_source = normalized_direct_source(source).into();
    zone.control_reason = "Direct/manual device control has priority".into();
    // Direct/pilot control is an explicit ownership takeover, not a temporary preset.
    // Keep it manual until the user explicitly chooses Resume automation. A global
    // ON/OFF command changes physical power only and must never erase manual ownership.
    // A schedule boundary must never silently take the unit back from the person controlling it.
    zone.device_manual_override_until = None;
    zone.control_resume_at = None;
    for field in fields {
        if !zone.device_manual_override_fields.iter().any(|existing| existing == &field) {
            zone.device_manual_override_fields.push(field);
        }
    }
    zone.demand = false;
    zone.demand_since = None;
    // Direct/manual takeover supersedes any thermostat task that was waiting behind
    // compressor protection; never leave a stale queued badge behind.
    rearm_compressor_queue(zone);
    zone.updated_at = now;
    state.db.save_zone(zone)?;
    state.broadcast("zone.updated", serde_json::to_value(&*zone)?);
    state.log("info", "zone.device_manual_override", &format!("Manual device control detected for {}", zone.name), json!({
        "zone_id": zone.id,
        "device_id": zone.device_id,
        "fields": zone.device_manual_override_fields,
        "source": source,
        "override_until": zone.device_manual_override_until,
    }));
    Ok(())
}

async fn detect_external_device_control(state: &AppState, before: &Device, after: &Device) -> Result<(), AppError> {
    if before.id != after.id { return Ok(()); }
    for mut zone in state.db.list_zones()?.into_iter().filter(|zone| zone.device_id == after.id) {
        let raw_fields = externally_changed_control_fields(before, after, &zone);
        let controller_settling = if raw_fields.is_empty() {
            json!({ "active": false, "reason": "no_changed_control_fields" })
        } else {
            controller_settling_diagnostics(state, &after.id).await
        };
        let fields = suppress_expected_controller_changes(
            state,
            after,
            raw_fields.clone(),
        ).await;
        // Once direct/pilot ownership has been detected, merely returning the device to a
        // previous physical state must not silently hand control back to schedules. Only the
        // explicit Resume automation action ends manual ownership.
        if fields.is_empty() { continue; }
        // A disabled zone is outside controller ownership. When its manually operated unit is
        // switched off there is no takeover left to display or remember.
        if !zone.enabled && !after.power {
            persist_manual_override_clear(state, &mut zone, "gree_poll", false)?;
            continue;
        }
        state.log("info", "device.remote_control_detected", &format!("External/pilot control detected for {}", zone.name), json!({
            "zone_id": zone.id.clone(),
            "device_id": zone.device_id.clone(),
            "raw_fields": raw_fields,
            "detected_fields": fields.clone(),
            "before": device_control_snapshot(before),
            "after": device_control_snapshot(after),
            "controller_settling": controller_settling,
            "source": "gree_poll",
            "zone_state": {
                "enabled": zone.enabled,
                "control_owner": zone.control_owner.clone(),
                "control_source": zone.control_source.clone(),
                "demand": zone.demand,
                "local_thermostat_power": zone.local_thermostat_power,
                "temporary_quick_thermostat": zone.temporary_quick_thermostat.clone(),
            },
        }));
        set_device_manual_override(state, &mut zone, fields, "gree_poll", before)?;
    }
    Ok(())
}

pub async fn send_manual_command(state: &AppState, device_id: &str, command: DeviceCommand, source: &str) -> Result<Device, AppError> {
    // Keep zone -> device lock ordering consistent with Quick Thermostat/full-zone edits.
    // A device belongs to at most one thermostat zone, but keep this generic for legacy data.
    let mut zone_ids: Vec<String> = state.db.list_zones()?.into_iter().filter(|zone| zone.device_id == device_id).map(|zone| zone.id).collect();
    zone_ids.sort();
    zone_ids.dedup();
    let mut _zone_guards = Vec::new();
    for zone_id in &zone_ids { _zone_guards.push(state.lock_zone_operation(zone_id).await); }
    // Keep the device lock until the zone takeover marker is persisted. Otherwise a poll
    // could observe our own just-sent command before the controller records manual ownership.
    let _device_guard = state.lock_device_operation(device_id).await;
    let before = state.db.get_device(device_id)?
        .ok_or_else(|| AppError::NotFound(format!("device {device_id}")))?;
    // An explicit direct-control request is an ownership action even when the requested
    // value already matches the cached device state. Derive takeover fields from the user's
    // request, not only from the physical delta, so clicking ON / entering the current target
    // still switches the thermostat zone to persistent manual control.
    let fields = command_manual_control_fields(&command);
    let updated = send_command_locked_inner(state, device_id, command, true, false).await?;
    if !fields.is_empty() {
        for mut zone in state.db.list_zones()?.into_iter().filter(|zone| zone.device_id == device_id) {
            if !zone.enabled && !updated.power {
                persist_manual_override_clear(state, &mut zone, source, false)?;
                continue;
            }
            set_device_manual_override(state, &mut zone, fields.clone(), source, &before)?;
        }
    }
    Ok(updated)
}

pub async fn force_house_power_off_device(state: &AppState, device_id: &str, _source: &str) -> Result<Device, AppError> {
    // Whole-house OFF physically forces the unit down after the API has persisted per-zone local OFF.
    // Keep zone -> device ordering so a concurrent local/manual action cannot race the frame.
    let mut zone_ids: Vec<String> = state.db.list_zones()?.into_iter()
        .filter(|zone| zone.device_id == device_id)
        .map(|zone| zone.id)
        .collect();
    zone_ids.sort();
    zone_ids.dedup();
    let mut _zone_guards = Vec::with_capacity(zone_ids.len());
    for zone_id in &zone_ids {
        _zone_guards.push(state.lock_zone_operation(zone_id).await);
    }
    let _device_guard = state.lock_device_operation(device_id).await;
    force_power_off_device_locked(state, device_id).await
}

pub async fn one_shot_house_power_on_device(state: &AppState, device_id: &str) -> Result<Device, AppError> {
    // Global ON releases per-zone OFF state in the API and must not create a local-ON ownership marker. For thermostat-managed
    // units it still respects compressor protection; a protected start is stored as a visible
    // queue item and executed at the protection deadline unless a newer intent cancels/replaces it.
    let mut zone_ids: Vec<String> = state.db.list_zones()?.into_iter()
        .filter(|zone| zone.device_id == device_id)
        .map(|zone| zone.id)
        .collect();
    zone_ids.sort();
    zone_ids.dedup();
    let mut _zone_guards = Vec::with_capacity(zone_ids.len());
    for zone_id in &zone_ids {
        _zone_guards.push(state.lock_zone_operation(zone_id).await);
    }
    let _device_guard = state.lock_device_operation(device_id).await;
    let device = state.db.get_device(device_id)?
        .ok_or_else(|| AppError::NotFound(format!("device {device_id}")))?;
    if !device.enabled { return Err(AppError::BadRequest("device is disabled".into())); }
    if device.power {
        return Ok(device);
    }

    let settings = state.settings.read().await.clone();
    if settings.compressor_protection_enabled {
        if let Some(zone_id) = zone_ids.first() {
            if let Some(mut zone) = state.db.get_zone(zone_id)? {
                let now = Utc::now();
                let protection = chrono::Duration::seconds(settings.compressor_protection_seconds as i64);
                if let Some(last_change) = zone.last_power_change_at {
                    let until = last_change + protection;
                    if until > now {
                        rearm_compressor_queue(&mut zone);
                        queue_compressor_action(&mut zone, "global_power_on".into(), until, "minimum_off_before_global_start");
                        zone.revision = zone.revision.saturating_add(1);
                        zone.updated_at = now;
                        state.db.save_zone(&zone)?;
                        state.broadcast("zone.updated", serde_json::to_value(&zone)?);
                        state.log("info", "zone.compressor_queue_queued", &format!("Queued global ON for {} behind compressor protection", zone.name), json!({
                            "zone_id": zone.id,
                            "device_id": zone.device_id,
                            "action": "global_power_on",
                            "resume_at": zone.compressor_pending_until,
                        }));
                        state.wake_zone_control();
                        return Ok(device);
                    }
                }
            }
        }
    }

    // The device lock is already held. This is physical global power only: no manual marker.
    send_command_locked(state, device_id, DeviceCommand { power: Some(true), ..Default::default() }).await
}

pub async fn force_power_off_device(state: &AppState, device_id: &str) -> Result<Device, AppError> {
    let _device_guard = state.lock_device_operation(device_id).await;
    send_command_locked_forced(state, device_id, DeviceCommand { power: Some(false), ..Default::default() }).await
}

/// Same safety transition for callers that already hold the per-device operation lock.
/// Keeping this separate avoids recursive lock acquisition during atomic configuration import.
pub async fn force_power_off_device_locked(state: &AppState, device_id: &str) -> Result<Device, AppError> {
    send_command_locked_forced(state, device_id, DeviceCommand { power: Some(false), ..Default::default() }).await
}

/// Technical device disable is a safety transition, not just a database flag. The unit is
/// explicitly powered off while it is still commandable, then removed from controller polling.
pub async fn disable_device_safely(state: &AppState, device_id: &str) -> Result<Device, AppError> {
    let _device_guard = state.lock_device_operation(device_id).await;
    let mut device = state.db.get_device(device_id)?
        .ok_or_else(|| AppError::NotFound(format!("device {device_id}")))?;
    if !device.enabled { return Ok(device); }
    device = send_command_locked_forced(
        state,
        device_id,
        DeviceCommand { power: Some(false), ..Default::default() },
    ).await?;
    device.enabled = false;
    device.updated_at = Utc::now();
    state.db.save_device(&device)?;
    state.broadcast("device.updated", serde_json::to_value(&device)?);
    state.wake_zone_control();
    Ok(device)
}