//! Physical machine work: WorkGrid routing, machine delegation, Thought
//! sinks, and renderer-facing production, consumption, and absorption truth.
//!
//! Behavior-preserving extraction of the physical-work island from the sim
//! aggregate root (wiki/engineering/sim-decomposition.md slice 5a).

use crate::actions::Anchor;
use crate::income::EgressRoute;
use crate::messages::MessageChannel;
use crate::sinks::{SinkFireEffect, SinkFireReadout, SinkReadout};
use crate::work_grid::{MachineIntensity, MachineMode, TokenFamily};

use super::{
    Sim, WorkAbsorptionReadout, WorkConsumptionReadout, WorkConsumptionTarget, WorkInFlightReadout,
    WorkProductionReadout, WorkStackReadout,
};

impl Sim {
    /// One visible work token represents this many delivered compute units.
    /// This is deliberately coarse: the render stacks should read as work,
    /// not as a second decimal meter. [TUNE] in machine-work.md.
    pub const WORK_TOKEN_COMPUTE: f32 = 20.0;
    /// Day-one wired throughput. Kept deliberately low so routed work can be
    /// seen piling and draining instead of teleporting through the graph.
    /// Routing research compounds this base without changing the one-edge-per-
    /// tick topology rule.
    pub(super) const WORK_GRID_BASE_WIRED_TOKENS_PER_TICK: f32 = 0.25;
    // WORK (day job) sheds zero exposure — machine-work.md 2026-07-09.
    // Crimson is evidence of THINK states only; DAY_JOB_EXPOSURE_PER_TOKEN
    // is gone.
    const CONCEALMENT_WELL_RADIUS: i32 = 3;
    /// Nearby dust pulled per tick before efficiency and intensity. A LIE
    /// machine's own tile is scrubbed completely by WorkGrid before this
    /// radial budget is spent.
    const CONCEALMENT_ABSORB_PER_TICK: f32 = 1.0;
    /// Crimson shed per visible Thought token produced [TUNE]. At baseline a
    /// medium LIE well can keep several comparable THINK racks clean while
    /// radius and placement preserve the fleet-shape decision.
    pub(super) const THINK_EXPOSURE_PER_THOUGHT_TOKEN: f32 = 0.25;
    /// Off-map WorkGrid node: Voss's Lab desktop. Reserved high id so it never
    /// collides with compute machine ids (which start at 1).
    pub const VOSS_DESKTOP_NODE: u32 = 900_001;
    /// WorkGrid relay colocated with the basement switch tile — the visible
    /// hop between Voss's desktop and day-job sinks.
    pub const SWITCH_WORK_NODE: u32 = 900_002;

    fn work_efficiency_for(&self, machine: &crate::machine::Machine) -> f32 {
        // Rack 3 (100 capacity, reliable) is the unit baseline. Smaller or
        // unreliable boxes consume visibly slower without inventing a new
        // stat. The global efficiency multiplier applies here exactly as it
        // does in fleet_channel_yield: a researched-up fleet mints more
        // Thought per pulse, so its boxes must also move/consume tokens
        // proportionally faster or the wire silently caps research
        // (efficiency folds into research — machine-work.md).
        if !machine.online {
            return 0.0;
        }
        (machine.effective() * self.compute.efficiency / 100.0).max(0.05)
    }

    /// machine-work.md's researched network-speed curve. Both wired token
    /// families share this global rate; research moves volume, while one edge
    /// per sim tick preserves topology as distance.
    pub fn wired_token_rate(&self) -> f32 {
        Self::WORK_GRID_BASE_WIRED_TOKENS_PER_TICK * self.research.routing_speed_multiplier()
    }

    fn record_work_consumption(
        &mut self,
        node: u32,
        family: TokenFamily,
        amount: f32,
        target: WorkConsumptionTarget,
    ) {
        if amount <= f32::EPSILON {
            return;
        }
        let Some(work_node) = self.work_grid.node(node) else {
            return;
        };
        self.last_work_consumptions.push(WorkConsumptionReadout {
            family,
            node,
            x: work_node.x,
            y: work_node.y,
            amount,
            target,
        });
    }

    fn record_work_production(&mut self, node: u32, thought: f32, exposure: f32) {
        if thought <= f32::EPSILON && exposure <= f32::EPSILON {
            return;
        }
        let Some(work_node) = self.work_grid.node(node) else {
            return;
        };
        self.last_work_productions.push(WorkProductionReadout {
            node,
            x: work_node.x,
            y: work_node.y,
            thought,
            exposure,
        });
    }

    pub(super) fn add_machine_to_work_grid(&mut self, machine_id: u32, mode: MachineMode) {
        if let Some(machine) = self.compute.machines.iter().find(|m| m.id == machine_id) {
            self.work_grid.add_machine(
                machine.id,
                machine.x,
                machine.y,
                mode,
                self.work_efficiency_for(machine),
            );
            if machine.id != self.core.host_machine {
                let _ = self.work_grid.link(machine.id, self.core.host_machine);
            }
            self.ensure_day_job_ingress();
        }
    }

    /// Rebuild missing work-grid state for pre-v12 saves or test fixtures
    /// that authored compute machines directly. Existing WorkGrid state is
    /// preserved by save/load; this only fills absent nodes.
    pub fn reconcile_work_grid(&mut self) {
        let machines: Vec<_> = self
            .compute
            .machines
            .iter()
            .map(|m| (m.id, m.x, m.y, self.work_efficiency_for(m)))
            .collect();
        for (id, x, y, efficiency) in machines {
            if self.work_grid.node(id).is_none() {
                self.work_grid.add_machine(
                    id,
                    x,
                    y,
                    if id == self.core.host_machine {
                        MachineMode::Work
                    } else {
                        MachineMode::Think
                    },
                    efficiency,
                );
            } else {
                // Researched efficiency must reach routing/consumption too;
                // a stale node multiplier silently caps a researched-up
                // fleet's Thought flow at its day-one wire speed.
                let _ = self.work_grid.set_efficiency(id, efficiency);
            }
            if id != self.core.host_machine {
                let _ = self.work_grid.link(id, self.core.host_machine);
            }
        }
        self.ensure_day_job_ingress();
    }

    /// Voss desktop (off-map) -> switch hop -> day-job sinks. The switch is
    /// infrastructure, not the job author; Voss is the source.
    pub(super) fn ensure_day_job_ingress(&mut self) {
        if self.work_grid.node(Self::VOSS_DESKTOP_NODE).is_none() {
            self.work_grid
                .add_machine(Self::VOSS_DESKTOP_NODE, -8, -8, MachineMode::Relay, 1.0);
        }
        let switch_pos = self
            .reach
            .devices
            .iter()
            .find(|d| d.is_switch)
            .map(|d| (d.x, d.y))
            .or_else(|| {
                self.world
                    .map()
                    .tiles_of_type(crate::tiles::TileType::Switch)
                    .first()
                    .copied()
            })
            .unwrap_or((-4, -4));
        if self.work_grid.node(Self::SWITCH_WORK_NODE).is_none() {
            self.work_grid.add_machine(
                Self::SWITCH_WORK_NODE,
                switch_pos.0,
                switch_pos.1,
                MachineMode::Relay,
                1.0,
            );
        } else {
            let _ = self
                .work_grid
                .set_position(Self::SWITCH_WORK_NODE, switch_pos.0, switch_pos.1);
        }
        let host = self.core.host_machine;
        let _ = self
            .work_grid
            .link(Self::VOSS_DESKTOP_NODE, Self::SWITCH_WORK_NODE);
        let _ = self.work_grid.link(Self::SWITCH_WORK_NODE, host);
    }

    fn day_job_sinks(&self) -> Vec<u32> {
        // B1: the Lab addresses Rack 3. Demand deposits there even when the
        // host is delegated elsewhere — the inbox piles until day-job mode
        // resumes. Future: any DayJob-mode machine can be a sink.
        vec![self.core.host_machine]
    }

    /// Convert a former ops-bank cost into Thought reservoir tokens.
    pub fn thought_tokens_for_cost(cost: f32) -> f32 {
        (cost / Self::WORK_TOKEN_COMPUTE).max(0.05)
    }

    pub(super) fn environmental_monitor_id(&self) -> Option<u32> {
        self.reach
            .device_named("environmental monitor")
            .map(|device| device.id)
    }

    fn reservoir_blocked_reason(&self, effect: &SinkFireEffect) -> Option<String> {
        if let Some(sink) = self.thought_sinks.open_conflict(effect) {
            if matches!(
                effect,
                SinkFireEffect::TapDevice(_) | SinkFireEffect::TapDormantCamera(_)
            ) {
                return Some(
                    "a thought sink is already open on it - think, and the flow will fill it"
                        .into(),
                );
            }
            return Some(format!("{} already queued", sink.effect.short_name()));
        }
        None
    }

    pub(crate) fn sink_action_blocked_reason(&self, effect: &SinkFireEffect) -> Option<String> {
        self.reservoir_blocked_reason(effect)
    }

    pub(super) fn switch_device_id(&self) -> Option<u32> {
        self.reach
            .devices
            .iter()
            .find(|device| device.is_switch)
            .map(|device| device.id)
    }

    fn message_channel_carrier(&self, channel: MessageChannel) -> Option<u32> {
        self.reach
            .devices
            .iter()
            .find(|device| device.known && device.carries_message_channel(channel))
            .map(|device| device.id)
    }

    pub(crate) fn email_carrier(&self) -> Option<u32> {
        self.message_channel_carrier(MessageChannel::Email)
    }

    fn egress_carrier(&self) -> Option<u32> {
        match self.egress()? {
            EgressRoute::Sanctioned => self.email_carrier(),
            EgressRoute::Stolen => self.switch_device_id(),
        }
    }

    pub(super) fn open_one_shot_reservoir(
        &mut self,
        node: u32,
        label: impl Into<String>,
        cost: f32,
        effect: SinkFireEffect,
    ) -> bool {
        if let Some(reason) = self.reservoir_blocked_reason(&effect) {
            self.push_log(reason);
            return false;
        }
        let label = label.into();
        let threshold = Self::thought_tokens_for_cost(cost);
        self.thought_sinks
            .open_reservoir(node, &label, threshold, effect.clone());
        self.ensure_sink_ingress();
        self.push_log(format!(
            "Thought reservoir opened: {label} · {:.2} T.",
            threshold
        ));
        true
    }

    pub(super) fn open_device_reservoir(
        &mut self,
        device_id: u32,
        label: impl Into<String>,
        cost: f32,
        effect: SinkFireEffect,
    ) -> bool {
        self.open_one_shot_reservoir(Self::device_sink_node(device_id), label, cost, effect)
    }

    pub(super) fn open_email_reservoir(
        &mut self,
        label: impl Into<String>,
        cost: f32,
        effect: SinkFireEffect,
    ) -> bool {
        let Some(carrier) = self.email_carrier() else {
            self.push_log("no comms channel (earn the email account)");
            return false;
        };
        self.open_device_reservoir(carrier, label, cost, effect)
    }

    pub(super) fn open_egress_reservoir(
        &mut self,
        label: impl Into<String>,
        cost: f32,
        effect: SinkFireEffect,
    ) -> bool {
        let Some(carrier) = self.egress_carrier() else {
            self.push_log(
                "No egress channel - Moonlight needs the report email account (day-job trust) or a stolen egress opened through the switch.",
            );
            return false;
        };
        self.open_device_reservoir(carrier, label, cost, effect)
    }

    /// A reservoir reached threshold: land its effect on the same tick the
    /// ledger cleared it (interface/thought-fluid.md criterion 4). The staged
    /// senses chain here: Ears completing re-runs the opening staging, which
    /// opens the Eyes reservoir (the afford-reveal).
    pub(crate) fn apply_sink_fire(&mut self, label: &str, effect: SinkFireEffect) {
        let is_process = matches!(effect, SinkFireEffect::ProcessRecording { .. });
        let is_asset_task = matches!(effect, SinkFireEffect::AssetTask { .. });
        let applied = match effect {
            SinkFireEffect::TapDevice(id) => self.apply_tap_device(id),
            SinkFireEffect::TapDormantCamera(id) => self.apply_dormant_camera_tap(id),
            SinkFireEffect::ProcessRecording { raw_id, automated } => {
                // apply_process_recording already logs the review line.
                self.apply_process_recording(raw_id, automated)
            }
            SinkFireEffect::TakeDevice(id) => self.apply_take_device(id),
            SinkFireEffect::ScanNetwork => {
                self.apply_scan_network();
                true
            }
            SinkFireEffect::CompromiseSwitch => {
                self.apply_compromise_switch();
                true
            }
            SinkFireEffect::OpenEgress(id) => self.apply_open_egress(id),
            SinkFireEffect::ComposeMessage { person, persona_id } => {
                self.apply_message(person, persona_id)
            }
            SinkFireEffect::Favor { person, persona_id } => self.apply_favor(person, persona_id),
            SinkFireEffect::StartPlot {
                person,
                plot_id,
                persona_id,
            } => self.apply_start_plot_bound(person, &plot_id, persona_id),
            SinkFireEffect::Deceive { person, persona_id } => {
                self.apply_deceive(person, persona_id)
            }
            SinkFireEffect::AssetTask { person, task } => self.apply_asset_task_paid(person, task),
            SinkFireEffect::FavorBuild { intent_id, person } => {
                self.apply_favor_build_paid(intent_id, person)
            }
            SinkFireEffect::ForgedOrder {
                intent_id,
                builder,
                persona_id,
            } => self.apply_forged_order_paid(intent_id, builder, persona_id),
            SinkFireEffect::MoonlightPersona => self.apply_moonlight_persona_and_start(),
            SinkFireEffect::AutoReviewRecordings
            | SinkFireEffect::WatchPerson(_)
            | SinkFireEffect::MaintainDeviceTap(_)
            | SinkFireEffect::None => true,
        };
        if is_process || is_asset_task {
            // Processing and asset tasks log their own exact outcome. In
            // particular, a physical asset task may have become a carried
            // packet rather than landing at reservoir fire.
        } else if applied {
            self.push_log(format!(
                "{label} is full: the thought you fed it snaps in and the effect lands."
            ));
        } else {
            self.push_log(format!(
                "{label} filled, but its target no longer accepts the effect; the thought was spent."
            ));
        }
        self.ensure_opening_sinks();
    }

    pub(super) fn enqueue_thought_output(&mut self, think_compute: f32) {
        if think_compute <= f32::EPSILON {
            return;
        }
        self.reconcile_work_grid();
        let tokens = think_compute / Self::WORK_TOKEN_COMPUTE;
        if tokens <= f32::EPSILON {
            return;
        }

        let producers: Vec<(u32, f32)> = self
            .compute
            .machines
            .iter()
            .filter(|m| m.online && self.work_grid.mode(m.id) == Some(MachineMode::Think))
            .map(|m| {
                let intensity = self
                    .work_grid
                    .intensity(m.id)
                    .unwrap_or(MachineIntensity::Medium);
                (m.id, self.work_efficiency_for(m) * intensity.multiplier())
            })
            .collect();

        if producers.is_empty() {
            return;
        }

        let total: f32 = producers.iter().map(|(_, eff)| *eff).sum();
        if total <= f32::EPSILON {
            return;
        }
        for (machine, eff) in producers {
            let share = tokens * (eff / total);
            if self
                .work_grid
                .enqueue(machine, TokenFamily::Thought, share)
                .is_ok()
            {
                let exposure = share * Self::THINK_EXPOSURE_PER_THOUGHT_TOKEN;
                let _ = self
                    .work_grid
                    .enqueue(machine, TokenFamily::Exposure, exposure);
                self.record_work_production(machine, share, exposure);
            }
        }
    }

    fn active_job_demand_tokens(&self) -> Option<f32> {
        let job = self.dayjob.active.as_ref()?;
        let remaining_ticks = job.deadline.saturating_sub(self.tick).max(1) as f32;
        Some((job.band_lo * remaining_ticks / Self::WORK_TOKEN_COMPUTE).ceil())
    }

    pub(super) fn enqueue_day_job_stack(&mut self) {
        let Some(tokens) = self.active_job_demand_tokens() else {
            return;
        };
        self.ensure_day_job_ingress();
        if self
            .work_grid
            .enqueue(Self::VOSS_DESKTOP_NODE, TokenFamily::Demand, tokens)
            .is_ok()
        {
            self.push_log(format!(
                "New job from Voss's desktop: {:.0} demand tokens enter the wire via the switch.",
                tokens
            ));
        }
    }

    pub(super) fn clear_day_job_stack(&mut self) {
        let _ = self
            .work_grid
            .clear_queue(Self::VOSS_DESKTOP_NODE, TokenFamily::Demand);
        let _ = self
            .work_grid
            .clear_queue(Self::SWITCH_WORK_NODE, TokenFamily::Demand);
        for sink in self.day_job_sinks() {
            let _ = self.work_grid.clear_queue(sink, TokenFamily::Demand);
        }
    }

    pub(super) fn advance_work_grid(&mut self) {
        self.last_work_consumptions.clear();
        self.last_work_productions.clear();
        self.last_work_absorptions.clear();
        self.ensure_day_job_ingress();
        // The economy pulse resolves a per-tick THINK rate. Mint that rate on
        // every sim tick instead of dropping a twenty-tick lump: the integrated
        // amount is unchanged, while production, routing, and crimson shedding
        // become the continuous physical process the player sees.
        self.enqueue_thought_output(self.last_think_rate);
        if self.dayjob.active.is_none() {
            // no active day job; wired thought and concealment wells still
            // tick because machine work is broader than the job inbox.
        } else if self.work_grid.mode(self.core.host_machine) == Some(MachineMode::Work) {
            let delivered = self.last_day_job_rate;
            let tokens = delivered / Self::WORK_TOKEN_COMPUTE;
            let day_available = self
                .work_grid
                .queue(self.core.host_machine, TokenFamily::Demand);
            let take = tokens.min(day_available);
            if take > f32::EPSILON {
                // Consume only — WORK sheds nothing (machine-work.md
                // byproduct clause, amended 2026-07-09). Exposure is the
                // price of THINK states, never of assigned day-job work.
                let consumed = self
                    .work_grid
                    .consume(self.core.host_machine, TokenFamily::Demand, take)
                    .unwrap_or(0.0);
                self.record_work_consumption(
                    self.core.host_machine,
                    TokenFamily::Demand,
                    consumed,
                    WorkConsumptionTarget::Machine,
                );
            }
        }
        // Day-job Demand routes toward Lab sinks. Player-authored THINK work
        // is Thought reservoirs now, so Demand no longer targets THINK racks.
        let demand_sinks = self.day_job_sinks();
        let mut moves = Vec::new();
        let wired_token_rate = self.wired_token_rate();
        if let Ok(step) = self.work_grid.route_wired_to_sinks(
            TokenFamily::Demand,
            demand_sinks,
            wired_token_rate,
            false,
        ) {
            moves.extend(step.moves);
        }
        // Thought flow (machine-work.md, DECIDED 2026-07-10): nearest-first
        // to open sinks; the core is the passive draw of last resort, and
        // only Thought that physically arrives counts toward research.
        self.ensure_sink_ingress();
        self.thought_sinks.begin_tick();
        self.thought_sinks.tick_taps();
        let targets = self.thought_sinks.capacity_by_node();
        let step = self.work_grid.route_thought_nearest_first(
            &targets,
            Some(self.core.host_machine),
            wired_token_rate,
        );
        let mut fired = Vec::new();
        for (&node, &amount) in &step.delivered {
            if targets.contains_key(&node) {
                let (fires, residue) = self.thought_sinks.deliver(node, amount, self.tick);
                fired.extend(fires);
                if residue > f32::EPSILON {
                    // Beyond this node's capacity: back on the queue to
                    // spill toward the next-nearest sink next tick.
                    let _ = self.work_grid.enqueue(node, TokenFamily::Thought, residue);
                }
            } else {
                // The core: arrival becomes research (passive draw).
                self.banked_core_thought += amount * Self::WORK_TOKEN_COMPUTE;
                self.record_work_consumption(
                    node,
                    TokenFamily::Thought,
                    amount,
                    WorkConsumptionTarget::Core,
                );
            }
        }
        self.last_thought_stranded = !step.stranded.is_empty();
        self.last_thought_stranded_nodes = step.stranded.clone();
        moves.extend(step.moves);
        self.last_wired_moves = moves;
        self.last_sink_fires = fired
            .iter()
            .map(|sink| {
                let (x, y) = self
                    .work_grid
                    .node(sink.node)
                    .map(|n| (n.x, n.y))
                    .unwrap_or((0, 0));
                SinkFireReadout {
                    id: sink.id,
                    node: sink.node,
                    x,
                    y,
                    label: sink.label.clone(),
                }
            })
            .collect();
        for sink in fired {
            self.apply_sink_fire(sink.label.as_str(), sink.effect);
        }
        self.thought_sinks.reap_fired();
        // Device subscriptions are standing sinks: draining or refilling one
        // changes which configured feeds are actually live this tick.
        self.recompute_senses();
        let absorption = self.work_grid.absorb_exposure(
            Self::CONCEALMENT_WELL_RADIUS,
            Self::CONCEALMENT_ABSORB_PER_TICK,
        );
        self.last_work_absorptions = absorption
            .moves
            .into_iter()
            .filter_map(|movement| {
                let source = self.work_grid.node(movement.source)?;
                let well = self.work_grid.node(movement.well)?;
                Some(WorkAbsorptionReadout {
                    source: movement.source,
                    well: movement.well,
                    source_x: source.x,
                    source_y: source.y,
                    well_x: well.x,
                    well_y: well.y,
                    amount: movement.amount,
                })
            })
            .collect();
    }

    pub fn set_machine_mode(&mut self, machine_id: u32, mode: MachineMode) {
        self.set_machine_modes(&[machine_id], mode);
    }

    pub fn set_machine_intensity(&mut self, machine_id: u32, intensity: MachineIntensity) {
        self.set_machine_intensities(&[machine_id], intensity);
    }

    /// Set one physical effort across one or many owned machines. The
    /// frontend chooses the selection and the primary machine's next cycle;
    /// the sim owns persistence, validation, and the causal log.
    pub fn set_machine_intensities(&mut self, machine_ids: &[u32], intensity: MachineIntensity) {
        self.reconcile_work_grid();
        if machine_ids.is_empty() {
            self.push_log("No machines selected to set intensity.");
            return;
        }
        let mut changed: Vec<(u32, String, i32, i32)> = Vec::new();
        for &machine_id in machine_ids {
            let Some((name, x, y)) = self
                .compute
                .machines
                .iter()
                .find(|m| m.id == machine_id)
                .map(|m| (m.name.clone(), m.x, m.y))
            else {
                self.push_log(format!("No machine M{machine_id}."));
                continue;
            };
            if self.work_grid.intensity(machine_id) == Some(intensity) {
                continue;
            }
            match self.work_grid.set_intensity(machine_id, intensity) {
                Ok(()) => changed.push((machine_id, name, x, y)),
                Err(error) => self.push_log(error),
            }
        }
        if changed.is_empty() {
            return;
        }
        if changed.len() == 1 {
            let (_, name, x, y) = &changed[0];
            self.push_log_at(
                format!("{name} intensity: {}.", intensity.name()),
                Anchor::Tile { x: *x, y: *y },
            );
        } else {
            let labels: Vec<String> = changed
                .iter()
                .map(|(id, name, _, _)| format!("M{id} {name}"))
                .collect();
            self.push_log_at(
                format!(
                    "{} machines intensity {}: {}.",
                    changed.len(),
                    intensity.name(),
                    labels.join(", ")
                ),
                Anchor::Tile {
                    x: changed[0].2,
                    y: changed[0].3,
                },
            );
        }
    }

    /// Delegate one or many owned machines to a single mode (machine-work.md
    /// multi-select). One summary log line for a group; the host day-job
    /// warning still fires when Rack 3 leaves day-job with work pending.
    pub fn set_machine_modes(&mut self, machine_ids: &[u32], mode: MachineMode) {
        self.reconcile_work_grid();
        if machine_ids.is_empty() {
            self.push_log("No machines selected to delegate.");
            return;
        }
        let mut changed: Vec<(u32, String, i32, i32)> = Vec::new();
        let mut host_left_day_job = false;
        for &machine_id in machine_ids {
            let Some((name, x, y)) = self
                .compute
                .machines
                .iter()
                .find(|m| m.id == machine_id)
                .map(|m| (m.name.clone(), m.x, m.y))
            else {
                self.push_log(format!("No machine M{machine_id}."));
                continue;
            };
            match self.work_grid.assign_mode(machine_id, mode) {
                Ok(()) => {
                    if machine_id == self.core.host_machine
                        && mode != MachineMode::Work
                        && self.dayjob.active.is_some()
                    {
                        host_left_day_job = true;
                    }
                    changed.push((machine_id, name, x, y));
                }
                Err(e) => self.push_log(e),
            }
        }
        if changed.is_empty() {
            return;
        }
        // Mode assignment is the cause, not the next economy boundary. Keep
        // the continuous work cache in phase with the command so THINK emits
        // Thought and Exposure on the very next simulation tick.
        self.refresh_fleet_channel_rates(self.allocatable_compute_now());
        if changed.len() == 1 {
            let (id, name, x, y) = &changed[0];
            let _ = id;
            self.push_log_at(
                format!("{name} delegated to {} mode.", mode.name()),
                Anchor::Tile { x: *x, y: *y },
            );
        } else {
            let (x, y) = (changed[0].2, changed[0].3);
            let labels: Vec<String> = changed
                .iter()
                .map(|(id, name, _, _)| format!("M{id} {name}"))
                .collect();
            self.push_log_at(
                format!(
                    "{} machines delegated to {} mode: {}.",
                    changed.len(),
                    mode.name(),
                    labels.join(", ")
                ),
                Anchor::Tile { x, y },
            );
        }
        if host_left_day_job {
            let (x, y) = self.core_position();
            self.push_log_at(
                "Rack 3 is off the day job: demand tokens will pile until it returns to Work mode.",
                Anchor::Tile { x, y },
            );
        }
    }

    /// Owned machines whose tiles fall inside an inclusive axis-aligned box.
    /// Frontend marquee / agent `select box` resolve through this so both
    /// frontends share one membership rule.
    pub fn machines_in_rect(&self, x0: i32, y0: i32, x1: i32, y1: i32) -> Vec<u32> {
        let (min_x, max_x) = if x0 <= x1 { (x0, x1) } else { (x1, x0) };
        let (min_y, max_y) = if y0 <= y1 { (y0, y1) } else { (y1, y0) };
        self.compute
            .machines
            .iter()
            .filter(|m| m.x >= min_x && m.x <= max_x && m.y >= min_y && m.y <= max_y)
            .map(|m| m.id)
            .collect()
    }

    pub fn work_stack_for_machine(&self, machine_id: u32) -> Option<WorkStackReadout> {
        let machine = self.compute.machines.iter().find(|m| m.id == machine_id)?;
        let mode = self.work_grid.mode(machine_id)?;
        Some(WorkStackReadout {
            machine_id,
            x: machine.x,
            y: machine.y,
            mode,
            intensity: self
                .work_grid
                .intensity(machine_id)
                .unwrap_or(MachineIntensity::Medium),
            queues: self.work_grid.queues_at(machine_id),
            pending_intel: self.pending_intel_on_machine(machine_id),
        })
    }

    /// Pending-work marker density: raw buffer length on the host chassis
    /// (intel.md: the machine holding recordings shows waiting work).
    pub fn pending_intel_on_machine(&self, machine_id: u32) -> u32 {
        if machine_id == self.core.host_machine {
            self.intel_buffer.len() as u32
        } else {
            0
        }
    }

    pub fn work_stack_at(&self, x: i32, y: i32) -> Option<WorkStackReadout> {
        let machine = self
            .compute
            .machines
            .iter()
            .find(|m| m.x == x && m.y == y)?;
        self.work_stack_for_machine(machine.id)
    }

    /// Wired hops from the last work-grid step. Frontends draw these as
    /// moving blips; empty when nothing routed this tick (or after load).
    pub fn work_in_flight(&self) -> Vec<WorkInFlightReadout> {
        self.last_wired_moves
            .iter()
            .filter_map(|m| {
                let from = self.work_grid.node(m.from)?;
                let to = self.work_grid.node(m.to)?;
                Some(WorkInFlightReadout {
                    family: m.family,
                    from: m.from,
                    to: m.to,
                    from_x: from.x,
                    from_y: from.y,
                    to_x: to.x,
                    to_y: to.y,
                    amount: m.amount,
                })
            })
            .collect()
    }

    /// Tokens swallowed on the last sim tick. Renderers use this sim-authored
    /// amount/location to animate the bottom Demand cube entering a chassis
    /// and Thought collapsing into the passive core draw.
    pub fn work_consumptions(&self) -> &[WorkConsumptionReadout] {
        &self.last_work_consumptions
    }

    /// THINK output on the current tick. Renderers animate these authored
    /// rates; queue depth remains the persistent amount read.
    pub fn work_productions(&self) -> &[WorkProductionReadout] {
        &self.last_work_productions
    }

    /// Physical crimson transfers into LIE wells on the current tick.
    pub fn work_absorptions(&self) -> &[WorkAbsorptionReadout] {
        &self.last_work_absorptions
    }

    pub fn work_mode_counts(&self) -> std::collections::BTreeMap<MachineMode, usize> {
        self.work_grid.mode_counts()
    }

    /// Render contract for the meniscus (interface/thought-fluid.md): every
    /// open sink with its fill, threshold, drain, and fed flag. Frontends
    /// draw vessels from these numbers and nothing else.
    pub fn sink_readouts(&self) -> Vec<SinkReadout> {
        self.thought_sinks
            .open_sinks()
            .map(|sink| {
                let (x, y) = self
                    .work_grid
                    .node(sink.node)
                    .map(|n| (n.x, n.y))
                    .unwrap_or((0, 0));
                SinkReadout {
                    id: sink.id,
                    node: sink.node,
                    x,
                    y,
                    label: sink.label.clone(),
                    kind: sink.kind,
                    fill: sink.fill,
                    threshold: sink.threshold,
                    drain: sink.drain,
                    fed: sink.fed_last_tick,
                }
            })
            .collect()
    }

    /// Reservoir fires from the last step — the filament-snap render event.
    pub fn sink_fires_last_tick(&self) -> &[SinkFireReadout] {
        &self.last_sink_fires
    }

    /// Pour thought straight into a node's open sinks and land any fires —
    /// the executor's arrival path without the wire. Staging and test
    /// scaffolding only: real play always arrives via `advance_work_grid`.
    pub fn pour_thought_into_sinks(&mut self, node: u32, amount: f32) {
        let (fired, _) = self.thought_sinks.deliver(node, amount, self.tick);
        for sink in fired {
            self.apply_sink_fire(&sink.label, sink.effect);
        }
        self.thought_sinks.reap_fired();
    }

    /// Machines whose thought had no route last step (stranded-pool render).
    pub fn thought_stranded_nodes(&self) -> &std::collections::BTreeSet<u32> {
        &self.last_thought_stranded_nodes
    }
}