Analytics Tracking Plan skill for Codex

Gives the Codex app a display name, a one-line description and a starter prompt for this skill.

interface:
  display_name: "Analytics Tracking Plan"
  short_description: "One event taxonomy, enforced in code"
  default_prompt: "$analytics-tracking-plan Create a tracking plan for our new team invites feature"

#!/usr/bin/env python3
"""Lint a JSON tracking plan and optionally diff it against tracking calls in code.

Usage:
  python3 lint_tracking_plan.py tracking-plan.json [--style snake|title] [--scan src/]

Plan format:
  {"events": [{"name": "project_created", "description": "...", "trigger": "...",
               "owner": "growth", "properties": {
                 "plan": {"type": "string", "required": true, "enum": ["free", "pro"]}}}]}

Checks: required fields, event naming style (snake: object_action, e.g.
project_created; title: "Project Created"), past-tense action, duplicates after
normalization, snake_case property names, valid property types, likely PII
property names, and (with --scan) events called in code but missing from the
plan or planned but never called. Exits 1 on errors; warnings alone exit 0.
Standard library only.
"""
import argparse
import json
import os
import re
import sys

TYPES = {"string", "number", "integer", "boolean", "array", "object", "datetime"}
PII = re.compile(r"(^|_)(email|phone|first_name|last_name|full_name|address|ip|ip_address|ssn|dob|birth|password|token)($|_)")
SNAKE_EVENT = re.compile(r"^[a-z][a-z0-9]*(_[a-z0-9]+)+$")
TITLE_EVENT = re.compile(r"^[A-Z][a-z0-9]*( [A-Z][a-z0-9]*)+$")
SNAKE_PROP = re.compile(r"^[a-z][a-z0-9]*(_[a-z0-9]+)*$")
CALLS = re.compile(
    r"""(?:\.track|\.capture|\.logEvent|trackEvent|gtag\(\s*['"]event['"]\s*,)\s*\(?\s*['"]([^'"]{2,80})['"]"""
)
CODE_EXT = {".js", ".jsx", ".ts", ".tsx", ".mjs", ".cjs", ".vue", ".svelte", ".py", ".rb", ".kt", ".swift", ".dart", ".go"}
SKIP_DIRS = {"node_modules", ".git", "dist", "build", ".next", "vendor", "__pycache__", ".venv"}


def norm(name):
    return re.sub(r"[^a-z0-9]", "", name.lower())


def past_tense(action):
    return action.endswith("ed") or action in {"sent", "viewed", "began", "won", "lost", "paid", "shown", "seen", "built", "made", "set", "reset", "upgrade_started"}


def lint(plan, style):
    errors, warns, seen = [], [], {}
    events = plan.get("events")
    if not isinstance(events, list) or not events:
        return ["plan must contain a non-empty 'events' list"], warns
    for i, ev in enumerate(events):
        name = ev.get("name", "")
        where = f"event[{i}] '{name}'"
        for field in ("name", "description", "trigger", "owner"):
            if not str(ev.get(field, "")).strip():
                errors.append(f"{where}: missing '{field}'")
        if style == "snake" and name and not SNAKE_EVENT.match(name):
            errors.append(f"{where}: expected object_action snake_case, e.g. project_created")
        if style == "title" and name and not TITLE_EVENT.match(name):
            errors.append(f"{where}: expected 'Object Actioned' title case, e.g. Project Created")
        if name:
            parts = name.replace(" ", "_").lower().split("_")
            if len(parts) >= 2 and not past_tense(parts[-1]):
                warns.append(f"{where}: action '{parts[-1]}' is not past tense; events describe what happened")
            key = norm(name)
            if key in seen:
                errors.append(f"{where}: duplicates '{seen[key]}' after normalization")
            seen[key] = name
        props = ev.get("properties", {})
        if not isinstance(props, dict):
            errors.append(f"{where}: 'properties' must be an object")
            continue
        for pname, spec in props.items():
            pw = f"{where} property '{pname}'"
            if not SNAKE_PROP.match(pname):
                errors.append(f"{pw}: use snake_case")
            ptype = (spec or {}).get("type") if isinstance(spec, dict) else None
            if ptype not in TYPES:
                errors.append(f"{pw}: type must be one of {sorted(TYPES)}")
            if PII.search(re.sub(r"(?<=[a-z0-9])([A-Z])", r"_\1", pname).lower()):
                warns.append(f"{pw}: looks like personal data; confirm consent, hashing or removal")
            if isinstance(spec, dict) and "required" not in spec:
                warns.append(f"{pw}: state required true/false")
    return errors, warns


def scan(root):
    found = {}
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for fn in filenames:
            if os.path.splitext(fn)[1] not in CODE_EXT:
                continue
            path = os.path.join(dirpath, fn)
            try:
                text = open(path, encoding="utf-8", errors="ignore").read()
            except OSError:
                continue
            for m in CALLS.finditer(text):
                line = text.count("\n", 0, m.start()) + 1
                found.setdefault(m.group(1), []).append(f"{path}:{line}")
    return found


def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("plan")
    ap.add_argument("--style", choices=["snake", "title"], default="snake")
    ap.add_argument("--scan", help="directory to search for tracking calls")
    args = ap.parse_args()
    try:
        with open(args.plan, encoding="utf-8") as fh:
            plan = json.load(fh)
    except (OSError, json.JSONDecodeError) as exc:
        sys.exit(f"error: cannot read plan: {exc}")

    errors, warns = lint(plan, args.style)
    if args.scan:
        planned = {e.get("name") for e in plan.get("events", []) if isinstance(e, dict)}
        in_code = scan(args.scan)
        for name, locs in sorted(in_code.items()):
            if name not in planned:
                errors.append(f"code calls unplanned event '{name}' at {locs[0]}" + (f" (+{len(locs) - 1} more)" if len(locs) > 1 else ""))
        for name in sorted(n for n in planned if n and n not in in_code):
            warns.append(f"planned event '{name}' not found in {args.scan} (may be server-side or not yet built)")

    for w in warns:
        print(f"warn:  {w}")
    for e in errors:
        print(f"error: {e}")
    total = len(plan.get("events", []) or [])
    print(f"\n{total} events checked: {len(errors)} error(s), {len(warns)} warning(s)")
    sys.exit(1 if errors else 0)


if __name__ == "__main__":
    main()