Function _call_claude_dual_or

plugins/security-guidance/hooks/llm.py:521–597

Run prompt through the model 2× in parallel and OR-merge the results. The second look samples the model again on the same prompt — independent sampling means borderline cases can flip between the legs, and the OR merge keeps any finding either leg surfaces. Trades higher API spend for

(prompt, output_schema, *, bool_key: str, list_key: str,
                         thinking_budget=10000, max_tokens=16000)

def _call_claude_dual_or(prompt, output_schema, *, bool_key: str, list_key: str,
                         thinking_budget=10000, max_tokens=16000):
    """Run prompt through the model 2× in parallel and OR-merge the results.

    The second look samples the model again on the same prompt — independent
    sampling means borderline cases can flip between the legs, and the OR
    merge keeps any finding either leg surfaces. Trades higher API spend for
    a chance to catch findings a single sample missed.

    bool_key/list_key name the schema's flag-field and findings-array. The
    merge unions the two arrays (exact-dict dedup) and ORs the flag. Each leg
    falls back to sonnet (with retries) independently if its primary call fails —
    529s are common under load and a single None leg would otherwise drop
    one of the two samples on that case. Honors SECURITY_REVIEW_MODEL override
    for both calls without fallback.

    Gated by _dual_or_enabled() — off by default to avoid the
    2× API cost. When disabled, short-circuits to a single _call_claude
    and wraps the result in the same {bool_key, list_key} envelope so
    callers don't need to branch.
    """
    from concurrent.futures import ThreadPoolExecutor

    explicit = os.environ.get("SECURITY_REVIEW_MODEL", "").strip()
    primary = explicit or SECURITY_REVIEW_MODEL

    if not _dual_or_enabled():
        # Single-call path. Reuse the same sonnet-fallback retry as a dual_or
        # leg so a 529/400 on the primary doesn't drop recall to zero.
        r = _call_claude(prompt, output_schema, thinking_budget=thinking_budget,
                         max_tokens=max_tokens, model=primary, retry_5xx=False)
        if r is None and not explicit:
            debug_log(f"single: {primary} failed, falling back to sonnet")
            r = _call_claude(prompt, output_schema, thinking_budget=thinking_budget,
                             max_tokens=max_tokens, model="claude-sonnet-4-6",
                             retry_5xx=True)
        return r

    def _leg():
        r = _call_claude(prompt, output_schema, thinking_budget=thinking_budget,
                         max_tokens=max_tokens, model=primary, retry_5xx=False)
        if r is None and not explicit:
            debug_log(f"dual_or: {primary} leg failed, falling back to sonnet")
            r = _call_claude(prompt, output_schema, thinking_budget=thinking_budget,
                             max_tokens=max_tokens, model="claude-sonnet-4-6",
                             retry_5xx=True)
        return r

    with ThreadPoolExecutor(max_workers=2) as ex:
        fa = ex.submit(_leg)
        fb = ex.submit(_leg)
        ra, rb = fa.result(), fb.result()

    if ra is None and rb is None:
        return None

    a_list = (ra or {}).get(list_key) or []
    b_list = (rb or {}).get(list_key) or []