backend/flowtwin/simulation/engine.py

"""The FlowTwin crowd simulator.

A mesoscopic, capacity-constrained pedestrian network model. Agents are
individuals with their own walking speed, destination, route and compliance,
but they move along graph edges rather than in free 2-D space. That choice is
deliberate: it keeps 40,000 agents inside a few milliseconds per step, which is
what makes counterfactual simulation — running five alternative futures from
the same frozen state while an operator waits — actually possible.

What the model reproduces, and why each part is needed:

* speed collapse under density        -> queues form instead of dots piling up
* per-minute throughput at gates      -> a degraded exit really is a bottleneck
* physical storage limits per corridor -> congestion spills back upstream
* first-come-first-served admission   -> queues behave like queues
* per-agent compliance                -> a reroute instruction is not obeyed by all

Every run is fully determined by (venue, scenario, seed, overrides). The RNG
state travels with the snapshot, so a counterfactual branch is reproducible and
two strategies are always compared against an identical starting state.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

import numpy as np

from ..config import Settings
from ..crowd.state import CrowdStateEngine
from ..routing.costs import CostModel
from ..routing.graph import RoutingTables, static_assignment
from ..venue.models import CompiledVenue, NodeType
from ..venue.scenario import Scenario, TimelineEvent
from .agents import (
    POLICY_ADAPTIVE,
    POLICY_SHORTEST,
    POLICY_STATIC,
    STATUS_ARRIVED,
    STATUS_ON_EDGE,
    STATUS_WAITING,
    AgentPopulation,
    build_population,
)
from .movement import CapacityBudget, admit, weidmann_speed

HEAD_EPSILON_M = 0.35


@dataclass
class RunOverrides:
    """Per-run parameters the operator can change from the What-If panel."""

    crowd_size: int | None = None
    release_ramp_s: float | None = None
    compliance_scale: float = 1.0
    routing_policy: int = POLICY_SHORTEST
    capacity_overrides: dict[str, float] = field(default_factory=dict)
    #: Replacement factors for scripted timeline events, keyed by event target.
    #: This is how the What-If panel retunes the scripted failure: the event
    #: still fires when the scenario says it does, but with the operator's
    #: severity instead of the authored one.
    event_factor_overrides: dict[str, float] = field(default_factory=dict)
    disable_timeline: bool = False

    def as_dict(self) -> dict[str, Any]:
        return {
            "crowd_size": self.crowd_size,
            "release_ramp_s": self.release_ramp_s,
            "compliance_scale": self.compliance_scale,
            "routing_policy": int(self.routing_policy),
            "capacity_overrides": dict(self.capacity_overrides),
            "event_factor_overrides": dict(self.event_factor_overrides),
            "disable_timeline": self.disable_timeline,
        }


@dataclass
class AppliedIntervention:
    """Record of an intervention actually applied to this simulation."""

    strategy_id: str
    label: str
    t_s: float
    detail: dict[str, Any] = field(default_factory=dict)
    agents_affected: int = 0


class Simulator:
    """Discrete-time crowd simulation over a venue graph."""

    def __init__(
        self,
        venue: CompiledVenue,
        scenario: Scenario,
        settings: Settings,
        seed: int | None = None,
        overrides: RunOverrides | None = None,
    ) -> None:
        self.venue = venue
        self.scenario = scenario
        self.settings = settings
        self.overrides = overrides or RunOverrides()
        self.seed = int(seed if seed is not None else scenario.default_seed)
        self.dt = settings.simulation.dt_s

        crowd = self.overrides.crowd_size or scenario.crowd_size
        if crowd > settings.simulation.max_agents:
            raise ValueError(
                f"crowd_size {crowd} exceeds the configured maximum "
                f"{settings.simulation.max_agents}"
            )

        self.rng = np.random.default_rng(self.seed)
        # A separate stream for interventions so that applying a strategy never
        # perturbs the population's own random draws.
        self.action_rng = np.random.default_rng(self.seed ^ 0x5F3759DF)

        self.pop, self.dest_indices, self.dest_ids = build_population(
            venue, scenario, self.rng, settings.movement,
            crowd_size=crowd,
            release_ramp_s=self.overrides.release_ramp_s,
            compliance_scale=self.overrides.compliance_scale,
            initial_policy=self.overrides.routing_policy,
        )
        self.n_agents = self.pop.size

        self.costs = CostModel(venue, settings.routing, settings.movement.free_speed_mps)
        self.tables = RoutingTables(venue, self.costs, self.dest_indices, settings.routing)
        self._prepare_static_routing()

        self.node_budget = CapacityBudget(venue.node_service_ppm)
        self.edge_budget = CapacityBudget(venue.edge_capacity_ppm)
        for target, factor in self.overrides.capacity_overrides.items():
            self._scale_capacity(target, factor)

        self.state = CrowdStateEngine(
            venue,
            settings.risk,
            settings.movement,
            settings.prediction.history_window,
            settings.prediction.growth_window_s,
            self.dt,
        )

        self.time = 0.0
        self.step_count = 0
        self.total_arrived = 0
        self.travel_time_sum = 0.0
        self.total_rerouted = 0
        self.total_reroute_decisions = 0
        self.fired_events: set[int] = set()
        self.event_log: list[dict[str, Any]] = []
        self.applied_interventions: list[AppliedIntervention] = []
        self.critical_edge_seconds = 0.0
        self.risk_integral = 0.0
        self.blocked_agents = 0

        # Warm the state engine so the first frame is not all zeros.
        self._cell_density = np.zeros(venue.n_cells)
        self._queue_len_m = np.zeros(venue.n_edges)
        self._queued_count = np.zeros(venue.n_edges)
        self._measure(np.zeros(venue.n_edges), np.zeros(venue.n_edges),
                      np.zeros(venue.n_nodes))

    # ------------------------------------------------------------------
    # setup
    # ------------------------------------------------------------------

    def _prepare_static_routing(self) -> None:
        """Build the frozen baseline routing tables.

        The static baseline runs a small method-of-successive-averages traffic
        assignment using the scenario's expected demand. It is a real
        pre-event plan: capacity-aware, but blind to what actually happens.
        """
        self.tables.costs.compute_static_costs(
            np.zeros(self.venue.n_edges), np.zeros(self.venue.n_nodes)
        )
        self.tables.build_static_tables()

        demand: list[tuple[int, int, float]] = []
        ramp = self.overrides.release_ramp_s or self.scenario.release.ramp_s
        window_min = max(ramp / 60.0, 1.0)
        total = self.pop.size
        for group, share in self.scenario.normalised_demand():
            origin = self.venue.node_index[group.origin]
            weight_sum = sum(group.destinations.values())
            for dest_id, w in group.destinations.items():
                dest_node = self.venue.node_index[dest_id]
                slot = self.dest_indices.index(dest_node)
                people = total * share * (w / weight_sum)
                demand.append((origin, slot, people / window_min))

        edge_vol, node_vol = static_assignment(self.venue, self.tables, demand)
        self.costs.compute_static_costs(edge_vol, node_vol)
        self.tables.build_static_tables()
        self.expected_edge_volume = edge_vol
        self.expected_node_volume = node_vol

    def _scale_capacity(self, target: str, factor: float) -> None:
        if target in self.venue.node_index:
            self.node_budget.multiplier[self.venue.node_index[target]] *= factor
            return
        touched = False
        for i, base in enumerate(self.venue.edge_base_id):
            if base == target:
                self.edge_budget.multiplier[i] *= factor
                touched = True
        if not touched:
            raise KeyError(f"unknown capacity target {target!r}")

    # ------------------------------------------------------------------
    # main loop
    # ------------------------------------------------------------------

    def step(self) -> None:
        dt = self.dt
        t = self.time
        pop = self.pop
        v = self.venue

        if not self.overrides.disable_timeline:
            self._fire_timeline_events(t)

        # -- 1. local density and walking speed, per cell ------------------
        #
        # Density is evaluated over ~12-metre cells rather than over a whole
        # corridor. A queue backing up from a degraded gate therefore slows
        # only the people who have actually reached it, and the congested
        # region grows upstream cell by cell — which is what a real queue does,
        # and what makes "peak local density" a meaningful operational number.
        on_edge = pop.status == STATUS_ON_EDGE
        edge_idx = pop.edge
        idx_on = np.flatnonzero(on_edge)

        occ = np.bincount(edge_idx[idx_on], minlength=v.n_edges).astype(np.float64)
        pair = v.pair_of
        has_pair = pair >= 0
        combined = occ.copy()
        combined[has_pair] += occ[pair[has_pair]]

        # The standing queue at the head of an edge is everyone who has stopped
        # or is barely shuffling — not only those formally at the stop line.
        #
        # This distinction is load-bearing. Discharge is governed by the gate's
        # throughput, so the queue must be a first-come-first-served pool that
        # the gate drains. If only the handful of agents literally at the stop
        # line counted, the queue would occupy almost no length, and everyone
        # behind would have to *walk* through a near-jammed corridor at a few
        # centimetres per second to reach it — throttling a 500/min gate to
        # under 200/min. Measuring the queue by who has actually stopped makes
        # its physical extent, and therefore where walkers join the back of it,
        # match what the crowd is really doing.
        n_queued = self._queued_count if self._queued_count is not None else np.zeros(v.n_edges)
        pack = self.settings.movement.queue_pack_density
        queue_len = np.minimum(n_queued / np.maximum(pack * v.edge_width, 1e-6),
                               v.edge_length * 0.99)
        self._queue_len_m = queue_len
        queue_start = v.edge_length - queue_len

        cell_of = np.zeros(0, dtype=np.int64)
        if idx_on.size:
            e = edge_idx[idx_on]
            eff_pos = pop.pos_m[idx_on].astype(np.float64)
            q = pop.blocked[idx_on]
            if np.any(q):
                spread = ((idx_on[q] * 40503) % 997) / 997.0
                eff_pos[q] = queue_start[e[q]] + spread * queue_len[e[q]]
            within = np.clip((eff_pos / v.edge_cell_size[e]).astype(np.int64),
                             0, v.edge_n_cells[e] - 1)
            cell_of = v.edge_cell_offset[e] + within
        cell_occ = np.bincount(cell_of, minlength=v.n_cells).astype(np.float64)
        cell_comb = cell_occ.copy()
        cp = v.cell_pair
        valid_pair = cp >= 0
        cell_comb[valid_pair] += cell_occ[cp[valid_pair]]
        cell_density = cell_comb / np.maximum(v.cell_area, 1e-6)
        cell_speed = weidmann_speed(cell_density, self.settings.movement)
        self._cell_density = cell_density

        # -- 2. advance the walking agents --------------------------------
        if idx_on.size:
            e = edge_idx[idx_on]
            free_mask = ~pop.blocked[idx_on]
            speed = cell_speed[cell_of] * pop.speed_factor[idx_on]
            new_pos = pop.pos_m[idx_on] + speed.astype(np.float32) * np.float32(dt)

            # A walker cannot step into a cell that is already packed solid.
            # Without this the model lets people accumulate past the physical
            # jam density at the head of a corridor; with it, the congestion
            # front propagates backwards one cell at a time, as it does in a
            # real crowd.
            within_now = (cell_of - v.edge_cell_offset[e]).astype(np.int64)
            has_next = within_now < (v.edge_n_cells[e] - 1)
            next_full = np.zeros(idx_on.size, dtype=bool)
            if np.any(has_next):
                nxt = cell_of[has_next] + 1
                next_full[has_next] = cell_density[nxt] >= (self.settings.movement.jam_density * 0.90)
            cell_ceiling = ((within_now + 1) * v.edge_cell_size[e] - 0.05).astype(np.float32)
            new_pos = np.where(next_full, np.minimum(new_pos, cell_ceiling), new_pos)

            # A walker stops when it reaches the back of the standing queue.
            stop_at = queue_start[e].astype(np.float32)
            reached = free_mask & (new_pos >= stop_at)
            pop.pos_m[idx_on] = np.where(free_mask, np.minimum(new_pos, stop_at),
                                         pop.pos_m[idx_on])
            pop.speed_now[idx_on] = np.where(free_mask & ~reached, speed, 0.0).astype(np.float32)
            newly = idx_on[reached]
            if newly.size:
                pop.blocked[newly] = True
                pop.pos_m[newly] = v.edge_length[edge_idx[newly]].astype(np.float32)

        # -- 3. build the transition candidate set ----------------------
        released = (pop.status == STATUS_WAITING) & (pop.release_t <= t)
        at_head = (pop.status == STATUS_ON_EDGE) & pop.blocked
        cand = np.flatnonzero(released | at_head)
        edge_inflow = np.zeros(v.n_edges, dtype=np.float64)
        edge_outflow = np.zeros(v.n_edges, dtype=np.float64)
        node_throughput = np.zeros(v.n_nodes, dtype=np.float64)

        if cand.size:
            fresh = np.isinf(pop.queue_since[cand])
            pop.queue_since[cand[fresh]] = np.float32(t)

            from_node = np.where(
                pop.status[cand] == STATUS_WAITING,
                pop.origin[cand],
                v.edge_dst[np.maximum(pop.edge[cand], 0)],
            ).astype(np.int32)

            arriving = from_node == pop.dest_node[cand]
            target = np.full(cand.size, -1, dtype=np.int32)
            moving = ~arriving
            if np.any(moving):
                target[moving] = self.tables.next_hop[
                    pop.policy[cand][moving], pop.dest_slot[cand][moving], from_node[moving]
                ]
            # No U-turns. A routing table that has just been re-weighted can
            # briefly make the corridor an agent is standing in look like the
            # cheapest way onward, which sends people back the way they came
            # and, with repeated interventions, leaves a residue bouncing
            # between two nodes. Crowds do not do this; fall back to the
            # baseline hop unless reversing is genuinely the only option.
            came_from = np.where(pop.status[cand] == STATUS_ON_EDGE,
                                 v.pair_of[np.maximum(pop.edge[cand], 0)],
                                 np.int32(-1))
            u_turn = moving & (target >= 0) & (target == came_from)
            if np.any(u_turn):
                fallback = self.tables.next_hop[
                    POLICY_SHORTEST, pop.dest_slot[cand][u_turn], from_node[u_turn]]
                keep = (fallback >= 0) & (fallback != came_from[u_turn])
                patched = target[u_turn]
                patched[keep] = fallback[keep]
                target[u_turn] = patched

            # Agents with no onward route are treated as arrived at a dead end
            # rather than being silently stuck forever.
            stranded = moving & (target < 0)
            arriving = arriving | stranded

            prio = pop.queue_since[cand]

            # Node throughput budget (gates, exits, transport interfaces).
            node_allow = self.node_budget.accrue(dt)
            self.node_budget.clamp_carry(3.0, dt)
            pass_node = admit(from_node, prio, node_allow)

            # Edge entry budget, then the receiving limit.
            #
            # A link does not accept people at its nominal capacity right up
            # until it is physically full. As it fills, the rate at which it
            # can take anyone new falls to zero — the congestion propagates
            # backwards at `backward_wave_mps`. This is what turns a degraded
            # exit into a queue that grows up the corridor and then out into
            # the concourse behind it, instead of a corridor that quietly
            # absorbs an impossible number of people.
            edge_allow = self.edge_budget.accrue(dt)
            self.edge_budget.clamp_carry(3.0, dt)
            space = np.maximum(v.edge_jam_occupancy - combined, 0.0)
            receiving_ppm = (self.settings.movement.backward_wave_mps * 60.0
                             * space / np.maximum(v.edge_length, 1e-6))
            receiving = np.floor(receiving_ppm * dt / 60.0).astype(np.int64)
            edge_allow = np.minimum(edge_allow, np.maximum(receiving, 0))
            headroom = np.floor(space).astype(np.int64)
            edge_allow = np.minimum(edge_allow, headroom)

            movers_mask = pass_node & ~arriving
            pass_edge = np.zeros(cand.size, dtype=bool)
            if np.any(movers_mask):
                sub = np.flatnonzero(movers_mask)
                ok = admit(target[sub], prio[sub], edge_allow)
                pass_edge[sub] = ok

            absorbers = pass_node & arriving
            movers = pass_edge

            used_nodes = np.bincount(from_node[absorbers | movers], minlength=v.n_nodes)
            self.node_budget.consume(used_nodes.astype(np.float64))
            if np.any(movers):
                used_edges = np.bincount(target[movers], minlength=v.n_edges)
                self.edge_budget.consume(used_edges.astype(np.float64))
                edge_inflow += used_edges
            node_throughput += used_nodes

            # -- apply absorptions -------------------------------------
            if np.any(absorbers):
                a = cand[absorbers]
                prev_edge = pop.edge[a]
                left = prev_edge >= 0
                if np.any(left):
                    edge_outflow += np.bincount(prev_edge[left], minlength=v.n_edges)
                pop.status[a] = STATUS_ARRIVED
                pop.arrive_t[a] = np.float32(t)
                pop.edge[a] = -1
                pop.node[a] = from_node[absorbers]
                pop.pos_m[a] = 0.0
                pop.speed_now[a] = 0.0
                pop.blocked[a] = False
                pop.queue_since[a] = np.inf
                entered = pop.enter_t[a]
                valid = ~np.isnan(entered)
                self.travel_time_sum += float(np.sum(t - entered[valid]))
                self.total_arrived += int(valid.sum())

            # -- apply moves --------------------------------------------
            if np.any(movers):
                m = cand[movers]
                prev_edge = pop.edge[m]
                left = prev_edge >= 0
                if np.any(left):
                    edge_outflow += np.bincount(prev_edge[left], minlength=v.n_edges)

                tgt = target[movers]
                # A route change is a decision that differs from the
                # shortest-path plan the agent would otherwise have followed.
                baseline_hop = self.tables.next_hop[
                    POLICY_SHORTEST, pop.dest_slot[m], from_node[movers]
                ]
                diverted = (pop.policy[m] != POLICY_SHORTEST) & (tgt != baseline_hop) & (baseline_hop >= 0)
                if np.any(diverted):
                    n_div = int(diverted.sum())
                    self.total_reroute_decisions += n_div
                    first_time = pop.reroute_count[m][diverted] == 0
                    self.total_rerouted += int(first_time.sum())
                    counts = pop.reroute_count[m]
                    counts[diverted] += 1
                    pop.reroute_count[m] = counts

                pop.status[m] = STATUS_ON_EDGE
                pop.edge[m] = tgt
                pop.pos_m[m] = 0.0
                pop.node[m] = from_node[movers]
                pop.blocked[m] = False
                pop.queue_since[m] = np.inf
                nan_enter = np.isnan(pop.enter_t[m])
                if np.any(nan_enter):
                    ent = pop.enter_t[m]
                    ent[nan_enter] = np.float32(t)
                    pop.enter_t[m] = ent

        # -- 4. measure -------------------------------------------------
        self._measure(edge_inflow, edge_outflow, node_throughput, None)

        # -- 5. refresh adaptive routing --------------------------------
        if (self.time - self.tables.last_refresh_t) >= self.settings.routing.refresh_interval_s:
            self.refresh_routing()

        self.time += dt
        self.step_count += 1

    def _measure(
        self,
        edge_inflow: np.ndarray,
        edge_outflow: np.ndarray,
        node_throughput: np.ndarray,
        _unused: Any = None,
    ) -> None:
        v = self.venue
        pop = self.pop

        on_edge = pop.status == STATUS_ON_EDGE
        idx_on = np.flatnonzero(on_edge)
        occ = np.bincount(pop.edge[idx_on], minlength=v.n_edges).astype(np.float64)
        speed_sum = np.bincount(pop.edge[idx_on], weights=pop.speed_now[idx_on].astype(np.float64),
                                minlength=v.n_edges)

        # "Queueing" means moving materially slower than a walk, not merely
        # standing on the stop line. A corridor where 3,000 people are shuffling
        # forward at 0.2 m/s is a queue of 3,000, and that is the number an
        # operator needs.
        queue_count = np.zeros(v.n_edges, dtype=np.float64)
        node_queue = np.zeros(v.n_nodes, dtype=np.float64)
        peak_local = np.zeros(v.n_edges, dtype=np.float64)
        if idx_on.size:
            e = pop.edge[idx_on]
            slow_cut = 0.35 * self.settings.movement.free_speed_mps
            stuck = pop.blocked[idx_on] | (pop.speed_now[idx_on] < slow_cut)
            if np.any(stuck):
                queue_count = np.bincount(e[stuck], minlength=v.n_edges).astype(np.float64)
                node_queue = np.bincount(v.edge_dst[e[stuck]], minlength=v.n_nodes).astype(np.float64)
        self._queued_count = queue_count
        cell_d = getattr(self, "_cell_density", None)
        if cell_d is not None and cell_d.size:
            peak_local = np.maximum.reduceat(cell_d, v.edge_cell_offset[:-1])

        waiting = pop.status == STATUS_WAITING
        node_occ = np.bincount(pop.origin[waiting], minlength=v.n_nodes).astype(np.float64)
        # People held at an origin whose departure time has passed are queueing
        # to leave, not sitting in a seat.
        ready = waiting & (pop.release_t <= self.time)
        if np.any(ready):
            node_queue += np.bincount(pop.origin[ready], minlength=v.n_nodes).astype(np.float64)

        self.state.update(
            edge_occupancy=occ,
            edge_speed_sum=speed_sum,
            edge_inflow_count=edge_inflow,
            edge_outflow_count=edge_outflow,
            edge_queue_count=queue_count,
            node_occupancy=node_occ,
            node_queue=node_queue,
            node_throughput_count=node_throughput,
            edge_peak_local=peak_local,
            warning_density=self.venue.venue.warning_density,
            critical_density=self.venue.venue.critical_density,
        )

        crit = self.state.critical_edge_count(self.venue.venue.critical_density)
        self.critical_edge_seconds += crit * self.dt
        self.risk_integral += float(np.sum(self.state.edge_risk)) * self.dt
        self.blocked_agents = int(queue_count.sum())

    def refresh_routing(self) -> None:
        """Recompute the adaptive next-hop table from the live crowd state.

        Intervention penalties relax back towards neutral each refresh. An
        operator who intervenes repeatedly would otherwise leave a permanently
        distorted cost surface, and the routing would keep chasing assets that
        recovered long ago.
        """
        self.costs.relax_penalties(self.settings.routing.penalty_decay)
        edge_cost = self.costs.dynamic_edge_cost(
            self.state.edge_velocity, self.state.phys_occupancy, self.state.edge_risk
        )
        node_cost = self.costs.dynamic_node_cost(self.state.node_queue)
        self.tables.refresh_adaptive(edge_cost, node_cost, apply_hysteresis=True)
        self.tables.last_refresh_t = self.time

    def run_for(self, seconds: float) -> None:
        steps = int(round(seconds / self.dt))
        for _ in range(steps):
            self.step()

    def run_until_complete(self, max_seconds: float | None = None) -> None:
        limit = max_seconds if max_seconds is not None else self.scenario.duration_s
        while self.time < limit and not self.is_complete:
            self.step()

    @property
    def is_complete(self) -> bool:
        return bool(np.all(self.pop.status == STATUS_ARRIVED))

    @property
    def remaining(self) -> int:
        return int(np.sum(self.pop.status != STATUS_ARRIVED))

    # ------------------------------------------------------------------
    # timeline
    # ------------------------------------------------------------------

    def _fire_timeline_events(self, t: float) -> None:
        for i, ev in enumerate(self.scenario.timeline):
            if i in self.fired_events or not ev.automatic or ev.t_s > t:
                continue
            self.trigger_event(i)

    def trigger_event(self, index: int) -> dict[str, Any]:
        """Apply a scenario timeline event (scripted or operator-triggered)."""
        if index in self.fired_events:
            return {"applied": False, "reason": "already fired"}
        ev: TimelineEvent = self.scenario.timeline[index]
        self.fired_events.add(index)
        factor = self.overrides.event_factor_overrides.get(ev.target, ev.factor)
        if ev.type == "capacity" and ev.target:
            self._scale_capacity(ev.target, factor)
        record = {
            "t_s": round(self.time, 1),
            "scheduled_t_s": ev.t_s,
            "type": ev.type,
            "target": ev.target,
            "factor": factor,
            "authored_factor": ev.factor,
            "label": (ev.label if factor == ev.factor
                      else f"{ev.target.replace('_', ' ')} throughput set to "
                           f"{factor * 100:.0f}% of nominal"),
            "detail": ev.detail,
            "severity": ev.severity,
            "index": index,
        }
        self.event_log.append(record)
        return {"applied": True, "event": record}

    # ------------------------------------------------------------------
    # interventions (used by the strategy engine)
    # ------------------------------------------------------------------

    def divert_flow(
        self,
        fraction: float,
        target_edges: set[int],
        target_nodes: set[int],
        penalty: float = 6.0,
    ) -> int:
        """Move a fraction of the affected crowd onto the adaptive routing plan.

        "Affected" means an agent whose current shortest-path route actually
        traverses the congested asset. Sending an instruction to people who
        were never going that way would inflate the intervention's apparent
        reach without changing anything.

        Compliance is per agent: an instruction reaches everyone selected, but
        only agents whose personal compliance clears a random draw act on it.
        """
        if fraction <= 0:
            return 0
        for e in target_edges:
            self.costs.penalise_edge(int(e), penalty)
            pair = int(self.venue.pair_of[int(e)])
            if pair >= 0:
                self.costs.penalise_edge(pair, penalty)
        for n in target_nodes:
            self.costs.penalise_node(int(n), penalty)

        matrix = self.tables.traversal_matrix(POLICY_SHORTEST, target_edges, target_nodes)
        pop = self.pop
        active = pop.status != STATUS_ARRIVED
        at_node = np.where(pop.status == STATUS_WAITING, pop.origin,
                           self.venue.edge_dst[np.maximum(pop.edge, 0)])
        affected = active & matrix[pop.dest_slot, at_node] & (pop.policy != POLICY_ADAPTIVE)

        candidates = np.flatnonzero(affected)
        if candidates.size == 0:
            self.refresh_routing()
            return 0

        self.action_rng.shuffle(candidates)
        take = int(round(fraction * candidates.size))
        chosen = candidates[:take]
        if chosen.size == 0:
            self.refresh_routing()
            return 0

        complies = self.action_rng.random(chosen.size) < pop.compliance[chosen]
        accepted = chosen[complies]
        pop.policy[accepted] = np.int8(POLICY_ADAPTIVE)
        self.refresh_routing()
        return int(accepted.size)

    def stagger_release(self, origin_ids: list[str], fraction: float, delay_s: float) -> int:
        """Hold back a fraction of not-yet-departed spectators.

        This is the demand-side lever: it flattens the departure peak instead of
        moving people sideways through the network.
        """
        if fraction <= 0 or delay_s <= 0:
            return 0
        pop = self.pop
        if origin_ids:
            origins = {self.venue.node_index[o] for o in origin_ids if o in self.venue.node_index}
            in_scope = np.isin(pop.origin, list(origins))
        else:
            in_scope = np.ones(self.n_agents, dtype=bool)
        eligible = np.flatnonzero((pop.status == STATUS_WAITING) & in_scope
                                  & (pop.release_t >= self.time - 1.0))
        if eligible.size == 0:
            return 0
        self.action_rng.shuffle(eligible)
        take = int(round(fraction * eligible.size))
        chosen = eligible[:take]
        if chosen.size == 0:
            return 0
        # Spread the held-back group across the delay window rather than
        # releasing them all at once when the hold ends.
        jitter = self.action_rng.random(chosen.size) * delay_s
        pop.release_t[chosen] = (pop.release_t[chosen] + np.float32(delay_s * 0.5)
                                 + jitter.astype(np.float32))
        return int(chosen.size)

    def open_alternate(self, node_id: str, factor: float) -> bool:
        """Bring contingency capacity online at an exit or transport interface."""
        if node_id not in self.venue.node_index:
            return False
        idx = self.venue.node_index[node_id]
        self.node_budget.multiplier[idx] *= factor
        # Make the newly opened asset attractive to the router.
        self.costs.penalise_node(idx, 1.0 / max(factor, 1e-6))
        self.refresh_routing()
        return True

    def redistribute_destinations(
        self, from_dest: str, to_dest: str, fraction: float
    ) -> int:
        """Send a fraction of one destination's demand to another.

        Operationally this is "your coach has been moved to the south apron":
        a change of where people are going, not merely how they get there.
        """
        if fraction <= 0:
            return 0
        vi = self.venue.node_index
        if from_dest not in vi or to_dest not in vi:
            return 0
        from_node, to_node = vi[from_dest], vi[to_dest]
        if to_node not in self.dest_indices:
            return 0
        to_slot = self.dest_indices.index(to_node)
        pop = self.pop
        eligible = np.flatnonzero((pop.status != STATUS_ARRIVED) & (pop.dest_node == from_node))
        if eligible.size == 0:
            return 0
        self.action_rng.shuffle(eligible)
        take = int(round(fraction * eligible.size))
        chosen = eligible[:take]
        if chosen.size == 0:
            return 0
        complies = self.action_rng.random(chosen.size) < pop.compliance[chosen]
        accepted = chosen[complies]
        pop.dest_node[accepted] = np.int32(to_node)
        pop.dest_slot[accepted] = np.int32(to_slot)
        pop.policy[accepted] = np.int8(POLICY_ADAPTIVE)
        self.refresh_routing()
        return int(accepted.size)

    def record_intervention(self, applied: AppliedIntervention) -> None:
        self.applied_interventions.append(applied)

    # ------------------------------------------------------------------
    # snapshot / restore
    # ------------------------------------------------------------------

    def snapshot(self) -> dict[str, Any]:
        """Exact, restorable copy of the entire simulation state."""
        return {
            "pop": self.pop.copy(),
            "time": self.time,
            "step_count": self.step_count,
            "total_arrived": self.total_arrived,
            "travel_time_sum": self.travel_time_sum,
            "total_rerouted": self.total_rerouted,
            "total_reroute_decisions": self.total_reroute_decisions,
            "critical_edge_seconds": self.critical_edge_seconds,
            "risk_integral": self.risk_integral,
            "blocked_agents": self.blocked_agents,
            "queued_count": self._queued_count.copy(),
            "fired_events": set(self.fired_events),
            "event_log": [dict(e) for e in self.event_log],
            "applied_interventions": list(self.applied_interventions),
            "node_budget": self.node_budget.state(),
            "edge_budget": self.edge_budget.state(),
            "costs": self.costs.state(),
            "tables": self.tables.state(),
            "crowd_state": self.state.state(),
            "rng": self.rng.bit_generator.state,
            "action_rng": self.action_rng.bit_generator.state,
        }

    def restore(self, snap: dict[str, Any]) -> None:
        self.pop = snap["pop"].copy()
        self.n_agents = self.pop.size
        self.time = snap["time"]
        self.step_count = snap["step_count"]
        self.total_arrived = snap["total_arrived"]
        self.travel_time_sum = snap["travel_time_sum"]
        self.total_rerouted = snap["total_rerouted"]
        self.total_reroute_decisions = snap["total_reroute_decisions"]
        self.critical_edge_seconds = snap["critical_edge_seconds"]
        self.risk_integral = snap["risk_integral"]
        self.blocked_agents = snap["blocked_agents"]
        self._queued_count = snap["queued_count"].copy()
        self.fired_events = set(snap["fired_events"])
        self.event_log = [dict(e) for e in snap["event_log"]]
        self.applied_interventions = list(snap["applied_interventions"])
        self.node_budget.restore(snap["node_budget"])
        self.edge_budget.restore(snap["edge_budget"])
        self.costs.restore(snap["costs"])
        self.tables.restore(snap["tables"])
        self.state.restore(snap["crowd_state"])
        self.rng.bit_generator.state = snap["rng"]
        self.action_rng.bit_generator.state = snap["action_rng"]

    def branch(self) -> "Simulator":
        """A detached copy of this simulation, for counterfactual roll-out."""
        clone = object.__new__(Simulator)
        clone.venue = self.venue
        clone.scenario = self.scenario
        clone.settings = self.settings
        clone.overrides = self.overrides
        clone.seed = self.seed
        clone.dt = self.dt
        clone.dest_indices = list(self.dest_indices)
        clone.dest_ids = list(self.dest_ids)
        clone.expected_edge_volume = self.expected_edge_volume
        clone.expected_node_volume = self.expected_node_volume
        clone.rng = np.random.default_rng(self.seed)
        clone.action_rng = np.random.default_rng(self.seed)
        clone.costs = CostModel(self.venue, self.settings.routing,
                                self.settings.movement.free_speed_mps)
        clone.tables = RoutingTables(self.venue, clone.costs, self.dest_indices,
                                     self.settings.routing)
        clone.node_budget = CapacityBudget(self.venue.node_service_ppm)
        clone.edge_budget = CapacityBudget(self.venue.edge_capacity_ppm)
        clone.state = CrowdStateEngine(
            self.venue, self.settings.risk, self.settings.movement,
            self.settings.prediction.history_window,
            self.settings.prediction.growth_window_s, self.dt,
        )
        clone.pop = self.pop.copy()
        clone.n_agents = clone.pop.size
        clone.restore(self.snapshot())
        return clone

    # ------------------------------------------------------------------
    # metrics
    # ------------------------------------------------------------------

    def metrics(self) -> dict[str, float]:
        """Cumulative run metrics. All measured, none assumed."""
        pop = self.pop
        arrived = pop.status == STATUS_ARRIVED
        travel = np.where(arrived & ~np.isnan(pop.enter_t) & ~np.isnan(pop.arrive_t),
                          pop.arrive_t - pop.enter_t, np.nan)
        finite = travel[~np.isnan(travel)]
        return {
            "sim_time_s": round(self.time, 2),
            "agents_total": int(self.n_agents),
            "agents_waiting": int(np.sum(pop.status == STATUS_WAITING)),
            "agents_moving": int(np.sum(pop.status == STATUS_ON_EDGE)),
            "agents_arrived": int(arrived.sum()),
            "throughput": int(arrived.sum()),
            "avg_travel_time_s": round(float(np.mean(finite)), 2) if finite.size else 0.0,
            "p95_travel_time_s": round(float(np.percentile(finite, 95)), 2) if finite.size else 0.0,
            "peak_density": round(float(np.max(self.state.peak_edge_density)), 3),
            "current_peak_density": round(float(np.max(self.state.edge_density)), 3),
            "critical_edge_seconds": round(self.critical_edge_seconds, 1),
            "max_queue": int(np.max(self.state.peak_node_queue)) if self.venue.n_nodes else 0,
            "current_max_queue": int(np.max(self.state.node_queue)) if self.venue.n_nodes else 0,
            "aggregate_risk": round(self.risk_integral, 1),
            "rerouted_agents": int(self.total_rerouted),
            "reroute_decisions": int(self.total_reroute_decisions),
            "blocked_agents": int(self.blocked_agents),
            "completion_pct": round(100.0 * float(arrived.sum()) / max(self.n_agents, 1), 1),
        }

    def dispersal_time(self, quantile: float = 0.95) -> float | None:
        """Sim time by which `quantile` of the crowd had reached a destination."""
        arrive = self.pop.arrive_t[~np.isnan(self.pop.arrive_t)]
        if arrive.size < max(1, int(quantile * self.n_agents)):
            return None
        return float(np.percentile(arrive, quantile * 100.0))

    # ------------------------------------------------------------------
    # rendering support
    # ------------------------------------------------------------------

    def agent_sample(self, budget: int) -> dict[str, list]:
        """A deterministic thinned sample of moving agents, for the map.

        Rendering every one of 40,000 agents is a browser problem, not a
        simulation problem. The simulation always runs the full population; the
        map draws an evenly spaced subset and reports the sampling ratio so the
        UI can be honest about what is on screen.
        """
        pop = self.pop
        idx = np.flatnonzero(pop.status == STATUS_ON_EDGE)
        total = idx.size
        if total == 0:
            return {"x": [], "y": [], "v": [], "sampled": 0, "total": 0, "ratio": 1.0}
        if total > budget:
            stride = int(np.ceil(total / budget))
            idx = idx[::stride]
        e = pop.edge[idx]
        frac = np.clip(pop.pos_m[idx] / np.maximum(self.venue.edge_length[e], 1e-6), 0.0, 1.0)

        # Queued agents are all held at pos == length internally. On the map
        # they are spread across the physical extent the queue actually
        # occupies, so a growing queue is visible as it backs up the corridor.
        qlen = getattr(self, "_queue_len_m", None)
        if qlen is not None:
            q = pop.blocked[idx]
            if np.any(q):
                spread = ((idx[q] * 40503) % 997) / 997.0
                length = np.maximum(self.venue.edge_length[e[q]], 1e-6)
                frac[q] = np.clip(1.0 - spread * (qlen[e[q]] / length), 0.0, 1.0)

        xs = np.empty(idx.size, dtype=np.float64)
        ys = np.empty(idx.size, dtype=np.float64)
        for edge_id in np.unique(e):
            m = e == edge_id
            x, y = self.venue.positions_on_edge(int(edge_id), frac[m])
            # Lateral spread across the corridor width, deterministic per agent.
            half = self.venue.edge_width[int(edge_id)] * 0.42
            dx, dy = self.venue.edge_direction(int(edge_id))
            offs = (((idx[m] * 2654435761) % 1000) / 1000.0 - 0.5) * 2.0 * half
            xs[m] = x - dy * offs
            ys[m] = y + dx * offs

        speed = pop.speed_now[idx] / max(self.settings.movement.free_speed_mps, 1e-6)
        return {
            "x": [round(float(a), 1) for a in xs],
            "y": [round(float(a), 1) for a in ys],
            "v": [round(float(a), 2) for a in np.clip(speed, 0.0, 1.0)],
            "sampled": int(idx.size),
            "total": int(total),
            "ratio": round(float(total) / max(idx.size, 1), 2),
        }