security-review/cross_review.py

#!/usr/bin/env python3
"""cross_review.py — Sea Haven cross-family (GPT-4.1) review CLI.

Re-homed from the archived Sea-Haven-Industries/orchestrator repo's
`cross_reviewer` agent. Router-less: one direct OpenAI SDK call, no langchain,
no routing logic. This is the mandatory cross-family reviewer for IAM/policy
and Lambda-handler-signature changes, and the reasoning backend for the
sh-plan-review / sh-security-audit / sh-build-review gates.

Usage:
    python3 cross_review.py "Review this diff for breaking changes: <diff>"

Auth: reads OPENAI_API_KEY from the environment, falling back to the
gitignored .env at the repo root. Never prints the key.
"""

import argparse
import os
import random
import sys
import time

# Model ID — single source of truth (ported from orchestrator/models.py).
DEFAULT_MODEL = "gpt-4.1"
TEMPERATURE = 0.2
MAX_ATTEMPTS = 3

# System prompt ported verbatim from the orchestrator's cross_reviewer agent
# (orchestrator/agents.py, CROSS_REVIEWER_PROMPT).
CROSS_REVIEWER_PROMPT = """You are a cross-family code review agent. You provide an independent review perspective.
Review the provided code for bugs, security issues, and improvements.
Categorize findings as BLOCK, FIX, or NIT. Be concise.
Focus on issues that might be missed by the primary development team."""


def load_api_key() -> str:
    """OPENAI_API_KEY from the environment, else hand-parsed from repo-root .env."""
    key = os.environ.get("OPENAI_API_KEY")
    if key:
        return key
    env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), ".env")
    try:
        with open(env_path, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line or line.startswith("#") or "=" not in line:
                    continue
                name, _, value = line.partition("=")
                if name.strip() == "OPENAI_API_KEY":
                    return value.strip().strip("'\"")
    except OSError:
        pass
    return ""


def run_review(task: str, model: str, api_key: str) -> str:
    """One review call with retries on transient API errors (3 attempts,
    exponential backoff with jitter — ported from orchestrator/models.py
    with_retries)."""
    # Lazy import so --help works without the openai package installed.
    import openai

    retriable = (
        openai.APIConnectionError,
        openai.RateLimitError,
        openai.InternalServerError,
    )
    client = openai.OpenAI(api_key=api_key)
    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            response = client.chat.completions.create(
                model=model,
                temperature=TEMPERATURE,
                messages=[
                    {"role": "system", "content": CROSS_REVIEWER_PROMPT},
                    {"role": "user", "content": task},
                ],
            )
            content = response.choices[0].message.content
            if not content:
                raise RuntimeError("model returned an empty review")
            return content
        except retriable as exc:
            if attempt == MAX_ATTEMPTS:
                raise
            delay = (2 ** (attempt - 1)) + random.uniform(0, 1)
            print(
                f"cross_review: transient API error ({type(exc).__name__}), "
                f"retrying in {delay:.1f}s (attempt {attempt}/{MAX_ATTEMPTS})",
                file=sys.stderr,
            )
            time.sleep(delay)
    raise RuntimeError("unreachable")


def main() -> int:
    parser = argparse.ArgumentParser(
        description=(
            "Cross-family GPT-4.1 review (direct OpenAI SDK, no router). "
            "Mandatory for IAM/policy and Lambda-handler-signature changes."
        )
    )
    parser.add_argument("task", help="review task: a diff, change description, or plan")
    parser.add_argument(
        "--model",
        default=DEFAULT_MODEL,
        help=f"OpenAI model id (default: {DEFAULT_MODEL})",
    )
    args = parser.parse_args()

    api_key = load_api_key()
    if not api_key:
        print(
            "cross_review: OPENAI_API_KEY not set and not found in repo-root .env",
            file=sys.stderr,
        )
        return 2

    try:
        review = run_review(args.task, args.model, api_key)
    except ImportError:
        print(
            "cross_review: the 'openai' package is not installed "
            "(pip install -r requirements.txt)",
            file=sys.stderr,
        )
        return 2
    except Exception as exc:  # noqa: BLE001 — CLI boundary: report and exit nonzero
        print(
            f"cross_review: review failed: {type(exc).__name__}: {exc}", file=sys.stderr
        )
        return 1

    print(review)
    return 0


if __name__ == "__main__":
    sys.exit(main())