""" Feature Flag Evaluator · Phase 2a+ Python port A-owned · session_a 1:1 port of evaluator.js (Phase 2a reference implementation). Same 9-step order · same djb2-xor hash · same response shape. Parity guarantee: For any (registry, request) pair, this module MUST produce the same (value, source, bucket) as evaluator.js would. Parity is enforced by verify_parity.py against evaluator_examples.json. Contract: ../feature-flags/evaluator_contract.json Reference: ../feature-flags/evaluator.js Examples: ../feature-flags/evaluator_examples.json HTTP binding: app.py (FastAPI) """ from __future__ import annotations from datetime import datetime, timezone from typing import Any VERSION = "2a_plus-py-port" _JS_PARITY_VERSION = "2a-js-module" # evaluator.js version we mirror def hash_bucket(s: str) -> int: """Deterministic 32-bit djb2-xor hash → mod 100 bucket. Must match evaluator.js::hashBucket exactly. The 32-bit force-truncation is the key parity detail: `h = h | 0` in JS; emulated here with a signed 32-bit wrap. """ h = 5381 INT32_MASK = 0xFFFFFFFF for ch in s: h = (((h << 5) + h) & INT32_MASK) ^ ord(ch) # Force signed 32-bit (mirror of `h = h | 0` in JS) h = h & INT32_MASK if h & 0x80000000: h = h - 0x100000000 return abs(h) % 100 def _find_flag(registry: dict | None, key: str): if not registry or not isinstance(registry.get("flags"), list): return None for f in registry["flags"]: if f.get("key") == key: return f return None def _now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.") + \ f"{datetime.now(timezone.utc).microsecond // 1000:03d}Z" def evaluate(registry: dict, request: dict, _depth: int = 0) -> dict: """Evaluate one flag. Returns a dict matching evaluator.js response_shape.""" trace: list[str] = [] deps_evaluated: list[dict] = [] start_at = (request or {}).get("now_iso") or _now_iso() def done(value, source, stage=None, bucket=None): return { "flag_key": (request or {}).get("flag_key"), "value": value, "source": source, "stage": stage, "bucket": bucket, "cached": False, "deps_evaluated": deps_evaluated, "trace": trace, "evaluated_at": start_at, "evaluator_version": VERSION, } # Step 0 · request validity if not request or not isinstance(request.get("flag_key"), str): trace.append("[0] request invalid: missing flag_key") return done(False, "error") # user_id required (string or number 0 OK) uid = request.get("user_id") if uid is None or uid == "": trace.append("[0] request invalid: missing user_id") return done(False, "error") if not request.get("env") or not request.get("tier"): trace.append("[0] request invalid: missing env or tier") return done(False, "error") # Step 1 · flag_exists_check flag = _find_flag(registry, request["flag_key"]) if not flag: trace.append(f"[1] flag_exists_check: UNKNOWN_FLAG key={request['flag_key']}") return done(False, "unknown_flag") trace.append(f"[1] flag_exists_check: found id={flag.get('id')} · stage={flag.get('rollout_stage')}") stage = flag.get("rollout_stage") default_value = flag.get("default_value") # Step 2 · rollout_stage short-circuit if stage in ("paused", "rolled_back"): trace.append(f"[2] rollout_stage_gate: SHORT-CIRCUIT stage={stage} → default") return done(default_value, stage, stage=stage) trace.append(f"[2] rollout_stage_gate: stage={stage} · proceed") # Step 3 · dependency_check (cycle guard depth 8) if _depth > 8: trace.append("[3] dependency_check: CYCLE depth>8 · treating as unsatisfied") return done(default_value, "dep_unsatisfied", stage=stage) deps = flag.get("dependencies") or [] if isinstance(deps, list) and len(deps) > 0: for dep in deps: sub_req = dict(request) sub_req["flag_key"] = dep.get("requires_flag") dep_result = evaluate(registry, sub_req, _depth=_depth + 1) deps_evaluated.append({ "flag_key": dep.get("requires_flag"), "wanted": dep.get("requires_value"), "got": dep_result["value"], "satisfied": dep_result["value"] == dep.get("requires_value"), }) if dep_result["value"] != dep.get("requires_value"): trace.append( f"[3] dependency_check: UNSATISFIED {dep.get('requires_flag')} " f"wanted={dep.get('requires_value')} got={dep_result['value']}" ) return done(default_value, "dep_unsatisfied", stage=stage) trace.append(f"[3] dependency_check: all {len(deps_evaluated)} satisfied") # Step 4 · approval_gate audit = flag.get("audit") or {} has_approval = audit.get("last_approval_ref") if flag.get("requires_approval") and not has_approval: trace.append("[4] approval_gate: requires_approval=true · no approval_ref → default") return done(default_value, "approval_missing", stage=stage) trace.append( "[4] approval_gate: " + (f"satisfied (ref={has_approval})" if flag.get("requires_approval") else "not required") ) overrides = (request.get("overrides") or {}) # Step 5 · user_override user_ov = overrides.get("user") or {} if flag["key"] in user_ov: trace.append(f"[5] user_override: HIT value={user_ov[flag['key']]}") return done(user_ov[flag["key"]], "user_override", stage=stage) trace.append("[5] user_override_lookup: no match") # Step 6 · tenant_override tenant_ov = overrides.get("tenant") or {} if flag["key"] in tenant_ov: trace.append(f"[6] tenant_override: HIT value={tenant_ov[flag['key']]}") return done(tenant_ov[flag["key"]], "tenant_override", stage=stage) trace.append("[6] tenant_override_lookup: no match") # Step 7 · cohort_bucketing (staged only) if stage == "staged": pct = flag.get("rollout_pct", 0) or 0 bucket = hash_bucket(f"{request['user_id']}|{flag['key']}") enabled = bucket < pct trace.append(f"[7] cohort_bucketing: pct={pct} bucket={bucket} → {enabled}") return done(enabled, "rollout_pct", stage=stage, bucket=bucket) # Step 8 · rollout_stage_map if stage == "internal": is_staff = request.get("tier") in ("staff", "admin") trace.append(f"[8] rollout_stage_map: internal · tier={request.get('tier')} → {is_staff}") return done(is_staff, "stage-internal", stage=stage) if stage == "pilot": trace.append("[8] rollout_stage_map: pilot · cohort check deferred · returning true (Phase 2a simplification)") return done(True, "stage-pilot", stage=stage) if stage == "full": trace.append("[8] rollout_stage_map: full → true") return done(True, "stage-full", stage=stage) trace.append(f"[8] rollout_stage_map: stage={stage} → default_value") # Step 9 · default_return trace.append(f"[9] default_return: {default_value}") return done(default_value, "default", stage=stage) def evaluate_batch(registry: dict, requests: list[dict]) -> list[dict]: out: list[dict] = [] if not isinstance(requests, list): return out for r in requests: try: out.append(evaluate(registry, r)) except Exception as e: # noqa: BLE001 out.append({ "flag_key": (r or {}).get("flag_key"), "value": False, "source": "error", "trace": [f"exception: {e}"], "evaluator_version": VERSION, }) return out def check_example(registry: dict, example: dict) -> dict: """Match evaluator.js::checkExample pass criteria exactly.""" try: got = evaluate(registry, example["request"]) exp = example["expected"] val_match = got["value"] == exp["value"] src_match = got["source"] == exp["source"] pass_ = val_match and src_match if exp.get("bucket") is not None: pass_ = pass_ and (got["bucket"] == exp["bucket"]) return { "id": example.get("id"), "name": example.get("name"), "pass": pass_, "got": {"value": got["value"], "source": got["source"], "bucket": got["bucket"]}, "expected": exp, "trace": got["trace"], } except Exception as e: # noqa: BLE001 return { "id": example.get("id"), "name": example.get("name"), "pass": False, "error": str(e), } __all__ = [ "VERSION", "hash_bucket", "evaluate", "evaluate_batch", "check_example", ]