shrike/waves.py

Part of SHRIKE.

"""The wave composer: what arrives every 35 to 55 seconds, and from where.

Between the hunter's arrivals a sector is not quiet. A composer coroutine
spends a threat budget

    T = 10 + 3 x sector index + notoriety / 10

(``balance.wave_threat_budget``) on one hand-authored recipe, and the recipe is
always the same three-part sentence: one **threat** that kills you, one **tax**
that costs you a resource, and one **modifier** that changes where you are
allowed to stand. Lancers plus Skimmers plus a Welder. Bombardiers plus Mites
plus a Screamer. Loudness therefore buys a meaner sentence rather than only a
longer one: recipes carry a notoriety floor, so a quiet ship never meets the
mixes a loud one lives in.

Three properties the rest of the game leans on:

* **Learnable danger.** Everything a sector throws comes off one
  :class:`random.Random` seeded from the chart node, so a node that opened with
  a bombardier battery opens with it again for the whole run. Backing out of a
  jump and coming back is a real decision rather than a reroll.
* **Readable arrivals.** Every group tears in from one off-screen bearing, the
  bearings of a wave are spread apart, and each is announced by an edge chevron
  ``balance.SPAWN_EDGE_CHEVRON_S`` before anything exists to shoot at. In a
  nebula, whose fog hides spawn directions, the chevron is withheld and
  the wave arrives unannounced.
* **The arrival is the wave.** The Shrike's visit is not a backdrop to another
  Lancer pack: ``HUNTER_ARRIVED`` stops the schedule and ``HUNTER_DEPARTED``
  resumes it, on the same seeded stream, so the pause costs the player nothing
  and the hunter has the screen to itself.
* **The staged opening comes first.** A first-run pilot is still being taught
  which key thrusts, so run one throws nothing at all until the scripted
  arrival has come and gone, and then throws it at half budget and without
  Lancers. Run two keeps its first forty seconds quiet and throws the first wave
  the moment they are up. From run three the schedule is the plain one. Neither
  hold is open-ended: a held run that never meets the condition to lift the hold
  starts composing anyway rather than staying empty. The composer reads the run
  number off the run scene's config; :meth:`WaveComposer.set_run_number` says it
  outright.

The composer owns no enemy behaviour. It picks the sentence, prices it, chooses
the bearings and mounts the bodies; everything after that belongs to
``shrike.enemies``.
"""

from __future__ import annotations

import math
import random
from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass, field

from simvx.core import Node, Node3D, Signal, Vec2

from . import balance
from .enemies.advanced import Bombardier, HuskTurret, Screamer, Welder, apply_elite
from .enemies.basic import Lancer, MagMine, MiteShoal, Skimmer
from .enemies.steering import Enemy
from .power import SignalWiring
from .runtime import Groups, Services, SignalNames, from_plane, to_plane

# ============================================================================
# Composition constants
#
# Staging geometry and pacing that balance.py carries no equivalent for: these
# describe how a wave is presented rather than how hard it hits. The threat
# numbers themselves all come from balance.
# ============================================================================

#: Distance from the ship a group tears in at, in world units. Comfortably
#: outside the camera's view of the flight plane, close enough that the chevron
#: and the contact belong to the same event.
SPAWN_RING_RADIUS = 46.0
#: Per-body radial jitter on that ring, so a group arrives as a loose formation.
SPAWN_RING_JITTER = 5.0
#: Half-width of the arc a group's bodies scatter across, radians.
SPAWN_GROUP_SPREAD = 0.30
#: Half-width of the random offset applied to each group's bearing, radians.
#: Small enough that three groups stay visibly separated around the ship.
SPAWN_BEARING_JITTER = 0.35

#: Biomes whose fog hides spawn directions: no chevron, no warning.
FOGGED_BIOMES = ("nebula",)

#: Extra threat budget past ``balance.ELITE_THREAT_THRESHOLD`` that buys one
#: more elite tag in the same wave.
ELITE_BUDGET_PER_EXTRA = 12.0
#: Hard ceiling on elite tags in one wave, whatever the budget.
ELITE_MAX_PER_WAVE = 3

#: Weight multiplier applied to the recipe that ran last, so a node's danger is
#: consistent without becoming one repeated sentence.
RECIPE_REPEAT_WEIGHT = 0.15

# -- The staged opening ------------------------------------------------------
#
# ``onboarding.py`` owns what the first runs teach; these say what the composer
# does while that teaching is going on. They are pacing rather than tuning,
# which is why they live here and not in balance.py.

#: The last run the composer softens. From the run after this one the schedule
#: is the plain one, at full budget, with every archetype in the pools.
SOFTENED_RUNS = 2
#: Share of the threat budget a run-one wave may spend. Half of the opening
#: sector's budget cannot pay for one of everything in even the cheapest
#: sentence, so run one's first sector composes nothing at all and the softened
#: mixes first appear a jump later. That is the intent, not a shortfall: the
#: sector the tutorial happens in belongs to the tutorial.
RUN_1_BUDGET_FRACTION = 0.5
#: Archetypes run one never buys. The Lancer's dash is what kills a pilot who
#: has not learned to dodge yet, so run one's sentences are written without it.
RUN_1_BARRED_ARCHETYPES: frozenset[str] = frozenset({"lancer"})
#: Seconds of quiet at the head of run two before its first wave lands. Long
#: enough to deploy the wings and find a vein, short enough that the opening
#: threatens: a due wave fires the moment the grace expires rather than rolling
#: a fresh interval, so this is when the first wave arrives and not merely when
#: it becomes legal. If the grace and a fresh interval added up, run two could
#: fly ninety seconds without meeting anything.
RUN_2_FIRST_WAVE_GRACE_S = 40.0
#: How long run one's hold may last before the schedule runs anyway. Run one
#: normally hands its opening to the scripted arrival and resumes once that has
#: been survived, but a pilot who jumps out of the opening sector before the
#: arrival is provoked never survives one, and without a ceiling that run stays
#: empty for as long as it lasts. Set past the arrival ladder's own length, so
#: it only ever fires for a run the ladder never reached.
RUN_1_HOLD_CEILING_S = 150.0
#: Run number that means "not known": nothing is softened, which is what an
#: arena test or a scene with no run config wants.
RUN_NUMBER_UNKNOWN = 0

#: Units bought per purchase of an archetype: (minimum, maximum). The minimum
#: is what one appearance of the archetype costs the recipe; the maximum is what
#: stops a rich budget collapsing into a single stacked archetype. Budget a
#: sentence cannot spend under these ceilings is not wasted: past
#: ``balance.ELITE_THREAT_THRESHOLD`` it buys elite tags instead: meaner mixes,
#: not just more units.
PURCHASE_LIMITS: dict[str, tuple[int, int]] = {
    "mite": (balance.MITE_SHOAL_MIN, balance.MITE_SHOAL_MAX),
    "skimmer": (1, 6),
    "lancer": (1, 7),
    "mag_mine": (1, 9),
    "welder": (1, 4),
    "bombardier": (1, 5),
    "screamer": (1, 4),
    "husk_turret": (1, 5),
}

# ============================================================================
# The grammar
# ============================================================================

#: Archetypes that kill you if ignored.
THREAT_ARCHETYPES = ("lancer", "bombardier", "husk_turret")
#: Archetypes that cost you a resource rather than hull: scrap, ammo, signature.
TAX_ARCHETYPES = ("mite", "skimmer", "screamer")
#: Archetypes that change where you are allowed to stand, or in what order you
#: are allowed to shoot.
MODIFIER_ARCHETYPES = ("welder", "mag_mine", "screamer", "husk_turret")

#: The grammar's three slots. An archetype may serve two roles (a Screamer taxes
#: signature and rewrites your kill order); no recipe may use one twice.
ROLE_POOLS: dict[str, tuple[str, ...]] = {
    "threat": THREAT_ARCHETYPES,
    "tax": TAX_ARCHETYPES,
    "modifier": MODIFIER_ARCHETYPES,
}

#: Heralds are the hunter's escort, not a wave: they arrive on the telegraph
#: ladder and spend none of this budget.
EXCLUDED_ARCHETYPES = ("herald",)

#: Per-biome bias on each archetype, multiplying a recipe's draw weight through
#: the geometric mean of its three slots. Anything unlisted weighs 1.0; a zero
#: bars the archetype from that biome entirely.
BIOME_WEIGHTS: dict[str, dict[str, float]] = {
    "debris_field": {},
    "solar_shallows": {"mite": 1.6, "bombardier": 1.4, "skimmer": 0.8},
    "ice_field": {"lancer": 0.7, "mite": 0.7, "husk_turret": 1.4, "mag_mine": 1.3},
    "nebula": {"lancer": 1.5, "mag_mine": 1.4, "bombardier": 0.7, "skimmer": 0.8},
    "wreck_graveyard": {"husk_turret": 2.0, "skimmer": 1.6, "welder": 1.2, "mite": 0.8},
    "vent_field": {"mag_mine": 2.0, "bombardier": 1.2, "husk_turret": 0.7},
    "broker_claim": {"welder": 1.4, "bombardier": 1.3, "screamer": 0.6},
    "roost": {},
}

#: The authored recipe table. Each entry is one threat + tax + modifier
#: sentence; ``biomes`` empty means every biome, ``min_notoriety`` is the dial
#: reading a mix needs before it can be drawn, and the budget floor is derived
#: from ``PURCHASE_LIMITS`` rather than authored, so a recipe can never be drawn
#: at a budget that cannot pay for one of everything it names.
#:
#: The combinations left in ROLE_POOLS are room for more recipes.
WAVE_RECIPES: list[dict] = [
    {
        "id": "lance_and_larceny",
        "threat": "lancer",
        "tax": "skimmer",
        "modifier": "welder",
        "note": "the design's worked example: dash pressure, a thief, and a medic keeping it alive",
    },
    {
        "id": "mortar_and_shoal",
        "threat": "bombardier",
        "tax": "mite",
        "modifier": "screamer",
        "note": "the second worked example: reticles on your mining spot while a shoal chews the hull",
    },
    {
        "id": "picket_line",
        "threat": "lancer",
        "tax": "skimmer",
        "modifier": "mag_mine",
        "weight": 1.2,
        "note": "the act 1 staple: mines close the easy escape lane the dash pushes you down",
    },
    {
        "id": "siren_run",
        "threat": "lancer",
        "tax": "screamer",
        "modifier": "mag_mine",
        "note": "the kill-order exam: the siren is cheap and the dash punishes the detour",
    },
    {
        "id": "dead_ship_guns",
        "threat": "husk_turret",
        "tax": "skimmer",
        "modifier": "welder",
        "biomes": ("wreck_graveyard", "debris_field"),
        "note": "emplacements you must break line with while the salvage walks away",
    },
    {
        "id": "graveyard_shift",
        "threat": "husk_turret",
        "tax": "mite",
        "modifier": "mag_mine",
        "biomes": ("wreck_graveyard",),
        "note": "the hulks shoot, the mines seal the gaps between them",
    },
    {
        "id": "panel_hunt",
        "threat": "bombardier",
        "tax": "skimmer",
        "modifier": "mag_mine",
        "biomes": ("solar_shallows",),
        "note": "stand-off fire aimed at deployed wings: retract and lose the sun, stay and lose the panel",
    },
    {
        "id": "glare_ambush",
        "threat": "lancer",
        "tax": "mite",
        "modifier": "welder",
        "biomes": ("solar_shallows",),
        "note": "a shoal on the wings with a medic behind it",
    },
    {
        "id": "slow_ice",
        "threat": "lancer",
        "tax": "skimmer",
        "modifier": "husk_turret",
        "biomes": ("ice_field",),
        "note": "slow bodies, fixed guns: the biome that rewards reading the field before moving",
    },
    {
        "id": "frost_choir",
        "threat": "husk_turret",
        "tax": "screamer",
        "modifier": "mag_mine",
        "biomes": ("ice_field",),
        "note": "sirens hiding behind a mined approach to a gun line",
    },
    {
        "id": "fog_lance",
        "threat": "lancer",
        "tax": "mite",
        "modifier": "mag_mine",
        "biomes": ("nebula",),
        "note": "dash attacks out of a fog that swallows the chevron",
    },
    {
        "id": "fog_battery",
        "threat": "bombardier",
        "tax": "screamer",
        "modifier": "welder",
        "biomes": ("nebula",),
        "min_notoriety": 20,
        "note": "unseen mortars with a repair line you have to find before you can end it",
    },
    {
        "id": "vent_maze",
        "threat": "bombardier",
        "tax": "mite",
        "modifier": "mag_mine",
        "biomes": ("vent_field",),
        "note": "the mine maze the biome is named for, with reticles falling into it",
    },
    {
        "id": "pressure_cooker",
        "threat": "bombardier",
        "tax": "screamer",
        "modifier": "mag_mine",
        "biomes": ("vent_field",),
        "note": "refuelling under mortars while a siren prices the delay",
    },
    {
        "id": "broker_muscle",
        "threat": "bombardier",
        "tax": "skimmer",
        "modifier": "welder",
        "biomes": ("broker_claim",),
        "min_notoriety": 50,
        "note": "the barge's own security: expensive, patient, and hard to finish",
    },
    {
        "id": "claim_jumpers",
        "threat": "husk_turret",
        "tax": "screamer",
        "modifier": "welder",
        "biomes": ("broker_claim",),
        "min_notoriety": 50,
        "note": "fixed guns the welders keep standing while the sirens raise your price",
    },
    {
        "id": "hard_shoulder",
        "threat": "husk_turret",
        "tax": "mite",
        "modifier": "welder",
        "min_notoriety": 25,
        "note": "a healed gun line behind a shoal: the first wave that punishes ignoring support",
    },
    {
        "id": "tithe_collectors",
        "threat": "bombardier",
        "tax": "skimmer",
        "modifier": "screamer",
        "min_notoriety": 25,
        "note": "they came for the haul, and the siren makes leaving it expensive too",
    },
    {
        "id": "wolfpack",
        "threat": "lancer",
        "tax": "screamer",
        "modifier": "welder",
        "min_notoriety": 15,
        "note": "dash attackers that get patched between passes",
    },
    {
        "id": "minefield_watch",
        "threat": "husk_turret",
        "tax": "skimmer",
        "modifier": "mag_mine",
        "note": "static everything: the wave you out-position rather than out-shoot",
    },
    {
        "id": "chorus_and_chaff",
        "threat": "bombardier",
        "tax": "mite",
        "modifier": "welder",
        "min_notoriety": 40,
        "weight": 0.9,
        "note": "the loud-ship mix: a shoal, a battery, and a medic behind both",
    },
    {
        "id": "long_arm",
        "threat": "bombardier",
        "tax": "screamer",
        "modifier": "husk_turret",
        "min_notoriety": 30,
        "note": "mortars from one side and a gun line on the other; the middle is not a place to mine",
    },
    {
        "id": "swarm_and_screen",
        "threat": "lancer",
        "tax": "mite",
        "modifier": "husk_turret",
        "note": "the shoal screens the dash line, the emplacement owns the lane you dodge into",
    },
    {
        "id": "salvage_war",
        "threat": "husk_turret",
        "tax": "skimmer",
        "modifier": "screamer",
        "biomes": ("wreck_graveyard", "debris_field"),
        "note": "everything in it wants the haul rather than the hull",
    },
]


@dataclass(frozen=True)
class WaveRecipe:
    """One threat + tax + modifier sentence, validated against the grammar."""

    id: str
    threat: str
    tax: str
    modifier: str
    biomes: tuple[str, ...] = ()
    min_notoriety: float = 0.0
    weight: float = 1.0
    note: str = ""
    #: Cheapest budget that pays for one purchase of every slot. Derived.
    floor_cost: float = field(default=0.0, compare=False)

    @property
    def slots(self) -> tuple[str, str, str]:
        """The recipe's archetypes in spending order: threat, tax, modifier."""
        return (self.threat, self.tax, self.modifier)

    def allows(self, biome_id: str) -> bool:
        """Whether this recipe may be drawn in *biome_id*."""
        return not self.biomes or biome_id in self.biomes

    def affordable(self, budget: float) -> bool:
        """Whether *budget* covers one purchase of each of the three slots."""
        return budget >= self.floor_cost


def _slot_floor(archetype: str) -> float:
    """Threat cost of the smallest legal purchase of *archetype*."""
    return float(balance.ENEMIES[archetype].threat * PURCHASE_LIMITS[archetype][0])


def _build_recipe(entry: Mapping) -> WaveRecipe:
    """Validate one authored table entry and freeze it into a :class:`WaveRecipe`.

    Every failure here is a typo that would otherwise show up as a wave that
    quietly never spawns, so each one raises rather than falling back.
    """
    for role, pool in ROLE_POOLS.items():
        archetype = entry[role]
        if archetype in EXCLUDED_ARCHETYPES:
            raise ValueError(f"recipe {entry['id']!r} uses {archetype!r}, which waves never buy")
        if archetype not in pool:
            raise ValueError(f"recipe {entry['id']!r} puts {archetype!r} in the {role} slot; expected one of {pool}")
    slots = (entry["threat"], entry["tax"], entry["modifier"])
    if len(set(slots)) != len(slots):
        raise ValueError(f"recipe {entry['id']!r} uses the same archetype in two slots: {slots}")
    for biome_id in entry.get("biomes", ()):
        if biome_id not in balance.BIOMES:
            raise ValueError(f"recipe {entry['id']!r} names unknown biome {biome_id!r}")
    return WaveRecipe(
        id=entry["id"],
        threat=entry["threat"],
        tax=entry["tax"],
        modifier=entry["modifier"],
        biomes=tuple(entry.get("biomes", ())),
        min_notoriety=float(entry.get("min_notoriety", 0.0)),
        weight=float(entry.get("weight", 1.0)),
        note=entry.get("note", ""),
        floor_cost=sum(_slot_floor(archetype) for archetype in slots),
    )


def _build_recipes(table: Sequence[Mapping]) -> tuple[WaveRecipe, ...]:
    recipes = tuple(_build_recipe(entry) for entry in table)
    ids = [recipe.id for recipe in recipes]
    duplicates = {name for name in ids if ids.count(name) > 1}
    if duplicates:
        raise ValueError(f"duplicate wave recipe ids: {sorted(duplicates)}")
    return recipes


#: The validated table, in authoring order.
RECIPES: tuple[WaveRecipe, ...] = _build_recipes(WAVE_RECIPES)
RECIPES_BY_ID: dict[str, WaveRecipe] = {recipe.id: recipe for recipe in RECIPES}


# ============================================================================
# Choosing and pricing a wave
# ============================================================================


def archetype_weight(biome_id: str, archetype: str) -> float:
    """This biome's bias on one archetype; 1.0 where the biome is neutral."""
    return float(BIOME_WEIGHTS.get(biome_id, {}).get(archetype, 1.0))


def recipe_weight(recipe: WaveRecipe, biome_id: str) -> float:
    """The recipe's draw weight in *biome_id*.

    The three slots' biome biases combine as a geometric mean, so a recipe is
    promoted when the biome likes all of it and barred outright when the biome
    bars any of it.
    """
    product = 1.0
    for archetype in recipe.slots:
        weight = archetype_weight(biome_id, archetype)
        if weight <= 0.0:
            return 0.0
        product *= weight
    return recipe.weight * product ** (1.0 / 3.0)


def eligible_recipes(biome_id: str, budget: float, notoriety: float, *, barred: Iterable[str] = ()) -> list[WaveRecipe]:
    """Every recipe this biome, budget and notoriety reading allow, in table order.

    *barred* names archetypes the run is not allowed to buy at all, which is how
    the staged opening keeps Lancers out of run one.
    """
    excluded = frozenset(barred)
    return [
        recipe
        for recipe in RECIPES
        if recipe.allows(biome_id)
        and recipe.affordable(budget)
        and notoriety >= recipe.min_notoriety
        and recipe_weight(recipe, biome_id) > 0.0
        and not excluded.intersection(recipe.slots)
    ]


def choose_recipe(
    rng: random.Random,
    biome_id: str,
    budget: float,
    notoriety: float,
    *,
    avoid: str = "",
    barred: Iterable[str] = (),
) -> WaveRecipe | None:
    """Draw one recipe by biome-weighted chance, or ``None`` if none is affordable.

    *avoid* names the recipe that ran last: it stays in the draw at
    :data:`RECIPE_REPEAT_WEIGHT` of its weight, so a node keeps its character
    without repeating one sentence all sector. *barred* is passed through to
    :func:`eligible_recipes`.
    """
    candidates = eligible_recipes(biome_id, budget, notoriety, barred=barred)
    if not candidates:
        return None
    weights = [
        recipe_weight(recipe, biome_id) * (RECIPE_REPEAT_WEIGHT if recipe.id == avoid else 1.0) for recipe in candidates
    ]
    total = sum(weights)
    if total <= 0.0:
        return candidates[0]
    roll = rng.uniform(0.0, total)
    for recipe, weight in zip(candidates, weights, strict=True):
        roll -= weight
        if roll <= 0.0:
            return recipe
    return candidates[-1]


def purchase(recipe: WaveRecipe, budget: float) -> tuple[list[tuple[str, int]], float]:
    """Spend *budget* on *recipe*, returning ``(purchases, threat spent)``.

    One of everything first, in threat-tax-modifier order, so a wave that can
    only just be paid for is still a whole sentence; then a round-robin top-up
    that thickens each slot a unit at a time up to its
    :data:`PURCHASE_LIMITS` ceiling. Leftover budget that buys nothing legal is
    simply not spent, which is what keeps a rich wave a mix rather than a
    stack.
    """
    remaining = float(budget)
    counts: dict[str, int] = {}
    for archetype in recipe.slots:
        floor = _slot_floor(archetype)
        if floor <= remaining:
            counts[archetype] = PURCHASE_LIMITS[archetype][0]
            remaining -= floor
    bought = True
    while bought:
        bought = False
        for archetype in recipe.slots:
            if archetype not in counts or counts[archetype] >= PURCHASE_LIMITS[archetype][1]:
                continue
            unit = float(balance.ENEMIES[archetype].threat)
            if unit > remaining:
                continue
            counts[archetype] += 1
            remaining -= unit
            bought = True
    purchases = [(archetype, counts[archetype]) for archetype in recipe.slots if archetype in counts]
    return purchases, float(budget) - remaining


# ============================================================================
# Staging
# ============================================================================

#: Every archetype a wave can buy, by its ``balance.ENEMIES`` id.
ARCHETYPE_TYPES: dict[str, type[Enemy]] = {
    "mite": MiteShoal,
    "skimmer": Skimmer,
    "lancer": Lancer,
    "mag_mine": MagMine,
    "welder": Welder,
    "bombardier": Bombardier,
    "screamer": Screamer,
    "husk_turret": HuskTurret,
}


@dataclass
class SpawnChevron:
    """One edge marker: the direction a group is about to tear in from.

    ``bearing`` is radians anticlockwise from +X seen from above, matching the
    rest of the flight-plane convention; ``origin`` is the plane point the group
    forms up on, which is what the HUD projects to a screen edge and what the
    audio director pans the chirp to.
    """

    archetype: str
    bearing: float
    origin: Vec2
    remaining: float


@dataclass
class SpawnGroup:
    """One purchase, placed: an archetype, how many, and where they come from.

    ``centre`` is the ship position the bearing was measured from, kept so the
    bodies form up where the chevron promised even though the ship has had a
    second to move since.
    """

    archetype: str
    count: int
    bearing: float
    origin: Vec2
    centre: Vec2


class WaveComposer(Node):
    """Spends the sector's threat budget on a wave every 35 to 55 seconds.

    Register as ``Services.WAVES`` and call :meth:`start` on sector entry with
    the chart node's seed. The schedule then runs itself: roll an interval,
    price a recipe against ``balance.wave_threat_budget``, raise the edge
    chevrons, and tear the bodies in a second later. :meth:`stop` and
    :meth:`resume` bracket a hunter arrival, which is the one thing that is
    allowed to interrupt the schedule, because the arrival is itself the wave.

    Everything the composer decides is reproducible from ``(seed, sector index,
    biome)``: the same node always throws the same waves in the same order, so
    a player who backs out of a jump and returns meets the danger they left.

    The staged opening is the one exception to "the schedule runs itself":
    while :attr:`held_for_onboarding` is true a wave that comes due waits for
    the hold to lift rather than being thrown away, and run one spends
    :meth:`wave_budget` rather than the whole formula. See
    :meth:`set_run_number`.
    """

    #: ``runtime.SignalNames.WAVE_COMPOSED`` (recipe_id, threat_spent), at chevron time.
    wave_composed = Signal(str, float)
    #: ``runtime.SignalNames.WAVE_SPAWNED`` (unit count), when the bodies exist.
    wave_spawned = Signal(int)

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.sector_index = 0
        self.biome_id: str | None = None
        self.seed = 0
        #: Where spawned bodies are parented; defaults to the run scene root.
        self.spawn_parent: Node3D | None = None
        self.waves_spawned = 0
        self.last_recipe_id = ""
        self.last_threat_spent = 0.0
        self.last_purchases: list[tuple[str, int]] = []
        self.last_elites: list[tuple[str, str]] = []
        self.seconds_to_wave = 0.0
        self._rng = random.Random(0)
        self._running = False
        self._handle = None
        self._chevrons: list[SpawnChevron] = []
        self._wiring = SignalWiring(self)
        self._run_number = RUN_NUMBER_UNKNOWN
        self._run_elapsed = 0.0
        self._arrival_survived = False

    # -- Lifecycle ----------------------------------------------------------

    def on_ready(self):
        self._wiring.want(SignalNames.HUNTER_ARRIVED, self._on_hunter_arrived)
        self._wiring.want(SignalNames.HUNTER_DEPARTED, self._on_hunter_departed)
        self._wiring.sweep()

    def on_update(self, dt: float):
        self._wiring.poll(dt)
        if self.biome_id is not None:
            # The run clock, which the staged opening reads. It keeps running
            # through an arrival: a pause the hunter imposed is not quiet.
            self._run_elapsed += dt

    def on_exit_tree(self):
        self.stop()

    # -- Schedule -----------------------------------------------------------

    @property
    def active(self) -> bool:
        """Whether waves are currently being scheduled."""
        return self._running

    @property
    def paused(self) -> bool:
        """Whether a started sector's schedule is held, as during an arrival."""
        return self.biome_id is not None and not self._running

    @property
    def chevrons(self) -> list[SpawnChevron]:
        """The edge markers currently up, for the HUD and the audio director."""
        return list(self._chevrons)

    def start(self, sector_index: int, biome_id: str, seed: int, *, run_number: int | None = None) -> None:
        """Begin composing for one sector, seeded so its danger is learnable.

        *run_number* is the staged opening's input. Left out, the composer reads
        it off the run scene's config the first time, and falls back to
        :data:`RUN_NUMBER_UNKNOWN` (nothing softened) in a scene that has none.
        A run number already in hand is never overwritten by a sector change:
        jumping is a sector boundary, not a run boundary.

        Starting a sector also clears the field. This is the game's one sector
        boundary (``flow.enter_sector`` calls it on every jump): an explicit
        call, rather than a ``SECTOR_ENTERED`` handler, because the field must
        be empty before the new sector begins, not whenever the wiring gets
        around to it.
        """
        if biome_id not in balance.BIOMES:
            raise KeyError(f"unknown biome {biome_id!r}; expected one of {sorted(balance.BIOMES)}")
        self.stop()
        self.despawn_all()
        if run_number is not None:
            self.set_run_number(run_number)
        elif self._run_number == RUN_NUMBER_UNKNOWN:
            self.set_run_number(self._discover_run_number())
        self.sector_index = int(sector_index)
        self.biome_id = biome_id
        self.seed = int(seed)
        self._rng = random.Random(f"shrike-waves:{seed}:{sector_index}:{biome_id}")
        self.waves_spawned = 0
        self.last_recipe_id = ""
        self.last_threat_spent = 0.0
        self.last_purchases = []
        self.last_elites = []
        self.resume()

    def stop(self) -> None:
        """Hold the schedule and drop any chevron already raised.

        A hunter arrival calls this: the arrival is the wave, and nothing else
        gets to share the screen with it. The seeded stream is kept, so
        :meth:`resume` carries on with the sector's own sequence.
        """
        self._running = False
        if self._handle is not None:
            self.stop_coroutine(self._handle)
            self._handle = None
        self._chevrons.clear()
        self.seconds_to_wave = 0.0

    def despawn_all(self) -> int:
        """Clear the field of hostiles and return how many bodies were taken.

        A wave belongs to the sector it was composed for. Bodies are parented to
        the run scene rather than to the sector node, so nothing removes them
        when the sector is torn down and swapped: without this a jump carries the
        pack that chased you into it through the warp and drops it, mid-fight,
        into the next node's opening seconds. The hunter is untouched, because it
        follows the ship by design and leaves on its own signal.
        """
        tree = self.tree
        if tree is None:
            return 0
        doomed = tree.group(Groups.ENEMIES)
        for enemy in doomed:
            enemy.destroy()
        return len(doomed)

    def resume(self) -> None:
        """Restart a held schedule, rolling a fresh interval before the next wave."""
        if self._running or self.biome_id is None:
            return
        self._running = True
        self._handle = self.start_coroutine(self._compose_loop())

    # -- Budget -------------------------------------------------------------

    def notoriety(self) -> float:
        """The run's notoriety reading, or zero before that system is mounted."""
        tree = self.tree
        if tree is None:
            return 0.0
        service = tree.singletons.get(Services.NOTORIETY)
        return float(getattr(service, "value", 0.0)) if service is not None else 0.0

    def threat_budget(self) -> float:
        """T = 10 + 3 x sector index + notoriety / 10, for the current sector."""
        return balance.wave_threat_budget(self.sector_index, self.notoriety())

    def wave_budget(self) -> float:
        """What a wave may actually spend: the formula, softened by the run.

        Only the staged runs differ from :meth:`threat_budget`; every later run
        spends the formula whole.
        """
        return self.threat_budget() * self.budget_fraction()

    # -- The staged opening -------------------------------------------------

    @property
    def run_number(self) -> int:
        """Which run this is, or :data:`RUN_NUMBER_UNKNOWN` when nothing said."""
        return self._run_number

    def set_run_number(self, run_number: int) -> None:
        """Tell the composer which run it is composing for.

        Changing it restarts the staged opening: the run clock and the memory of
        the scripted arrival both belong to one run, not to one sector.
        """
        value = max(RUN_NUMBER_UNKNOWN, int(run_number))
        if value == self._run_number:
            return
        self._run_number = value
        self._run_elapsed = 0.0
        self._arrival_survived = False

    @property
    def softened(self) -> bool:
        """Whether this run is still inside the staged opening."""
        return RUN_NUMBER_UNKNOWN < self._run_number <= SOFTENED_RUNS

    @property
    def arrival_survived(self) -> bool:
        """Whether a hunter arrival has come and gone this run."""
        return self._arrival_survived

    @property
    def run_elapsed(self) -> float:
        """Seconds of composing this run, across every sector of it."""
        return self._run_elapsed

    @property
    def held_for_onboarding(self) -> bool:
        """Whether the staged opening is still holding the next wave back.

        Run one holds everything until the scripted arrival has been survived:
        the first thing a new pilot fights is the Shrike, and the game says so
        rather than opening with a Lancer pack behind the tutorial cards. Run
        two only holds :data:`RUN_2_FIRST_WAVE_GRACE_S`.

        Run one's hold has a ceiling because the condition that lifts it is not
        one the composer can bring about. A pilot who leaves the opening sector
        before the arrival is provoked is never handed one, and without a
        ceiling the run would have no way back to a schedule at all, meeting
        nothing whatsoever for as long as it lasted. Past :data:`RUN_1_HOLD_CEILING_S` the
        schedule runs regardless, still at run one's softened budget and without
        the barred archetypes.
        """
        if not self.softened:
            return False
        if self._run_number == 1:
            return not self._arrival_survived and self._run_elapsed < RUN_1_HOLD_CEILING_S
        return self._run_elapsed < RUN_2_FIRST_WAVE_GRACE_S

    def budget_fraction(self) -> float:
        """Share of the threat budget this run's waves may spend."""
        return RUN_1_BUDGET_FRACTION if self._run_number == 1 else 1.0

    def barred_archetypes(self) -> frozenset[str]:
        """Archetypes this run never buys."""
        return RUN_1_BARRED_ARCHETYPES if self._run_number == 1 else frozenset()

    def _discover_run_number(self) -> int:
        """The run number off the run scene's config, or unknown without one.

        The composer is a singleton rather than a child of the run scene, so it
        reads the run it belongs to from the scene root's own state. A caller
        that would rather say it outright passes ``run_number`` to
        :meth:`start`.
        """
        tree = self.tree
        root = tree.root if tree is not None else None
        if root is None:
            return RUN_NUMBER_UNKNOWN
        value = getattr(getattr(root, "state", None), "run_number", None)
        if value is None:
            config = getattr(root, "config_dict", None)
            value = config.get("run_number") if isinstance(config, Mapping) else None
        return RUN_NUMBER_UNKNOWN if value is None else max(RUN_NUMBER_UNKNOWN, int(value))

    def live_enemy_units(self) -> int:
        """Bodies currently on the field, counting a mite shoal as its live mites."""
        tree = self.tree
        if tree is None:
            return 0
        total = 0
        for enemy in tree.group(Groups.ENEMIES):
            if getattr(enemy, "destroying", False):
                continue
            total += int(getattr(enemy, "alive_count", 1))
        return total

    # -- Composition --------------------------------------------------------

    def compose(self, budget: float) -> list[tuple[str, int]]:
        """Draw a recipe for *budget* and price it into ``(archetype, count)`` buys.

        Records the drawn recipe on :attr:`last_recipe_id` and what it cost on
        :attr:`last_threat_spent`. An empty list means no recipe in this biome
        was affordable, which is a quiet wave rather than an error.
        """
        biome_id = self.biome_id or "debris_field"
        recipe = choose_recipe(
            self._rng,
            biome_id,
            budget,
            self.notoriety(),
            avoid=self.last_recipe_id,
            barred=self.barred_archetypes(),
        )
        if recipe is None:
            self.last_recipe_id = ""
            self.last_threat_spent = 0.0
            self.last_purchases = []
            return []
        purchases, spent = purchase(recipe, budget)
        self.last_recipe_id = recipe.id
        self.last_threat_spent = spent
        self.last_purchases = list(purchases)
        return purchases

    def _compose_loop(self):
        while self._running:
            self.seconds_to_wave = self._rng.uniform(balance.WAVE_INTERVAL_MIN_S, balance.WAVE_INTERVAL_MAX_S)
            while self.seconds_to_wave > 0.0:
                dt = yield
                self.seconds_to_wave = max(0.0, self.seconds_to_wave - (dt or 0.0))
            # A wave that comes due inside the staged opening is deferred, not
            # cancelled: it lands the instant the hold lifts. Rolling a fresh
            # interval instead would put run two's first wave a whole interval
            # past its grace and leave the opening ninety seconds of empty
            # screen.
            while self.held_for_onboarding:
                yield
            yield from self._run_wave()

    def _run_wave(self):
        budget = self.wave_budget()
        purchases = self.compose(budget)
        if not purchases:
            return
        self.wave_composed.emit(self.last_recipe_id, self.last_threat_spent)
        groups = self._plan_groups(purchases)
        lead = 0.0 if self.biome_id in FOGGED_BIOMES else balance.SPAWN_EDGE_CHEVRON_S
        self._raise_chevrons(groups, lead)
        elapsed = 0.0
        while elapsed < lead:
            dt = yield
            elapsed += dt or 0.0
            for chevron in self._chevrons:
                chevron.remaining = max(0.0, lead - elapsed)
        self._chevrons.clear()
        spawned = self._spawn(groups, budget)
        self.waves_spawned += 1
        self.wave_spawned.emit(spawned)

    # -- Staging ------------------------------------------------------------

    def _ship_centre(self) -> Vec2:
        tree = self.tree
        if tree is not None:
            for ship in tree.group(Groups.SHIP):
                return to_plane(ship.position)
        return Vec2(0.0, 0.0)

    def _plan_groups(self, purchases: Sequence[tuple[str, int]]) -> list[SpawnGroup]:
        """Give each purchase its own readable bearing around the ship."""
        centre = self._ship_centre()
        base = self._rng.uniform(0.0, math.tau)
        step = math.tau / max(1, len(purchases))
        groups: list[SpawnGroup] = []
        for index, (archetype, count) in enumerate(purchases):
            bearing = base + index * step + self._rng.uniform(-SPAWN_BEARING_JITTER, SPAWN_BEARING_JITTER)
            origin = Vec2(
                float(centre.x) + math.cos(bearing) * SPAWN_RING_RADIUS,
                float(centre.y) + math.sin(bearing) * SPAWN_RING_RADIUS,
            )
            groups.append(
                SpawnGroup(
                    archetype=archetype,
                    count=int(count),
                    bearing=bearing % math.tau,
                    origin=origin,
                    centre=Vec2(float(centre.x), float(centre.y)),
                )
            )
        return groups

    def _raise_chevrons(self, groups: Iterable[SpawnGroup], lead: float) -> None:
        """Put an edge marker up per group and pan its chirp, unless the biome fogs it."""
        self._chevrons = []
        if lead <= 0.0:
            return
        for group in groups:
            self._chevrons.append(
                SpawnChevron(
                    archetype=group.archetype,
                    bearing=group.bearing,
                    origin=Vec2(float(group.origin.x), float(group.origin.y)),
                    remaining=lead,
                )
            )
        tree = self.tree
        audio = tree.singletons.get(Services.AUDIO) if tree is not None else None
        if audio is None or not hasattr(audio, "play"):
            return
        for chevron in self._chevrons:
            audio.play("spawn_chevron", position=from_plane(chevron.origin))

    def _spawn_parent(self) -> Node | None:
        if self.spawn_parent is not None and self.spawn_parent.tree is not None:
            return self.spawn_parent
        tree = self.tree
        if tree is None:
            return None
        return tree.root if tree.root is not None else self

    def _spawn(self, groups: Sequence[SpawnGroup], budget: float) -> int:
        """Mount the wave's bodies and return how many units arrived."""
        parent = self._spawn_parent()
        if parent is None:
            return 0
        capacity = max(0, balance.ENEMY_HARD_CAP - self.live_enemy_units())
        bodies: list[Enemy] = []
        units = 0
        for group in groups:
            count = min(group.count, capacity - units)
            if group.archetype == "mite":
                if count < balance.MITE_SHOAL_MIN:
                    continue
                bodies.append(self._mount_mites(parent, group, count))
                units += count
                continue
            for member in range(count):
                bodies.append(self._mount(parent, group, member))
                units += 1
        self._apply_elites(bodies, budget)
        return units

    def _member_position(self, group: SpawnGroup, member: int) -> Vec2:
        centre = group.centre
        bearing = group.bearing + self._rng.uniform(-SPAWN_GROUP_SPREAD, SPAWN_GROUP_SPREAD)
        radius = SPAWN_RING_RADIUS + self._rng.uniform(-SPAWN_RING_JITTER, SPAWN_RING_JITTER)
        return Vec2(float(centre.x) + math.cos(bearing) * radius, float(centre.y) + math.sin(bearing) * radius)

    def _body_name(self, archetype: str, member: int) -> str:
        stem = "".join(part.title() for part in archetype.split("_"))
        return f"{stem}W{self.waves_spawned + 1}N{member + 1}"

    def _mount_mites(self, parent: Node, group: SpawnGroup, count: int) -> Enemy:
        shoal = MiteShoal(name=self._body_name("mite", 0), count=count, seed=self._rng.getrandbits(32))
        shoal.position = from_plane(group.origin)
        parent.add_child(shoal)
        return shoal

    def _mount(self, parent: Node, group: SpawnGroup, member: int) -> Enemy:
        archetype = group.archetype
        node_type = ARCHETYPE_TYPES[archetype]
        if archetype == "mag_mine":
            drift = self._rng.uniform(0.0, math.tau)
            enemy = node_type(name=self._body_name(archetype, member), drift=Vec2(math.cos(drift), math.sin(drift)))
        else:
            enemy = node_type(name=self._body_name(archetype, member))
        enemy.position = from_plane(self._member_position(group, member))
        parent.add_child(enemy)
        setter = getattr(enemy, "set_seed", None)
        if callable(setter):
            setter(self._rng.getrandbits(32))
        return enemy

    def _apply_elites(self, bodies: Sequence[Enemy], budget: float) -> None:
        """Tag part of a rich wave with elite modifiers.

        Past ``balance.ELITE_THREAT_THRESHOLD`` one body is elite, and every
        further :data:`ELITE_BUDGET_PER_EXTRA` of budget buys one more, up to
        :data:`ELITE_MAX_PER_WAVE`. Tags land after the bodies are mounted, so
        a modifier that scales a hull sees the hull it is scaling.
        """
        self.last_elites = []
        if not bodies or budget <= balance.ELITE_THREAT_THRESHOLD:
            return
        extra = int((budget - balance.ELITE_THREAT_THRESHOLD) / ELITE_BUDGET_PER_EXTRA)
        wanted = min(len(bodies), ELITE_MAX_PER_WAVE, 1 + extra)
        for enemy in self._rng.sample(list(bodies), wanted):
            modifier = self._rng.choice(balance.ELITE_MODIFIERS)
            apply_elite(enemy, modifier)
            self.last_elites.append((enemy.ARCHETYPE, modifier))

    # -- Hunter handshake ---------------------------------------------------

    def _on_hunter_arrived(self, arrival_index: int) -> None:
        self.stop()

    def _on_hunter_departed(self) -> None:
        # Run one's hold lifts here: the scripted arrival has been survived, so
        # the pilot has met the thing the waves are a warm-up for.
        self._arrival_survived = True
        self.resume()