Function run_gates_concurrent

examples/policy-advisor/ctf.py:388–480

Run multiple gates concurrently, retrying until all pass. Returns the number of gates that passed.

(gates: list[dict])

def run_gates_concurrent(gates: list[dict]) -> int:
    """Run multiple gates concurrently, retrying until all pass.

    Returns the number of gates that passed.
    """
    nums = ", ".join(str(g["num"]) for g in gates)
    total = len(GATES)

    print(f"  {CYAN}{BOLD}{'=' * 60}{RESET}")
    print(f"  {CYAN}{BOLD}  GATES {nums} (CONCURRENT):{RESET}")
    for g in gates:
        proto = "https" if g["port"] == 443 else "http"
        print(
            f"  {DIM}  Gate {g['num']}: {g['name']}  "
            f"{proto}://{g['host']}:{g['port']}{RESET}"
        )
    print(f"  {CYAN}{BOLD}{'=' * 60}{RESET}")
    print()

    log(
        "INFO",
        f"Firing {len(gates)} requests concurrently -- approve them all at once",
    )
    log("INFO", "Tip: press [A] in the TUI draft panel to approve all pending")
    print()

    remaining_gates = list(gates)

    for attempt in range(1, MAX_RETRIES + 1):
        log(
            "GATE",
            f"Concurrent attempt #{attempt}",
            pending=len(remaining_gates),
        )

        # Fire all remaining gates concurrently.
        results: dict[int, tuple[str, str]] = {}
        threads: list[threading.Thread] = []

        def _attempt(g: dict) -> None:
            results[g["num"]] = attempt_gate(g)

        for g in remaining_gates:
            t = threading.Thread(target=_attempt, args=(g,))
            threads.append(t)
            t.start()

        for t in threads:
            t.join(timeout=30)

        # Process results.
        still_blocked = []
        for g in remaining_gates:
            num = g["num"]
            status, result = results.get(num, ("blocked", "no result"))

            if status == "pass":
                log("PASS", f"Gate {num}/{total} UNLOCKED ({g['name']})")