zc2 0.0.29

//! Shared state of one `zc agent` and the operations the local API and the
//! run loops call (spec §6.3-6.7).

use crate::agent::config::AgentConfig;
use crate::agent::files::{self, DesiredState, CHILDREN_FILE, STATE_FILE};
use crate::agent::hub::{self, CredWatch, Credentials, HubClient, HubError};
use crate::agent::hubwire::HubSummary;
use crate::agent::merge::{self, LocalView, MergeInput};
use crate::agent::plan::{self, Desired};
use crate::agent::summary::{HubStatus, Problem, Summary};
use crate::agent::supervisor::{self, HttpBrokerApi, SpawnSpec, Supervisor};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::sync::Notify;

/// A local-API error: `{"error":{"code","message"}}` with `status` (spec §6.3).
#[derive(Debug, Clone, PartialEq)]
pub struct ApiError {
    pub status: u16,
    pub code: &'static str,
    pub message: String,
}

impl ApiError {
    fn new(status: u16, code: &'static str, message: impl Into<String>) -> Self {
        Self {
            status,
            code,
            message: message.into(),
        }
    }
    pub fn invalid(m: impl Into<String>) -> Self {
        Self::new(400, "invalid_input", m)
    }
    pub fn conflict(m: impl Into<String>) -> Self {
        Self::new(409, "conflict", m)
    }
    pub fn hub(m: impl Into<String>) -> Self {
        Self::new(502, "hub_error", m)
    }
    pub fn prerequisite(m: impl Into<String>) -> Self {
        Self::new(503, "prerequisite_missing", m)
    }
    pub fn internal(m: impl Into<String>) -> Self {
        Self::new(500, "internal", m)
    }
}

#[derive(Default)]
struct HubCache {
    last_good: Option<HubSummary>,
    fetched_at: Option<Instant>,
    last_ok: Option<String>,
    error: Option<String>,
    failures: u32,
    reachable: bool,
    unauthorized: bool,
    too_old: bool,
}

/// Lock order, where nested: `hub` before `desired` and `local_problems`.
/// `supervisor` is only ever taken alone, from the blocking reconcile thread.
pub struct Core {
    pub cfg: AgentConfig,
    node_pubkey: String,
    node_name: String,
    desired: Mutex<DesiredState>,
    hub: Mutex<HubCache>,
    creds: Mutex<CredWatch>,
    local: Mutex<LocalView>,
    local_problems: Mutex<Vec<Problem>>,
    supervisor: Mutex<Supervisor>,
    /// Where the starting time zone for a summary poll comes from; a test can
    /// replace this with a fixed value instead of the real machine zone.
    tz_source: Mutex<Box<dyn Fn() -> String + Send + Sync>>,
    /// Once a summary poll gets a 422 for its zone, this session switches to
    /// `UTC` and every later poll goes straight there instead of trying again.
    tz_override: Mutex<Option<String>>,
    pub(crate) broker_unmanaged: AtomicBool,
    pub wake_reconcile: Notify,
    pub wake_hub: Notify,
    /// Shared with the `Supervisor` behind `supervisor` (see `Supervisor::stop_flag`).
    /// Set by `shutdown_children` before it takes the supervisor lock, so an
    /// in-flight reconcile tick blocked inside a `terminate` wait notices
    /// immediately and releases the lock instead of running out its grace
    /// period.
    stop_requested: Arc<AtomicBool>,
}

impl Core {
    pub fn new(cfg: AgentConfig) -> std::io::Result<Arc<Self>> {
        files::ensure_dir(&cfg.dir)?;
        let desired =
            files::load_json::<DesiredState>(&cfg.dir.join(STATE_FILE)).unwrap_or_default();
        let mut sup = Supervisor::new(SpawnSpec::from_config(&cfg));
        if let Some(children) = files::load_json(&cfg.dir.join(CHILDREN_FILE)) {
            sup.adopt(&children, &supervisor::proc_start_time);
        }
        let stop_requested = sup.stop_flag();
        Ok(Arc::new(Self {
            node_pubkey: cfg.node_pubkey(),
            node_name: crate::broker::node_name_or_default(),
            desired: Mutex::new(desired),
            hub: Mutex::new(HubCache::default()),
            creds: Mutex::new(CredWatch::new(cfg.state_dir.join("credentials"))),
            local: Mutex::new(LocalView::default()),
            local_problems: Mutex::new(vec![]),
            supervisor: Mutex::new(sup),
            tz_source: Mutex::new(Box::new(hub::system_tz)),
            tz_override: Mutex::new(None),
            broker_unmanaged: AtomicBool::new(false),
            wake_reconcile: Notify::new(),
            wake_hub: Notify::new(),
            stop_requested,
            cfg,
        }))
    }

    pub fn desired(&self) -> DesiredState {
        self.desired.lock().unwrap().clone()
    }

    pub fn hub_failures(&self) -> u32 {
        self.hub.lock().unwrap().failures
    }

    fn credentials(&self) -> Option<Credentials> {
        self.creds.lock().unwrap().current()
    }

    fn api_url(&self) -> String {
        self.credentials()
            .and_then(|c| c.api_url)
            .unwrap_or_else(|| self.cfg.api_url.clone())
    }

    fn client(&self) -> Option<HubClient> {
        let c = self.credentials()?;
        let base = c
            .api_url
            .clone()
            .unwrap_or_else(|| self.cfg.api_url.clone());
        Some(HubClient::new(
            &base,
            &c.api_key,
            self.cfg.timings.hub_timeout,
        ))
    }

    fn save_desired(&self, d: &DesiredState) -> Result<(), ApiError> {
        files::save_json(&self.cfg.dir.join(STATE_FILE), d)
            .map_err(|e| ApiError::internal(e.to_string()))
    }

    pub fn summary(&self) -> Summary {
        let logged_in = self.credentials().is_some();
        let api_url = self.api_url();
        let desired = self.desired();
        let mut local = self.local.lock().unwrap().clone();
        local.sharing = desired.sharing;
        local.workers.desired = desired.workers;
        local.workers.max = self.cfg.max_workers;

        let hub = self.hub.lock().unwrap();
        let mut problems = Vec::new();
        if !logged_in || hub.unauthorized {
            problems.push(Problem::new("not_logged_in"));
        } else if hub.too_old {
            problems.push(Problem::new("hub_too_old"));
        } else if hub.failures > 0 {
            problems.push(Problem::new("hub_unreachable"));
        }
        for p in self.local_problems.lock().unwrap().iter() {
            if !problems.iter().any(|q| q.code == p.code) {
                problems.push(p.clone());
            }
        }
        merge::build_summary(MergeInput {
            agent_version: env!("CARGO_PKG_VERSION"),
            now: chrono::Utc::now(),
            api_url: &api_url,
            hub: hub.last_good.as_ref(),
            hub_status: HubStatus {
                reachable: hub.reachable,
                last_ok: hub.last_ok.clone(),
                error: hub.error.clone(),
            },
            problems,
            node_pubkey: &self.node_pubkey,
            node_name: &self.node_name,
            local: &local,
        })
    }

    pub fn set_sharing(&self, on: bool) -> Result<Summary, ApiError> {
        if on && self.broker_unmanaged.load(Ordering::Relaxed) {
            return Err(ApiError::conflict(
                "a broker the agent didn't start is running on this Mac; run `zc down` so the agent can manage sharing",
            ));
        }
        {
            let mut d = self.desired.lock().unwrap();
            d.sharing = on;
            self.save_desired(&d)?;
        }
        self.wake_reconcile.notify_one();
        self.wake_hub.notify_one(); // spec §6.7: re-poll the hub right after every write
        Ok(self.summary())
    }

    /// A worker is draining: the run loop then ticks every `drain_poll` (1 s)
    /// instead of every `reconcile` (2 s), as spec §6.5 step 3 asks.
    pub fn is_draining(&self) -> bool {
        self.local.lock().unwrap().workers.draining > 0
    }

    pub fn set_workers(&self, count: u32) -> Result<Summary, ApiError> {
        if count > self.cfg.max_workers {
            return Err(ApiError::invalid(format!(
                "count must be between 0 and {}",
                self.cfg.max_workers
            )));
        }
        {
            let mut d = self.desired.lock().unwrap();
            d.workers = count;
            self.save_desired(&d)?;
        }
        self.wake_reconcile.notify_one();
        self.wake_hub.notify_one();
        Ok(self.summary())
    }

    /// Spec §6.7: `device` fans out to every one of this Mac's worker rows,
    /// `default` sets the account default. `None` clears the price.
    pub async fn set_price(&self, scope: &str, price: Option<f64>) -> Result<Summary, ApiError> {
        if scope != "device" && scope != "default" {
            return Err(ApiError::invalid(format!(
                r#"scope must be "device" or "default", got {scope:?}"#
            )));
        }
        let bounds = self
            .hub
            .lock()
            .unwrap()
            .last_good
            .as_ref()
            .map(HubSummary::bounds)
            .unwrap_or_default();
        if let Some(p) = price {
            if !p.is_finite() || p < bounds.min as f64 || p > bounds.max as f64 {
                return Err(ApiError::invalid(format!(
                    "price_per_hour must be between {} and {}",
                    bounds.min, bounds.max
                )));
            }
        }
        let client = self
            .client()
            .ok_or_else(|| ApiError::prerequisite("this Mac is not signed in: run `zc login`"))?;

        if scope == "default" {
            client
                .put_default_price(price)
                .await
                .map_err(|e| ApiError::hub(e.message()))?;
        } else {
            let stale = self
                .hub
                .lock()
                .unwrap()
                .fetched_at
                .is_none_or(|t| t.elapsed() > Duration::from_secs(120));
            if stale {
                self.refresh_hub().await;
            }
            let (ids, hub_error): (Vec<i64>, Option<String>) = {
                let h = self.hub.lock().unwrap();
                match h.last_good.as_ref() {
                    Some(summary) => (
                        summary
                            .this_mac_workers(&self.node_pubkey)
                            .iter()
                            .map(|w| w.id)
                            .collect(),
                        None,
                    ),
                    None => (vec![], h.error.clone()),
                }
            };
            if ids.is_empty() {
                // The hub never answered at all: say so, rather than telling
                // the owner to start sharing when this Mac may already be.
                if let Some(msg) = hub_error {
                    return Err(ApiError::hub(msg));
                }
                return Err(ApiError::conflict(
                    "this Mac has no workers on the hub yet: start sharing first",
                ));
            }
            hub::set_device_price(&client, &ids, price)
                .await
                .map_err(|failed| {
                    let detail: Vec<String> =
                        failed.iter().map(|(id, e)| format!("{id}: {e}")).collect();
                    ApiError::hub(format!(
                        "could not price {} worker(s): {}",
                        failed.len(),
                        detail.join("; ")
                    ))
                })?;
        }
        self.refresh_hub().await;
        Ok(self.summary())
    }

    /// The time zone this session sends with `GET /api/provide/summary`: the
    /// machine's own zone (or a test's stand-in) until the hub rejects it,
    /// then `UTC` for the rest of the run.
    fn tz_for_summary(&self) -> String {
        if let Some(tz) = self.tz_override.lock().unwrap().clone() {
            return tz;
        }
        (self.tz_source.lock().unwrap())()
    }

    /// Replace the starting time zone with a fixed value, so a test does not
    /// depend on the real machine's zone.
    #[cfg(test)]
    fn set_tz_source_for_test(&self, tz: &'static str) {
        *self.tz_source.lock().unwrap() = Box::new(move || tz.to_string());
    }

    /// Poll the hub once, fold the result into the cache, then seed prices on
    /// worker rows this Mac has never seen before (spec §6.7). A hub that
    /// answers a non-`UTC` zone with 422 is asked again, immediately, with
    /// `UTC`; once that happens every later poll in this run goes straight to
    /// `UTC`. A 422 on `UTC` itself is not retried.
    pub async fn refresh_hub(&self) {
        let Some(client) = self.client() else {
            let mut h = self.hub.lock().unwrap();
            h.reachable = false;
            h.error = Some("not signed in".to_string());
            return;
        };
        let tz = self.tz_for_summary();
        let mut result = client.summary(&tz).await;
        if tz != "UTC" && matches!(result, Err(HubError::Status(422, _))) {
            *self.tz_override.lock().unwrap() = Some("UTC".to_string());
            result = client.summary("UTC").await;
        }
        {
            let mut h = self.hub.lock().unwrap();
            match result {
                Ok(summary) => {
                    h.last_good = Some(summary);
                    h.fetched_at = Some(Instant::now());
                    h.last_ok =
                        Some(chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true));
                    h.error = None;
                    h.failures = 0;
                    h.reachable = true;
                    h.unauthorized = false;
                    h.too_old = false;
                }
                Err(e) => {
                    h.unauthorized = e == HubError::Unauthorized;
                    h.too_old = e == HubError::NotFound;
                    h.reachable = h.too_old; // a 404 is an old hub, not an unreachable one
                    h.error = Some(e.message());
                    h.failures += 1;
                }
            }
        }
        self.seed_new_workers(&client).await;
    }

    /// A worker row this Mac has never seen before enters `known_worker_ids`
    /// once there is nothing left to do for it: either it was never a seed
    /// target (no agreed sibling price, or the row is already known), or its
    /// seed price write to the hub succeeded. A row whose seed write fails
    /// stays unknown, so the next poll tries to seed it again instead of
    /// quietly leaving it at the account default forever.
    async fn seed_new_workers(&self, client: &HubClient) {
        let (targets, all_ids) = {
            let h = self.hub.lock().unwrap();
            let Some(summary) = h.last_good.as_ref() else {
                return;
            };
            let mine = summary.this_mac_workers(&self.node_pubkey);
            let known = self.desired.lock().unwrap().known_worker_ids.clone();
            let min = summary.price_bounds.min;
            (
                hub::seed_targets(&mine, &known, min),
                mine.iter().map(|w| w.id).collect::<Vec<_>>(),
            )
        };
        let mut seeded = Vec::new();
        for (id, price) in &targets {
            match client.put_worker_price(*id, Some(*price)).await {
                Ok(()) => seeded.push(*id),
                Err(e) => eprintln!(
                    "  [AGENT] seeding price on worker row {id}: {}",
                    e.message()
                ),
            }
        }
        let target_ids: Vec<i64> = targets.iter().map(|(id, _)| *id).collect();
        let mut d = self.desired.lock().unwrap();
        let before = d.known_worker_ids.len();
        for id in all_ids {
            let now_known = !target_ids.contains(&id) || seeded.contains(&id);
            if now_known && !d.known_worker_ids.contains(&id) {
                d.known_worker_ids.push(id);
            }
        }
        if d.known_worker_ids.len() != before {
            let _ = files::save_json(&self.cfg.dir.join(STATE_FILE), &*d);
        }
        drop(d);
        if !seeded.is_empty() {
            self.wake_hub.notify_one(); // re-poll so the summary shows the seeded price
        }
    }

    /// One reconcile tick (spec §6.5). Blocking: run it on a blocking thread.
    ///
    /// Returns immediately, before touching the supervisor, once a shutdown
    /// has been requested: no tick may start (or restart) a child after
    /// `terminate_all` has begun taking them down.
    pub fn reconcile_once(&self) {
        if self.stop_requested.load(Ordering::Relaxed) {
            return;
        }
        let d = self.desired();
        let desired = Desired {
            sharing: d.sharing,
            workers: d.workers,
        };
        let path = std::env::var("PATH").unwrap_or_default();
        // Resolved fresh on every tick (a cheap file check), not once at
        // startup: a fresh clone, or an edited `<zc dir>/env`, must clear
        // `zakuro_dir_missing` on the very next tick, without a restart.
        // `state_dir` (never the process-global `credentials::dir()`),
        // `worker_dir_env` and `worker_dir_legacy_candidates` (all captured
        // once, from `from_env`) keep this independent of whatever the real
        // process environment or host `$HOME`/cwd/dev checkout happen to be.
        let zakuro_dir = crate::up::resolve_zakuro_dir_for(
            self.cfg.worker_dir_env.clone(),
            Some(&self.cfg.state_dir),
            &self.cfg.worker_dir_legacy_candidates,
        );
        let prerequisites = plan::prerequisite_problems(
            plan::find_in_path("uv", &path).is_some(),
            zakuro_dir.is_ok(),
            zakuro_dir.as_ref().err().and_then(|m| m.detail()),
            self.credentials().is_some(),
            self.cfg.worker_template_overridden,
        );
        let now = Instant::now();
        let mut sup = self.supervisor.lock().unwrap();
        sup.set_worker_cwd(zakuro_dir.ok());
        let api = HttpBrokerApi {
            port: sup.broker_port(),
            worker_key: std::env::var("ZAKURO_WORKER_KEY").ok(),
        };
        sup.reconcile(&desired, &api, prerequisites.is_empty(), now);
        let _ = files::save_json(&self.cfg.dir.join(CHILDREN_FILE), &sup.children());
        let mut problems = prerequisites;
        problems.extend(sup.problems(now));
        let view = sup.local_view(&desired, self.cfg.max_workers);
        self.broker_unmanaged
            .store(sup.broker_unmanaged(), Ordering::Relaxed);
        drop(sup);
        *self.local.lock().unwrap() = view;
        *self.local_problems.lock().unwrap() = problems;
    }

    /// Tell any in-flight reconcile tick to give up early on whatever
    /// `terminate` wait it is blocked in, instead of running out its full
    /// grace period. Safe to call repeatedly and from outside the supervisor
    /// lock; `shutdown_children` also sets this itself before it locks.
    pub fn request_stop(&self) {
        self.stop_requested.store(true, Ordering::Relaxed);
    }

    /// Agent SIGTERM: stop every child (no drain), then record that none remain.
    ///
    /// Sets the stop flag before taking the supervisor lock, so a reconcile
    /// tick that is currently blocked in a `terminate` wait (stopping the
    /// broker, killing a worker) returns immediately instead of running out
    /// its grace period, and this call does not sit behind it. The lock
    /// itself is taken with `unwrap_or_else` so a mutex a previous panic
    /// poisoned still yields its contents: children must still be signalled
    /// on the way out even after something else went wrong.
    pub fn shutdown_children(&self) {
        self.request_stop();
        let mut sup = self
            .supervisor
            .lock()
            .unwrap_or_else(|poisoned| poisoned.into_inner());
        sup.terminate_all(self.cfg.timings.shutdown_wait);
        let _ = files::save_json(&self.cfg.dir.join(CHILDREN_FILE), &sup.children());
    }
}

#[cfg(test)]
pub(crate) mod tests {
    use super::*;
    use crate::agent::files::tests::tmp;
    use crate::agent::hub::tests::{mock_hub, Seen};

    fn block_on<F: std::future::Future>(f: F) -> F::Output {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(f)
    }

    /// A Core on a fresh temp state dir, not signed in.
    pub(crate) fn core_with(max_workers: u32) -> Arc<Core> {
        let dir = tmp("core");
        std::fs::create_dir_all(&dir).unwrap();
        let mut cfg = AgentConfig::for_dirs(dir);
        cfg.max_workers = max_workers;
        Core::new(cfg).unwrap()
    }

    /// Write `~/.zakuro/credentials` pointing at `hub` (picked up by mtime).
    pub(crate) fn sign_in(core: &Core, hub: &str) {
        std::fs::write(
            core.cfg.state_dir.join("credentials"),
            format!("api_key=zk_1_test\napi_url={hub}\n"),
        )
        .unwrap();
    }

    /// A hub summary whose only device is the Mac with node key `pk`.
    pub(crate) fn hub_json(pk: &str, worker_rows: &str) -> String {
        format!(
            r#"{{"account":{{"username":"jean","email":"jean@zakuro-ai.com","credits_balance":1240.5}},"default_price_per_hour":null,"price_bounds":{{"min":1,"max":120,"step":1}},"earnings":{{"today":1.0,"last_7d":2.0,"tz":"UTC"}},"devices":[{{"node_pubkey":"{pk}","name":"mac","freshness":"fresh","last_seen_at":null,"workers":[{worker_rows}]}}],"web_base":"https://stg.hub.zakuro-ai.com"}}"#
        )
    }

    pub(crate) fn row(id: i64, price: Option<f64>) -> String {
        let p = price.map_or("null".to_string(), |p| p.to_string());
        let e = price.unwrap_or(3.6);
        format!(
            r#"{{"id":{id},"worker_id":"fp-w{id}","status":"online","last_seen":null,"price_per_hour":{p},"reported_price_per_hour":3.6,"effective_price_per_hour":{e},"disabled":false}}"#
        )
    }

    fn puts(seen: &Seen) -> Vec<(String, String)> {
        seen.lock()
            .unwrap()
            .iter()
            .filter(|e| e.0 == "PUT")
            .map(|e| (e.1.clone(), e.2.clone()))
            .collect()
    }

    #[test]
    fn workers_are_persisted_and_bounded_by_max() {
        let core = core_with(3);
        let s = core.set_workers(2).unwrap();
        assert_eq!((s.this_mac.workers.desired, s.this_mac.workers.max), (2, 3));
        let on_disk: DesiredState = files::load_json(&core.cfg.dir.join(STATE_FILE)).unwrap();
        assert_eq!(on_disk.workers, 2);
        let err = core.set_workers(4).unwrap_err();
        assert_eq!((err.status, err.code), (400, "invalid_input"));
    }

    #[test]
    fn sharing_on_is_refused_while_an_unmanaged_broker_runs() {
        let core = core_with(3);
        core.broker_unmanaged.store(true, Ordering::Relaxed);
        assert_eq!(core.set_sharing(true).unwrap_err().status, 409);
        assert!(!core.set_sharing(false).unwrap().this_mac.sharing);
        core.broker_unmanaged.store(false, Ordering::Relaxed);
        assert!(core.set_sharing(true).unwrap().this_mac.sharing);
    }

    #[test]
    fn before_sign_in_the_summary_says_not_logged_in() {
        let s = core_with(3).summary();
        assert!(s.account.is_none());
        assert_eq!(s.problems[0].code, "not_logged_in");
    }

    #[test]
    fn refresh_keeps_the_last_snapshot_and_names_the_problem() {
        let core = core_with(3);
        let pk = core.node_pubkey.clone();
        let status = Arc::new(Mutex::new(200u16));
        let st = status.clone();
        let (hub, _) = mock_hub(move |_, _, _| {
            let s = *st.lock().unwrap();
            (
                s,
                if s == 200 {
                    hub_json(&pk, &row(11, None))
                } else {
                    "{}".into()
                },
            )
        });
        sign_in(&core, &hub);

        block_on(core.refresh_hub());
        let s = core.summary();
        assert_eq!(s.account.as_ref().unwrap().username, "jean");
        assert!(s.devices.as_ref().unwrap()[0].this_mac);
        assert!(s.hub.reachable && s.problems.is_empty(), "{:?}", s.problems);

        *status.lock().unwrap() = 500;
        block_on(core.refresh_hub());
        let s = core.summary();
        assert!(!s.hub.reachable);
        assert!(s.account.is_some(), "the last good snapshot is kept");
        assert!(s.problems.iter().any(|p| p.code == "hub_unreachable"));
        assert_eq!(core.hub_failures(), 1);

        *status.lock().unwrap() = 404;
        block_on(core.refresh_hub());
        let s = core.summary();
        assert!(s.hub.reachable, "an old hub is still a reachable hub");
        assert!(s.problems.iter().any(|p| p.code == "hub_too_old"));

        *status.lock().unwrap() = 401;
        block_on(core.refresh_hub());
        assert!(core
            .summary()
            .problems
            .iter()
            .any(|p| p.code == "not_logged_in"));
    }

    #[test]
    fn device_price_fans_out_then_new_rows_are_seeded() {
        let core = core_with(3);
        let pk = core.node_pubkey.clone();
        let rows = Arc::new(Mutex::new(format!("{},{}", row(11, None), row(12, None))));
        let r = rows.clone();
        let (hub, seen) = mock_hub(move |method, url, _| match method {
            "GET" if url.starts_with("/api/provide/summary") => {
                (200, hub_json(&pk, &r.lock().unwrap()))
            }
            "PUT" => (200, "{}".into()),
            _ => (404, "{}".into()),
        });
        sign_in(&core, &hub);
        block_on(core.refresh_hub());
        assert_eq!(core.desired().known_worker_ids, vec![11, 12]);

        block_on(core.set_price("device", Some(18.0))).unwrap();
        let body = r#"{"price_per_hour":18.0}"#.to_string();
        // The fan-out is concurrent, so the rows arrive in either order.
        let mut fanned_out = puts(&seen);
        fanned_out.sort();
        assert_eq!(
            fanned_out,
            vec![
                ("/api/workers/11/price".to_string(), body.clone()),
                ("/api/workers/12/price".to_string(), body.clone()),
            ]
        );

        // A scale-up created row 13; the hub shows 11 and 12 priced.
        *rows.lock().unwrap() = format!(
            "{},{},{}",
            row(11, Some(18.0)),
            row(12, Some(18.0)),
            row(13, None)
        );
        block_on(core.refresh_hub());
        assert_eq!(
            puts(&seen).last().unwrap(),
            &("/api/workers/13/price".to_string(), body)
        );
        assert_eq!(core.desired().known_worker_ids, vec![11, 12, 13]);
    }

    #[test]
    fn price_writes_are_validated_before_touching_the_hub() {
        let core = core_with(3);
        let status = |r: Result<Summary, ApiError>| r.unwrap_err().status;
        assert_eq!(status(block_on(core.set_price("bogus", Some(10.0)))), 400);
        assert_eq!(
            status(block_on(core.set_price("device", Some(500.0)))),
            400,
            "outside {{1,120}}"
        );
        assert_eq!(status(block_on(core.set_price("default", Some(0.5)))), 400);
        assert_eq!(
            status(block_on(core.set_price("default", Some(10.0)))),
            503,
            "not signed in"
        );
        // A hub that can't even be reached or decoded leaves no snapshot at
        // all: that's a hub problem, not "this Mac has no workers yet".
        let (hub, _) = mock_hub(|_, _, _| (200, "{}".into()));
        sign_in(&core, &hub);
        assert_eq!(
            status(block_on(core.set_price("device", Some(10.0)))),
            502,
            "the hub never produced a usable snapshot"
        );
    }

    #[test]
    fn a_price_write_gets_409_only_once_the_hub_confirms_this_mac_has_no_workers() {
        let core = core_with(3);
        let pk = core.node_pubkey.clone();
        let (hub, _) = mock_hub(move |_, _, _| (200, hub_json(&pk, "")));
        sign_in(&core, &hub);
        let err = block_on(core.set_price("device", Some(10.0))).unwrap_err();
        assert_eq!(err.status, 409);
    }

    /// A row whose seed price write fails stays unknown, so the very next
    /// poll tries to seed it again instead of leaving it at the account
    /// default forever.
    #[test]
    fn a_failed_seed_put_keeps_the_row_unknown_until_it_succeeds() {
        let core = core_with(3);
        let pk = core.node_pubkey.clone();
        let rows = Arc::new(Mutex::new(format!(
            "{},{}",
            row(11, Some(18.0)),
            row(12, Some(18.0))
        )));
        let r = rows.clone();
        let (hub, seen) = mock_hub(move |method, url, nth| match method {
            "GET" if url.starts_with("/api/provide/summary") => {
                (200, hub_json(&pk, &r.lock().unwrap()))
            }
            "PUT" if url == "/api/workers/13/price" && nth == 1 => (500, "{}".into()),
            "PUT" => (200, "{}".into()),
            _ => (404, "{}".into()),
        });
        sign_in(&core, &hub);
        block_on(core.refresh_hub());
        assert_eq!(core.desired().known_worker_ids, vec![11, 12]);

        *rows.lock().unwrap() = format!(
            "{},{},{}",
            row(11, Some(18.0)),
            row(12, Some(18.0)),
            row(13, None)
        );
        block_on(core.refresh_hub());
        assert_eq!(
            core.desired().known_worker_ids,
            vec![11, 12],
            "the failed seed keeps row 13 unknown"
        );
        assert!(puts(&seen)
            .iter()
            .any(|(u, _)| u == "/api/workers/13/price"));

        block_on(core.refresh_hub());
        assert_eq!(
            core.desired().known_worker_ids,
            vec![11, 12, 13],
            "the retried seed succeeds and 13 becomes known"
        );
    }

    /// The mock starts by rejecting every zone but `UTC` with 422, mirroring
    /// a hub that has never heard of the local zone.
    #[test]
    fn a_422_on_the_local_zone_retries_once_with_utc_then_sticks_to_it() {
        let core = core_with(3);
        core.set_tz_source_for_test("Asia/Tokyo");
        let pk = core.node_pubkey.clone();
        let requested_tz: TzLog = Default::default();
        let seen = requested_tz.clone();
        let (hub, _) = mock_hub(move |_, url, _| {
            let raw = url.split("tz=").nth(1).unwrap_or("");
            let tz = raw.replace("%2F", "/");
            seen.lock().unwrap().push(tz.clone());
            if tz == "UTC" {
                (200, hub_json(&pk, &row(11, None)))
            } else {
                (422, "{}".into())
            }
        });
        sign_in(&core, &hub);

        block_on(core.refresh_hub());
        let s = core.summary();
        assert!(s.hub.reachable, "{:?}", s.problems);
        assert!(s.account.is_some());
        assert_eq!(
            *requested_tz.lock().unwrap(),
            vec!["Asia/Tokyo".to_string(), "UTC".to_string()],
            "exactly one retry, with UTC"
        );

        requested_tz.lock().unwrap().clear();
        block_on(core.refresh_hub());
        assert_eq!(
            *requested_tz.lock().unwrap(),
            vec!["UTC".to_string()],
            "later polls go straight to UTC"
        );
    }

    /// A hub that rejects `UTC` itself is not retried: there is nowhere left
    /// to fall back to.
    #[test]
    fn a_422_on_utc_itself_is_not_retried() {
        let core = core_with(3);
        core.set_tz_source_for_test("UTC");
        let calls = Arc::new(Mutex::new(0u32));
        let c = calls.clone();
        let (hub, _) = mock_hub(move |_, _, _| {
            *c.lock().unwrap() += 1;
            (422, "{}".into())
        });
        sign_in(&core, &hub);

        block_on(core.refresh_hub());
        assert_eq!(
            *calls.lock().unwrap(),
            1,
            "no retry when the zone is already UTC"
        );
        assert!(!core.summary().hub.reachable);
    }

    type TzLog = Arc<Mutex<Vec<String>>>;

    /// `zakuro_dir_missing` is resolved fresh on every reconcile tick, not
    /// once at startup: a fresh clone appearing under `<zc dir>/zak-zakuro`
    /// clears the problem on the very next tick, no restart needed.
    /// `sharing` stays off throughout, so this never spawns a real broker or
    /// worker -- `reconcile_once` only ever checks/records the prerequisite.
    #[test]
    fn zakuro_dir_missing_clears_on_the_next_tick_once_the_directory_appears() {
        // `AgentConfig::for_dirs` leaves `worker_dir_legacy_candidates` empty
        // and `reconcile_once` resolves against `cfg.state_dir` (never the
        // process-global `credentials::dir()`, `$HOME` or the cwd), so this
        // is deterministic on any host, whatever checkouts exist there, and
        // mutates no environment variable -- no `HOME_ENV_LOCK` needed.
        let dir = tmp("zakuro-dir-appears");
        std::fs::create_dir_all(&dir).unwrap();
        let cfg = AgentConfig::for_dirs(dir.clone());
        let core = Core::new(cfg).unwrap();

        core.reconcile_once();
        assert!(
            core.summary()
                .problems
                .iter()
                .any(|p| p.code == "zakuro_dir_missing"),
            "no zak-zakuro directory exists yet"
        );

        std::fs::create_dir_all(dir.join("zak-zakuro/zakuro/worker")).unwrap();
        std::fs::write(dir.join("zak-zakuro/zakuro/worker/server.py"), b"").unwrap();
        core.reconcile_once();
        assert!(
            !core
                .summary()
                .problems
                .iter()
                .any(|p| p.code == "zakuro_dir_missing"),
            "the very next tick must see the freshly cloned directory"
        );
    }
}