zc2 0.0.28

//! Pure merge of the last hub snapshot and the local view into Summary v1 (spec §6.3).

use crate::agent::hubwire::{HubSummary, HubWorker};
use crate::agent::summary::*;
use chrono::{DateTime, Utc};

/// What the supervisor knows about this Mac, independent of the hub.
#[derive(Debug, Clone, PartialEq)]
pub struct LocalView {
    pub sharing: bool,
    pub broker: String,
    pub workers: WorkerCounts,
    pub requests: Requests,
}

impl Default for LocalView {
    fn default() -> Self {
        Self {
            sharing: false,
            broker: "stopped".to_string(),
            workers: WorkerCounts::default(),
            requests: Requests::default(),
        }
    }
}

pub struct MergeInput<'a> {
    pub agent_version: &'a str,
    pub now: DateTime<Utc>,
    pub api_url: &'a str,
    /// The last successful hub snapshot, kept while the hub is unreachable.
    pub hub: Option<&'a HubSummary>,
    pub hub_status: HubStatus,
    pub problems: Vec<Problem>,
    pub node_pubkey: &'a str,
    /// Fallback name for this Mac until the hub names it.
    pub node_name: &'a str,
    pub local: &'a LocalView,
}

pub fn env_for(api_url: &str) -> &'static str {
    let u = api_url.trim_end_matches('/');
    if u == crate::credentials::PROD_API_URL {
        "production"
    } else if u == crate::credentials::STAGING_API_URL {
        "staging"
    } else {
        "custom"
    }
}

/// Percent-encode everything outside RFC 3986's unreserved set.
pub fn encode_component(s: &str) -> String {
    s.bytes()
        .map(|b| match b {
            b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'.' | b'_' | b'~' => {
                (b as char).to_string()
            }
            _ => format!("%{b:02X}"),
        })
        .collect()
}

/// Summary v1's one timestamp form: RFC 3339 in UTC, whole seconds, with a
/// `Z` (`2026-09-13T01:41:30Z`, like `as_of`), whatever RFC 3339 form the hub
/// sent: any fractional precision, `Z` or an offset. Fractions are cut, not
/// rounded. `None` for a value that doesn't parse, so no second format ever
/// reaches the app.
pub fn normalize_timestamp(raw: &str) -> Option<String> {
    DateTime::parse_from_rfc3339(raw.trim()).ok().map(|t| {
        t.with_timezone(&Utc)
            .to_rfc3339_opts(chrono::SecondsFormat::Secs, true)
    })
}

fn min_max(values: impl Iterator<Item = f64>) -> Option<PriceRange> {
    values.fold(None, |acc, p| match acc {
        None => Some(PriceRange { min: p, max: p }),
        Some(r) => Some(PriceRange {
            min: r.min.min(p),
            max: r.max.max(p),
        }),
    })
}

/// `set` when every one of this Mac's non-disabled workers has the same
/// explicit price, `inherited` when all of them are null, `mixed` for
/// anything else (spec §6.3) — including any mix of disabled and priced
/// workers — and `disabled` when every worker is disabled on the
/// hub. The agent never surfaces a disabled worker's price: a disabled row
/// never contributes to `effective_per_hour`, and `per_hour`/
/// `effective_per_hour` are always `None` in the `disabled` state.
pub fn this_mac_price(workers: &[&HubWorker]) -> ThisMacPrice {
    if workers.is_empty() {
        return ThisMacPrice {
            state: "inherited".to_string(),
            per_hour: None,
            effective_per_hour: None,
        };
    }
    if workers.iter().all(|w| w.disabled) {
        return ThisMacPrice {
            state: "disabled".to_string(),
            per_hour: None,
            effective_per_hour: None,
        };
    }

    let priced: Vec<&HubWorker> = workers.iter().filter(|w| !w.disabled).copied().collect();
    let has_disabled = priced.len() != workers.len();
    let effective = min_max(
        priced
            .iter()
            .filter_map(|w| w.effective_price_per_hour)
            .filter(|p| *p >= 0.0),
    )
    .map(|r| r.min);

    if has_disabled {
        // A mix of disabled and priced workers is always "mixed"; only the
        // priced workers' minimum ever surfaces.
        return ThisMacPrice {
            state: "mixed".to_string(),
            per_hour: None,
            effective_per_hour: effective,
        };
    }

    let explicit: Vec<Option<f64>> = priced.iter().map(|w| w.price_per_hour).collect();
    let (state, per_hour) = match explicit.first() {
        None => ("inherited", None),
        Some(None) if explicit.iter().all(Option::is_none) => ("inherited", None),
        Some(Some(p)) if explicit.iter().all(|q| *q == Some(*p)) => ("set", Some(*p)),
        _ => ("mixed", None),
    };
    ThisMacPrice {
        state: state.to_string(),
        per_hour,
        effective_per_hour: effective,
    }
}

pub fn build_summary(i: MergeInput) -> Summary {
    let web_base = i.hub.map(|h| h.web_base.trim_end_matches('/').to_string());
    let link_for = |worker_id: &str| {
        web_base
            .as_ref()
            .map(|b| format!("{b}/provide?device={}", encode_component(worker_id)))
    };

    let devices: Option<Vec<Device>> = i.hub.map(|h| {
        h.devices
            .iter()
            .map(|d| Device {
                node_pubkey: d.node_pubkey.clone(),
                name: d.name.clone(),
                this_mac: d.node_pubkey == i.node_pubkey,
                freshness: d.freshness.clone(),
                last_seen_at: d.last_seen_at.as_deref().and_then(normalize_timestamp),
                workers: d.workers.len() as u32,
                effective_price_per_hour: min_max(
                    d.workers.iter().filter_map(|w| w.effective_price_per_hour),
                ),
                link: d
                    .workers
                    .iter()
                    .find(|w| w.status == "online")
                    .and_then(|w| link_for(&w.worker_id)),
            })
            .collect()
    });
    let devices_total = devices.as_ref().map_or(0, |d| d.len() as u32);
    let devices_online = devices.as_ref().map_or(0, |d| {
        d.iter().filter(|x| x.freshness == "fresh").count() as u32
    });
    let this_device = devices.as_ref().and_then(|d| d.iter().find(|x| x.this_mac));

    let this_mac_workers = i
        .hub
        .map(|h| h.this_mac_workers(i.node_pubkey))
        .unwrap_or_default();

    Summary {
        v: SUMMARY_VERSION,
        agent_version: i.agent_version.to_string(),
        as_of: i.now.to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
        env: env_for(i.api_url).to_string(),
        hub: i.hub_status,
        problems: i.problems,
        account: i.hub.map(|h| Account {
            username: h.account.username.clone(),
            email: h.account.email.clone(),
            credits_balance: h.account.credits_balance,
        }),
        earnings: i.hub.map(|h| Earnings {
            today: h.earnings.today,
            last_7d: h.earnings.last_7d,
            tz: h.earnings.tz.clone(),
        }),
        prices: Prices {
            bounds: i.hub.map(HubSummary::bounds).unwrap_or_default(),
            // A negative hub default (the hub's "disabled" sentinel) never
            // surfaces as a price.
            default_per_hour: i
                .hub
                .and_then(|h| h.default_price_per_hour)
                .filter(|p| *p >= 0.0),
            this_mac: this_mac_price(&this_mac_workers),
        },
        this_mac: ThisMac {
            node_pubkey: i.node_pubkey.to_string(),
            name: this_device
                .map(|d| d.name.clone())
                .unwrap_or_else(|| i.node_name.to_string()),
            sharing: i.local.sharing,
            broker: i.local.broker.clone(),
            workers: i.local.workers,
            requests: i.local.requests,
        },
        links: Links {
            hub: web_base.as_ref().map(|b| format!("{b}/provide")),
            this_mac: this_device.and_then(|d| d.link.clone()),
        },
        devices_online,
        devices_total,
        devices,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::agent::hubwire::{tests::HUB_JSON, HubSummary, HubWorker};

    fn hub() -> HubSummary {
        serde_json::from_str(HUB_JSON).unwrap()
    }

    fn worker(id: i64, price: Option<f64>, effective: Option<f64>, status: &str) -> HubWorker {
        HubWorker {
            id,
            worker_id: format!("a1b2c3d4-w{id}"),
            status: status.into(),
            last_seen: None,
            price_per_hour: price,
            reported_price_per_hour: Some(3.6),
            effective_price_per_hour: effective,
            disabled: false,
        }
    }

    fn disabled_worker(id: i64) -> HubWorker {
        HubWorker {
            id,
            worker_id: format!("a1b2c3d4-w{id}"),
            status: "online".into(),
            last_seen: None,
            price_per_hour: None,
            reported_price_per_hour: Some(3.6),
            effective_price_per_hour: None,
            disabled: true,
        }
    }

    fn input<'a>(hub: Option<&'a HubSummary>, local: &'a LocalView) -> MergeInput<'a> {
        MergeInput {
            agent_version: "0.0.26",
            now: chrono::DateTime::parse_from_rfc3339("2026-09-13T01:42:00Z")
                .unwrap()
                .into(),
            api_url: "https://stg.api.zakuro-ai.com",
            hub,
            hub_status: HubStatus {
                reachable: hub.is_some(),
                last_ok: None,
                error: None,
            },
            problems: vec![],
            node_pubkey: "PK_THIS",
            node_name: "local-hostname",
            local,
        }
    }

    #[test]
    fn before_the_first_hub_poll_hub_fields_are_null() {
        let local = LocalView::default();
        let s = build_summary(input(None, &local));
        assert_eq!(s.v, 1);
        assert_eq!(s.as_of, "2026-09-13T01:42:00Z");
        assert_eq!(s.env, "staging");
        assert!(s.account.is_none() && s.earnings.is_none() && s.devices.is_none());
        assert_eq!((s.devices_online, s.devices_total), (0, 0));
        assert_eq!(
            s.links,
            Links {
                hub: None,
                this_mac: None
            }
        );
        assert_eq!(s.prices.bounds, PriceBounds::default());
        assert_eq!(s.this_mac.name, "local-hostname");
        assert_eq!(s.this_mac.broker, "stopped");
        assert_eq!(s.prices.this_mac.state, "inherited");
    }

    #[test]
    fn merges_account_devices_links_and_local_state() {
        let h = hub();
        let local = LocalView {
            sharing: true,
            broker: "running".into(),
            workers: WorkerCounts {
                desired: 1,
                running: 1,
                busy: 0,
                draining: 0,
                max: 5,
            },
            requests: Requests {
                last_5h: 3,
                last_1w: 9,
            },
        };
        let s = build_summary(input(Some(&h), &local));
        assert_eq!(s.account.as_ref().unwrap().email, "jean@zakuro-ai.com");
        assert_eq!(s.earnings.as_ref().unwrap().today, 12.4);
        assert_eq!(s.prices.default_per_hour, Some(18.0));
        let d = &s.devices.as_ref().unwrap()[0];
        assert!(d.this_mac);
        assert_eq!(d.workers, 1);
        assert_eq!(
            d.effective_price_per_hour,
            Some(PriceRange {
                min: 18.0,
                max: 18.0
            })
        );
        assert_eq!(
            d.link.as_deref(),
            Some("https://stg.hub.zakuro-ai.com/provide?device=a1b2c3d4-w0")
        );
        assert_eq!((s.devices_online, s.devices_total), (1, 1));
        assert_eq!(
            s.links.hub.as_deref(),
            Some("https://stg.hub.zakuro-ai.com/provide")
        );
        assert_eq!(s.links.this_mac, d.link);
        assert_eq!(s.this_mac.name, "jeans-mbp", "the hub's device name wins");
        assert_eq!(s.this_mac.workers.running, 1);
        assert_eq!(s.this_mac.requests.last_1w, 9);
    }

    #[test]
    fn this_mac_price_states() {
        let (a, b) = (
            worker(1, Some(18.0), Some(18.0), "online"),
            worker(2, Some(18.0), Some(18.0), "online"),
        );
        let set = this_mac_price(&[&a, &b]);
        assert_eq!(
            (set.state.as_str(), set.per_hour, set.effective_per_hour),
            ("set", Some(18.0), Some(18.0))
        );

        let (c, d) = (
            worker(1, None, Some(12.0), "online"),
            worker(2, None, Some(12.0), "online"),
        );
        let inherited = this_mac_price(&[&c, &d]);
        assert_eq!(
            (
                inherited.state.as_str(),
                inherited.per_hour,
                inherited.effective_per_hour
            ),
            ("inherited", None, Some(12.0))
        );

        let (e, f) = (
            worker(1, Some(18.0), Some(18.0), "online"),
            worker(2, None, Some(12.0), "online"),
        );
        let mixed = this_mac_price(&[&e, &f]);
        assert_eq!(
            (
                mixed.state.as_str(),
                mixed.per_hour,
                mixed.effective_per_hour
            ),
            ("mixed", None, Some(12.0))
        );
    }

    /// Every worker on this Mac disabled -> "disabled", no prices leak through.
    #[test]
    fn this_mac_price_all_disabled_is_disabled_state_with_no_price() {
        let (a, b) = (disabled_worker(1), disabled_worker(2));
        let r = this_mac_price(&[&a, &b]);
        assert_eq!(r.state, "disabled");
        assert_eq!(r.per_hour, None);
        assert_eq!(r.effective_per_hour, None);
    }

    /// A mix of disabled and priced workers is "mixed", and the disabled
    /// worker's (null) price never counts toward the minimum.
    #[test]
    fn this_mac_price_mixed_disabled_and_priced_ignores_disabled() {
        let priced = worker(1, Some(18.0), Some(18.0), "online");
        let cheaper_priced = worker(2, Some(9.0), Some(9.0), "online");
        let disabled = disabled_worker(3);
        let r = this_mac_price(&[&priced, &cheaper_priced, &disabled]);
        assert_eq!(r.state, "mixed");
        assert_eq!(r.per_hour, None);
        assert_eq!(
            r.effective_per_hour,
            Some(9.0),
            "min of priced workers only, disabled excluded"
        );
    }

    #[test]
    fn an_offline_device_has_no_link_and_is_not_online() {
        let mut h = hub();
        h.devices[0].freshness = "gone".into();
        h.devices[0].workers = vec![worker(1, None, Some(3.6), "offline")];
        let local = LocalView::default();
        let s = build_summary(input(Some(&h), &local));
        assert_eq!(s.devices.as_ref().unwrap()[0].link, None);
        assert_eq!((s.devices_online, s.devices_total), (0, 1));
    }

    #[test]
    fn env_follows_the_api_url() {
        assert_eq!(env_for("https://hub.zakuro-ai.com"), "production");
        assert_eq!(env_for("https://hub.zakuro-ai.com/"), "production");
        assert_eq!(env_for("https://stg.api.zakuro-ai.com"), "staging");
        assert_eq!(env_for("http://127.0.0.1:8000"), "custom");
    }

    #[test]
    fn device_ids_are_url_encoded() {
        assert_eq!(encode_component("a1b2c3d4-w0"), "a1b2c3d4-w0");
        assert_eq!(encode_component("a b/c"), "a%20b%2Fc");
    }

    /// A negative hub default (the hub's "disabled" sentinel) never surfaces as a price.
    #[test]
    fn negative_default_price_maps_to_none() {
        let mut h = hub();
        h.default_price_per_hour = Some(-1.0);
        let local = LocalView::default();
        let s = build_summary(input(Some(&h), &local));
        assert_eq!(s.prices.default_per_hour, None);
    }

    /// The hub may send any RFC 3339 form (pydantic emits microseconds, other
    /// paths an explicit offset). Summary v1 always carries whole-second UTC
    /// with a `Z`, the same form as `as_of`, and drops a value it can't parse
    /// instead of passing it through.
    #[test]
    fn hub_timestamps_become_whole_second_utc() {
        let local = LocalView::default();
        let seen_at = |raw: Option<&str>| {
            let mut h = hub();
            h.devices[0].last_seen_at = raw.map(str::to_string);
            build_summary(input(Some(&h), &local)).devices.unwrap()[0]
                .last_seen_at
                .clone()
        };
        let want = Some("2026-09-13T01:41:30Z".to_string());
        assert_eq!(seen_at(Some("2026-09-13T01:41:30Z")), want, "unchanged");
        assert_eq!(seen_at(Some("2026-09-13T01:41:30.123456Z")), want);
        assert_eq!(
            seen_at(Some("2026-09-13T01:41:30.9Z")),
            want,
            "cut, not rounded"
        );
        assert_eq!(seen_at(Some("2026-09-13T01:41:30+00:00")), want);
        assert_eq!(seen_at(Some("2026-09-13T10:41:30.5+09:00")), want);
        assert_eq!(seen_at(Some("yesterday")), None);
        assert_eq!(seen_at(Some("")), None);
        assert_eq!(seen_at(None), None);
    }

    /// `devices` is `None` only before the first successful hub poll; a poll
    /// with zero devices is `Some(vec![])`, not `None`.
    #[test]
    fn zero_devices_is_some_empty_not_none() {
        let mut h = hub();
        h.devices = vec![];
        let local = LocalView::default();
        let s = build_summary(input(Some(&h), &local));
        assert_eq!(s.devices, Some(vec![]));
        assert_eq!(s.devices_total, 0);
        assert_eq!(s.devices_online, 0);

        let s_none = build_summary(input(None, &local));
        assert_eq!(s_none.devices, None);
    }
}