Flagship

#!/usr/bin/env bash
set -euo pipefail

usage() {
  cat <<'EOF'
Usage:
  evaluate_policy.sh \
    --manifest <path> \
    --metrics <path> \
    --budget <path> \
    --recommendation <path|literal> \
    --output <path> \
    [--min-sample <n>] \
    [--guardrail-max-degradation <decimal>]

Description:
  Evaluate deterministic policy gates and emit final Flagship decision payload.
EOF
}

manifest=""
metrics=""
budget=""
recommendation=""
output=""
min_sample="500"
guardrail_max_degradation="2.0"

while [[ $# -gt 0 ]]; do
  case "$1" in
    --manifest) manifest="$2"; shift 2 ;;
    --metrics) metrics="$2"; shift 2 ;;
    --budget) budget="$2"; shift 2 ;;
    --recommendation) recommendation="$2"; shift 2 ;;
    --output) output="$2"; shift 2 ;;
    --min-sample) min_sample="$2"; shift 2 ;;
    --guardrail-max-degradation) guardrail_max_degradation="$2"; shift 2 ;;
    -h|--help) usage; exit 0 ;;
    *)
      echo "Unknown argument: $1" >&2
      usage
      exit 1
      ;;
  esac
done

if [[ -z "$manifest" || -z "$metrics" || -z "$budget" || -z "$recommendation" || -z "$output" ]]; then
  echo "Missing required arguments." >&2
  usage
  exit 1
fi

python3 - "$manifest" "$metrics" "$budget" "$recommendation" "$output" "$min_sample" "$guardrail_max_degradation" <<'PY'
import json
import subprocess
import sys
from pathlib import Path

def load_yaml(path: Path):
    try:
        import yaml  # type: ignore
        return yaml.safe_load(path.read_text())
    except ImportError:
        try:
            raw = subprocess.check_output(
                [
                    "ruby",
                    "-ryaml",
                    "-rjson",
                    "-e",
                    "print JSON.dump(YAML.load_file(ARGV[0]))",
                    str(path),
                ],
                text=True,
            )
            return json.loads(raw)
        except Exception as exc:
            print(
                "Unable to parse YAML. Install pyyaml or ensure ruby with psych is available.",
                file=sys.stderr,
            )
            raise exc

manifest_path = Path(sys.argv[1])
metrics_path = Path(sys.argv[2])
budget_path = Path(sys.argv[3])
recommendation_arg = sys.argv[4]
output_path = Path(sys.argv[5])
min_sample = int(sys.argv[6])
guardrail_max = float(sys.argv[7])

manifest = load_yaml(manifest_path) or {}
metrics = json.loads(metrics_path.read_text())
budget = json.loads(budget_path.read_text())

rec_path = Path(recommendation_arg)
if rec_path.exists():
    recommendation_raw = json.loads(rec_path.read_text())
else:
    recommendation_raw = {
        "agent_recommendation": recommendation_arg,
        "confidence": 0.5,
        "reasoning_summary": "Literal recommendation input",
    }

allowed_recommendations = {"ITERATE", "KEEP_TREATMENT", "KEEP_CONTROL", "HOLD", "STOP"}
fail_reasons = []

budget_before = float(
    budget.get("budget_remaining_usd", budget.get("remaining_usd", budget.get("budget_before_usd", 0.0)))
)
if budget_before <= 0:
    fail_reasons.append("Budget hard stop: remaining budget <= 0")

control_n = int(metrics.get("sample_size_control", 0))
treatment_n = int(metrics.get("sample_size_treatment", 0))
if control_n < min_sample or treatment_n < min_sample:
    fail_reasons.append(
        f"Minimum sample not met: control={control_n}, treatment={treatment_n}, required={min_sample}"
    )

guardrail_config = {}
for entry in manifest.get("guardrails", []) or []:
    if not isinstance(entry, dict):
        continue
    name = entry.get("name")
    if not name:
        continue
    guardrail_config[name] = {
        "direction": entry.get("direction", "lower_is_better"),
        "max_degradation_pct": float(entry.get("max_degradation_pct", guardrail_max)),
    }

raw_guardrails = metrics.get("raw_guardrails", {}) or {}
raw_control = raw_guardrails.get("control", {}) or {}
raw_treatment = raw_guardrails.get("treatment", {}) or {}

for key, delta in (metrics.get("guardrail_deltas", {}) or {}).items():
    cfg = guardrail_config.get(
        key,
        {"direction": "lower_is_better", "max_degradation_pct": guardrail_max},
    )
    direction = cfg["direction"]
    limit_pct = float(cfg["max_degradation_pct"])

    c_val = raw_control.get(key)
    t_val = raw_treatment.get(key)
    if c_val is not None and t_val is not None:
        c_val = float(c_val)
        t_val = float(t_val)
        signed_delta = t_val - c_val
        pct_delta = abs((signed_delta / c_val) * 100.0) if c_val != 0 else (100.0 if signed_delta != 0 else 0.0)

        if direction == "lower_is_better" and signed_delta > 0 and pct_delta > limit_pct:
            fail_reasons.append(
                f"Guardrail breach: {key} degraded by {pct_delta:.4f}% (limit={limit_pct:.4f}%)"
            )
        elif direction == "higher_is_better" and signed_delta < 0 and pct_delta > limit_pct:
            fail_reasons.append(
                f"Guardrail breach: {key} degraded by {pct_delta:.4f}% (limit={limit_pct:.4f}%)"
            )
        elif direction not in {"lower_is_better", "higher_is_better"} and pct_delta > limit_pct:
            fail_reasons.append(
                f"Guardrail breach: {key} moved by {pct_delta:.4f}% (limit={limit_pct:.4f}%)"
            )
    else:
        if float(delta) > guardrail_max:
            fail_reasons.append(
                f"Guardrail breach: {key} delta={float(delta):.6f} exceeds fallback limit={guardrail_max:.6f}"
            )

drift_reasons = metrics.get("manifest_posthog_drift_reasons", [])
if not isinstance(drift_reasons, list):
    drift_reasons = [str(drift_reasons)]
drift_detected = bool(metrics.get("manifest_posthog_drift_detected", False)) or bool(drift_reasons)
if drift_detected:
    # Keep drift visibility in logs but do not hard-block decisions.
    if drift_reasons:
        print(
            "Manifest/PostHog drift warnings: " + "; ".join(str(reason) for reason in drift_reasons),
            file=sys.stderr,
        )
    else:
        print("Manifest/PostHog drift warning detected", file=sys.stderr)

agent_recommendation = str(recommendation_raw.get("agent_recommendation", "HOLD")).upper()
if agent_recommendation not in allowed_recommendations:
    fail_reasons.append(f"Invalid recommendation: {agent_recommendation}")
    agent_recommendation = "HOLD"

if budget_before <= 0:
    final_action = "STOP"
elif fail_reasons:
    final_action = "HOLD"
else:
    final_action = agent_recommendation

policy_result = "PASS" if not fail_reasons else "FAIL"

payload = {
    "experiment_id": manifest.get("experiment_id"),
    "window_start_utc": metrics.get("window_start_utc"),
    "window_end_utc": metrics.get("window_end_utc"),
    "kpi_control": metrics.get("kpi_control"),
    "kpi_treatment": metrics.get("kpi_treatment"),
    "guardrail_deltas": metrics.get("guardrail_deltas", {}),
    "agent_recommendation": agent_recommendation,
    "policy_result": policy_result,
    "policy_fail_reasons": fail_reasons,
    "final_action": final_action,
    "budget_before_usd": budget_before,
    "budget_after_usd": budget_before,
    "confidence": float(recommendation_raw.get("confidence", 0.5)),
}

output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
print(str(output_path))
PY

#!/usr/bin/env bash
set -euo pipefail

usage() {
  cat <<'EOF'
Usage:
  fetch_metrics.sh --manifest <path> --mcp-output <path> --output <path> [--window-start <iso8601>] [--window-end <iso8601>]

Description:
  Normalize raw PostHog MCP metric output into the Flagship metrics schema.
EOF
}

manifest=""
mcp_output=""
output=""
window_start=""
window_end=""

while [[ $# -gt 0 ]]; do
  case "$1" in
    --manifest) manifest="$2"; shift 2 ;;
    --mcp-output) mcp_output="$2"; shift 2 ;;
    --output) output="$2"; shift 2 ;;
    --window-start) window_start="$2"; shift 2 ;;
    --window-end) window_end="$2"; shift 2 ;;
    -h|--help) usage; exit 0 ;;
    *)
      echo "Unknown argument: $1" >&2
      usage
      exit 1
      ;;
  esac
done

if [[ -z "$manifest" || -z "$mcp_output" || -z "$output" ]]; then
  echo "Missing required arguments." >&2
  usage
  exit 1
fi

python3 - "$manifest" "$mcp_output" "$output" "$window_start" "$window_end" <<'PY'
import json
import subprocess
import sys
from pathlib import Path

def load_yaml(path: Path):
    try:
        import yaml  # type: ignore
        return yaml.safe_load(path.read_text())
    except ImportError:
        try:
            raw = subprocess.check_output(
                [
                    "ruby",
                    "-ryaml",
                    "-rjson",
                    "-e",
                    "print JSON.dump(YAML.load_file(ARGV[0]))",
                    str(path),
                ],
                text=True,
            )
            return json.loads(raw)
        except Exception as exc:
            print(
                "Unable to parse YAML. Install pyyaml or ensure ruby with psych is available.",
                file=sys.stderr,
            )
            raise exc

manifest_path = Path(sys.argv[1])
mcp_path = Path(sys.argv[2])
output_path = Path(sys.argv[3])
window_start_arg = sys.argv[4]
window_end_arg = sys.argv[5]

manifest = load_yaml(manifest_path)
raw = json.loads(mcp_path.read_text())

manifest_feature_flag = manifest.get("feature_flag", {}) or {}
manifest_posthog = manifest.get("posthog", {}) or {}
raw_experiment = raw.get("experiment", {}) or raw.get("posthog_experiment", {}) or {}

def resolve_variant(container, names):
    if not isinstance(container, dict):
        return {}

    for name in names:
        if name and isinstance(container.get(name), dict):
            return container[name]

    normalized = {str(name).lower() for name in names if name}
    for key, value in container.items():
        if str(key).lower() in normalized and isinstance(value, dict):
            return value

    return {}

def resolve_variants(payload, control_name, treatment_name):
    candidates = [
        payload.get("variants"),
        payload.get("cohorts"),
        payload.get("results"),
        (payload.get("metrics", {}) or {}).get("variants"),
    ]

    control_names = [control_name, "control", "baseline", "a"]
    treatment_names = [treatment_name, "treatment", "variant", "b"]

    control_variant = {}
    treatment_variant = {}
    for container in candidates:
        if not isinstance(container, dict):
            continue

        if not control_variant:
            control_variant = resolve_variant(container, control_names)
        if not treatment_variant:
            treatment_variant = resolve_variant(container, treatment_names)
        if control_variant and treatment_variant:
            break

    return control_variant, treatment_variant

control, treatment = resolve_variants(
    raw,
    manifest_feature_flag.get("control_variant"),
    manifest_feature_flag.get("treatment_variant"),
)

def read_nested(data, dotted):
    cur = data
    for key in dotted.split("."):
        if not isinstance(cur, dict) or key not in cur:
            return None
        cur = cur[key]
    return cur

def read_raw_experiment(*candidates):
    for source in [raw_experiment, raw]:
        if not isinstance(source, dict):
            continue
        for key in candidates:
            if "." in key:
                value = read_nested(source, key)
            else:
                value = source.get(key)
            if value is not None:
                return value
    return None

drift_reasons = []

def add_drift_reason(field_name, manifest_value, posthog_value):
    if manifest_value is None or posthog_value is None:
        return
    if str(manifest_value) != str(posthog_value):
        drift_reasons.append(
            f"{field_name} mismatch (manifest={manifest_value!r}, posthog={posthog_value!r})"
        )

add_drift_reason(
    "posthog.experiment_id",
    manifest_posthog.get("experiment_id"),
    read_raw_experiment("experiment_id", "id"),
)
add_drift_reason(
    "feature_flag.provider",
    manifest_feature_flag.get("provider"),
    read_raw_experiment("provider"),
)
add_drift_reason(
    "feature_flag.key",
    manifest_feature_flag.get("key"),
    read_raw_experiment("feature_flag_key", "feature_flag.key", "flag.key"),
)
add_drift_reason(
    "feature_flag.control_variant",
    manifest_feature_flag.get("control_variant"),
    read_raw_experiment("control_variant", "variants.control.name"),
)
add_drift_reason(
    "feature_flag.treatment_variant",
    manifest_feature_flag.get("treatment_variant"),
    read_raw_experiment("treatment_variant", "variants.treatment.name"),
)

def read_number(variant, keys, default=0.0):
    if not isinstance(variant, dict):
        return float(default)

    for key in keys:
        value = variant.get(key)
        if value is not None:
            try:
                return float(value)
            except (TypeError, ValueError):
                pass

    nested_metrics = variant.get("metrics", {}) or {}
    if isinstance(nested_metrics, dict):
        for key in keys:
            value = nested_metrics.get(key)
            if value is not None:
                try:
                    return float(value)
                except (TypeError, ValueError):
                    pass

    return float(default)

def read_int(variant, keys, default=0):
    return int(read_number(variant, keys, default=default))

def read_guardrails(variant):
    if not isinstance(variant, dict):
        return {}

    direct = variant.get("guardrails")
    if isinstance(direct, dict):
        return direct

    nested_metrics = variant.get("metrics", {}) or {}
    if isinstance(nested_metrics, dict):
        nested_guardrails = nested_metrics.get("guardrails")
        if isinstance(nested_guardrails, dict):
            return nested_guardrails

    return {}

control_guardrails = read_guardrails(control)
treatment_guardrails = read_guardrails(treatment)

guardrail_names = set(control_guardrails.keys()) | set(treatment_guardrails.keys())
guardrail_deltas = {}
for name in sorted(guardrail_names):
    c_val = control_guardrails.get(name)
    t_val = treatment_guardrails.get(name)
    if c_val is None or t_val is None:
        continue
    guardrail_deltas[name] = float(t_val) - float(c_val)

out = {
    "experiment_id": manifest.get("experiment_id"),
    "window_start_utc": window_start_arg or raw.get("window_start_utc"),
    "window_end_utc": window_end_arg or raw.get("window_end_utc"),
    "kpi_control": read_number(control, ["kpi", "primary_kpi", "value", "metric", "conversion_rate"]),
    "kpi_treatment": read_number(treatment, ["kpi", "primary_kpi", "value", "metric", "conversion_rate"]),
    "sample_size_control": read_int(control, ["sample_size", "n", "count", "users"]),
    "sample_size_treatment": read_int(treatment, ["sample_size", "n", "count", "users"]),
    "guardrail_deltas": guardrail_deltas,
    "raw_guardrails": {
        "control": control_guardrails,
        "treatment": treatment_guardrails,
    },
    "manifest_posthog_drift_detected": len(drift_reasons) > 0,
    "manifest_posthog_drift_reasons": drift_reasons,
}

output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(out, indent=2, sort_keys=True) + "\n")
print(str(output_path))
PY

#!/usr/bin/env bash
set -euo pipefail

usage() {
  cat <<'EOF'
Usage:
  validate_manifest.sh --manifest <path> [--previous <path>] [--output <path>]

Description:
  Validate required experiment manifest fields and immutable field constraints.
EOF
}

manifest=""
previous=""
output=""

while [[ $# -gt 0 ]]; do
  case "$1" in
    --manifest) manifest="$2"; shift 2 ;;
    --previous) previous="$2"; shift 2 ;;
    --output) output="$2"; shift 2 ;;
    -h|--help) usage; exit 0 ;;
    *)
      echo "Unknown argument: $1" >&2
      usage
      exit 1
      ;;
  esac
done

if [[ -z "$manifest" ]]; then
  echo "Missing required argument: --manifest" >&2
  usage
  exit 1
fi

python3 - "$manifest" "$previous" "$output" <<'PY'
import json
import subprocess
import sys
from pathlib import Path

def load_yaml(path: Path):
    try:
        import yaml  # type: ignore
        return yaml.safe_load(path.read_text())
    except ImportError:
        try:
            raw = subprocess.check_output(
                [
                    "ruby",
                    "-ryaml",
                    "-rjson",
                    "-e",
                    "print JSON.dump(YAML.load_file(ARGV[0]))",
                    str(path),
                ],
                text=True,
            )
            return json.loads(raw)
        except Exception as exc:
            print(
                "Unable to parse YAML. Install pyyaml or ensure ruby with psych is available.",
                file=sys.stderr,
            )
            raise exc

manifest_path = Path(sys.argv[1])
previous_path = Path(sys.argv[2]) if sys.argv[2] else None
output_path = Path(sys.argv[3]) if sys.argv[3] else None

manifest = load_yaml(manifest_path) or {}
errors = []

required_root = [
    "experiment_id",
    "title",
    "objective",
    "primary_kpi",
    "guardrails",
    "max_budget_usd",
    "posthog",
    "feature_flag",
    "status",
    "created_at_utc",
]

for field in required_root:
    if field not in manifest:
        errors.append(f"Missing required field: {field}")

feature_flag = manifest.get("feature_flag", {}) or {}
for field in ["key", "control_variant", "treatment_variant"]:
    if field not in feature_flag:
        errors.append(f"Missing required field: feature_flag.{field}")

posthog = manifest.get("posthog", {}) or {}
if "project_id" not in posthog:
    errors.append("Missing required field: posthog.project_id")

cohorts = posthog.get("cohorts", {}) or {}
for field in ["control", "treatment"]:
    if field not in cohorts:
        errors.append(f"Missing required field: posthog.cohorts.{field}")

allowed_status = {"draft", "active", "paused", "winner_selected", "completed", "stopped"}
status = manifest.get("status")
if status not in allowed_status:
    errors.append(f"Invalid status: {status}. Allowed: {sorted(allowed_status)}")

budget = manifest.get("max_budget_usd")
if budget is None:
    errors.append("max_budget_usd cannot be null")
else:
    try:
        if float(budget) <= 0:
            errors.append("max_budget_usd must be greater than 0")
    except (TypeError, ValueError):
        errors.append("max_budget_usd must be numeric")

if previous_path:
    prev = load_yaml(previous_path) or {}
    for immutable_key in ["objective", "primary_kpi", "max_budget_usd"]:
        if prev.get(immutable_key) != manifest.get(immutable_key):
            errors.append(
                f"Immutable field changed: {immutable_key} "
                f"(old={prev.get(immutable_key)!r}, new={manifest.get(immutable_key)!r})"
            )

result = {
    "valid": len(errors) == 0,
    "errors": errors,
    "manifest_path": str(manifest_path),
}

payload = json.dumps(result, indent=2, sort_keys=True) + "\n"

if output_path:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(payload)

sys.stdout.write(payload)
sys.exit(0 if result["valid"] else 1)
PY