"""core/measure_resolve.py — HOST-NEUTRAL measure resolution (EXIT-5, SHELL.md ladder step 3).

The `_cl_measure_offer` / `_cl_resolve_measures` / `_cl_measure_columns` family, OUT of app.py:
one implementation, two hosts. The Streamlit adapter passes its `st.session_state` memos and its
pending-view read; the API passes runtime-level dicts and no pending view (over HTTP the
`view_upsert` lands via `/grid/events` BEFORE the workspace re-read, so the saved views already
carry the freshest rule — there is no mid-run component value to read).

No streamlit import, no app import — harness only, so both adapters can call it.

Memo discipline (unchanged from the app.py originals):
  * keys carry (stamp, scope, POOL IDENTITY, question) — the pool's identity, never its size,
    because Cohort passes a different pool per cohort and two same-sized cohorts must not be
    handed each other's answers.
  * `column_values` memoises permanent failures as None but NEVER a transient one (datastore
    still warming after a restart) — a memoised transient means blank measure cells that stay
    blank for the whole session long after the data landed.
  * both memos are bounded by clearing wholesale past a cap — session/runtime-lifetime caches,
    not leaks.
"""
import hashlib


def offer(team_id=None, on_error=None):
    """The measures the condition builder / measure columns may use, or [] if the semantic
    store is unavailable. The caller's BU narrows the offer (a company-level measure cannot be
    answered for one business unit)."""
    try:
        from harness import measure_filter as _mf
        return _mf.measure_fields(team_id)
    except Exception as e:                      # no store / model problem -> offer nothing
        if on_error:
            on_error('measures', e)
        return []


def _pool_key(allowed_pids):
    """The pool's identity, hashed. Part of every ANSWER, not just the question: the zero-group
    is reconstructed from it (a customer with no orders in the window has no row for SQL to
    group, so `Sales < 100` would miss exactly the lapsed customers the filter is for)."""
    return hashlib.blake2b(
        repr(sorted(allowed_pids)).encode(), digest_size=8).hexdigest()


def condition_sets(view_configs, pending_cfg, team_id, allowed_pids, today, stamp, memo,
                   on_error=None):
    """Every measure condition across `view_configs` (+ the optional pending config, which goes
    LAST and overwrites — the freshest statement of what the user wants) -> `{ruleId: [pid,…]}`.

    A rule that is incomplete (mid-edit), unresolvable, or errors is left OUT of the answer —
    the client reads a missing id as PENDING and says "Calculating…" rather than showing a count
    it cannot stand behind. Never resolve an incomplete rule: `to_num(None)` is 0 by the
    engines' shared contract, so a valueless `Sales > …` would become `Sales > 0` — a wrong set
    under a confident count.
    """
    from harness import measure_filter as _mf
    by_id = {}
    for cfg in list(view_configs or []) + [pending_cfg or {}]:
        for rule in _mf.collect((cfg or {}).get('filters')):
            rid = str(rule.get('id') or '')
            if rid:
                by_id[rid] = rule
    rules = list(by_id.values())
    if not rules:
        return {}
    pk = _pool_key(allowed_pids)
    out = {}
    for rule in rules:
        if not _mf.rule_complete(rule):
            continue
        key = (stamp, team_id, pk, _mf.signature(rule))
        if key not in memo:
            try:
                memo[key] = _mf.resolve_rule(rule, today, team_id, allowed_pids)
            except Exception as e:
                # An unresolvable condition must NOT become "everything" or "nothing" silently.
                if on_error:
                    on_error('measure-resolve', e)
                continue
        out[str(rule['id'])] = sorted(memo[key])
    if len(memo) > 200:                          # a lifetime cache, not a leak
        memo.clear()
    return out


def series_values(fields, bucket, buckets, team_id, allowed_pids, today, stamp,
                  memo, on_error=None):
    """`{field key: {'values': {bucket_start: v}, 'agg': kind}}` for every measure column
    among `fields`, over `buckets` = [(start_iso, end_iso), …] — the C-TS channel, C-TSWIN
    semantics since wave 14: each field's OWN window slides across the buckets
    (`today := bucket end`), so two fields on the same measure with different windows are
    DIFFERENT questions — the memo key carries the window signature, which the wave-13 key
    (measure-only) silently conflated. `custom` fixed-range windows are the CALLER's to drop
    (`window_fixed`) before calling; one that slips through memoises as a permanent failure.

    Same memo discipline as `column_values`: keys carry (stamp, scope, POOL IDENTITY, the
    question); a permanent failure memoises as None (the caller reports that field as
    dropped), a TRANSIENT one (datastore still warming) never memoises, so the next call
    retries instead of freezing a blank panel for the session's lifetime.
    """
    out = {}
    mfields = [f for f in fields or [] if isinstance(f.get('measure'), dict)]
    if not mfields or not buckets:
        return out
    from harness import measure_filter as _mf
    from harness import windows as _wn
    pk = _pool_key(allowed_pids)
    span_from, span_to = str(buckets[0][0]), str(buckets[-1][1])
    for f in mfields:
        spec = f.get('measure') or {}
        mkey = spec.get('key')
        w = _wn.normalize(spec.get('window'))
        wsig = None if w is None else tuple(sorted(w.items()))
        key = (stamp, team_id, pk, mkey, bucket, span_from, span_to, wsig)
        if key not in memo:
            try:
                memo[key] = _mf.resolve_series_slid(mkey, spec.get('window'), buckets,
                                                    today, team_id, allowed_pids)
            except Exception as e:
                if on_error:
                    on_error('measure-series', e)
                transient = False
                try:
                    from harness import datastore as _ds
                    transient = not _ds.ready()
                except Exception:
                    transient = False
                if transient:
                    continue                     # no memo entry -> the next call tries again
                memo[key] = None
        ans = memo[key]
        if ans is None:
            continue
        out[f['key']] = ans
    if len(memo) > 200:
        memo.clear()
    return out


def column_values(fields, team_id, allowed_pids, today, stamp, memo, on_error=None):
    """Values for every FORMULA-MEASURE column among `fields` -> `{pid: {field key: value}}`,
    handed to `rows_from_pool(derived=…)` beside the cohort cells.

    A column that cannot be computed (store down, BU scope on a company-level measure,
    unresolvable window) degrades to BLANK — the value dict simply omits that field key, and
    the client renders empty cells, never $0: blank is "could not compute", 0 is a real zero.
    """
    mfields = [f for f in fields or [] if isinstance(f.get('measure'), dict)]
    if not mfields:
        return {}
    from harness import measure_filter as _mf
    from harness import windows as _wn
    pk = _pool_key(allowed_pids)
    out = {}
    for f in mfields:
        spec = f['measure']
        w = _wn.normalize(spec.get('window'))
        key = (stamp, team_id, pk, spec.get('key'),
               None if w is None else tuple(sorted(w.items())))
        if key not in memo:
            try:
                memo[key] = _mf.resolve_values(spec.get('key'), spec.get('window'),
                                               today, team_id, allowed_pids)
            except Exception as e:
                if on_error:
                    on_error('measure-column', e)
                # ⚠ ONLY A PERMANENT FAILURE MAY BE MEMOISED. "The data cache is still warming
                # up after a restart" is TRANSIENT and self-healing — memoising it means a user
                # who opened the page during that window gets blank measure cells that STAY
                # blank for the whole session. A permanent failure (an unadmitted measure, an
                # unresolvable window) still memoises, so the retry-storm this guard was
                # written for cannot come back.
                transient = False
                try:
                    from harness import datastore as _ds
                    transient = not _ds.ready()
                except Exception:
                    transient = False
                if transient:
                    continue                     # no memo entry -> the next call tries again
                memo[key] = None
        vals = memo[key]
        if vals is None:
            continue
        fkey = f['key']
        for pid, v in vals.items():
            out.setdefault(pid, {})[fkey] = v
    if len(memo) > 100:
        memo.clear()
    return out