Three CI/CD
Gates

jobs:
  instrumentation-gate:
    docker:
      - image: cimg/python:3.11
    steps:
      - checkout
      - run:
          name: Check instrumentation for active bets
          command: |
            python3 scripts/check_instrumentation.py \
              --registry .outcomes/registry.json \
              --src src/

import json, sys, subprocess, argparse
from datetime import date

def load_registry(path):
    with open(path) as f:
        return json.load(f)

def is_active(bet):
    today = date.today().isoformat()
    return bet["measurement_window_open"] <= today <= bet["measurement_window_close"]

def check_event_in_src(event_name, src_path):
    result = subprocess.run(
        ["grep", "-r", event_name, src_path],
        capture_output=True, text=True
    )
    return result.returncode == 0

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--registry", required=True)
    parser.add_argument("--src", required=True)
    args = parser.parse_args()

    registry = load_registry(args.registry)
    failures = []

    for bet in registry["bets"]:
        if not is_active(bet):
            continue
        for event in bet["required_events"]:
            if not check_event_in_src(event, args.src):
                failures.append({"bet": bet["id"], "missing_event": event})

    if failures:
        print("INSTRUMENTATION GATE FAILED")
        print("The following events are required by active bets but not found in the codebase:")
        for f in failures:
            print(f"  Bet: {f['bet']} | Missing event: {f['missing_event']}")
        print("")
        print("This deployment cannot proceed until these events are wired.")
        print("If this bet is no longer active, update the registry.")
        sys.exit(1)

    print(f"Instrumentation gate passed. {len(registry['bets'])} bets checked.")

if __name__ == "__main__":
    main()

jobs:
  validation-gate:
    docker:
      - image: cimg/python:3.11
    steps:
      - checkout
      - run:
          name: Check validation records for active P0/P1 bets
          command: |
            python3 scripts/check_validation.py \
              --registry .outcomes/registry.json

import json, os, sys, argparse
from datetime import date

def load_registry(path):
    with open(path) as f:
        return json.load(f)

def is_active(bet):
    today = date.today().isoformat()
    return bet["measurement_window_open"] <= today <= bet["measurement_window_close"]

def requires_validation(bet):
    return bet.get("classification") in ("P0", "P1")

def validation_record_exists(bet):
    ref = bet.get("validation_record_ref")
    return bool(ref) and os.path.exists(ref)

def validation_record_is_signed(bet):
    ref = bet.get("validation_record_ref")
    if not ref or not os.path.exists(ref):
        return False
    with open(ref) as f:
        content = f.read()
    return "Signed-off-by:" in content

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--registry", required=True)
    args = parser.parse_args()

    registry = load_registry(args.registry)
    failures = []

    for bet in registry["bets"]:
        if not is_active(bet) or not requires_validation(bet):
            continue
        if not validation_record_exists(bet):
            failures.append({"bet": bet["id"], "reason": "No validation record found at: " + str(bet.get("validation_record_ref"))})
        elif not validation_record_is_signed(bet):
            failures.append({"bet": bet["id"], "reason": "Validation record exists but has no Signed-off-by line"})

    if failures:
        print("VALIDATION GATE FAILED")
        for f in failures:
            print(f"  Bet: {f['bet']} | {f['reason']}")
        sys.exit(1)

    print("Validation gate passed.")

if __name__ == "__main__":
    main()

# Validation Record: [Bet ID]

**Bet:** [Bet name]
**Validation method:** [Prototype test / shadow mode / staged cohort / other]
**Date:** [YYYY-MM-DD]

## Evidence
[Description of what was tested, how, and what was observed]

## Conclusion
[Proceed / Do not proceed / Proceed with modifications]

[One to three sentences explaining the conclusion]

**Signed-off-by:** [Name, role]
**Date signed:** [YYYY-MM-DD]

jobs:
  confirmation-integration:
    docker:
      - image: cimg/python:3.11
    steps:
      - checkout
      - run:
          name: Write confirmation integration record
          command: |
            python3 scripts/write_confirmation_record.py \
              --registry .outcomes/registry.json \
              --deploy-sha $CIRCLE_SHA1 \
              --deploy-time $(date -u +"%Y-%m-%dT%H:%M:%SZ") \
              --output .outcomes/deployments/

import json, os, sys, argparse
from datetime import date

def load_registry(path):
    with open(path) as f:
        return json.load(f)

def is_active(bet):
    today = date.today().isoformat()
    return bet["measurement_window_open"] <= today <= bet["measurement_window_close"]

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--registry", required=True)
    parser.add_argument("--deploy-sha", required=True)
    parser.add_argument("--deploy-time", required=True)
    parser.add_argument("--output", required=True)
    args = parser.parse_args()

    registry = load_registry(args.registry)
    active_bets = [b for b in registry["bets"] if is_active(b)]

    if not active_bets:
        print("No active bets. No confirmation integration record written.")
        return

    os.makedirs(args.output, exist_ok=True)

    record = {
        "deploy_sha": args.deploy_sha,
        "deploy_time": args.deploy_time,
        "active_bets": [
            {
                "id": bet["id"],
                "name": bet["name"],
                "classification": bet["classification"],
                "required_events": bet["required_events"],
                "success_threshold": bet["success_threshold"],
                "measurement_window_open": bet["measurement_window_open"],
                "measurement_window_close": bet["measurement_window_close"],
                "confirmation_owner": bet["confirmation_owner"],
                "validation_record_ref": bet.get("validation_record_ref")
            }
            for bet in active_bets
        ]
    }

    filename = f"{args.deploy_sha[:8]}-{args.deploy_time[:10]}.json"
    output_path = os.path.join(args.output, filename)

    with open(output_path, "w") as f:
        json.dump(record, f, indent=2)

    print(f"Confirmation integration record written to {output_path}")
    print(f"Linked {len(active_bets)} active bet(s) to deploy {args.deploy_sha[:8]}")

if __name__ == "__main__":
    main()