"""Mesh model + runtime configuration.

Layers, weakest to strongest:

1. the network model — ``netmodel.json`` (the canonical v1 constants and duty
   catalog) with the host's overlay merged over it, see :mod:`.host`;
2. ``SZPONTNET_*`` environment overrides for the protocol knobs — how the
   tests run whole meshes on loopback with fast timeouts without touching the
   model;
3. gossiped last-writer-wins *mesh-wide overrides* (per-duty strategy /
   token-awareness / platform spread, plus the preferred WAN transport, edited
   live from a topology UI) — see :class:`PlacementOverrides`.
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field, replace

from . import env
from .host import model


def _has_non_finite(v: object) -> bool:
    """Whether ``v`` contains a non-finite float (∞/NaN) anywhere, recursing through
    dicts and lists. A gossiped placement override's duty dict is kept VERBATIM (see
    :meth:`PlacementOverrides.from_dict`) and re-serialized into the shared snapshot, so
    a signed peer that slips a non-finite float into any key would poison ``state.json``:
    ``json.dumps`` (allow_nan=True default) writes it as the bare RFC 8259-invalid token
    ``Infinity``/``NaN`` that a strict reader rejects WHOLESALE. Mirrors the advert-side
    guard in ``protocol.NodeInfo.from_dict``."""
    if isinstance(v, float):
        return not math.isfinite(v)
    if isinstance(v, dict):
        return any(_has_non_finite(x) for x in v.values())
    if isinstance(v, list):
        return any(_has_non_finite(x) for x in v)
    return False


# Env override suffixes (see :mod:`.env` for the ``SZPONTNET_`` prefix), mapped onto
# the network model's "protocol" keys. Values are parsed with the type of the default
# they replace.
_ENV_KEYS = {
    "MCAST_GROUP": "multicastGroup",
    "MCAST_PORT": "multicastPort",
    "TCP_BASE": "tcpPortBase",
    "TCP_SPAN": "tcpPortSpan",
    "BEACON_SECS": "beaconIntervalSecs",
    "REDIAL_SECS": "redialIntervalSecs",
    "HEARTBEAT_SECS": "heartbeatIntervalSecs",
    "STALE_SECS": "peerStaleSecs",
    "TIMEOUT_SECS": "peerTimeoutSecs",
    "ACK_SECS": "dispatchAckTimeoutSecs",
    "STATE_SECS": "stateWriteIntervalSecs",
    "RESULT_RETRY_SECS": "foreignResultRetryIntervalSecs",
    "RESULT_MAX_SECS": "foreignResultMaxSecs",
    "FOREIGN_TIMEOUT_SECS": "foreignJobTimeoutSecs",
    "COMPLETION_DEADLINE_SECS": "foreignCompletionDeadlineSecs",
    "REMINDER_GRACE_SECS": "foreignReminderGraceSecs",
}


def protocol() -> dict:
    """The protocol constants with any SZPONTNET_* env overrides applied."""
    out = dict(model()["protocol"])
    for suffix, key in _ENV_KEYS.items():
        raw = env.get(suffix)
        if raw is None:
            continue
        default = out.get(key)
        try:
            if isinstance(default, bool):  # not used today; guard against int() eating it
                out[key] = raw == "1"
            elif isinstance(default, int):
                out[key] = int(raw)
            elif isinstance(default, float):
                out[key] = float(raw)
            else:
                out[key] = raw
        except ValueError:
            pass  # a malformed override falls back to the model default
    return out


def loopback_only() -> bool:
    """SZPONTNET_LOOPBACK=1 keeps every socket on 127.0.0.1 — used by the
    integration tests (and demos) to run a whole mesh on one machine without
    touching the real LAN."""
    return env.get("LOOPBACK") == "1"


def secret() -> str:
    """Optional pre-shared join token (SZPONTNET_SECRET, same value on every
    machine + in the CLI/panel environment). A node with a secret refuses peer
    links, control sessions, and therefore dispatches that don't present it.

    This is a fence, not cryptography — the token rides plaintext on the LAN.
    It keeps a stray machine (or a colleague's mesh on the same office network)
    from joining yours and receiving jobs; it does not defend against a hostile
    network. Empty (the default) = open mesh, fine for a home LAN.
    """
    return env.get("SECRET", "")


# Spellings that turn OFF a switch which is on by default. Deliberately lenient:
# for an opt-in knob a misspelt value fails safe (the feature stays off), but for a
# default-ON one it fails the wrong way — an operator who wrote
# ``SZPONTNET_TOR=false`` and got a tor process anyway has no way to read that as
# anything but a bug. The empty string counts: ``SZPONTNET_TOR=`` is how a unit file
# or a wrapper script clears a variable it cannot unset.
_OFF_VALUES = frozenset({"0", "false", "no", "off", ""})

# The mirror of :data:`_OFF_VALUES`, for a switch that is OFF by default. An opt-in
# knob must recognise only these: anything else — a typo, a stray quote, the word
# "enabled" — leaves the feature off, which is the direction that fails safe.
_ON_VALUES = frozenset({"1", "true", "yes", "on"})


def iroh_enabled() -> bool:
    """Whether this node runs the iroh QUIC transport — off by default, on only for
    an explicit ``SZPONTNET_IROH`` in :data:`_ON_VALUES`.

    The transport is complementary to the LAN, not an alternative to it: the node
    binds a permanent endpoint (an Ed25519 public key it advertises inside its signed
    advert) and dials known-but-unseen personal peers by that key with exponential
    backoff, which is what joins several LANs into one wide-area mesh. Multicast
    discovery and direct TCP links are untouched, and remain the path between peers
    that share a network.

    It reaches the same peers as the Tor transport (:func:`tor_enabled`) without a
    daemon, a multi-minute bootstrap, or a rendezvous circuit per dial, and a node
    running both prefers it. A node may run either, both, or neither.

    On is an *intent*, not a promise: a machine without the optional ``iroh`` package,
    or one whose endpoint never comes online, simply does not get this transport — it
    degrades, it never stops a node from starting. See irohnet.py."""
    return (env.get("IROH", "") or "").strip().lower() in _ON_VALUES


def iroh_online_timeout() -> float:
    """How long to wait for the iroh endpoint to come online before giving up and
    staying LAN-only (SZPONTNET_IROH_ONLINE_SECS). The node stays fully usable on the
    LAN in the meantime."""
    try:
        v = float(env.get("IROH_ONLINE_SECS", "30"))
    except ValueError:
        return 30.0
    # Reject non-finite / non-positive: a non-finite timeout makes the wait block
    # FOREVER — the opposite of "give up and stay LAN-only" — and a non-positive one
    # is meaningless.
    return v if math.isfinite(v) and v > 0 else 30.0


def tor_enabled() -> bool:
    """Whether this node runs the Tor onion-service transport — **on by default**,
    off only for an explicit ``SZPONTNET_TOR`` in :data:`_OFF_VALUES`.

    The transport is complementary to the LAN, not an alternative to it: the node
    runs a persistent onion service (a permanent ``.onion`` it advertises inside its
    signed advert) and dials known-but-unseen personal peers over Tor with
    exponential backoff, which is what joins several LANs into one wide-area mesh.
    Multicast discovery and direct TCP links are untouched, and remain the path
    between peers that share a network.

    On is an *intent*, not a promise: a machine with no ``tor`` binary, or one whose
    tor fails to bootstrap, simply does not get this transport — it degrades, it
    never stops a node from starting. See tor.py."""
    return (env.get("TOR", "1") or "").strip().lower() not in _OFF_VALUES


def wan_transports() -> tuple[str, ...]:
    """Every WAN transport this implementation can run, in the model's default
    preference order (``wan.transports``). The vocabulary the mesh's preferred
    transport is picked from, and the tie-break order behind it."""
    raw = model().get("wan", {}).get("transports", [])
    names = tuple(str(t) for t in raw if isinstance(t, str) and t)
    return names or ("iroh", "tor")


def wan_preferred(overrides: "PlacementOverrides | None" = None) -> str:
    """The mesh's preferred WAN transport: the gossiped choice
    (:attr:`PlacementOverrides.wan`) when one has been made, else the model's first
    entry. Never returns a transport this implementation doesn't know — an override
    naming one is ignored, which is what keeps a newer mesh's choice from silencing
    an older node's WAN entirely."""
    names = wan_transports()
    picked = overrides.wan if overrides else ""
    return picked if picked in names else names[0]


def wan_order(overrides: "PlacementOverrides | None" = None) -> tuple[str, ...]:
    """Transport names in the order a WAN dial should try them: the mesh's preferred
    one first, the rest in model order.

    Preference *orders*, it never excludes: a peer that runs only the less-preferred
    transport is still dialed over that one, so an edge uses the most-preferred
    transport BOTH ends actually run."""
    pref = wan_preferred(overrides)
    return (pref, *(t for t in wan_transports() if t != pref))


def tor_bootstrap_timeout() -> float:
    """How long to wait for Tor to bootstrap before giving up and staying LAN-only
    (SZPONTNET_TOR_BOOTSTRAP_SECS). Tor's first bootstrap can be slow; the node
    stays fully usable on the LAN in the meantime."""
    try:
        v = float(env.get("TOR_BOOTSTRAP_SECS", "90"))
    except ValueError:
        return 90.0
    # Reject non-finite / non-positive (e.g. "inf" from 1e999, "nan", "-1", "0"): a
    # non-finite timeout makes asyncio.wait block FOREVER — the opposite of the
    # docstring's "give up and stay LAN-only" — and a non-positive one is meaningless.
    return v if math.isfinite(v) and v > 0 else 90.0


def server_mode() -> bool:
    """SZPONTNET_SERVER=1 makes this node a dedicated **server**: it accepts and
    runs requests but NEVER originates a dispatch to peers. A request it is asked
    to route (via a control session or the CLI) runs on **itself** instead of
    being fanned out. Combined with :func:`api_key` this is the spec's
    accept-only, optionally API-key-authenticated server role — a beefy shared
    box that takes work but never pushes work onto anyone else."""
    return env.get("SERVER") == "1"


def api_key() -> str:
    """Optional per-node API key (SZPONTNET_API_KEY). When set, this node
    requires a matching ``apiKey`` field on inbound control sessions and inbound
    dispatch requests, refusing any that lack it. It is **independent of** the
    mesh-wide join :func:`secret`: the secret fences who may *join* the mesh; the
    API key authenticates who may submit *work* to this (typically server) node,
    without granting mesh membership or device trust. Empty = no API-key gate."""
    return env.get("API_KEY", "")


def foreign_spawn() -> str:
    """Optional **confinement runner** (SZPONTNET_FOREIGN_SPAWN) — the command
    template a node uses to run a *foreign* SzpontRequest under zero trust. Its
    presence is what turns a foreign request from **declined** into **confined,
    response-only execution**: the untrusted ``prompt`` runs inside the operator's
    own sandbox (a container/VM/jailed process — *the node's own responsibility to
    isolate*), which the template names, with ``{prompt_file}`` and ``{result_file}``
    substituted. The node then returns the sandbox's ``{result_file}`` to the
    originator as a ``job-result``; the originator performs any social action itself.

    Empty (the default) = **no foreign execution**: a foreign request is declined,
    exactly as before. A node only ever runs a stranger's compute when the operator
    has explicitly supplied the jail to run it in. See
    szpontnet-spec/docs/13-foreign-execution.md."""
    return env.get("FOREIGN_SPAWN", "")


def on_result() -> str:
    """Optional **result handler** (SZPONTNET_ON_RESULT) — the command template an
    originator runs when a foreign executor returns a ``job-result`` for a request it
    dispatched. This is where the **social action runs under the originator's own
    identity** (e.g. ``gh pr review``): ``{result_file}`` holds the executor's
    computed artifact plus the job metadata. Empty = the originator just records the
    result (no automatic action). See szpontnet-spec/docs/13-foreign-execution.md."""
    return env.get("ON_RESULT", "")


def extend_decider() -> str:
    """Optional **extension decider** (SZPONTNET_EXTEND_DECIDER) — the command
    template an originator runs to judge a late foreign executor's ``job-progress``
    plea: whether "still working" is a valid reason to extend its completion
    deadline. ``{job_file}`` is substituted with a JSON file carrying the case (the
    job, who the executor is, when it accepted, extensions so far, and the plea);
    exit status 0 extends (re-arms the full deadline window), anything else — a
    non-zero exit, a crash, or a timeout — bans. The operator typically points this
    at an agent, the same pattern as SZPONTNET_FOREIGN_SPAWN / ON_RESULT.

    Empty (the default) = **no extension is ever granted**: a progress plea then
    cannot save a late executor, and it is banned — the zero-trust default. See
    szpontnet-spec/docs/13-foreign-execution.md#the-extension-decision."""
    return env.get("EXTEND_DECIDER", "")


def default_trust() -> str:
    """The trust level a node applies to an **unknown** device — one whose proven
    fingerprint is not in the local allowlist (and to any unverified/keyless peer,
    which can never match the allowlist). ``SZPONTNET_DEFAULT_TRUST`` overrides the
    shipped baseline in the network model (``trust.default``).

    Ships as **foreign** (zero-trust by default): a device you have not explicitly
    marked *personal* is untrusted — its requests are declined (or, with a
    confinement runner, run confined and response-only), it cannot mutate this node
    via ``set-attr``, and it can never own work. The operator promotes the devices
    they trust one at a time (the allowlist is that set of exceptions). Set this to
    ``personal`` to restore the pre-trust **full-altruism** mode — every unlisted
    peer is trusted, exactly as a fresh mesh behaved before the default became
    configurable. An unrecognised value falls back to ``foreign`` (the safe default).
    This is the node-wide baseline; the operator's live choice is persisted in
    ``trusted.json`` (:func:`trust.load_default_level`) and edited from the panel."""
    raw = env.get("DEFAULT_TRUST", "").strip().lower()
    if raw in ("personal", "foreign"):
        return raw
    baseline = str(model().get("trust", {}).get("default", "foreign")).strip().lower()
    return baseline if baseline in ("personal", "foreign") else "foreign"


def accounts() -> dict:
    """The subscription-plan + accounting knobs (plan weights, capacity, quota
    window, usage time-constant) behind per-node load balancing."""
    return model().get("accounts", {})


def plan_weight(plan_id: str) -> float:
    """Quota capacity of a plan relative to Pro (Max 5× → 5, Max 20× → 20).
    An unknown plan weighs 1.0 (Pro-equivalent) — safe, never an error."""
    for p in accounts().get("plans", []):
        if p.get("id") == plan_id:
            try:
                return float(p.get("weight", 1.0))
            except (TypeError, ValueError, OverflowError):
                return 1.0
    return 1.0


def job_cost_units() -> float:
    """How much quota one spawned SzpontRequest books, in capacity units."""
    try:
        return float(accounts().get("jobCostUnits", 1.0))
    except (TypeError, ValueError, OverflowError):
        return 1.0


def dispatch_strategy() -> str:
    """The ranking a dispatcher uses to pick a target — the load-balancing
    decision, made unilaterally from its own view (no consensus). Defaults to
    surplus-first so requests flow to whoever has the most spare quota."""
    return str(model().get("dispatchStrategy", "surplus-first"))


def duty_ids() -> list[str]:
    return [d["id"] for d in model()["duties"]]


def duty_by_id(duty_id: str) -> dict | None:
    return next((d for d in model()["duties"] if d["id"] == duty_id), None)


def tier_bounds() -> tuple[int, int, int]:
    """(min, max, default) machine tier from the shared model."""
    t = model()["tiers"]
    return t["min"], t["max"], t["default"]


def tier_label(tier: int) -> str:
    """Human word for a strength tier ('Very strong' … 'Very light'), from the
    shared model's ``tiers.labels``; falls back to ``tier N`` if unlabelled."""
    labels = model()["tiers"].get("labels", {})
    return labels.get(str(tier), f"tier {tier}")


def tokens_per_weight() -> float:
    """Heuristic per-window token ceiling for a weight-1 (Pro) plan. Scaled by a
    plan's weight to get its ceiling (see usage.py)."""
    try:
        return float(accounts().get("tokensPerWeight", 2_000_000))
    except (TypeError, ValueError, OverflowError):
        return 2_000_000.0


def usage_window_hours() -> float:
    """Trailing window over which local token consumption is measured."""
    try:
        return float(accounts().get("usageWindowHours", 5.0)) or 5.0
    except (TypeError, ValueError, OverflowError):
        return 5.0


def low_threshold() -> float:
    """Remaining-fraction boundary below which the token state drops to 'low'."""
    try:
        return float(accounts().get("lowThreshold", 0.34))
    except (TypeError, ValueError, OverflowError):
        return 0.34


# MARK: - Placement (per-duty policy) + LWW overrides


@dataclass(frozen=True)
class Placement:
    """The resolved placement policy for one duty."""

    strategy: str
    token_aware: bool
    # [(platform, count)] the duty must cover; empty = any one node.
    spread: tuple[tuple[str, int], ...] = ()

    @classmethod
    def from_dict(cls, d: object) -> "Placement":
        if not isinstance(d, dict):
            # A per-duty placement VALUE can arrive over gossip (overrides) as junk
            # (a non-object) and is dereferenced here at assign time via placement_for
            # — tolerate it exactly like _parse_spread tolerates its entries, or one
            # poisoned override crashes every _recompute (a persistent, mesh-wide DoS).
            # A non-mapping resolves to the schema default.
            d = {}
        return cls(
            strategy=d.get("strategy", model()["defaultStrategy"]),
            token_aware=bool(d.get("tokenAware", True)),
            spread=cls._parse_spread(d.get("spread", [])),
        )

    @staticmethod
    def _parse_spread(raw: object) -> tuple[tuple[str, int], ...]:
        """Parse the ``[(platform, count)]`` spread, tolerating malformed entries
        the way the rest of the wire layer tolerates garbage. This resolves on
        every ``_recompute`` from placement dicts that can arrive over gossip
        (``overrides``) or a ctl edit, so an entry that isn't a mapping, names no
        platform, or carries a non-integer count must be SKIPPED / defaulted, not
        allowed to crash assignment. A bad or non-positive count falls back to the
        schema default 1."""
        out: list[tuple[str, int]] = []
        if not isinstance(raw, list):
            return ()
        for s in raw:
            if not isinstance(s, dict):
                continue
            platform = s.get("platform")
            if not isinstance(platform, str) or not platform:
                continue
            try:
                count = int(s.get("count", 1))
            except (TypeError, ValueError, OverflowError):
                count = 1
            if count < 1:
                # A non-positive count is a semantically bad count — a spread entry
                # dispatches to >= 1 node — so fall back to the schema default 1, the
                # same as the parse-exception branch above. Left unnormalized, a valid
                # negative/zero int diverges the two placement consumers that both read
                # this spread: assign_duty counts the duty SATISFIED (its `got == count`
                # loop never trips and `got < count` is false, so it records no
                # shortfall) while slot_candidates yields range(count) == 0 slots and
                # dispatches to NOBODY — a duty that looks placed but silently never runs.
                count = 1
            out.append((platform, count))
        return tuple(out)

    def to_dict(self) -> dict:
        return {
            "strategy": self.strategy,
            "tokenAware": self.token_aware,
            "spread": [{"platform": p, "count": c} for p, c in self.spread],
        }


@dataclass(frozen=True)
class PlacementOverrides:
    """Mesh-wide settings edits, gossiped last-writer-wins.

    ``rev`` is a lamport-ish counter: every edit bumps it past the highest rev
    seen anywhere, so concurrent edits converge on the same winner everywhere
    (ties broken by ``updated_by``). ``duties`` maps duty id → placement dict
    (the full policy, not a diff), and ``wan`` names the mesh's preferred WAN
    transport — one record, so however many settings ride on it they converge
    under one comparison.
    """

    rev: int = 0
    updated_by: str = ""
    duties: dict = field(default_factory=dict)
    # The mesh's preferred WAN transport, or "" while nobody has chosen and every node
    # uses the model default. It ORDERS the transports a node runs rather than
    # selecting one (:func:`wan_order`), and is held as gossiped — resolving a name
    # this build cannot honour is :func:`wan_preferred`'s job, not this field's.
    wan: str = ""
    # Base64 Ed25519 signature by the ``updated_by`` node over this override's
    # canonical form, so a relay can't forge or tamper with a mesh-wide placement
    # edit. Empty for a legacy/keyless editor (then unauthenticated). See node.py.
    sig: str = ""

    @classmethod
    def from_dict(cls, d: dict | None) -> "PlacementOverrides":
        d = d or {}
        raw_duties = d.get("duties")
        return cls(
            rev=cls._as_rev(d.get("rev", 0)),
            updated_by=str(d.get("updatedBy", "")),
            # Keep only mapping duty VALUES, and only ones free of non-finite floats:
            # a non-object value is junk that would crash placement_for at assign time,
            # and a signed peer can slip a bare ∞/NaN into any key of an otherwise-valid
            # duty dict (kept VERBATIM here and re-serialized into the snapshot) to write
            # the RFC 8259-invalid tokens Infinity/NaN into state.json and blank a strict
            # reader's topology mesh-wide — the override-path twin of the advert-side
            # dutiesEnabled/stats guard. Either way the offending duty is dropped at
            # ingestion and falls back to its catalog default.
            duties={k: v for k, v in raw_duties.items()
                    if isinstance(v, dict) and not _has_non_finite(v)}
            if isinstance(raw_duties, dict) else {},
            wan=cls._as_wan(d.get("wan")),
            sig=str(d.get("sig", "")),
        )

    @staticmethod
    def _as_wan(raw: object) -> str:
        """The gossiped WAN preference, tolerating garbage the way the rest of the
        wire layer does: a non-string becomes "" (no choice).

        A *name* is kept whatever it says, including one from a newer model this build
        has no transport for. Dropping it here would be lossy at exactly the wrong
        place: this record is relayed by re-serializing it, so a field removed on the
        way through no longer matches the editor's signature and the hop behind this
        node loses the whole revision. :func:`wan_preferred` is where a name this node
        cannot honour falls back to the local default."""
        return raw if isinstance(raw, str) else ""

    @staticmethod
    def _as_rev(raw: object) -> int:
        """The LWW rev, tolerating garbage the way the rest of the wire layer does
        (cf. :meth:`Placement._parse_spread`): a gossiped override arrives from a
        peer/hello and a non-numeric ``rev`` (null, a list, a string) must default
        to 0, not raise — a malformed edit is simply the default (no) edit."""
        try:
            return int(raw)
        except (TypeError, ValueError, OverflowError):
            # OverflowError: a JSON rev of 1e999 parses to float('inf'), and int(inf)
            # raises it (an ArithmeticError, not a ValueError) — treat as the default 0.
            return 0

    def to_dict(self) -> dict:
        d = {"rev": self.rev, "updatedBy": self.updated_by, "duties": self.duties}
        # Omitted while unchosen, so a mesh that never picks a WAN transport gossips
        # exactly the bytes it did before the field existed.
        if self.wan:
            d["wan"] = self.wan
        if self.sig:
            d["sig"] = self.sig
        return d

    def wins_over(self, other: "PlacementOverrides") -> bool:
        return (self.rev, self.updated_by) > (other.rev, other.updated_by)

    def with_duty(self, duty_id: str, placement: Placement, by: str) -> "PlacementOverrides":
        duties = dict(self.duties)
        duties[duty_id] = placement.to_dict()
        return PlacementOverrides(rev=self.rev + 1, updated_by=by, duties=duties,
                                  wan=self.wan)

    def with_wan(self, transport: str, by: str) -> "PlacementOverrides":
        """This record with the mesh's preferred WAN transport set — a new revision,
        carrying the duty policies forward untouched (the whole record is what wins
        the LWW comparison, so dropping them would reset every duty edit)."""
        return PlacementOverrides(rev=self.rev + 1, updated_by=by,
                                  duties=dict(self.duties), wan=transport)

    def signed(self, sig: str) -> "PlacementOverrides":
        return replace(self, sig=sig)


def placement_for(duty_id: str, overrides: PlacementOverrides | None = None) -> Placement:
    """The effective placement for a duty: the gossiped override if present,
    else the network model's default."""
    if overrides and duty_id in overrides.duties:
        return Placement.from_dict(overrides.duties[duty_id])
    duty = duty_by_id(duty_id)
    return Placement.from_dict(duty["placement"] if duty else {})
