fleet.py

"""Fleet engine — run the whole fleet concurrently under ONE rate limiter.

The orchestration layer protoTrader-in-space uses to manage many ships toward a
goal (default: maximise credits/hour). Every ship's job is an async coroutine; the
orchestrator runs them all at once via asyncio over the *single* rate-limited
client (`client.call` self-paces to ~2 req/s) — so a fleet shares the one
per-account API budget correctly, which separate processes can't. Ships spend most
of their time waiting on travel/cooldowns, so the budget easily covers a fleet.

Jobs are deterministic (the proven travel→buy→deliver→fulfill / survey→extract
loops) — the LLM's role is strategy (which job for which ship), not clicking every
leg. See `autopilot()` in tools.py for the agent-facing entry point.
"""

from __future__ import annotations

import asyncio
import re

from . import client as C
from . import events as E
from . import roles as R
from . import tools as T
# Control surface (knobs + presets + per-ship pins + decision log) on the shared protoAgent
# SDK helpers — see knobs.py. Imports graph.sdk transitively, so fleet only loads at runtime
# (when the engine starts), not at plugin-register time. The engine reads KNOBS.get(...) live.
from .knobs import (  # noqa: F401
    DLOG, KNOBS, apply_strategy, current_strategy, decisions, knobs,
    overrides, set_knob, set_override,
)

# The engine signals "trade round trip completed" by a result string starting with this — the
# one place the success marker lives, so a caller (e.g. _trade_route_loop) checks against it
# rather than re-hardcoding the literal (keeps job_trade's message and its readers in sync).
_TRADE_OK = "traded"

# Waypoint → (x, y), refreshed from the window-open marketplace scan (coordinates are
# static within a reset). Feeds analysis.rank_routes' time-aware lane scoring and the
# ride-along picker's nearby-sink check WITHOUT threading coords through every loop
# signature — the ranker itself stays pure (the map is an argument there; this cache is
# just the engine-side plumbing). Missing entries only mean that route/sink falls back
# to the time-blind score / delivery-stop-only sinks, never an error.
_WP_COORDS: dict = {}


async def _ship(sym: str) -> dict:
    return await C.call("GET", f"/my/ships/{sym}")


async def _credits() -> int:
    a = await C.call("GET", "/my/agent")
    return a["credits"]


async def _wait_arrival(sym: str, poll: float = 8.0) -> dict:
    """Block (paced) until the ship is no longer IN_TRANSIT."""
    while True:
        s = await _ship(sym)
        if s["nav"]["status"] != "IN_TRANSIT":
            return s
        await asyncio.sleep(poll)


async def _wait_cooldown(sym: str, poll: float = 5.0) -> None:
    while True:
        s = await _ship(sym)
        cd = s.get("cooldown", {}).get("remainingSeconds", 0)
        if not cd:
            return
        await asyncio.sleep(min(cd + 1, poll if poll > cd else cd + 1))


async def _refuel_if_low(sym: str, frac: float = 0.6, *, log=None) -> None:
    """Top off a ship's fuel if it's below ``frac`` and the current waypoint sells
    fuel. Keeps ships from trickling down to DRIFT-only — call on arrival anywhere."""
    s = await _ship(sym)
    fuel = s.get("fuel", {})
    cap = fuel.get("capacity", 0)
    if cap == 0 or fuel.get("current", 0) >= frac * cap:   # probe (0 fuel) or already full enough
        return
    wp = s["nav"]["waypointSymbol"]
    try:
        m = await C.call("GET", f"/systems/{T._system_of(wp)}/waypoints/{wp}/market")
        if any(g.get("symbol") == "FUEL" for g in m.get("tradeGoods", [])):
            await C.call("POST", f"/my/ships/{sym}/dock")
            r = await C.call("POST", f"/my/ships/{sym}/refuel")
            if log:
                log(f"{sym}: refueled → {r['fuel']['current']}/{r['fuel']['capacity']}")
    except C.SpaceTradersError:
        pass


async def travel_to(sym: str, dest: str, *, max_hops: int = 12, log=None) -> bool:
    """Get a ship parked at ``dest``, looping st_travel through any fuel stops, and
    top off fuel on arrival wherever it's sold.

    st_travel issues one hop (CRUISE / DRIFT / fuel-station detour); we wait out
    each leg and call again until the ship is actually at the destination.
    """
    for _ in range(max_hops):
        s = await _ship(sym)
        nav = s["nav"]
        if nav["status"] != "IN_TRANSIT" and nav["waypointSymbol"] == dest:
            await _refuel_if_low(sym, log=log)   # arrived — top off if fuel is sold here
            return True
        if nav["status"] == "IN_TRANSIT":
            await _wait_arrival(sym)
            continue
        out = await T.st_travel.ainvoke({"ship": sym, "destination": dest})
        if log:
            log(f"{sym}: {out.splitlines()[0]}")
        if out.startswith("REFUSED"):   # too far to auto-DRIFT — don't loop, let the job skip it
            return False
        await _wait_arrival(sym)
    return False


async def run_fleet(jobs: dict, *, log=None) -> dict:
    """Run a {ship_symbol: coroutine} map concurrently; return {ship: result}.

    One shared client → one rate budget. A job that raises is captured as an
    ``error: ...`` string so one ship's failure never sinks the fleet.
    """
    results: dict[str, str] = {}

    async def _run(sym: str, coro):
        try:
            results[sym] = await coro
        except C.SpaceTradersError as e:
            results[sym] = f"error: {e}"
        except Exception as e:  # noqa: BLE001 — isolate one ship's failure
            results[sym] = f"error: {type(e).__name__}: {e}"
        if log:
            log(f"{sym} done: {results[sym]}")

    await asyncio.gather(*[_run(s, c) for s, c in jobs.items()])
    return results


# ── shared job primitives ────────────────────────────────────────────────────


async def _held(sym: str, good: str) -> int:
    s = await _ship(sym)
    return sum(i["units"] for i in s["cargo"]["inventory"] if i["symbol"] == good)


async def _buy(sym: str, good: str, target: int, *, max_price: float | None = None,
               floor: int = 0, log=None) -> None:
    """Buy up to ``target`` units of ``good`` — bounded by three guards:

      * profitability: if ``max_price`` is set and the live unit price exceeds it, buy NOTHING
        (a buy above the ceiling is a guaranteed loss — a saturated market or over-priced
        contract good);
      * working capital: never commit more than ``max_spend_frac`` of current credits to this
        buy, so one high-value good can't drain the treasury into cargo and crater us when the
        sell leg doesn't realize (analysis.affordable_units — the documented ASSAULT_RIFLES crash);
      * treasury floor: with ``floor`` set, no chunk may take LIVE credits below it. A
        caller's budget is a snapshot, but the treasury is fleet-SHARED and run_fleet
        spends it concurrently — only a per-chunk live check can actually hold a
        guaranteed floor (the gate rung's max(reserve_floor, buy_buffer/2): sized off
        a round-start budget, its chunks used to keep buying while a trade hauler's
        own leg drained the same treasury mid-flight).
    """
    from .analysis import affordable_units
    await C.call("POST", f"/my/ships/{sym}/dock")
    # Live unit price — needed for BOTH guards. Fetched once up front (a buy is one round trip,
    # so the extra market read is cheap relative to the leg).
    s = await _ship(sym)
    wp = s["nav"]["waypointSymbol"]
    try:
        m = await C.call("GET", f"/systems/{T._system_of(wp)}/waypoints/{wp}/market")
        price = next((g["purchasePrice"] for g in m.get("tradeGoods", []) if g["symbol"] == good), None)
    except C.SpaceTradersError:
        price = None
    if max_price is not None and price is not None and price > max_price:
        if log:
            log(f"{sym}: SKIP buy {good} @ {price} > ceiling {max_price:.0f} (would lose money)")
        return
    frac = KNOBS.get("max_spend_frac")
    if frac is None:                       # explicit: a missing knob disables the cap, NOT a 0.0 value
        frac = 1.0
    while await _held(sym, good) < target:
        s = await _ship(sym)
        room = s["cargo"]["capacity"] - s["cargo"]["units"]
        want = min(target - await _held(sym, good), room, 20)
        if price:   # cash guards, re-checked against the LIVE (shared) treasury each chunk
            credits_now = await _credits()
            want = affordable_units(credits_now, price, room, want, max_spend_frac=frac)
            if floor:
                want = min(want, max(0, (credits_now - int(floor)) // int(price)))
        if want <= 0:
            if log and price:
                log(f"{sym}: holding cash — {good} @ {price} exceeds the per-trade cap "
                    f"({frac:g}× credits) or the cash floor; bought {await _held(sym, good)}/{target}")
            break
        try:
            r = await C.call("POST", f"/my/ships/{sym}/purchase",
                             json={"symbol": good, "units": want})
            if log:
                log(f"{sym}: bought {r['transaction']['units']}×{good} @ {r['transaction']['pricePerUnit']}")
        except C.SpaceTradersError as e:
            if log:
                log(f"{sym}: buy stopped — {e}")
            break


async def _sell(sym: str, good: str, units: int, *, log=None) -> tuple:
    """Sell ``units`` of ``good`` at the current docked market, CHUNKED to the market's
    per-transaction volume limit (``tradeVolume``) — a low-limit good (e.g. CLOTHING cap 20)
    would otherwise error [4604] when dumped all at once and strand the ship. Returns
    ``(sold, revenue)`` summed from THIS ship's transaction receipts — never a /my/agent
    delta: the treasury is fleet-global and run_fleet sells concurrently, so a balance
    bracket books other ships' sales as this ship's."""
    s = await _ship(sym)
    wp = s["nav"]["waypointSymbol"]
    try:
        m = await C.call("GET", f"/systems/{T._system_of(wp)}/waypoints/{wp}/market")
        vol = next((g.get("tradeVolume", 20) for g in m.get("tradeGoods", []) if g["symbol"] == good), 20)
    except C.SpaceTradersError:
        vol = 20
    vol = vol or 20
    sold, revenue = 0, 0
    while sold < units:
        try:
            r = await C.call("POST", f"/my/ships/{sym}/sell",
                             json={"symbol": good, "units": min(units - sold, vol)})
            sold += r["transaction"]["units"]
            revenue += r["transaction"]["totalPrice"]
            if log:
                log(f"{sym}: sold {r['transaction']['units']}×{good} @ {r['transaction']['pricePerUnit']}")
        except C.SpaceTradersError as e:
            if log:
                log(f"{sym}: sell stopped at {sold}/{units} {good} — {e}")
            break
    return sold, revenue


async def _dump_except(sym: str, keep: str, *, log=None) -> None:
    """Sell (or jettison) everything that isn't ``keep`` to free the hold."""
    try:
        await C.call("POST", f"/my/ships/{sym}/dock")
    except C.SpaceTradersError:
        pass
    s = await _ship(sym)
    for it in list(s["cargo"]["inventory"]):
        if it["symbol"] == keep:
            continue
        await _sell(sym, it["symbol"], it["units"], log=log)   # chunked to the market limit
        left = await _held(sym, it["symbol"])
        if left > 0:  # market won't buy it → jettison to free the hold
            try:
                await C.call("POST", f"/my/ships/{sym}/jettison",
                             json={"symbol": it["symbol"], "units": left})
            except C.SpaceTradersError:
                pass


# ── jobs ─────────────────────────────────────────────────────────────────────


async def _ride_along_load(sym: str, deliver_wp: str, exclude: str, *, log=None) -> dict | None:
    """Fill the hold's SPARE room with a ride-along good before a contract leg departs:
    something THIS market exports that the delivery stop (or a market a trivial hop from
    it) imports — the trip is already paid for, so an empty corner of the hold is margin
    left on the table (analysis.ride_along_pick; v2.7 research round). Bounded by every
    existing buy guard: max_price = the confirmed resale (never a loss leg), the
    max_spend_frac working-capital cap inside ``_buy``, and reserve_floor as the hard
    per-chunk cash floor. ``exclude`` is the CONTRACT good — extra units of it would
    blend into the delivery count. Best-effort BY CONTRACT (in both senses): any failure
    returns None and the contract flow proceeds exactly as if this feature didn't exist —
    the ride-along must never block, delay, or fail the contract leg."""
    try:
        s = await _ship(sym)
        room = s["cargo"]["capacity"] - s["cargo"]["units"]
        if room <= 0:
            return None
        wp = s["nav"]["waypointSymbol"]
        system = T._system_of(wp)
        from . import prices as _P
        from .analysis import ride_along_pick
        m = await C.call("GET", f"/systems/{system}/waypoints/{wp}/market")
        pick = ride_along_pick(
            m.get("tradeGoods") or [],
            _P.price_map(system, max_age=KNOBS.get("route_max_age") or 900.0),
            deliver_wp, min_margin=KNOBS.get("min_margin") or 1, exclude=exclude,
            coords=_WP_COORDS or None)
        if not pick:
            return None
        prior = await _held(sym, pick["good"])
        await _buy(sym, pick["good"], prior + room, max_price=pick["sell_price"],
                   floor=KNOBS.get("reserve_floor") or 0, log=log)
        if await _held(sym, pick["good"]) <= prior:
            return None                      # guards said no — plain contract flow
        if log:
            log(f"{sym}: ride-along {pick['good']} → {pick['sell_at']} "
                f"(+{pick['margin']}/unit alongside the contract leg)")
        return pick
    except Exception as e:  # noqa: BLE001 — opportunistic: degrade to the plain contract flow
        if log:
            log(f"{sym}: ride-along skipped ({e})")
        return None


async def _ride_along_sell(sym: str, ride: dict, *, log=None) -> None:
    """Realize the ride-along AFTER the contract delivery landed (never before — the
    contract leg owns the schedule). Best-effort like the load: an unreachable or
    refusing sink just leaves the cargo aboard for the next buy stop's ``_dump_except``
    to realize — never a wedge, never a jettison."""
    try:
        held = await _held(sym, ride["good"])
        if held <= 0:
            return
        at = (await _ship(sym))["nav"]["waypointSymbol"]
        if at != ride["sell_at"] and not await travel_to(sym, ride["sell_at"], log=log):
            return
        await C.call("POST", f"/my/ships/{sym}/dock")
        await _sell(sym, ride["good"], held, log=log)
    except Exception as e:  # noqa: BLE001 — the ride-along must never fail the contract job
        if log:
            log(f"{sym}: ride-along sell skipped ({e})")


async def job_contract(sym: str, contract_id: str, *, log=None) -> str:
    """Work a procurement contract end to end: accept → buy → deliver → fulfill.
    Spare hold room on the buy→deliver leg carries a RIDE-ALONG good when one clears
    the margin/space/budget guards (v2.7) — sold after each delivery, and any
    ride-along failure silently degrades to the plain contract flow."""
    cs = await C.call("GET", "/my/contracts")
    ct = next((x for x in cs if x["id"] == contract_id), None)
    if not ct:
        return f"contract {contract_id} not found"
    if not ct["accepted"]:
        await C.call("POST", f"/my/contracts/{contract_id}/accept")
    dv = ct["terms"]["deliver"][0]
    good, deliver_wp, req = dv["tradeSymbol"], dv["destinationSymbol"], dv["unitsRequired"]
    system = T._system_of(deliver_wp)
    buys, _ = await T._good_markets(system, good)
    if not buys:
        return f"no market sells {good} in {system} (needs mining) — skipped"
    buy_wp = buys[0]
    # Profitability ceiling: the contract pays (advance + fulfillment) over the
    # required units. Buying a unit above that is a net loss — _buy refuses to.
    pay = ct["terms"].get("payment", {})
    pay_per_unit = (pay.get("onAccepted", 0) + pay.get("onFulfilled", 0)) / max(req, 1)
    cap = (await _ship(sym))["cargo"]["capacity"]
    ride = None   # this leg's ride-along pick (loaded at the buy stop, sold post-delivery)
    while True:
        cs = await C.call("GET", "/my/contracts")
        ct = next(x for x in cs if x["id"] == contract_id)
        dv = ct["terms"]["deliver"][0]
        done = dv["unitsFulfilled"]
        if done >= req or ct["fulfilled"]:
            break
        # Buy: only purchase if we're actually AT the buy market (else we'd loop on a
        # bad-state error). If already holding enough of the good, skip the buy leg.
        if await _held(sym, good) < min(req - done, cap):
            if not await travel_to(sym, buy_wp, log=log):
                return f"could not reach buy market {buy_wp} for {good} (stuck at {done}/{req})"
            await _dump_except(sym, good, log=log)
            await _buy(sym, good, min(req - done, cap), max_price=pay_per_unit, log=log)
            if await _held(sym, good) == 0:
                return (f"skipped {good} contract at {done}/{req}: buy price exceeds the "
                        f"{pay_per_unit:.0f}/unit the contract pays (would lose money)")
            # Spare room rides along (best-effort; guards inside) — AFTER the contract
            # good is aboard, so the contract's own load is never squeezed.
            ride = await _ride_along_load(sym, deliver_wp, good, log=log)
        # Deliver: NEVER dock+deliver unless we actually reached the destination —
        # otherwise the API rejects it ([4510] not at delivery waypoint) and the loop
        # spins on the error (the recurring K93→J66 stall).
        if not await travel_to(sym, deliver_wp, log=log):
            return f"could not reach delivery waypoint {deliver_wp} (stuck at {done}/{req}, holding {good})"
        await C.call("POST", f"/my/ships/{sym}/dock")
        u = await _held(sym, good)
        await C.call("POST", f"/my/contracts/{contract_id}/deliver",
                     json={"shipSymbol": sym, "tradeSymbol": good, "units": u})
        if log:
            log(f"{sym}: delivered {u}×{good}")
        if ride:                              # contract leg done — NOW realize the ride-along
            await _ride_along_sell(sym, ride, log=log)
            ride = None
    r = await C.call("POST", f"/my/contracts/{contract_id}/fulfill")
    return f"fulfilled {contract_id} ({req} {good}); credits {r['agent']['credits']:,}"


async def job_mining(sym: str, asteroid: str, ore: str, sell_wp: str, *, log=None) -> str:
    """Fill the hold with an ore at an asteroid, haul it to a market, and sell."""
    await travel_to(sym, asteroid, log=log)
    await C.call("POST", f"/my/ships/{sym}/orbit")
    s = await _ship(sym)
    cap = s["cargo"]["capacity"]
    dry = 0
    while (await _ship(sym))["cargo"]["units"] < cap and dry < 4:
        await _wait_cooldown(sym)
        out = await T.st_extract.ainvoke({"ship": sym, "prefer": ore})
        if "Error" in out:
            break
        if ore not in out:
            # survey to try to target the ore
            await T.st_survey.ainvoke({"ship": sym})
            dry += 1
        else:
            dry = 0
    await _dump_except(sym, ore, log=log)
    held = await _held(sym, ore)
    if held == 0:
        return f"{asteroid} yielded no {ore} (mine elsewhere or buy it)"
    await travel_to(sym, sell_wp, log=log)
    await C.call("POST", f"/my/ships/{sym}/dock")
    r = await C.call("POST", f"/my/ships/{sym}/sell", json={"symbol": ore, "units": held})
    return f"mined+sold {held}×{ore} for {r['transaction']['totalPrice']:,} cr"


async def _good_price(wp: str, good: str) -> tuple:
    """(purchasePrice, sellPrice) for ``good`` at ``wp`` — needs a ship present; (None, None) if unknown."""
    p, s, _ = await _good_quote(wp, good)
    return (p, s)


async def _good_quote(wp: str, good: str) -> tuple:
    """(purchasePrice, sellPrice, tradeVolume) for ``good`` at ``wp`` — needs a ship present;
    (None, None, None) if unknown. tradeVolume is the per-transaction cap the saturation
    guard sizes a delivery against."""
    try:
        m = await C.call("GET", f"/systems/{T._system_of(wp)}/waypoints/{wp}/market")
        g = next((x for x in m.get("tradeGoods", []) if x["symbol"] == good), None)
        return (g.get("purchasePrice"), g.get("sellPrice"), g.get("tradeVolume")) if g else (None, None, None)
    except C.SpaceTradersError:
        return (None, None, None)


async def job_trade(sym: str, good: str, buy_wp: str, sell_wp: str, *,
                    expected_sell: float | None = None, sink_vol: int | None = None,
                    log=None) -> str:
    """One buy-low / sell-high round trip for a good.

    Requires a CONFIRMED sink: ``expected_sell`` is the importer's sell price from the FRESH
    price map (carried on the ranked route). Without it we never commit a buy — that's the
    no-blind-dispatch rule that kills the dead-end loop (a recalled / since-moved sink whose
    price can't be confirmed). It also bounds the buy: we never pay above the resale we have
    evidence for. Sized so it doesn't crash the sink — buy at most
    ``sink_volume_mult × sink_vol`` (the importer's tradeVolume), so one delivery moves the
    import price ~one tier-step instead of cratering it.
    """
    cap = (await _ship(sym))["cargo"]["capacity"]
    # Confirmed-sink guard: only trade a route whose sink we have a fresh sell price for. No
    # price → don't buy blind (the route is recalled/stale or the sink moved).
    if not expected_sell:
        return f"skipped {good} trade: no confirmed sell price at sink {sell_wp} — won't buy blind"
    if not await travel_to(sym, buy_wp, log=log):   # too far to auto-DRIFT → skip, don't buy elsewhere
        return f"skipped {good} trade: couldn't reach buy waypoint {buy_wp} (too far)"
    # Profitability guard: the ship is AT buy_wp now, so its buy price is live/real — bail if the
    # spread vs the confirmed sink price has closed since the route was ranked.
    buy_price, _ = await _good_price(buy_wp, good)
    if buy_price and expected_sell <= buy_price:
        return (f"skipped {good} trade: buy {buy_price} ≥ confirmed sell {expected_sell:.0f} "
                f"({buy_wp}→{sell_wp}) — no margin, would lose money")
    target = cap
    if sink_vol:                                  # saturation cap — keep a delivery ≈ one tier-step
        target = min(cap, max(1, round(KNOBS.get("sink_volume_mult") * sink_vol)))
    await _dump_except(sym, good, log=log)
    await _buy(sym, good, target, max_price=expected_sell, log=log)
    held = await _held(sym, good)
    if held == 0:
        return f"could not buy {good} at {buy_wp} (price ≥ resale or below cash cap — guarded)"
    if not await travel_to(sym, sell_wp, log=log):   # bought but the sink is too far to auto-DRIFT
        return f"bought {held}×{good} but couldn't reach sell waypoint {sell_wp} (too far) — holding"
    await C.call("POST", f"/my/ships/{sym}/dock")
    sold, _ = await _sell(sym, good, held, log=log)
    E.emit("trade_executed", {"ship": sym, "good": good, "buy_at": buy_wp,
                              "sell_at": sell_wp, "units": sold})
    return f"{_TRADE_OK} {sold}×{good} (of {held} hauled)"


async def job_scout(sym: str, market_waypoints: list, deadline: float, *, log=None) -> dict:
    """Visit markets and record live per-unit prices into the persistent price map (a
    ship present unlocks them) — this is what fills the map the trade finder reasons over.

    Stops at the window ``deadline`` so EVERY ship's job finishes together and the engine
    loops to a fresh window promptly — otherwise a long, unbounded sweep strands the faster
    contract/trade ships idle (they finish, then wait on the scouts). Resumes next window;
    the price map persists, so coverage accrues across windows either way.
    """
    from . import prices as _pricemem
    seen: dict[str, dict] = {}
    for wp in market_waypoints:
        if _now() >= deadline:
            break
        if not await travel_to(sym, wp, log=log):
            continue
        await C.call("POST", f"/my/ships/{sym}/dock")
        system = T._system_of(wp)
        m = await C.call("GET", f"/systems/{system}/waypoints/{wp}/market")
        tg = m.get("tradeGoods", [])
        _pricemem.record_market(system, wp, tg)
        seen[wp] = {g["symbol"]: (g["purchasePrice"], g["sellPrice"]) for g in tg}
    return seen


# ── autopilot — drive the whole fleet toward an objective for a time window ──


def _now() -> float:
    return asyncio.get_event_loop().time()


async def _cruise_reachable(system: str, frm: str, to: str, fuel_cap) -> bool:
    """Can a ship CRUISE from ``frm`` to ``to`` (fast) — directly within one tank, or via a fuel
    station within CRUISE range to hop from? False ⇒ only a multi-hour DRIFT would reach it (the
    leg that wedges a contract ship for hours, e.g. the J58 delivery). Fuel-free ships (probes)
    are always reachable. Best-effort: on any lookup failure it returns True, so the guard never
    false-declines a workable contract."""
    if not fuel_cap:
        return True
    try:
        if await T._distance(system, frm, to) <= fuel_cap:
            return True                      # one CRUISE covers it
        fstop = await T._nearest_fuel(system, frm)
    except C.SpaceTradersError:
        return True                          # can't tell → don't false-decline
    if not fstop or fstop[0] == frm:
        return False                         # no fuel station to break the leg up → long DRIFT only
    return fstop[1] <= fuel_cap              # a fuel station is within one CRUISE → can hop + continue


async def _contract_loop(sym: str, deadline: float, claimed: set, lock, *, log=None) -> str:
    """A cargo ship: claim/negotiate a procurement contract, work it, repeat."""
    hq = (await C.call("GET", "/my/agent")).get("headquarters")  # negotiate at a faction waypoint
    n = 0
    while _now() < deadline:
        async with lock:
            cs = await C.call("GET", "/my/contracts")
            # Prefer an accepted, unfulfilled contract; else pick up an OFFERED one — a
            # fresh agent STARTS with an offered contract, and the API refuses to
            # negotiate a new one (4511) while an offer is pending, so we must ACCEPT the
            # offer, not negotiate around it (the bug that parked the starter contract).
            ct = (next((c for c in cs if c["accepted"] and not c["fulfilled"]
                        and c["id"] not in claimed), None)
                  or next((c for c in cs if not c["accepted"] and not c["fulfilled"]
                           and c["id"] not in claimed), None))
            if ct is None:
                try:
                    # Negotiate only while DOCKED at a faction waypoint — travel to HQ
                    # and dock first, or it errors "not docked".
                    if hq:
                        await travel_to(sym, hq, log=log)
                        try:
                            await C.call("POST", f"/my/ships/{sym}/dock")
                        except C.SpaceTradersError:
                            pass
                    ct = (await C.call("POST", f"/my/ships/{sym}/negotiate/contract"))["contract"]
                except C.SpaceTradersError as e:
                    return f"{n} contract(s) done; no more available ({e})"
            # Sourceability guard (offered + negotiated alike): don't accept a contract
            # whose good no market in-system sells — it's un-fulfillable, and an accepted
            # contract can't be cancelled, so it would block this ship until it expires.
            ndv = ct["terms"]["deliver"][0]
            g = ndv["tradeSymbol"]
            buys, _ = await T._good_markets(T._system_of(ndv["destinationSymbol"]), g)
            if not buys:
                return (f"{n} done; declined an un-sourceable {g} contract "
                        f"(no market sells it in-system) — parking, not hauling dead weight")
            # Reachability guard: decline a contract whose delivery is beyond ~one tank from
            # the source. The ship CAN technically DRIFT there (1 fuel), but a 700u+ DRIFT
            # takes HOURS and won't finish inside an engine window — it just wedges the only
            # cargo ship on an un-fulfillable accepted contract (the J66/DRUGS trap). Work
            # in-range contracts + supply-chain trade instead, and bank toward a longer-range
            # hauler. (We tried value-aware accept of far lucrative ones — the occasional 169k
            # win wasn't worth the repeated wedging; range is a SHIP problem, not a guard one.)
            try:
                deliver_wp = ndv["destinationSymbol"]
                fuel_cap = (await _ship(sym))["fuel"].get("capacity") or 400
                if not await _cruise_reachable(T._system_of(deliver_wp), buys[0], deliver_wp, fuel_cap):
                    return (f"{n} done; declined {g} contract — delivery {deliver_wp} is only "
                            f"reachable from the source ({buys[0]}) by a multi-hour DRIFT (no CRUISE "
                            f"route within fuel range). A far contract just wedges the ship for hours; "
                            f"working in-range trade + banking toward a longer-range hauler instead.")
            except Exception:  # noqa: BLE001 — the guard must never crash the loop
                pass
            cid = ct["id"]
            if not ct["accepted"]:
                try:
                    await C.call("POST", f"/my/contracts/{cid}/accept")
                except C.SpaceTradersError as e:
                    return f"{n} contract(s) done; couldn't accept {g} contract ({e})"
            claimed.add(cid)
        if log:
            log(f"{sym}: working contract {cid}")
        res = await job_contract(sym, cid, log=log)
        if log:
            log(f"{sym}: {res}")
        n += 1
        if "skipped" in res or "could not" in res:
            return f"{n-1} done; stuck: {res}"
    return f"completed {n} contract(s)"


def _traded_symbols(m: dict) -> set:
    """Every good a market lists (live ``tradeGoods`` when a ship is present, plus the
    static import/export/exchange lists) — a sell of anything else is a guaranteed 400."""
    return {g["symbol"] for k in ("tradeGoods", "imports", "exports", "exchange")
            for g in m.get(k) or []}


async def _sell_all_here(sym: str, *, log=None) -> tuple:
    """Sell every cargo good the current docked market actually TRADES, splitting what
    remains by WHY it remains. The market's trade list is checked FIRST — a sell for an
    unlisted good is a guaranteed 400, and a full hold of them replays it every run (the
    CE5C trap: 28 straight rejected sells at a market that didn't buy the ore). Returns
    ``(credits_earned, refused, kept)``: ``refused`` = goods missing from the trade list
    (never sellable HERE — recovery candidates); ``kept`` = goods the market lists but
    didn't absorb this visit (saturation [4604] / a transient error) — those stay in the
    hold, because a glut clears (saturation damping is documented-normal) and destroying
    a listed good would turn every demand dip into cargo loss. A per-good failure logs
    and moves on — an engine job must never crash the ops loop. An unreadable market
    skips the pre-check (attempt everything) and every failure lands in ``kept`` — we
    can't PROVE a market refuses a good without reading its trade list, and jettison
    needs proof. The trade list is re-read at every stop, never cached — import lists
    evolve with the economy, and a stale allowlist would re-arm the very trap this
    check removes. ``credits_earned`` is summed from THIS ship's sell receipts, never
    a /my/agent bracket: the loops' zero-progress strike-out rides on it, and the
    treasury is fleet-global — with run_fleet selling concurrently, another ship's
    sale landing inside the bracket would read as progress here and reset the
    strike-out forever (the CE5C spin, back under normal fleet load)."""
    s = await _ship(sym)
    wp = s["nav"]["waypointSymbol"]
    try:
        m = await C.call("GET", f"/systems/{T._system_of(wp)}/waypoints/{wp}/market")
        traded = _traded_symbols(m)
    except C.SpaceTradersError:
        traded = None
    earned, refused, kept = 0, {}, {}
    for it in list(s["cargo"]["inventory"]):
        good, units = it["symbol"], it["units"]
        if traded is not None and good not in traded:
            refused[good] = units
            if log:
                log(f"{sym}: {wp} doesn't trade {good} — skipping sell ({units} held)")
            continue
        try:
            _, revenue = await _sell(sym, good, units, log=log)   # chunked to the market limit
            earned += revenue
            left = await _held(sym, good)
        except C.SpaceTradersError as e:
            left = units          # can't confirm a sale → count the stack as still held
            if log:
                log(f"{sym}: sell {good} failed — {e}")
        if left > 0:
            kept[good] = left
    return earned, refused, kept


async def _reroute_or_jettison(sym: str, market: str, refused: dict, *,
                               must_trade: set, allow_reroute: bool,
                               buyers: dict, log=None) -> tuple:
    """Bounded recovery when the assigned sell market's trade list REFUSED goods (the
    CE5C trap). EVERY refused good first gets a buyer-existence scan (T._good_markets
    — importers plus exchange, the same definition the assignment-time _markets_buying
    pool uses), cached per window in ``buyers`` so no good is scanned twice: that scan
    is the PROOF jettison requires. 'Reroute budget spent' alone must never destroy
    cargo an in-system market provably takes — mixed tails are the live norm (the
    incident hold was 15/15 mixed ore) and the tail can be the high-value part. Then
    at most ONE side-trip per window to the nearest market buying the biggest refused
    stack. A good whose scan PROVED no buyer anywhere is jettisoned — a hold of
    unsellable ore blocks every future extract, so freeing it beats hoarding it. A
    good with a known buyer, or an unproven scan (a transient failure anywhere in it
    — a waypoints page OR a single market GET — is not evidence; it retries next
    run), rides along in the hold instead; the
    loop's zero-progress strike-out bounds the window if held goods wedge it shut.
    Only ``refused`` goods are recovery candidates: a good the market LISTS is a sale
    deferred, not a sale denied — the same gate applies AT the side-trip market, so a
    [4604] glut there is spared too. Any failure ON the side-trip itself (drift-cap
    REFUSED travel, a dock/sell error at the buyer — one exhausted 429 retry was
    enough live) returns ok=False with NOTHING jettisoned: the failure happened at
    the very market that buys the haul, and destroying it there would convert a
    transient into total loss.
    Returns ``(extra_credits, new_market | None, ok, spent)``. ``new_market`` proposes
    moving the standing sell stop and is set ONLY when the side-trip market also trades
    everything in ``must_trade`` (the yield goods the assigned market WAS buying) —
    without that gate a 3-unit tail hijacks the stop and every later run strands the
    primary ore the assigned market still buys. ``spent`` says the window's side-trip
    was consumed — by the TRIP, not the scans: the ``buyers`` cache already stops the
    re-scan rate leak (~a page of GETs per good per run) that motivated spending on
    the attempt, so an attempt that never launched no longer burns the one trip.
    ok=False ⇒ the side-trip failed or even jettison did, and the caller bails out for
    the window exactly like the extract-rejection guard — never an unbounded retry
    storm."""
    extra, new_market, spent, deferred = 0, None, False, set()
    system = T._system_of(market)
    for good in refused:
        if good in buyers:
            continue
        try:
            _, sellers = await T._good_markets(system, good)   # importers/exchangers pay
        except C.SpaceTradersError:
            continue        # unproven — spared below, and the scan retries next run
        buyers[good] = [w for w in sellers if w != market]
    if allow_reroute:
        # Biggest stack first — the one side-trip should recover the most cargo it can.
        for good in sorted(refused, key=refused.get, reverse=True):
            sellers = buyers.get(good) or []
            if not sellers:
                continue
            target = sellers[0]
            if len(sellers) > 1:               # nearest buyer — don't cross the system for it
                try:
                    dists = [(await T._distance(system, market, w), w) for w in sellers]
                    target = min(dists)[1]
                except C.SpaceTradersError:
                    pass
            spent = True
            if log:
                log(f"{sym}: {market} won't buy {good} — side-trip to {target} (buys it)")
            if not await travel_to(sym, target, log=log):
                if log:
                    log(f"{sym}: couldn't reach side-trip buyer {target} — "
                        f"stopping for the window (haul kept)")
                return extra, None, False, spent
            try:
                await C.call("POST", f"/my/ships/{sym}/dock")
                extra, _, t_kept = await _sell_all_here(sym, log=log)
                deferred = set(t_kept)   # listed HERE, just not absorbed — spare below
                # Adopt the target as the standing stop only if it covers every
                # good the assigned market was buying from this yield — else this
                # stays a side-trip and the loop returns to the assigned market.
                try:
                    tm = await C.call("GET",
                                      f"/systems/{system}/waypoints/{target}/market")
                    if must_trade <= _traded_symbols(tm):
                        new_market = target
                except C.SpaceTradersError:
                    pass
                # Decision-logged, not just narrated: the DecisionLog tail is what the
                # window's lesson synthesis distills, so the KB learns which markets
                # refuse which goods — the narrative log dies with the window.
                DLOG.record("sell-guard",
                            f"{sym}: {market} refused {good} — "
                            + (f"sell stop moved to {target}" if new_market else
                               f"sold at {target}, stop stays {market} (it buys goods "
                               f"{target} doesn't)"))
            except C.SpaceTradersError as e:
                if log:
                    log(f"{sym}: reroute sell at {target} failed — {e} "
                        f"(stopping for the window, haul kept)")
                return extra, None, False, spent
            break                              # ONE trip, win or lose — bounded
    ok = True
    for good in list(refused):
        if good in deferred:
            # The side-trip market LISTS this good — the chunked sell just hit its
            # demand limit ([4604]) partway. Same evidence gate as `kept` at the
            # assigned stop: it rides along (when the stop moved here the next visit
            # sells it; otherwise the next run's spent-budget pass settles it).
            continue
        left = await _held(sym, good)
        if left <= 0:
            continue
        if good not in buyers or buyers[good]:
            # A known in-system buyer — or no proof there isn't one — spares the good:
            # jettison is reserved for scan-proven no-buyer cargo. It rides along; if
            # held goods wedge the hold shut, the strike-out ends the window haul-intact.
            if log:
                why = (f"{buyers[good][0]} buys it" if buyers.get(good)
                       else "buyer scan unproven")
                log(f"{sym}: holding {left}×{good} ({why}) — jettison needs a "
                    f"scan-proven no-buyer")
            continue
        try:
            await C.call("POST", f"/my/ships/{sym}/jettison",
                         json={"symbol": good, "units": left})
            DLOG.record("sell-guard",
                        f"{sym}: jettisoned {left}×{good} — {market} refused it and the "
                        f"buyer scan proved no in-system market takes it")
            if log:
                log(f"{sym}: jettisoned {left}×{good} — no in-system market buys it "
                    f"(freeing the hold so extraction can continue)")
        except C.SpaceTradersError as e:
            ok = False
            if log:
                log(f"{sym}: couldn't jettison {good} ({e})")
    return extra, new_market, ok, spent


async def _sell_stop(sym: str, market: str, *, allow_reroute: bool, buyers: dict,
                     log=None) -> tuple:
    """Dock + sell the haul at ``market``; goods its trade list REFUSES enter the
    bounded recovery — goods it lists but didn't absorb ride along for the next visit.
    Shared by the mining and siphon loops (identical trap, identical fix). ``buyers``
    is the loop's per-window buyer-scan cache. Returns
    ``(credits, new_market | None, ok, reroute_spent)``."""
    await C.call("POST", f"/my/ships/{sym}/dock")
    # Snapshot the yield BEFORE selling: the goods this market trades from it (sold or
    # kept) are the ones a reroute target must also cover before the stop may move.
    haul = {it["symbol"] for it in (await _ship(sym))["cargo"]["inventory"]}
    earned, refused, kept = await _sell_all_here(sym, log=log)
    if kept and log:
        log(f"{sym}: {market} trades {', '.join(kept)} but didn't absorb it — "
            f"keeping the tail for the next run")
    if not refused:
        return earned, None, True, False
    extra, new_market, ok, spent = await _reroute_or_jettison(
        sym, market, refused, must_trade=haul - set(refused),
        allow_reroute=allow_reroute, buyers=buyers, log=log)
    return earned + extra, new_market, ok, spent


# ── survey-fed mining (v2.7) ─────────────────────────────────────────────────────────────
# Surveyed extracts pull from the survey's deposit signature (typically a 2–3× value lift
# over blind extraction), so when a hull on the rock carries a MOUNT_SURVEYOR_* it charts
# the asteroid once per fill cycle and EVERY miner on that rock extracts against the best
# cached survey. The cache is per-ASTEROID, module-level: surveys are per-rock server
# objects, not per-ship, so one frigate's chart feeds the whole dig. Surveys expire and
# exhaust server-side — every survey-shaped failure degrades to a plain extract (the
# fall-back doctrine everywhere in this engine), never a bail; and a fleet whose surveyor
# hull isn't free (it may be the contract lead) just mines plain, exactly as before.
_ASTEROID_SURVEYS: dict[str, list] = {}   # asteroid → cached survey objects (API shape)
_SURVEY_SIZE = {"SMALL": 0, "MODERATE": 1, "LARGE": 2}


def _survey_fresh(sv: dict) -> bool:
    """True while a survey's ``expiration`` is in the future. Unknown/unparseable dates
    count as fresh — the API is the real arbiter, and a spent survey costs only one
    rejected extract that immediately falls back to plain. Pure — host-free testable."""
    exp = (sv or {}).get("expiration")
    if not exp:
        return True
    try:
        from datetime import datetime, timezone
        return datetime.fromisoformat(exp.replace("Z", "+00:00")) > datetime.now(timezone.utc)
    except ValueError:
        return True


def _best_survey(asteroid: str) -> dict | None:
    """The best still-fresh cached survey for ``asteroid`` (largest deposit size wins),
    pruning expired ones on the way. None ⇒ mine plain."""
    fresh = [sv for sv in _ASTEROID_SURVEYS.get(asteroid) or [] if _survey_fresh(sv)]
    _ASTEROID_SURVEYS[asteroid] = fresh
    if not fresh:
        return None
    return max(fresh, key=lambda sv: _SURVEY_SIZE.get(sv.get("size"), 0))


def _drop_survey(asteroid: str, sv: dict) -> None:
    """Discard one survey the API rejected (expired/exhausted) so no extract replays it."""
    _ASTEROID_SURVEYS[asteroid] = [x for x in _ASTEROID_SURVEYS.get(asteroid) or []
                                   if x is not sv]


async def _refresh_surveys(sym: str, asteroid: str, *, log=None) -> None:
    """Chart the rock when the cache holds no fresh survey (called once per fill cycle —
    a still-valid chart is never re-burned: the survey shares the extract cooldown, so a
    redundant one costs a mining swing). Best-effort: any failure logs and leaves plain
    extraction untouched — a survey is an optimizer, never a gate.

    The survey POST inherits the extract path's ONE cooldown-skew retry (v2.7.1): the
    survey shares the mining cooldown, so the same seconds-wide clock skew that costs an
    extract a swing 409s the survey too — and without the retry the chart almost never
    lands (LIVE: the first survey-fed dig 409'd every attempt, mining plain forever). On
    that 409 alone, wait the server-stated remainder and retry once; every other
    rejection stays best-effort log-and-continue."""
    if _best_survey(asteroid) is not None:
        return
    for attempt in range(2):
        try:
            await _wait_cooldown(sym)          # survey and extract share the one cooldown
            d = await C.call("POST", f"/my/ships/{sym}/survey")
            _ASTEROID_SURVEYS[asteroid] = list(d.get("surveys") or [])
            if log:
                log(f"{sym}: surveyed {asteroid} — {len(_ASTEROID_SURVEYS[asteroid])} site(s) "
                    f"charted for the diggers")
            return
        except C.SpaceTradersError as e:
            if attempt == 0 and _is_cooldown_conflict(str(e)):
                await asyncio.sleep(_conflict_wait_s(str(e)))   # server's clock is authoritative
                continue
            if log:
                log(f"{sym}: survey failed at {asteroid} ({e}) — mining plain this cycle")
            return


async def _extract_once(sym: str, asteroid: str) -> str:
    """One extract swing — survey-targeted when a fresh chart for this rock is cached,
    plain otherwise. Returns the same string shape ``st_extract`` does (``Error: …`` on a
    rejection) so the loop's terminal guard reads both paths identically. A survey-shaped
    rejection (expired/exhausted server-side — the API says which by naming the survey)
    drops the dead chart and falls back to a PLAIN extract in the same swing: survey
    trouble must never cost a mining window."""
    sv = _best_survey(asteroid)
    if sv is not None:
        try:
            d = await C.call("POST", f"/my/ships/{sym}/extract", json={"survey": sv})
            y = d.get("extraction", {}).get("yield", {})
            return (f"{sym} extracted {y.get('units', '?')}×{y.get('symbol', '?')} "
                    f"(survey-targeted)")
        except C.SpaceTradersError as e:
            if "survey" in str(e).lower():     # spent chart — same fallback as st_extract's
                _drop_survey(asteroid, sv)
            else:
                return f"Error: {e}"
    return await T.st_extract.ainvoke({"ship": sym})


def _is_cooldown_conflict(out: str) -> bool:
    """True iff a rejected extract/siphon is the 409 cooldown-skew shape — the server's
    cooldown clock and ours disagree by a hair ([4000] "ship … is on cooldown", or a bare
    HTTP 409 with no envelope). LIVE: five of these in one night, each one tripping the
    bail-on-reject guard and killing the miner's whole window over a seconds-wide skew.
    That guard exists for TERMINAL rejections (no mount / not an asteroid — the J58 spam),
    so only this one transient class earns a bounded retry; every other rejection stays
    terminal. Pure — host-free testable."""
    t = out or ""
    return "[4000]" in t or "HTTP 409" in t or "on cooldown" in t.lower()


def _conflict_wait_s(out: str, cap: float = 70.0) -> float:
    """Seconds to wait before the ONE cooldown-skew retry — the SERVER-stated remainder
    parsed from the rejection itself ("… on cooldown for N second(s)"), plus a second of
    slack. The whole point of the skew is that OUR cooldown read said ready while the
    server's clock disagreed — so re-reading our side can wait ZERO and burn the single
    retry on an instant second 409 (the same window-killer, one call later). The
    rejection is the server's authoritative clock, so trust it first; ``_wait_cooldown``
    still runs after as the idiomatic backstop. Capped (extract/survey cooldowns top out
    around 70 s — a parse of some unrelated number must not stall the window); an
    unparseable message → 0 (the _wait_cooldown re-read stays the only wait, exactly the
    pre-parse behavior). Pure — host-free testable."""
    m = re.search(r"(\d+(?:\.\d+)?)\s*second", out or "")
    return min(float(m.group(1)) + 1.0, cap) if m else 0.0


async def _mining_loop(sym: str, deadline: float, asteroid: str, market: str, *, log=None) -> str:
    """A mining ship: fill the hold at an asteroid, sell at a market, repeat.

    Guarded so an impossible job can't spam the rate budget: a ship with no mining mount
    (a pin can force a laser-less hull here), a failed travel to the rock, or an extract the
    API rejects all bail out cleanly instead of re-extracting wherever the ship sits — that
    was the J58 case (a pinned light-hauler 'extracting' at a marketplace, looping on 3001).
    The SELL side is guarded too (the CE5C trap — a market that doesn't buy the ore left the
    hold full, so extraction skipped and the loop replayed sell-400s until the deadline):
    goods the market's trade list refuses get ONE reroute to a buying market, a jettison only
    when a per-good buyer scan PROVED no in-system market takes them, else a clean bail —
    while listed-but-unabsorbed goods stay held for the next visit, and an unreachable sell
    market ends the window with the haul intact (selling from the rock is the same 400-spam
    trap, and cargo the market DOES buy must never be destroyed over travel). A zero-progress
    strike-out closes the last spin path: a market that lists the ore but never absorbs it
    (or one whose data can't be read) produces sell stops that neither earn nor shrink the
    hold — three straight of those end the window, haul kept, instead of replaying rejected
    sells until the deadline (the CE5C shape with the refusal check passed)."""
    me = await _ship(sym)
    if not R.can_mine(me):
        if log:
            log(f"{sym}: no mining mount — can't mine (pin a mining-capable ship instead)")
        return f"{sym} has no mining laser — skipped mining"
    surveyor = R.can_survey(me)   # a MOUNT_SURVEYOR_* hull charts its rock for every digger
    runs, earned, stalls = 0, 0, 0
    rerouted = False   # one sell-reroute per window — bounded recovery, not a wander
    buyers: dict = {}  # good → in-system seller wps: the window's buyer-scan proof cache
    while _now() < deadline:
        if not await travel_to(sym, asteroid, log=log):   # never extract unless we REACHED the rock
            if log:
                log(f"{sym}: couldn't reach asteroid {asteroid} — skipping mining this window")
            return f"could not reach {asteroid} — no mining"
        await C.call("POST", f"/my/ships/{sym}/orbit")
        if surveyor:
            # Once per fill cycle: refresh the shared per-asteroid chart if none is fresh.
            # Best-effort — a failed survey mines plain; a fleet with no free surveyor
            # (the frigate may be on contract duty) never reaches here at all.
            await _refresh_surveys(sym, asteroid, log=log)
        cap = (await _ship(sym))["cargo"]["capacity"]
        conflicted = False   # one 409 (cooldown-skew) retry per streak — see _is_cooldown_conflict
        while (await _ship(sym))["cargo"]["units"] < cap and _now() < deadline:
            await _wait_cooldown(sym)
            out = await _extract_once(sym, asteroid)
            if "Error" in out:
                if _is_cooldown_conflict(out) and not conflicted:
                    # Cooldown skew (LIVE: 5 windows killed in one night): wait out the
                    # SERVER-stated remainder from the rejection (our own cooldown read
                    # said ready — that's the skew), re-read as backstop, retry ONCE. A
                    # second consecutive rejection — 409 or otherwise — is terminal below.
                    conflicted = True
                    if log:
                        log(f"{sym}: extract hit cooldown skew (409) — waiting it out, one retry")
                    await asyncio.sleep(_conflict_wait_s(out))
                    await _wait_cooldown(sym)
                    continue
                # The API rejected the extract (not an asteroid / no mount / etc.) — terminal,
                # not transient: stop instead of retrying it every cooldown and burning budget.
                if log:
                    log(f"{sym}: extract rejected ({out.strip()[:60]}) — stopping mining")
                return f"{sym} can't extract at {asteroid}: {out.strip()[:80]}"
            conflicted = False
        if not await travel_to(sym, market, log=log):
            # Never sell (or recover) without REACHING the marketplace: docked at the
            # rock every sell is a 400 and the pre-check can't read a market, so the
            # recovery would destroy a perfectly sellable haul — bail like the extract
            # guard instead, cargo intact for the next window.
            if log:
                log(f"{sym}: couldn't reach sell market {market} — stopping mining (haul kept)")
            return (f"could not reach sell market {market} — stopping mining "
                    f"({runs} run(s), +{earned:,} cr)")
        units_before = (await _ship(sym))["cargo"]["units"]
        got, new_market, ok, spent = await _sell_stop(sym, market, allow_reroute=not rerouted,
                                                      buyers=buyers, log=log)
        earned += got
        rerouted = rerouted or spent   # the trip spends the budget; buyer scans are cached
        if new_market:
            # The reroute found a buyer that covers everything this market was buying
            # too — sell THERE for the rest of the window instead of re-tripping the
            # trap each run. (A partial buyer stays a side-trip: _reroute_or_jettison
            # only proposes the move when nothing sellable is lost by it.)
            market = new_market
        if not ok:
            return (f"{sym} sell recovery failed at {market} — stopping mining "
                    f"({runs} run(s), +{earned:,} cr, haul kept where possible)")
        runs += 1
        # Progress is measured on THIS ship's facts alone — its sell receipts (`got`)
        # and its hold: a stop that neither earned nor shrank the hold (sold or
        # jettisoned) is a stall. The fleet-global treasury must never feed this
        # check — other ships sell concurrently, and their earnings would reset the
        # count forever. Gluts are documented-normal and usually clear, so THREE
        # straight stalls — not one — end the window, haul kept, instead of
        # replaying rejected sells to deadline.
        if got > 0 or (await _ship(sym))["cargo"]["units"] < units_before:
            stalls = 0
        else:
            stalls += 1
            if stalls >= 3:
                return (f"no sell progress at {market} for {stalls} straight runs — "
                        f"stopping mining ({runs} run(s), +{earned:,} cr, haul kept)")
        if log:
            log(f"{sym}: mining run {runs} done (+{earned:,} cr total)")
    return f"{runs} mining run(s), +{earned:,} cr"


async def _siphon_loop(sym: str, deadline: float, giant: str, market: str, *, log=None) -> str:
    """A siphon ship: fill the hold at a gas giant, sell at a market, repeat. Guarded exactly
    like _mining_loop — no gas-siphon mount, an unreachable giant, a rejected siphon, an
    unsellable haul (the CE5C trap), or a market that never absorbs the gas (the zero-progress
    strike-out) all bail/recover cleanly instead of spamming the rate budget."""
    if not R.can_siphon(await _ship(sym)):
        if log:
            log(f"{sym}: no gas-siphon mount — can't siphon (pin a siphon-capable ship instead)")
        return f"{sym} has no gas siphon — skipped siphoning"
    runs, earned, stalls = 0, 0, 0
    rerouted = False   # one sell-reroute per window — bounded recovery, not a wander
    buyers: dict = {}  # good → in-system seller wps: the window's buyer-scan proof cache
    while _now() < deadline:
        if not await travel_to(sym, giant, log=log):
            if log:
                log(f"{sym}: couldn't reach gas giant {giant} — skipping siphon this window")
            return f"could not reach {giant} — no siphon"
        await C.call("POST", f"/my/ships/{sym}/orbit")
        cap = (await _ship(sym))["cargo"]["capacity"]
        conflicted = False   # one 409 (cooldown-skew) retry per streak — same as the mining loop
        while (await _ship(sym))["cargo"]["units"] < cap and _now() < deadline:
            await _wait_cooldown(sym)
            out = await T.st_siphon.ainvoke({"ship": sym})
            if "Error" in out:
                if _is_cooldown_conflict(out) and not conflicted:
                    conflicted = True
                    if log:
                        log(f"{sym}: siphon hit cooldown skew (409) — waiting it out, one retry")
                    await asyncio.sleep(_conflict_wait_s(out))   # server-stated remainder first
                    await _wait_cooldown(sym)
                    continue
                if log:
                    log(f"{sym}: siphon rejected ({out.strip()[:60]}) — stopping siphon")
                return f"{sym} can't siphon at {giant}: {out.strip()[:80]}"
            conflicted = False
        if not await travel_to(sym, market, log=log):
            # Same guard as the mining loop: no marketplace reached ⇒ no sell, no
            # recovery, no jettison — the haul is kept and the window ends cleanly.
            if log:
                log(f"{sym}: couldn't reach sell market {market} — stopping siphon (haul kept)")
            return (f"could not reach sell market {market} — stopping siphon "
                    f"({runs} run(s), +{earned:,} cr)")
        units_before = (await _ship(sym))["cargo"]["units"]
        got, new_market, ok, spent = await _sell_stop(sym, market, allow_reroute=not rerouted,
                                                      buyers=buyers, log=log)
        earned += got
        rerouted = rerouted or spent   # the trip spends the budget; buyer scans are cached
        if new_market:
            market = new_market
        if not ok:
            return (f"{sym} sell recovery failed at {market} — stopping siphon "
                    f"({runs} run(s), +{earned:,} cr, haul kept where possible)")
        runs += 1
        # Same strike-out as the mining loop, on the same per-ship facts (receipts +
        # hold, never the shared treasury): three sell stops that neither earn nor
        # shrink the hold end the window, haul kept — never a rejected-sell replay.
        if got > 0 or (await _ship(sym))["cargo"]["units"] < units_before:
            stalls = 0
        else:
            stalls += 1
            if stalls >= 3:
                return (f"no sell progress at {market} for {stalls} straight runs — "
                        f"stopping siphon ({runs} run(s), +{earned:,} cr, haul kept)")
        if log:
            log(f"{sym}: siphon run {runs} done (+{earned:,} cr total)")
    return f"{runs} siphon run(s), +{earned:,} cr"


# Engine route cache — the ranked trade routes, refreshed each window (a ship present at a
# market unlocks live prices). The control surface (knobs/presets/pins/decision-log) moved to
# knobs.py on the shared SDK helpers and is imported at the top of this module.
_ROUTE_CACHE: dict = {"at": -1e9, "routes": []}


async def _ranked_routes(system: str, market_wps: list) -> list:
    """All currently-profitable arbitrage routes, ranked best-first (a ship present unlocks
    live prices). Cached 60s — scanning every market is costly. Returns a LIST so several
    haulers can diversify across the top-N (anti-saturation) instead of stacking on one."""
    now = _now()
    if _ROUTE_CACHE["routes"] and now - _ROUTE_CACHE["at"] < 60:
        return _ROUTE_CACHE["routes"]
    from . import prices, routes
    from .analysis import rank_routes
    # Refresh the persistent PRICE MAP for any currently-lit markets, then rank arbitrage
    # over the RECORDED map (built up as probes sweep the system) — not just what's lit
    # right now, which is almost never two markets at once. REMEMBER the best for next wipe.
    for wp in market_wps:
        try:
            m = await C.call("GET", f"/systems/{system}/waypoints/{wp}/market")
        except C.SpaceTradersError:
            continue
        if m.get("tradeGoods"):
            prices.record_market(system, wp, m["tradeGoods"])
    # Rank ONLY over FRESH prices (route_max_age) — a stale leg is not a live route. Prices
    # drift fast (the strategist watched a +3,086/unit margin collapse to +81 within windows),
    # so dispatch a hauler only on a freshly-confirmed export→import pair. Coordinates
    # (window-open waypoint scan → _WP_COORDS) make the score time-aware net-per-second;
    # any unmapped endpoint falls back to the time-blind score per route.
    ranked = rank_routes(prices.price_map(system, max_age=KNOBS.get("route_max_age")),
                         min_margin=KNOBS.get("min_margin"),
                         sink_supply_cutoff=KNOBS.get("sink_supply_cutoff"),
                         coords=_WP_COORDS or None)
    if ranked:
        top = ranked[0]
        await routes.remember_route(system, top["good"], top["buy_at"], top["sell_at"],
                                    top["profit_per_unit"])
    # Recalled routes (routes.recall_routes) are NO LONGER dispatched to haulers: a remembered
    # route can be from a prior wipe or a since-moved market, so buying on it blind is exactly the
    # dead-end loss (the "D40 attractor"). Recall now only SEEDS exploration — probes re-light
    # remembered markets, and a hauler trades a route only once it's re-confirmed fresh above.
    # When the fresh scan is dry the hauler idles (waits for probes), which never loses money.
    # (PR 3 will add probe stationing to hold the active route's endpoints live continuously.)
    _ROUTE_CACHE.update(at=now, routes=ranked)
    return ranked


async def _best_trade_route(system: str, market_wps: list, rank: int = 0) -> dict | None:
    """The ``rank``-th best profitable route (0 = best). The plan's reconcile hands each hauler
    a distinct top-N route so they spread out instead of piling on the single best route and
    saturating it; this rank arg serves the fallback re-ranking loop."""
    ranked = await _ranked_routes(system, market_wps)
    if not ranked:
        return None
    return ranked[min(rank, len(ranked) - 1)]


async def _clear_stranded_hold(sym: str, system: str, *, log=None) -> bool:
    """Free a trader parked with cargo no local market bought — the stranded-full gap that
    left a hauler dead ~1h live (issue #55). Called ONLY from _trade_loop's no-route branch,
    so it never pre-empts a real run. Sells the biggest stranded stack at the NEAREST in-system
    market that buys it, accepting sub-margin — freeing the ship beats hoarding dead capital.
    Guards: never act mid-flight; NEVER liquidate contract-deliver cargo (a stuck contract lead
    reaches _trade_loop still holding its haul — /my/contracts is the authoritative signal, and
    an unreadable list protects EVERYTHING, fail-safe); an unproven buyer scan (T._good_markets
    raises) or a provably-empty buyer set leaves the cargo aboard — #55 SELLS stranded cargo, it
    never jettisons (that proof-gated destruction is the mining loop's job, not this). At most one
    side-trip per window. Returns True iff it sold, so the caller can look for a route again."""
    try:
        s = await _ship(sym)
    except C.SpaceTradersError:
        return False
    if s["nav"]["status"] == "IN_TRANSIT":
        return False
    here = s["nav"]["waypointSymbol"]
    held = {i["symbol"]: i["units"] for i in s["cargo"]["inventory"] if i["units"] > 0}
    if not held:
        return False
    try:
        contracts = await C.call("GET", "/my/contracts")
    except C.SpaceTradersError:
        return False        # can't prove it isn't contract cargo → protect everything
    protected = {d.get("tradeSymbol") for c in contracts
                 if c.get("accepted") and not c.get("fulfilled")
                 for d in (c.get("terms", {}).get("deliver") or [])}
    # Biggest non-protected stack first — the one side-trip should recover the most cargo.
    for good in sorted(held, key=held.get, reverse=True):
        if good in protected:
            continue
        try:
            _, sellers = await T._good_markets(system, good)   # importers/exchangers pay
        except C.SpaceTradersError:
            continue        # unproven scan — leave it aboard, retry next window
        sellers = [w for w in sellers if w != here]            # 'here' already refused it
        if not sellers:
            continue        # proven no in-system buyer — hold it (never jettison)
        target = sellers[0]
        if len(sellers) > 1:                                   # nearest — don't cross the system
            try:
                dists = [(await T._distance(system, here, w), w) for w in sellers]
                target = min(dists)[1]
            except C.SpaceTradersError:
                pass
        if log:
            log(f"{sym}: stranded {held[good]}×{good} at {here} — repositioning to {target} to "
                f"sell (issue #55)")
        if not await travel_to(sym, target, log=log):
            continue        # too far to auto-DRIFT — leave aboard, try again next window
        await C.call("POST", f"/my/ships/{sym}/dock")
        sold, _ = await _sell(sym, good, held[good], log=log)
        return bool(sold)
    return False


async def _trade_loop(sym: str, deadline: float, system: str, market_wps: list,
                      *, rank: int = 0, log=None) -> str:
    """A cargo ship: run a profitable arbitrage route (its diversify ``rank``), re-evaluating
    each round so it adapts when a market saturates. job_trade is spread-guarded + sized to
    the sink's tradeVolume (no loss legs, no self-crash)."""
    runs = 0
    cleared = False                           # one stranded-hold rescue per window (issue #55)
    while _now() < deadline:
        route = await _best_trade_route(system, market_wps, rank=rank)
        if not route:
            if not cleared:                   # nothing profitable AND maybe parked full — free it
                cleared = True
                if await _clear_stranded_hold(sym, system, log=log):
                    continue                  # hold freed — re-check for a route we can now buy
            await asyncio.sleep(15)           # wait for probes to scout a profitable spread
            continue
        res = await job_trade(sym, route["good"], route["buy_at"], route["sell_at"],
                              expected_sell=route.get("sell_price"),
                              sink_vol=route.get("sink_volume"), log=log)
        if log:
            log(f"{sym}: {res}")
        runs += 1
        if "skipped" in res:                 # this route went unprofitable (saturation)
            _ROUTE_CACHE["at"] = -1e9         # force a re-scan next round
    return f"completed {runs} trade run(s)" if runs else "no profitable route this window — idled"


# The unworkable-contract exit markers _contract_loop returns — shared by the trade-fallback
# trigger and the strikeout classifier so the two predicates can't drift apart.
_CONTRACT_STUCK = ("declined", "unreachable", "no more available", "could not", "couldn't",
                   "skipped")


def _contract_struck_out(res: str) -> bool:
    """True iff a contract window produced NOTHING: the loop exited on an unworkable-contract
    path with ZERO contracts fulfilled. Every unworkable exit leads with the fulfilled count
    ("0 done; declined …"), so a lead that fulfilled one and THEN ran dry is a success — the
    demotion in plan.reconcile must only count windows where the contract role earned nothing
    at all. A deadline-exhausted window ("completed 0 contract(s)" — mid-haul, still working)
    is not a strikeout either. Pure — host-free testable."""
    m = re.match(r"\s*(\d+)", res or "")
    if m and int(m.group(1)) > 0:
        return False
    return any(k in (res or "") for k in _CONTRACT_STUCK)


async def _contract_then_trade(sym: str, deadline: float, claimed: set, lock, system: str,
                               market_wps: list, *, rank: int = 0, log=None) -> str:
    """The lead cargo ship works contracts, but if there's no WORKABLE contract this window
    (declined as un-sourceable/unreachable, or a slot blocked by an un-fulfillable accepted
    one), it FALLS BACK to trade routes for the rest of the window instead of sitting idle —
    on ``rank``, the best route the plan did NOT already hand another hauler (no saturation
    pile-up). The window's outcome is reported into the persistent plan first
    (plan.report_contract_window), so the next reconcile can demote a lead that keeps
    striking out instead of re-pinning it to a dead contract slot every window."""
    res = await _contract_loop(sym, deadline, claimed, lock, log=log)
    from . import plan as _plan
    _plan.report_contract_window(_contract_struck_out(res))
    stuck = any(k in res for k in _CONTRACT_STUCK)
    if stuck and _now() < deadline and system and market_wps:
        if log:
            log(f"{sym}: no workable contract ({res}) — trading instead")
        tres = await _trade_loop(sym, deadline, system, market_wps, rank=rank, log=log)
        return f"{res}; fell back to trade → {tres}"
    return res


# Ship-cost estimates (cr) — affordability gates before a reinvest buy (the purchase itself
# always pays the yard's LIVE price). v2.7 research round corrected the price book to the
# observed typical yard prices — probe ≈28k, light hauler ≈393k — the old conservative rows
# (40k/450k) made every rung wait a window-plus of earnings past the point the treasury
# could actually afford the hull (compounding delayed for nothing).
_SHIP_COST = {"SHIP_PROBE": 28_000, "SHIP_LIGHT_HAULER": 393_000, "SHIP_HEAVY_FREIGHTER": 1_500_000}
_LONG_RANGE_FUEL = 1500   # a hull with this fuel cap can reach far contracts/routes a hauler can't


async def _yard_stock(system: str) -> tuple[dict, bool]:
    """``({ship_type: yard_waypoint}, proven)`` for every hull type an in-system shipyard
    actually SELLS — scanned ONCE per reinvest pass and never cached across windows
    (yards restock and change). The ladder consults this so an affordable-but-unsold
    rung can fall through to the next rung the SAME window instead of ending
    reinvestment (the live ≥1.5M wedge: the heavy-freighter rung matched, no X1-FX92
    yard sells one, the buy skipped, and the elif chain never reached the light-hauler
    rung — reinvest bought NOTHING every window).

    ``proven`` is True only when EVERY yard read succeeded: a flaked per-yard GET
    (exhausted-retry 429s are the live norm) leaves that yard's types simply ABSENT
    from the dict, and reading absent-because-unread as not-sold is the empty-side lie
    (the same possibly-incomplete-is-unproven doctrine as tools._good_markets) — except
    here the consequence of the lie is a wrong-rung PURCHASE, not a skipped sale."""
    stock: dict = {}
    try:
        yards = await C.call("GET", f"/systems/{system}/waypoints",
                             params={"traits": "SHIPYARD", "limit": 20})
    except C.SpaceTradersError:
        return stock, False
    proven = True
    for y in yards or []:
        try:
            sy = await C.call("GET", f"/systems/{system}/waypoints/{y['symbol']}/shipyard")
        except C.SpaceTradersError:
            proven = False   # this yard's stock is UNKNOWN, not empty
            continue
        for t in sy.get("shipTypes", []) or []:
            stock.setdefault(t.get("type"), y["symbol"])
    return stock, proven


async def _buy_ship_at_yard(ships: list, system: str, ship_type: str, *,
                            yard: str | None = None, log=None) -> bool:
    """Ferry a free-flying probe (or any ship) to a shipyard that SELLS ``ship_type`` and buy it.
    Best-effort. Picks a yard whose ``shipTypes`` includes the type — buying at the first
    shipyard regardless used to silently error + no-op when that yard didn't stock it.
    Pass ``yard`` when the caller already scanned availability (``_yard_stock``) so the
    ladder doesn't re-walk every shipyard per rung."""
    if yard is None:
        stock, _ = await _yard_stock(system)   # a miss here only SKIPS — safe on a flake
        yard = stock.get(ship_type)
    if yard is None:
        if log:
            log(f"reinvest skipped: no shipyard in {system} sells {ship_type}")
        return False
    mover = next((s for s in ships if s["fuel"].get("capacity", 0) == 0), None) or (ships[0] if ships else None)
    if mover is None or not await travel_to(mover["symbol"], yard, log=log):
        return False
    try:
        r = await C.call("POST", "/my/ships", json={"shipType": ship_type, "waypointSymbol": yard})
        if log:
            log(f"reinvest: bought {r['ship']['symbol']} ({ship_type}) @ {yard}")
        E.emit("ship_purchased", {"ship": r["ship"]["symbol"], "type": ship_type, "yard": yard})
        from . import watches as _W
        _W.stamp_deployment()   # st-idle-capital's ground truth: capital was deployed
        # A hull purchase is deliberate deployment too: step the st-drawdown high-water
        # mark down by the price (the API receipt when present, the conservative book
        # estimate otherwise) so buying the fleet the ladder exists FOR never reads as
        # a treasury crash. The ShipyardTransaction receipt's amount field is ``price``
        # (st_buy_ship in tools.py reads this same endpoint's receipt as t['price']);
        # ``totalPrice`` exists only on MARKET (cargo) transactions — reading it here
        # made the receipt branch dead live, so an expensive hull (a 2.3M freighter vs
        # the 1.5M book row) stepped the mark by the estimate and st-drawdown fired on
        # the exact deliberate deployment this reconcile licenses.
        _W.reconcile_deployment(int(((r.get("transaction") or {}).get("price") or 0)
                                    or _SHIP_COST.get(ship_type, 0)))
        return True
    except C.SpaceTradersError as e:
        if log:
            log(f"reinvest skipped ({ship_type}): {e}")
        return False


async def _maybe_reinvest(ships: list, system: str, *, log=None) -> None:
    """Self-scaling reinvestment: fill the price map FIRST with cheap probes (parallel
    scouting → arbitrage surfaces sooner), then scale trade with haulers once the map is
    covered and capital is comfortable. Every buy also keeps credits at/above reserve_floor.

    AVAILABILITY-AWARE LADDER (v2.5): the rungs are tried in priority order as a LOOP,
    not an elif chain — a rung whose hull no in-system yard SELLS cannot EXECUTE, so it
    falls through to the next rung the SAME window. The old chain wedged live at ≥1.5M
    credits: the heavy-freighter rung matched (affordable), no X1-FX92 yard sells one,
    the buy skipped, and control never reached the light-hauler rung — reinvest bought
    nothing every window while capital piled up. Yard stock is scanned once per pass
    (``_yard_stock``), never cached across windows — yards restock/change."""
    if len(ships) >= KNOBS.get("max_ships"):
        return
    credits = await _credits()
    floor = KNOBS.get("reserve_floor")

    def _affordable(ship_type: str, buffer_knob: str) -> bool:
        # reserve_floor is the CANONICAL guard: the buy must not dip below it. The per-tier
        # *_buffer is now just an OPTIONAL extra comfort threshold on top — raising reserve_floor
        # alone already tightens every reinvest, so the cash story is one knob (the buffers are
        # legacy comfort dials). Both conditions must hold.
        return credits >= KNOBS.get(buffer_knob) and credits - _SHIP_COST.get(ship_type, 0) >= floor

    from . import prices
    coverage = prices.stats(system).get("markets", 0)
    probes = sum(1 for s in ships if s["fuel"].get("capacity", 0) == 0)
    have_long_range = any(s["fuel"].get("capacity", 0) >= _LONG_RANGE_FUEL for s in ships)
    # The rungs, priority-first — the same policy the elif chain encoded (scouting
    # coverage, then range, then scale), each ``(type, comfort-buffer knob, gate)``:
    #   • probe rung while the price map is thin (arbitrage needs coverage);
    #   • range-upgrade HEAVY_FREIGHTER (2300 fuel + big hold) once very comfortable and
    #     still all short-range — far contracts/routes the drift-cap declines open up;
    #   • LIGHT_HAULER — the scaling default.
    ladder = [
        ("SHIP_PROBE", "probe_buffer",
         coverage < KNOBS.get("map_target") and probes < KNOBS.get("max_probes")),
        ("SHIP_HEAVY_FREIGHTER", "heavy_buffer", not have_long_range),
        ("SHIP_LIGHT_HAULER", "buy_buffer", True),
    ]
    stock: dict | None = None   # lazy: scan yards only once a rung actually needs them
    proven = True               # every yard read succeeded — absent really means unsold
    for ship_type, buffer_knob, gate in ladder:
        if not gate or not _affordable(ship_type, buffer_knob):
            continue
        if stock is None:
            stock, proven = await _yard_stock(system)
        if ship_type not in stock:
            if not proven:
                # UNKNOWN ≠ unsold: a flaked yard read left this rung's absence
                # unproven, and falling through would BUY the next rung — converting
                # one exhausted-retry 429 into an irreversible wrong purchase (a 450k
                # hauler while the fleet was deliberately banking for the heavy). The
                # old chain's safe no-buy is the right move: skip the pass, retry on
                # a clean scan next window.
                if log:
                    log(f"reinvest: yard scan incomplete — can't prove {ship_type} "
                        f"is unsold in {system}; holding this window")
                return
            # Execution-skip: the rung provably can't run HERE, so the next rung gets
            # its turn this window (the wedge fix) — never a silent end to reinvestment.
            if log:
                log(f"reinvest: no shipyard in {system} sells {ship_type} — falling through")
            continue
        # One buy attempt per window, win or lose: a failed travel/purchase is a
        # different problem than an unsold hull, and cascading buy attempts here
        # could stack purchases the cash guards sized independently.
        await _buy_ship_at_yard(ships, system, ship_type, yard=stock[ship_type], log=log)
        return


# ── gate supply — surplus capital builds the jump gate (v2.5 capital deployment) ─────
# The system's jump gate under construction (X1-FX92-I59: 1,280 FAB_MATS + 400
# ADVANCED_CIRCUITRY) is the only path to heavy freighters and multi-system trade —
# and until v2.5 no engine policy spent surplus on it despite the st_construction /
# st_supply_construction tools existing. This rung closes that: when the treasury
# clears gate_buffer, ONE spare hauler per window buys missing materials in-system
# and delivers them, floors respected, so gate spend can never starve trading.


def _remember_gate_sighting(gate_wp: str | None) -> None:
    """Persist the last CONFIRMED gate sighting into the watch-state file (the same
    epoch-scoped home as the drawdown mark — a universe wipe clears it via
    ``watches.clear_epoch_state``). Disk, not module state, on purpose: the restart
    scenario that strands gate cargo mid-window (watchdog re-kick, hot reload) is
    exactly the one that wipes process-lifetime caches. A successful read that shows
    NO gate under construction drops the whole memory — sighting, material list AND
    the stamped gate cargo — a finished gate must never keep drafting carriers off a
    stale sighting (residue cargo goes back to the normal sell-off path)."""
    from . import watches as _W
    st = _W.load_state()
    if gate_wp:
        if st.get("gate_wp_seen") != gate_wp:
            st["gate_wp_seen"] = gate_wp
            _W.save_state(st)
    elif "gate_wp_seen" in st or "gate_goods_seen" in st or "gate_cargo" in st:
        st.pop("gate_wp_seen", None)
        st.pop("gate_goods_seen", None)
        st.pop("gate_cargo", None)
        _W.save_state(st)
        _GATE_GOODS_SEEN.clear()


async def _gate_under_construction(system: str) -> tuple[str | None, bool]:
    """``(gate_wp, live)`` — the system's under-construction JUMP_GATE waypoint (or
    None) and whether the answer came from a LIVE read. One cheap filtered waypoint
    call — made once per dispatch (the carrier check below must run even when the
    treasury is under gate_buffer, or leftover materials get handed to a trade loop).

    On a transient read failure (exhausted-retry 429s are the live norm) the last
    PERSISTED sighting answers with ``live=False``: the entire carrier-first no-dump
    guarantee sits downstream of this one GET, and returning None here handed a
    LOADED hauler to a trade loop whose _dump_except fire-sold the paid-for
    materials — a full spend-budget destroyed by one flaky read. A stale sighting is
    safe: if the gate completed since, job_gate_supply's own site read finds no gaps
    and exits cleanly."""
    try:
        wps = await C.call("GET", f"/systems/{system}/waypoints",
                           params={"type": "JUMP_GATE", "limit": 20})
    except C.SpaceTradersError:
        from . import watches as _W
        return _W.load_state().get("gate_wp_seen"), False
    from .exploration import gate_under_construction
    gate_wp = gate_under_construction(wps or [])
    _remember_gate_sighting(gate_wp)
    return gate_wp, True


def _gate_ship(traders: list, gate_wp: str | None, credits: int, gate_buffer) -> str | None:
    """The ONE hauler this window's gate rung may use — the last spare trader whose hold
    isn't FULL, never the contract lead (traders[0]), and only when a spare exists (≥2
    traders). None ⇒ no gate run this window. The one-hauler cap is the starvation guard:
    contracts and every other route run exactly as before. A full hold is skipped because
    it's trade cargo the sell-guard is holding for its own buyer — gate shopping couldn't
    load a unit, and the ship's plan route is what sells the hold. Pure — host-free
    testable."""
    if not gate_wp or credits < (gate_buffer or 0) or len(traders) < 2:
        return None
    for s in reversed(traders[1:]):
        cargo = s.get("cargo") or {}
        cap, units = int(cargo.get("capacity") or 0), int(cargo.get("units") or 0)
        if cap and units >= cap:
            continue
        return s.get("symbol")
    return None


# The gate's still-missing material LIST from the last successful site read — the
# carrier check's fallback when a construction GET transiently fails: a build's
# material list is static even though the amounts move, so a stale set can only keep
# the no-dump invariant alive across a flaky read, never draft the wrong hauler for
# goods a gate was never built from. Hot cache only — the disk mirror below is the
# ground truth a restart comes back to.
_GATE_GOODS_SEEN: set = set()


def _remember_gate_goods(goods: set) -> None:
    """Refresh the material-list memory: module cache + the watch-state file. The
    restart that strands gate cargo (watchdog re-kick mid-window) also wipes module
    state, so the only fallback must live on disk — same epoch-scoped home as the
    gate sighting (cleared on a universe wipe)."""
    _GATE_GOODS_SEEN.clear()
    _GATE_GOODS_SEEN.update(goods)
    from . import watches as _W
    st = _W.load_state()
    if set(st.get("gate_goods_seen") or []) != goods:
        st["gate_goods_seen"] = sorted(goods)
        _W.save_state(st)


def _recall_gate_goods() -> set:
    """The last successful site read's material list — hot cache first, disk after a
    restart. A stale list beats dumping paid-for cargo (the list is static; only the
    amounts move)."""
    if _GATE_GOODS_SEEN:
        return set(_GATE_GOODS_SEEN)
    from . import watches as _W
    return set(_W.load_state().get("gate_goods_seen") or [])


# ── gate-cargo provenance — stamped at BUY time, consumed at delivery ────────────────
# Cargo symbols cannot tell a cut gate window's leftovers from trade capital, and the
# plan route cannot either: a plan-LESS spare (reconcile ran out of ranked routes →
# route None; _trade_loop picks routes dynamically and never writes the plan) or a
# hauler whose route was SWAPPED after strikes can hold a gap-good tail that is pure
# trade capital — and inferring provenance from the mutable route drafted both shapes
# treasury-blind and donated the paid-for load to the site at zero revenue. So the
# shopping leg stamps what it buys ({ship: {good: units}}) in the watch-state file —
# the same epoch-scoped home as the gate sighting (a universe wipe or a finished gate
# clears it; disk, because the restart that strands gate cargo mid-window wipes module
# state) — and ONLY stamped units are ever treated as deliverable gate cargo.


def _stamped_gate_cargo(sym: str) -> dict:
    """``{good: units}`` of ``sym``'s hold the gate rung BOUGHT — the provenance
    ground truth the carrier check and both delivery legs consult."""
    from . import watches as _W
    return dict(((_W.load_state().get("gate_cargo") or {}).get(sym)) or {})


def _stamp_gate_cargo(sym: str, good: str, units: int) -> None:
    """Record a gate-shopping buy the moment it lands (not at delivery — a cut window
    or failed gate leg is exactly when provenance matters most)."""
    if int(units) <= 0:
        return
    from . import watches as _W
    st = _W.load_state()
    held = st.setdefault("gate_cargo", {}).setdefault(sym, {})
    held[good] = int(held.get(good) or 0) + int(units)
    _W.save_state(st)


def _consume_gate_cargo(sym: str, good: str, units: int) -> None:
    """A delivery landed — its stamped units are spent provenance. Leaving them would
    let a later TRADE buy of the same good on the same hull inherit gate provenance
    and be donated."""
    from . import watches as _W
    st = _W.load_state()
    ships = st.get("gate_cargo") or {}
    held = ships.get(sym) or {}
    left = int(held.get(good) or 0) - int(units)
    if left > 0:
        held[good] = left
    else:
        held.pop(good, None)
        if not held:
            ships.pop(sym, None)
    _W.save_state(st)


def _plan_route_good(sym: str) -> str | None:
    """The good ``sym``'s PERSISTENT plan route trades, or None. The gate's gap goods
    are ordinary market goods (the rung's whole premise is an in-system exporter), so
    a FAB_MATS route can rank and be plan-assigned — and then cargo symbols alone
    cannot tell a cut gate window's leftovers from that route's bought-not-yet-sold
    load (the sell-guard deliberately holds it across cut windows for its known sink).
    The BUY-TIME stamp (``_stamp_gate_cargo``) is the gate rung's provenance ground
    truth — the route can't vouch for a plan-less or route-swapped hauler — but the
    route good still rules two spots on top of it: the shopping leg never buys it on
    this hull, and the carrier/carry legs never ship it even when stamped units of it
    exist (an orphaned stamp must not donate the route's working cargo)."""
    from . import plan as _plan
    a = ((_plan.load() or {}).get("ships") or {}).get(sym) or {}
    return (a.get("route") or {}).get("good") or None


async def _gate_carrier(spares: list, system: str, gate_wp: str) -> str | None:
    """The spare hauler already HOLDING one of the gate's still-missing materials — it
    gets gate duty THIS window regardless of the treasury (carrier-first dispatch).

    job_gate_supply delivers held materials before its budget gate, but that guarantee
    only fires if the loaded ship actually RE-ENTERS gate duty. Two live shapes defeated
    the old credits>=gate_buffer arm gate: a shopping run that starts near the buffer
    necessarily ENDS below it (so the next window handed the loaded hauler to a trade
    loop whose _dump_except sold the paid-for materials at the buy market's bid, or
    jettisoned them), and a same-window reinvest buy shifting traders[-1] onto the new
    hull did the same even with credits above the buffer. Costs one construction GET,
    and only when a spare actually holds STAMPED gate cargo.

    PROVENANCE GUARD (buy-time stamps): only cargo the gate rung's own shopping leg
    STAMPED (``_stamp_gate_cargo``) counts — a gap-good tail without a stamp is trade
    capital whatever the plan says (a plan-less spare's dynamic ``_trade_loop`` buy, a
    route swapped after strikes with the old good's unsold tail still aboard), and
    drafting it (treasury-blind, by design) would hand it to the zero-revenue delivery
    leg. Belt-and-suspenders on top: the ship's CURRENT plan-route good is never
    drafted even when stamped units of it exist (an orphaned stamp must not let the
    route's working cargo ship) — its route sells that hold at its confirmed sink."""
    holders = []
    for s in spares:
        stamped = _stamped_gate_cargo(s.get("symbol"))
        if stamped and any(int(i.get("units") or 0) > 0
                           and int(stamped.get(i.get("symbol")) or 0) > 0
                           for i in (s.get("cargo") or {}).get("inventory") or []):
            holders.append((s, stamped))
    if not holders:
        return None
    from .exploration import construction_gaps
    try:
        site = await C.call("GET", f"/systems/{system}/waypoints/{gate_wp}/construction")
        gap_goods = {g["good"] for g in construction_gaps(site if isinstance(site, dict) else {})}
        _remember_gate_goods(gap_goods)
    except C.SpaceTradersError:
        gap_goods = _recall_gate_goods()    # a stale list beats dumping paid-for cargo
    for s, stamped in holders:
        own = _plan_route_good(s.get("symbol"))
        if any(i.get("symbol") in gap_goods and i.get("symbol") != own
               and min(int(i.get("units") or 0), int(stamped.get(i.get("symbol")) or 0)) > 0
               for i in (s.get("cargo") or {}).get("inventory") or []):
            return s.get("symbol")
    return None


async def _gate_dispatch(traders: list, system: str | None) -> tuple[str | None, str | None]:
    """Pick THIS window's gate-duty hauler: ``(ship, gate_wp)``, or ``(None, gate_wp/None)``.

    Three rules the window-open credits snapshot can't serve:

      * carrier-first — a spare already holding STAMPED gate materials (buy-time
        provenance, never a symbol match) is drafted regardless of
        the treasury (``_gate_carrier``): the leftover-first delivery inside
        job_gate_supply is the no-dump guarantee, and it only runs if the loaded ship
        re-enters gate duty;
      * live credits — autopilot's ``start_credits`` goes stale the moment
        ``_maybe_reinvest`` buys a hull (the ships list is refreshed one line later; the
        treasury never was). Arming on the stale number drafted a hauler whose
        job_gate_supply budget (from LIVE credits) was 0 — it exited on its first round
        and idled the whole window while its plan route went unrun;
      * flake-proof re-entry — the gate-waypoint read falls back to the last persisted
        sighting (``_gate_under_construction`` live=False), so one exhausted-retry 429
        can't skip the carrier check and hand a loaded hauler to a trade loop. The
        stale path drafts ONLY a carrier, never a fresh budget-armed hauler.
    """
    if not system or len(traders) < 2:
        return None, None
    gate_wp, live = await _gate_under_construction(system)
    if not gate_wp:
        return None, None
    carrier = await _gate_carrier(traders[1:], system, gate_wp)
    if carrier:
        return carrier, gate_wp
    if not live:
        # A stale (persisted) sighting licenses PROTECTING loaded cargo — the carrier
        # check above — never NEW spending: the gate may have completed since the last
        # confirmed read, and a fresh draft would idle a hauler on a finished site.
        return None, gate_wp
    return _gate_ship(traders, gate_wp, await _credits(), KNOBS.get("gate_buffer")), gate_wp


async def job_gate_supply(sym: str, gate_wp: str, deadline: float, *, log=None) -> str:
    """Buy the gate's missing materials at in-system markets and deliver them to the
    construction site, round trip after round trip until the window ends or a guard
    stops it. Every round re-reads the site (other agents supply it too) and budgets
    from LIVE credits (``gate_spend_budget`` — never below max(reserve_floor,
    buy_buffer/2), never more than max_spend_frac of the treasury); the budget is
    RE-derived after the travel leg lands and the floor rides into ``_buy`` as a hard
    per-chunk guard — the treasury is fleet-shared and run_fleet spends it
    concurrently, so a round-start snapshot aged minutes by a travel leg licensed
    chunk-buys straight through the guaranteed floor. Materials a prior
    (cut) window already bought are delivered FIRST — before the budget gate — and
    non-gate cargo is NEVER dumped: a drafted hauler's trade tail (the sell-guard holds
    unsold cargo across windows for its known buyer) rides along, shopping fills only
    the free room, and a FULL hold hands the window back to the ship's route. Sources
    resolve from the cached price map when it knows a seller, falling to the next
    candidate when one shows no live price (a stale map row), and to ONE live
    ``_good_markets`` sweep per good per window when the cache runs dry — the sweep is
    the engine's burstiest call pattern, so a per-round rescan would tax the whole
    fleet's rate budget. A missing good with no workable source is logged ONCE per
    window (price-fail path included) and skipped — never a wedge; when no gap good has
    a source the job ends cleanly. A gate spend steps the st-drawdown high-water mark down
    in TWO parts (``watches.reconcile_deployment``), because a gate EXPORT lowers the
    realizable treasury at two moments: units·(P − S) at BUY (the spread the metric marks off
    the instant credits become cargo it values at the in-system sell price S), and units·S at
    DELIVERY (construction consumes the cargo for no credits back). Both read S from the
    DURABLE 1-day price window — the drawdown verifier's own window — so a gate source gone
    >1h stale still supplies the basis and the licensed spend never reads as a crash; a
    buy→deliver window left half-unreconciled would have false-tripped on the exact incident
    (engine down, capital parked in in-flight cargo).

    PROVENANCE GUARD: only units the shopping leg STAMPED at buy time
    (``_stamp_gate_cargo``) are ever delivered — an unstamped gap-good tail is trade
    capital whatever its symbol says (a plan-less hauler's dynamic trade buy, a
    swapped route's unsold load), and the mutable plan route cannot vouch for it. On
    top of the stamps, the ship's OWN plan-route good is never shopped on THIS hull
    (``_plan_route_good`` — a same-symbol gate buy riding next to the route's held
    tail invites exactly the blend the stamps exist to prevent) and never delivered
    by the carry leg. The route sells that hold at its confirmed sink; a hauler on a
    different route supplies the good instead."""
    from . import prices as _P
    from .exploration import construction_gaps, gate_spend_budget, market_sources
    system = T._system_of(gate_wp)
    own_good = _plan_route_good(sym)
    runs, delivered = 0, 0
    own_skip: set = set()    # the route-good skip, logged once per window
    no_source: set = set()   # goods with NO workable source this window — logged once each
    sources: dict = {}       # good → REMAINING candidate sellers (cached map first, live
    #                          sweep as fallback): listings are stable within a window
    #                          (only PRICES drift, and the buy leg re-reads the live price
    #                          anyway), so re-sweeping per round would just re-tax the
    #                          shared rate budget for the same answer
    tried: dict = {}         # good → sources whose LIVE price read failed (a stale map
    #                          row: the good left the listings, or the GET 429-exhausted)
    #                          — never revisited, never re-offered by the sweep fallback
    swept: set = set()       # goods that already got their ONE live sweep this window
    sweep_unproven: dict = {}  # good → error, when the sweep itself couldn't finish
    while _now() < deadline:
        try:
            site = await C.call("GET", f"/systems/{system}/waypoints/{gate_wp}/construction")
        except C.SpaceTradersError as e:
            return f"gate supply stopped — couldn't read the site ({e})"
        gaps = construction_gaps(site if isinstance(site, dict) else {})
        if not gaps:
            return (f"gate {gate_wp} needs nothing — construction complete "
                    f"({runs} run(s), {delivered} units)")
        # Leftover-first: a cut window / failed gate leg leaves BOUGHT materials in the
        # hold. Deliver them before the budget gate could hold them hostage — the
        # treasury already funded this cargo, and delivering costs nothing and frees
        # the hauler either way. (The dispatch side guarantees re-entry: _gate_carrier
        # drafts a loaded spare regardless of the treasury, so this path actually runs
        # after a cut window instead of the cargo landing in a trade loop's
        # _dump_except.)
        s = await _ship(sym)
        gap_goods = {g["good"] for g in gaps}
        # Only STAMPED units are gate leftovers (buy-time provenance) — a same-symbol
        # unstamped tail is trade capital and rides along untouched; own_good is
        # excluded on top (see the docstring's guard).
        stamped = _stamped_gate_cargo(sym)
        carry = next((it for it in s["cargo"]["inventory"]
                      if it["symbol"] in gap_goods and it["symbol"] != own_good
                      and min(it["units"], int(stamped.get(it["symbol"]) or 0)) > 0), None)
        if carry is not None:
            # Capped at the LIVE gap: another agent may have supplied the site since
            # this cargo was bought, and over-delivering errors the supply call. Any
            # residue simply stops matching a gap next round and is sold off normally.
            gap = next(g["missing"] for g in gaps if g["good"] == carry["symbol"])
            good, held = carry["symbol"], min(carry["units"],
                                              int(stamped.get(carry["symbol"]) or 0), gap)
        else:
            room = s["cargo"]["capacity"] - s["cargo"]["units"]
            if room <= 0:
                # A drafted hauler can arrive with a FULL hold of trade cargo the
                # sell-guard is holding for its own buyer (windows are deadline-cut
                # mid-route by design). Gate shopping can't load a single unit, and
                # dumping the hold at a source market is exactly the fire-sale /
                # jettison the v2.4.x sell-guard exists to prevent — hand the window
                # back instead; the plan kept the ship's route, so the next window
                # sells the hold at its confirmed sink.
                return (f"gate supply skipped — hold full of trade cargo held for its "
                        f"buyer ({runs} run(s), {delivered} units)")
            budget = gate_spend_budget(await _credits(), KNOBS.get("reserve_floor") or 0,
                                       KNOBS.get("buy_buffer") or 0, KNOBS.get("max_spend_frac"),
                                       KNOBS.get("gate_buffer") or 0)
            if budget <= 0:
                return (f"gate supply holding — treasury at/below the gate floors "
                        f"({runs} run(s), {delivered} units delivered)")
            # First gap good an in-system market actually SELLS (st_find_market's buy
            # side — exports/exchange). The CACHED price map answers first (free — the
            # probes' recorded market visits carry the export/exchange types), and the
            # live _good_markets sweep runs once per good per window when the cache is
            # blind OR every cached seller failed its live price read (a stale row:
            # record_market only upserts goods PRESENT in a new read, so a delisted
            # good's row survives until age-out — writing the good off on it would
            # stall gate supply for up to an hour with a FALSE 'no source' story). An
            # unproven sweep (incomplete pagination / a lost market GET) skips THIS
            # window and retries next — never buy blind.
            pick = None
            for g in gaps:
                good = g["good"]
                if good == own_good:
                    # Never shopped on THIS hull: a gate buy of the route's own good
                    # would blend with any held trade tail (the supply call sends
                    # every held unit) and donate both — provenance guard.
                    if good not in own_skip:
                        own_skip.add(good)
                        if log:
                            log(f"{sym}: gate needs {g['missing']}×{good} but that is "
                                f"{sym}'s own route cargo good — provenance guard, "
                                f"skipping it on this hull")
                    continue
                if good in no_source:
                    continue
                if good not in sources:
                    sources[good] = [w for w in market_sources(_P.price_map(system), good)
                                     if w not in tried.get(good, ())]
                if not sources[good] and good not in swept:
                    swept.add(good)
                    try:
                        live, _ = await T._good_markets(system, good)
                        sources[good] = [w for w in live if w not in tried.get(good, ())]
                    except C.SpaceTradersError as e:
                        sweep_unproven[good] = str(e)
                if sources[good]:
                    pick = (good, g["missing"], sources[good][0])
                    break
                no_source.add(good)
                if log:
                    if good in sweep_unproven:
                        why = f"source scan unproven ({sweep_unproven[good]})"
                    elif tried.get(good):
                        why = ("no live price at any known source "
                               f"({', '.join(sorted(tried[good]))})")
                    else:
                        why = "no in-system market sells it"
                    log(f"{sym}: gate needs {g['missing']}×{good} but {why} — "
                        f"skipping it this window")
            if pick is None:
                missing = ", ".join(g["good"] for g in gaps)
                why = ("every missing good is unsourceable or this hull's own route "
                       "cargo (provenance guard)" if own_skip
                       else "no in-system source for any missing good")
                return (f"gate supply skipped — {why} "
                        f"({missing}) ({runs} run(s), {delivered} units)")
            good, missing, src = pick
            if not await travel_to(sym, src, log=log):
                return (f"gate supply stopped — couldn't reach source {src} "
                        f"({runs} run(s), {delivered} units)")
            price, _ = await _good_price(src, good)
            if not price:
                # No live price HERE — the good left this market's listings since the
                # map row was recorded, or the read 429-exhausted (the live norm). The
                # old code wrote the whole GOOD off as no-source on the first dead
                # seller, silently: drop just THIS candidate, say so, and re-enter the
                # round — the next candidate (or the one-per-window sweep above) gets
                # its turn.
                tried.setdefault(good, set()).add(src)
                sources[good] = [w for w in sources[good] if w != src]
                if log:
                    log(f"{sym}: no live price for {good} at {src} — trying the next source")
                continue
            s = await _ship(sym)
            room = s["cargo"]["capacity"] - s["cargo"]["units"]
            # RE-budget from LIVE credits now the travel leg is behind us: the round-
            # start budget aged minutes in flight while other ships spent the same
            # shared treasury — sizing off the stale snapshot chunk-bought below the
            # guaranteed floor (a single concurrent job_trade buy can commit up to
            # max_spend_frac of the treasury mid-travel).
            budget = gate_spend_budget(await _credits(), KNOBS.get("reserve_floor") or 0,
                                       KNOBS.get("buy_buffer") or 0, KNOBS.get("max_spend_frac"),
                                       KNOBS.get("gate_buffer") or 0)
            want = min(room, missing, budget // price)
            if want <= 0:
                return (f"gate supply holding — budget {budget:,} cr won't buy a unit of "
                        f"{good} @ {price} ({runs} run(s), {delivered} units)")
            prior = await _held(sym, good)
            # The floor rides into _buy as a hard per-chunk guard against the LIVE
            # treasury — the budget above is still a snapshot the fleet keeps spending
            # under (gate_spend_budget's floor, kept in lockstep). Target is
            # prior + want: _buy counts TOTAL held units, and an unstamped same-symbol
            # trade tail riding along must not eat the shopping allowance.
            await _buy(sym, good, prior + want,
                       floor=max(int(KNOBS.get("reserve_floor") or 0),
                                 int(KNOBS.get("buy_buffer") or 0) // 2), log=log)
            total = await _held(sym, good)
            bought = total - prior
            if bought > 0:
                # Stamp provenance the moment the buy lands (a cut window / failed gate leg
                # is exactly when it matters): only these units may ever be delivered — a
                # pre-existing same-symbol tail is trade capital and stays aboard for its
                # route.
                _stamp_gate_cargo(sym, good, bought)
                # Reconcile the BUY half of the st-drawdown mark NOW. Gate materials are
                # EXPORT goods: paid at purchasePrice `price` (P), but the treasury metric
                # values the resulting cargo at the (lower) in-system SELL price S — often 0
                # when no in-system market buys the export. So the buy DROPS the realizable
                # treasury by bought·(P − S) the instant it lands (credits → cargo marked to
                # S), NOT at delivery. This is the blocker fix: a delivery-only step left the
                # buy→deliver window (engine down, capital parked in cargo the metric already
                # marks to S) unreconciled and false-tripped the alarm on the exact incident
                # this metric exists for. The residual bought·S drops at CONSUMPTION (delivery
                # below). S is read from the DURABLE window — the same book the verifier values
                # this cargo against — so the mark tracks the metric exactly; unknown → S=0 →
                # the full spend is the buy-time drop.
                from . import watches as _W_buy
                sell = _W_buy.best_sell_price(_P.price_map(system, max_age=86_400.0), good)
                _W_buy.reconcile_deployment(bought * max(0, int(price) - int(sell)))
            held = min(total, int(_stamped_gate_cargo(sym).get(good) or 0), missing)
            if held <= 0:
                return (f"gate supply stopped — couldn't buy {good} at {src} "
                        f"({runs} run(s), {delivered} units)")
        if not await travel_to(sym, gate_wp, log=log):
            return f"holding {held}×{good} but couldn't reach the gate {gate_wp}"
        try:
            await C.call("POST", f"/my/ships/{sym}/dock")
            await C.call("POST",
                         f"/systems/{system}/waypoints/{gate_wp}/construction/supply",
                         json={"shipSymbol": sym, "tradeSymbol": good, "units": held})
        except C.SpaceTradersError as e:
            return f"gate delivery failed at {gate_wp} ({e}) — holding {held}×{good}"
        runs += 1
        delivered += held
        _consume_gate_cargo(sym, good, held)   # spent provenance — a later trade buy
        #                                        of the same good must not inherit it
        from . import watches as _W
        _W.stamp_deployment()   # st-idle-capital's other deploy path
        # Reconcile the DELIVERY half of the st-drawdown mark HERE, at consumption. The buy
        # already stepped the mark by units·(P − S) — the spread the export lost the instant
        # it became cargo the metric marks to its in-system SELL price S. Consumption drops
        # the REMAINING units·S: the delivered materials leave the holds and construction
        # returns no credits, so the credits+cargo metric falls by exactly the sell value the
        # cargo was carrying. Buy step + delivery step = units·P total (the full licensed
        # spend), each landing when that portion of the treasury actually falls. S is read
        # from the DURABLE window — the drawdown verifier's window, and the SAME S the buy
        # step used — so a gate source gone >1h stale still supplies the basis (a 1h lookup
        # would find nothing and reconcile by 0). When no in-system market buys the export
        # (S=0) the whole drop already landed at buy and this step is a no-op. Covers the
        # carry leg cleanly: a prior window's buy stepped its (P − S), so this window's
        # delivery reconciles only its residual S, exactly once, here.
        sell = _W.best_sell_price(_P.price_map(system, max_age=86_400.0), good)
        _W.reconcile_deployment(int(held) * int(sell))
        E.emit("construction_supplied",
               {"ship": sym, "good": good, "units": held, "site": gate_wp})
        DLOG.record("gate-supply", f"{sym}: delivered {held}×{good} to {gate_wp}")
        if log:
            log(f"{sym}: delivered {held}×{good} to the gate ({delivered} this window)")
    return f"{runs} gate-supply run(s), {delivered} units to {gate_wp}"


def _d2(a: dict, b: dict) -> int:
    return (a.get("x", 0) - b.get("x", 0)) ** 2 + (a.get("y", 0) - b.get("y", 0)) ** 2


# What extraction can yield — the sell half of a mining/siphon assignment must BUY at least
# one of these (imports/exchange) or the loop's sell stop rejects the whole haul (the CE5C
# trap survives at runtime, but the right sell target avoids the detour entirely). Kept
# BROAD (common minerals + the precious/exotic ores richer rocks drop): a missing symbol
# here can only shrink the buying pool toward the all-markets fallback — the trap again.
_ORE_GOODS = {"IRON_ORE", "COPPER_ORE", "ALUMINUM_ORE", "SILICON_CRYSTALS", "QUARTZ_SAND",
              "ICE_WATER", "AMMONIA_ICE", "PRECIOUS_STONES", "DIAMONDS",
              "SILVER_ORE", "GOLD_ORE", "PLATINUM_ORE", "URANITE_ORE", "MERITIUM_ORE"}
_GAS_GOODS = {"HYDROCARBON", "LIQUID_HYDROGEN", "LIQUID_NITROGEN"}


async def _markets_buying(system: str, markets: list, goods: set) -> list:
    """The subset of ``markets`` whose imports/exchange intersect ``goods`` — readable with
    no ship present, so assignment can check BEFORE dispatching a miner. Falls back to ALL
    of them when none qualifies: a nearest-market pick plus the runtime sell guard beats
    stranding every extractor on a scan gap."""
    buying = []
    for w in markets:
        try:
            m = await C.call("GET", f"/systems/{system}/waypoints/{w['symbol']}/market")
        except C.SpaceTradersError:
            continue
        if {g["symbol"] for k in ("imports", "exchange") for g in m.get(k) or []} & goods:
            buying.append(w)
    return buying or markets


async def _mining_targets(system: str) -> list:
    """Mineable (asteroid, sell-market) pairs in ``system``, ranked by SHORTEST extract→sell
    round trip first — so the engine mines the most CENTRAL asteroids (near the markets =
    near the system core = reachable without a marathon DRIFT), not an arbitrary ``asteroids[0]``
    that a low-fuel drone can't get to (which, with the travel drift-cap, left miners idle).
    Sell markets are drawn from those that actually BUY ore (_markets_buying) — pairing a
    miner with a market that won't take its yield is the CE5C trap.

    Returns a list so the autopilot can SPREAD several miners across different asteroids/markets
    instead of piling them all on one rock + one buyer (which craters that ore's price). Empty
    list ⇒ no asteroid or no market; the miner falls back to trade. Prefers ENGINEERED_ASTEROIDs.
    """
    async def _of_type(t: str) -> list:
        try:
            return await C.call("GET", f"/systems/{system}/waypoints",
                                params={"type": t, "limit": 20})
        except C.SpaceTradersError:
            return []
    asteroids = (await _of_type("ENGINEERED_ASTEROID")
                 or await _of_type("ASTEROID_FIELD")
                 or await _of_type("ASTEROID"))
    if not asteroids:
        return []
    # One page (~20) is enough for a CANDIDATE pool — a wrong pick here just means a
    # farther market, and the runtime sell guard recovers it. Only the jettison-proof
    # scan must see every marketplace page (T._good_markets paginates for that).
    try:
        markets = await C.call("GET", f"/systems/{system}/waypoints",
                               params={"traits": "MARKETPLACE", "limit": 20})
    except C.SpaceTradersError:
        return []
    if not markets:
        return []
    pool = await _markets_buying(system, markets, _ORE_GOODS)
    pairs = []
    for ast in asteroids:
        mkt = min(pool, key=lambda w: _d2(w, ast))
        pairs.append((ast["symbol"], mkt["symbol"], _d2(ast, mkt)))
    pairs.sort(key=lambda p: p[2])                       # shortest round trip / most central first
    return [(a, m) for a, m, _ in pairs[:4]]


async def _siphon_targets(system: str) -> list:
    """Gas-giant (siphon, sell-market) pairs in ``system``, ranked by shortest round trip —
    same shape as ``_mining_targets`` but for GAS_GIANT waypoints (and gas goods for the
    buys-the-yield check). Empty ⇒ no gas giant/market."""
    try:
        giants = await C.call("GET", f"/systems/{system}/waypoints",
                              params={"type": "GAS_GIANT", "limit": 20})
        markets = await C.call("GET", f"/systems/{system}/waypoints",
                               params={"traits": "MARKETPLACE", "limit": 20})
    except C.SpaceTradersError:
        return []
    if not giants or not markets:
        return []
    pool = await _markets_buying(system, markets, _GAS_GOODS)
    pairs = []
    for gg in giants:
        mkt = min(pool, key=lambda w: _d2(w, gg))
        pairs.append((gg["symbol"], mkt["symbol"], _d2(gg, mkt)))
    pairs.sort(key=lambda p: p[2])
    return [(g, m) for g, m, _ in pairs[:4]]


# ── persistent-plan jobs — hold positions across the window and across windows ──────────────
_STATION_POLL = 60.0   # seconds between a stationed probe's market re-reads


async def _station_loop(sym: str, deadline: float, waypoint: str, *, log=None) -> str:
    """Hold a probe ON STATION at a route endpoint, re-polling its market so the price stays
    LIVE for the route ranker — the fix for the stale/blind-sink problem (a route's two ends
    are rarely both lit by roving probes). Free for a probe (no fuel). It does NOT roam, so the
    endpoint never goes dark mid-window."""
    if not await travel_to(sym, waypoint, log=log):
        return f"could not reach station {waypoint}"
    from . import prices
    polls, errors = 0, 0
    while _now() < deadline:
        try:
            await C.call("POST", f"/my/ships/{sym}/dock")
            system = T._system_of(waypoint)
            m = await C.call("GET", f"/systems/{system}/waypoints/{waypoint}/market")
            if m.get("tradeGoods"):
                prices.record_market(system, waypoint, m["tradeGoods"])
                polls += 1
            errors = 0
        except C.SpaceTradersError as e:
            errors += 1
            if log:
                log(f"{sym}: station read failed at {waypoint} ({e}) [{errors}/3]")
            if errors >= 3:   # waypoint invalid / persistent failure — don't hold a dead post all window
                return f"abandoned station {waypoint} after {errors} consecutive errors ({polls} refreshes)"
        await asyncio.sleep(_STATION_POLL)
    return f"stationed at {waypoint} ({polls} price refreshes)"


async def _trade_route_loop(sym: str, deadline: float, route: dict, *, log=None) -> str:
    """Trade a FIXED route (from the persistent plan) round trip after round trip until the
    window ends — no per-round re-ranking, so the hauler STAYS on its route (and, via the plan,
    across windows) instead of hopping. job_trade still re-checks the live spread each trip and
    bails if it closed; that miss is what strikes the route at the next reconcile."""
    runs = 0
    while _now() < deadline:
        res = await job_trade(sym, route["good"], route["buy_at"], route["sell_at"],
                              expected_sell=route.get("sell_price"),
                              sink_vol=route.get("sink_volume"), log=log)
        if log:
            log(f"{sym}: {res}")
        # Key the bail on the SUCCESS marker (job_trade returns _TRADE_OK only on a completed
        # round trip) rather than matching a failure string — any non-trade outcome (skipped /
        # couldn't reach / holding cash) stops the loop, and the miss strikes the route at the
        # next reconcile.
        if not res.startswith(_TRADE_OK):
            return f"completed {runs} fixed-route trade run(s); stopped: {res}"
        runs += 1
    return f"completed {runs} fixed-route trade run(s)"


async def _market_scan(system: str) -> list:
    """Every MARKETPLACE waypoint symbol in the system, coordinates cached into
    ``_WP_COORDS`` on the way. PAGINATED via tools._all_waypoints — the API caps a page
    at 20 and the documented live system holds more (the sell guard once 'proved' no
    buyer existed because the only importer sat on page 2; _good_markets was hardened
    for exactly this). A single limit-20 GET here silently truncates BOTH surfaces this
    scan feeds: ``market_wps`` (scout coverage + the per-window price refresh) and
    ``_WP_COORDS`` (the time-aware lane score, where a missing endpoint demotes a route
    to the pessimistic unmapped-distance score — so every page-2 market's routes would
    rank as unknown-distance for the process lifetime, while the sqlite price map keeps
    recording them). Errors propagate: a half scan must read as failed, not shorter."""
    wps = await T._all_waypoints(system, {"traits": "MARKETPLACE"})
    _WP_COORDS.update({w["symbol"]: (w["x"], w["y"]) for w in wps
                       if w.get("x") is not None and w.get("y") is not None})
    return [w["symbol"] for w in wps]


async def autopilot(minutes: float, *, log=None) -> dict:
    """Run the fleet toward max credits for ``minutes`` — the zero-to-million engine.
    Contracts are per-AGENT (one active), so ONE trader works contracts (the capital
    base) while the OTHERS run profitable trade routes (the scaling lever, each
    independent + spread-guarded); ships with a mining laser run the extract→sell loop
    at the nearest asteroid; probes scout markets for price intel; profit is reinvested
    into haulers when capital allows. One shared rate budget. Returns a credits/hr
    summary + per-ship results."""
    start_credits = await _credits()
    deadline = _now() + minutes * 60.0
    ships = await C.call("GET", "/my/ships")
    claimed: set = set()
    lock = asyncio.Lock()

    system = ships[0]["nav"]["systemSymbol"] if ships else None
    market_wps = []
    if system:
        # Paginated marketplace scan; carries every marketplace's coordinates so the
        # route ranking is time-aware (net-per-second) and the ride-along picker can
        # prove a sink is a trivial hop from a delivery stop. Static within a reset.
        # See _market_scan for why one limit-20 GET is not enough here.
        market_wps = await _market_scan(system)

    # Growth engine (zero-to-million): reinvest profit first, then assign by role
    # (roles.assign_roles — classified by CAPABILITY, not just "has a hold"):
    #   • probes SCOUT markets — free, and it builds the price map trade needs;
    #   • MINERS (a mining laser) run the survey→extract→sell loop at the nearest
    #     asteroid — a dedicated mining drone finally mines instead of being mis-cast
    #     as a hauler (its 15-unit hold used to sweep it into the trade rotation);
    #   • ONE trader works CONTRACTS (the capital base — capped at 1/agent); an
    #     unworkable window falls back to trade, and the plan DEMOTES a lead that
    #     keeps striking out (cooldown) so it never idles on a dead contract slot;
    #   • every OTHER trader runs the best profitable TRADE route (the scaling lever
    #     — independent + spread-guarded; idles, never loses, if none yet).
    if system:
        await _maybe_reinvest(ships, system, log=log)   # probes to fill the map, then haulers
        ships = await C.call("GET", "/my/ships")   # refresh after a possible buy
    jobs = {}
    # Strategy preset (mining on/off) + per-ship pins (st_assign) drive the split.
    role = R.assign_roles(ships, mining_enabled=KNOBS.get("mining"), overrides=overrides())
    probes, miners, siphoners, traders = (
        role["probes"], role["miners"], role["siphoners"], role["traders"])
    # The PERSISTENT PLAN is the engine: reconcile a stable assignment so positions STICK —
    # haulers keep their route across windows, probes hold the active route's endpoints live.
    # (This replaced the old stateless re-plan-every-window dispatch, which let positions
    # evaporate each window — the cause of the multi-hour flatline.)
    from . import plan as _plan
    _pa: dict = {}
    ranked: list = []
    if system and market_wps:
        ranked = await _ranked_routes(system, market_wps)
        pl = _plan.reconcile(_plan.load(), role, ranked, route_strikes=KNOBS.get("route_strikes"),
                             contract_strikeouts=KNOBS.get("contract_strikeouts"),
                             contract_cooldown=KNOBS.get("contract_cooldown"),
                             route_rotate_cycles=KNOBS.get("route_rotate_cycles"))
        _plan.save(pl)
        _pa = pl["ships"]
    # Probes: STATION the active route's endpoints (stable plan), the rest round-robin SCOUT the
    # non-stationed markets so they cover ground / fill the price map fast.
    station_wps = {a["station"] for a in _pa.values() if a.get("role") == "station" and a.get("station")}
    scout_targets = [w for w in market_wps if w not in station_wps] or market_wps
    scout_probes = [p for p in probes if _pa.get(p["symbol"], {}).get("role") != "station"]
    for p in probes:
        if _pa.get(p["symbol"], {}).get("role") == "station":
            jobs[p["symbol"]] = _station_loop(p["symbol"], deadline, _pa[p["symbol"]]["station"], log=log)
    for i, p in enumerate(scout_probes):
        if scout_targets:
            share = scout_targets[i::len(scout_probes)] or scout_targets
            jobs[p["symbol"]] = job_scout(p["symbol"], share, deadline, log=log)
    # Miners dig at a CENTRAL asteroid and sell the ore — spread round-robin across the ranked
    # (asteroid, market) pairs so several miners don't pile on one rock + one buyer (saturation)
    # and so each gets a reachable target. If the system has no asteroid/market, fall back to
    # trade so the ship still earns — never stranded.
    if miners:
        targets = await _mining_targets(system) if system else []
        for i, s in enumerate(miners):
            if targets:
                ast, mkt = targets[i % len(targets)]
                jobs[s["symbol"]] = _mining_loop(s["symbol"], deadline, ast, mkt, log=log)
            elif system and market_wps:
                jobs[s["symbol"]] = _trade_loop(s["symbol"], deadline, system, market_wps, log=log)
    # Siphoners (pin-only, st_assign siphon) work gas giants — same spread/reachability model.
    if siphoners:
        gtargets = await _siphon_targets(system) if system else []
        for i, s in enumerate(siphoners):
            if gtargets:
                gg, mkt = gtargets[i % len(gtargets)]
                jobs[s["symbol"]] = _siphon_loop(s["symbol"], deadline, gg, mkt, log=log)
            elif system and market_wps:
                jobs[s["symbol"]] = _trade_loop(s["symbol"], deadline, system, market_wps, log=log)
    if traders:
        # Gate-supply rung (v2.5): once the home system's jump gate is under
        # construction, ONE spare hauler (never the contract lead — _gate_ship enforces
        # both caps) hauls construction materials this window instead of its trade
        # route. Dispatch is carrier-first and armed from LIVE credits (_gate_dispatch
        # — the window-open snapshot goes stale the moment reinvest buys). The plan
        # KEEPS the ship's route assignment (position hysteresis intact) — the ship
        # just serves gate duty for the window, exactly like the mining fallback
        # borrows a hull without rewriting the plan.
        gate_ship, gate_wp = await _gate_dispatch(traders, system)
        if gate_ship:
            jobs[gate_ship] = job_gate_supply(gate_ship, gate_wp, deadline, log=log)
        lead = traders[0]["symbol"]
        la = _pa.get(lead) or {}
        if la.get("role") == "trade":
            # reconcile DEMOTED a striking-out lead (contract cooldown) — it trades this
            # window on its plan route, exactly like any other hauler.
            if la.get("route"):
                jobs[lead] = _trade_route_loop(lead, deadline, la["route"], log=log)
            elif system and market_wps:
                jobs[lead] = _trade_loop(lead, deadline, system, market_wps, log=log)
        else:
            # The unworkable-contract fallback trades the FIRST ranked route no assignment
            # holds (plan.unclaimed_rank) — NOT the count of routed ships: keep/hold
            # hysteresis lets a hauler carry a route that slid down the fresh ranking, so
            # the count can name a rank a hauler already sits on (a saturation pile-up).
            jobs[lead] = _contract_then_trade(lead, deadline, claimed, lock, system, market_wps,
                                              rank=_plan.unclaimed_rank(_pa, ranked), log=log)
        # Every other hauler runs its PLAN route — reconcile already spread them across distinct
        # top-N routes (no saturation pile-up). A hauler the plan left without a fresh route
        # (fewer fresh routes than haulers) grabs the best available via the re-ranking loop.
        for s in traders[1:]:
            if s["symbol"] == gate_ship:   # on gate duty this window — job already set
                continue
            if not (system and market_wps):
                continue
            route = _pa.get(s["symbol"], {}).get("route")
            if route:
                jobs[s["symbol"]] = _trade_route_loop(s["symbol"], deadline, route, log=log)
            else:
                jobs[s["symbol"]] = _trade_loop(s["symbol"], deadline, system, market_wps, log=log)

    results = await run_fleet(jobs, log=log)
    end_credits = await _credits()
    gained = end_credits - start_credits
    rate = round(gained / (minutes / 60.0)) if minutes else 0
    # Window summary on the bus — ship COUNT only (per-ship detail stays in the return;
    # bus payloads are for dashboards/feeds, keep them light).
    E.emit("window_closed", {"minutes": minutes, "credits_start": start_credits,
                             "credits_end": end_credits, "gained": gained,
                             "per_hour": rate, "ships": len(results)})
    return {
        "minutes": minutes,
        "credits_start": start_credits,
        "credits_end": end_credits,
        "gained": gained,
        "per_hour": rate,
        "ships": results,
    }


# ── background ops — the engine runs as a SUPERVISED background task (graph.sdk.supervise) ─
# The autopilot blocks for its whole window; an agent over A2A can't tie up a turn for 20
# minutes. So the engine runs as a self-perpetuating background task with a watchdog — re-kick
# a crash, restart a stall, recover a universe reset — all from the shared Supervisor helper
# (protoAgent #1025): we supply only the window + the predicates. The agent supervises via
# ops_status() between turns. The deterministic watchdog is the heartbeat; the LLM agent is the
# exception handler, not the other way around.
_LOG: list = []      # the running progress log (capped) — for status + the dashboard
_LOGSEQ = 0          # monotonic log counter — the supervisor's "progress" signal for stalls
_ENGINE = None       # the Supervisor singleton (lazy — start() needs a running loop)


def _record(msg: str) -> None:
    global _LOGSEQ
    _LOG.append(msg)
    del _LOG[:-200]
    _LOGSEQ += 1


async def _window() -> dict:
    """One autopilot window — the supervised unit of work. Reads the window length from the knob
    each time, so the strategist can retune the OODA cadence live (st_tune window_minutes)."""
    return await autopilot(KNOBS.get("window_minutes") or 15.0, log=_record)


async def _stalled() -> bool:
    """Confirm a REAL stall: the engine's log is frozen (progress) AND no ship is in transit — a
    long DRIFT legitimately freezes the log, so don't false-trip on it."""
    try:
        ships = await C.call("GET", "/my/ships", params={"limit": 20})
        return not any(s["nav"]["status"] == "IN_TRANSIT" for s in ships)
    except Exception:  # noqa: BLE001 — can't tell → don't false-trip
        return False


# Re-register backoff for riding out a universe wipe. SpaceTraders 503s ("rebuilding") on
# /register for the first minutes after a reset. on_crash gets exactly ONE shot per crash —
# the supervisor won't re-invoke it until the runner runs again (graph/supervisor.py: the
# `recovered` latch), and the runner can't run without a fresh token — so a single failed
# re-register would strand the engine on a dead token. Retry IN PLACE across the rebuild
# window instead. First delay is 0 (a server that's already back recovers with no wait).
# ~9 min total covers an observed wipe (2026-07-05: server back within ~6 min).
_RESET_REREGISTER_BACKOFF = (0.0, 15.0, 30.0, 45.0, 60.0, 90.0, 90.0, 90.0, 90.0)


def _reset_recovery_terminal(status: str) -> bool:
    """A re-register failure that retrying can't fix — the operator must act: the call sign is
    already claimed this cycle, or no account token / call sign is configured. Everything else
    (a 503 while the universe rebuilds, a network blip) is transient and worth another try."""
    s = status.lower()
    return ("already claimed" in s or "no account token" in s
            or "no call sign" in s or "can't auto-recover" in s)


async def _recover(result) -> bool:
    """on_crash: a universe reset (4113) kills the token — re-register the call sign for a fresh
    token, then let the supervisor re-kick. Any other crash → re-kick. The server 503s while it
    rebuilds the universe, so retry the re-register across that window (bounded backoff) — a
    single failed try would strand the engine on a dead token, because the supervisor gives
    on_crash only one shot per crash. Give up (return False → supervisor stops the storm) only
    on a TERMINAL failure (call sign claimed / unconfigured) or after the retry budget."""
    err = (result or {}).get("error") or ""
    if "4113" in err or "reset_date" in err:
        from . import plan as _plan
        # 4113 means the universe HAS already wiped — the plan's waypoints/ship-ids are invalid
        # NOW, so clear it regardless of whether the re-registration below succeeds (a failed
        # recovery would only re-kick into the same wall; a later window rebuilds the plan fresh).
        # The learned-route memory (routes.py) survives the wipe; the per-reset plan does not.
        _plan.clear()
        status = ""
        n = len(_RESET_REREGISTER_BACKOFF)
        for i, delay in enumerate(_RESET_REREGISTER_BACKOFF):
            if delay:
                await asyncio.sleep(delay)
            try:
                status = await C.recover_from_reset()
            except Exception as e:  # noqa: BLE001
                status = f"reset recovery errored: {e}"
            if status.startswith("re-registered"):
                _record(f"universe reset — {status}")
                E.emit("reset_recovered", {"status": status})
                return True
            if _reset_recovery_terminal(status):
                _record(f"universe reset — UNRECOVERABLE: {status}")
                return False
            _record(f"universe reset — server still rebuilding (try {i + 1}/{n}), retrying: {status}")
        _record(f"universe reset — re-register still failing after {n} tries; stopping ({status})")
        return False
    return True


def _engine():
    """The Supervisor singleton (lazy: start() creates tasks, so build it in the async context
    of the first start)."""
    global _ENGINE
    if _ENGINE is None:
        from graph.sdk import supervise
        _ENGINE = supervise(_window, name="fleet", interval=90, breath=3.0, stall_ticks=3,
                            progress=lambda: _LOGSEQ, stall_check=_stalled, on_crash=_recover)
    return _ENGINE


# ── autopilot intent (v2.9) — survives module reload / app restart so the fleet-engine
# surface can RESUME a running autopilot the reload interrupted, but NEVER auto-start one
# the operator deliberately stopped. Persisted in the watch-state file (the proven
# per-agent store); epoch-scoped, so a universe wipe drops it and st-reset's recovery
# playbook is what re-starts the engine on the fresh epoch (not a blind resume before a
# fleet exists). The intent tracks OPERATOR/agent decisions only — a lifecycle stop
# (reload/shutdown) leaves it untouched, which is the whole point.
def _set_autopilot_intent(wanted: bool) -> None:
    # The failure modes are asymmetric: a lost START-write (intent stays absent → no resume)
    # is safe, but a lost STOP-write leaves the flag True and a later reload would RESURRECT
    # a deliberately-stopped engine — the one thing this feature must never do. save_state
    # RETURNS whether it landed (it never raises; the swallow-and-WARN is deliberate), so we
    # gate on BOTH the write-success bool AND a read-back, retry once, and on a persistent
    # failure to CLEAR delete the key (absence reads as not-wanted → safe). If even the
    # delete can't persist — a totally dead state layer (ENOSPC / read-only mount) — the flag
    # is genuinely stuck and a reload MAY resume a stopped engine: a rare, recoverable
    # annoyance (the operator re-stops it). You can't clear a flag on a dead disk; the best
    # we can do is make that LOUD, not silent (the WARN below is now reachable — the old
    # code hung it off an exception save_state never throws, so it never fired).
    from . import watches as _W
    for _ in range(2):
        st = _W.load_state()
        st["autopilot_wanted"] = bool(wanted)
        if _W.save_state(st) and bool(_W.load_state().get("autopilot_wanted")) == bool(wanted):
            return
    if not wanted:
        st = _W.load_state()
        st.pop("autopilot_wanted", None)
        if not _W.save_state(st) or _W.load_state().get("autopilot_wanted"):
            _LOG.append("WARN: could not clear autopilot intent (dead state layer) — a reload "
                        "may resume the engine; stop it again after any restart")


def autopilot_intended() -> bool:
    """True iff the operator/agent last START-ed the autopilot (and hasn't stopped it) —
    the flag the surface consults to decide whether a reload/restart should RESUME it."""
    try:
        from . import watches as _W
        return bool(_W.load_state().get("autopilot_wanted"))
    except Exception:  # noqa: BLE001
        return False


def request_stop() -> None:
    """Wind the engine down gracefully after the current window (no re-kick) — called by the
    plugin's goal on_achieved hook when the operator's target is reached."""
    _set_autopilot_intent(False)   # a goal-achieved wind-down is a DELIBERATE stop — don't resume
    _engine().request_stop()
    E.emit("engine_stopped", {"reason": "goal achieved — winding down after this window"})


def start_ops(minutes: float) -> str:
    set_knob("window_minutes", minutes)   # seed the knob so st_report/st_tune agree
    _LOG.clear()
    # Record intent BEFORE the idempotency guard: calling st_autopilot_start on an
    # ALREADY-running engine must still (re)assert "operator wants it running" — otherwise
    # the intent can never be repaired while the engine is up (e.g. after an epoch recovery
    # that kept it running), and the next restart wouldn't resume.
    _set_autopilot_intent(True)
    if "already running" in _engine().start():
        return "Fleet ops already running — check status before starting another."
    E.emit("engine_started", {"window_minutes": minutes})
    E.navigate("fleet")   # flip the console to the Fleet dashboard (ADR 0044)
    from . import watches
    watches.arm_all()     # tripwires up whenever the engine is (idempotent replace)
    return (f"Fleet ops started in the background (~{minutes:g} min windows, watchdog keeping it "
            f"alive). The whole fleet works under one rate budget. Check st_autopilot_status.")


def stop_ops() -> str:
    _set_autopilot_intent(False)   # a DELIBERATE operator/agent stop — the surface won't resume it
    if _ENGINE is None or not _engine().running():
        return "No fleet ops running."
    _engine().stop()
    E.emit("engine_stopped", {"reason": "stopped by operator/agent"})
    return "Stopping fleet ops."


def lifecycle_stop() -> None:
    """Stop the engine task for a server shutdown / plugin reload WITHOUT touching the
    operator's intent — so the fleet-engine surface can resume it on the next start if it
    was running. Distinct from stop_ops() (a deliberate operator stop, which clears intent):
    a reload must not read as 'the operator turned it off'. Best-effort; never raises."""
    try:
        if _ENGINE is not None and _engine().running():
            _engine().stop()
    except Exception:  # noqa: BLE001 — shutdown must not raise
        pass


def ops_status() -> dict:
    st = _engine().status() if _ENGINE is not None else {
        "running": False, "want_running": False, "watchdog": False, "result": None, "events": []}
    return {
        "running": st["running"],
        "want_running": st["want_running"],
        "watchdog": st["watchdog"],
        "started_minutes": KNOBS.get("window_minutes"),
        "recent_log": _LOG[-6:],
        "watchdog_log": st.get("events", [])[-5:],
        "result": st.get("result"),
    }


def _eta_hours(credits: int, per_hour: float, target: int):
    """Hours to ``target`` credits at the current rate, or None if already there / idle."""
    if per_hour <= 0 or credits >= target:
        return None
    return round((target - credits) / per_hour, 1)


async def report() -> dict:
    """Rich fleet telemetry for the OODA strategist's OBSERVE step — credits/hr trajectory,
    per-ship role + health, engine state, the live knobs + strategy + pins, and a few
    deterministic HINTS (stranded ships, idle capital, dead routes) to seed ORIENT. Costs
    one /my/agent + one /my/ships call; everything else is local engine state."""
    try:
        agent = await C.call("GET", "/my/agent")
        ships = await C.call("GET", "/my/ships", params={"limit": 20})
    except C.SpaceTradersError as e:
        return {"ok": False, "error": str(e)}
    credits = agent.get("credits", 0)
    ops = ops_status()
    last = ops.get("result") or {}
    per_hour = last.get("per_hour", 0) or 0

    # Label each ship with the role the engine WOULD assign it this window (same call the
    # autopilot makes), so the report and the engine never disagree.
    role = R.assign_roles(ships, mining_enabled=KNOBS.get("mining"), overrides=overrides())
    label = {}
    for p in role["probes"]:
        label[p["symbol"]] = "scout"
    for m in role["miners"]:
        label[m["symbol"]] = "miner"
    for x in role["siphoners"]:
        label[x["symbol"]] = "siphon"
    for i, t in enumerate(role["traders"]):
        label[t["symbol"]] = "contract" if i == 0 else "trader"
    rows, hints = [], []
    # The plan can DEMOTE a striking-out contract lead to trade for a cooldown — mirror it
    # (or the report would show "contract" for a ship the engine dispatched to trade) and
    # SAY so: the live pinned-idle lead went unnoticed for hours because no telemetry
    # surfaced the contract slot's state, and demotion is a policy event ORIENT must see.
    # Only a REAL demotion, though: a demoted lead always carries contract_cooldown >= 1,
    # while a plain hauler's entry is role "trade" with no contract keys — without the
    # cooldown guard, re-pinning ships mid-window (the incident's manual workflow) would
    # make the report fabricate a demotion for whichever hauler just became traders[0].
    from . import plan as _plan
    for sym, a in (_plan.load().get("ships") or {}).items():
        if (label.get(sym) == "contract" and a.get("role") == "trade"
                and a.get("contract_cooldown", 0) > 0):
            label[sym] = "trader"
            hints.append(f"contract lead {sym} demoted after strikeouts — trading "
                         f"{a.get('contract_cooldown', 0)} more window(s) before re-pinning contract")
    for s in ships:
        sym, nav, cargo = s["symbol"], s["nav"], s["cargo"]
        units, cap = cargo.get("units", 0), cargo.get("capacity", 0)
        in_transit = nav["status"] == "IN_TRANSIT"
        r = label.get(sym) or ("idle" if sym in overrides() else "—")
        stranded = (not in_transit) and cap > 0 and units >= cap   # parked with a full hold
        rows.append({"symbol": sym, "role": r, "pinned": overrides().get(sym),
                     "status": nav["status"], "at": nav["waypointSymbol"],
                     "cargo": f"{units}/{cap}", "stranded": stranded})
        if stranded:
            hints.append(f"{sym} parked FULL ({units}/{cap}) at {nav['waypointSymbol']} — may be stranded")

    if ops["running"] and last and per_hour <= 0:
        hints.append("engine not earning this window — check routes/saturation or lower min_margin")
    if credits >= KNOBS.get("buy_buffer") and len(ships) < KNOBS.get("max_ships") and ops["running"]:
        hints.append(f"{credits:,} cr idle with room to grow — reinvest (buy_buffer {KNOBS.get('buy_buffer'):,})")
    if not _ROUTE_CACHE.get("routes") and any(v in ("trader", "contract") for v in label.values()):
        hints.append("no profitable route mapped — scout more markets or lower min_margin")

    # API-error tally (st-api-errors' ground truth) — surfaced HERE because the trip
    # prompt sends the agent to st_report; below the trip floor a repeating shape is
    # still worth a look (3+ same-shape failures isn't ambient flake anymore).
    err_stats = C.api_error_stats()
    if err_stats and max(err_stats.values()) >= 3:
        from . import watches as W
        _, err_detail = W.api_errors_met(err_stats, 10, C.api_error_samples())
        hints.append(f"API errors (30 min): {err_detail}")

    # Build the standard telemetry envelope (status / metrics / hints / decisions / sections),
    # on the shared SDK helper (protoAgent #1027), carrying the existing keys as extras so the
    # st_report tool + dashboard consume it unchanged.
    from graph.sdk import telemetry
    eta_1m = _eta_hours(credits, per_hour, 1_000_000)
    status = (f"{'running' if ops['running'] else 'stopped'} · {credits:,} cr · {per_hour:,} cr/hr"
              + (f" · ~{eta_1m}h to 1M" if eta_1m else ""))
    fleet_section = {
        "title": "Fleet",
        "columns": ["ship", "role", "status", "at", "cargo"],
        "rows": [[r["symbol"], r["role"] + (" ⚠" if r["stranded"] else ""),
                  r["status"], r["at"], r["cargo"]] for r in rows],
    }
    return telemetry(
        status=status,
        metrics={"credits": credits, "ships": len(ships), "cr/hr": per_hour},
        hints=hints,
        decisions=decisions()[-6:],
        sections=[fleet_section],
        # extras (consumed by the st_report tool / any caller):
        ok=True, credits=credits, ships=len(ships), strategy=current_strategy()["name"],
        mining=KNOBS.get("mining"),
        engine={"running": ops["running"], "window_min": KNOBS.get("window_minutes"),
                "last_gained": last.get("gained"), "per_hour": per_hour,
                "recent_log": ops["recent_log"][-4:]},
        projection={"to_250k": _eta_hours(credits, per_hour, 250_000), "to_1m": eta_1m},
        knobs=knobs(), overrides=overrides(), fleet=rows,
    )