[BUG] Race condition in security-guidance's _agentic_review_with_race causes redundant security analysis

Preflight Checklist

• [x] I have searched existing issues and this hasn't been reported yet
• [x] This is a single bug report (please file separate reports for different bugs)
• [x] I am using the latest version of Claude Code

What's Wrong?

In plugins/security-guidance/hooks/security_reminder_hook.py, the function _agentic_review_with_race runs two threads in a race: an agentic reviewer thread (_agentic) and a delayed fallback reviewer thread (_fallback). They communicate their completion using a shared queue q of size 1.

If the agentic thread finishes first, it puts its result into q. The main thread immediately retrieves this result using q.get(), which empties the queue, and then returns.

When the fallback thread wakes up after its delay (delay_s), it checks if the agentic thread finished by calling if not q.empty(): return. Because the main thread already consumed the item from q, the queue is empty. The fallback thread incorrectly assumes the agentic thread hasn't finished, so it proceeds to run the expensive analyze_code_security function anyway. This results in redundant LLM/security analysis calls and wasted resources — silently, with no error or indication anything went wrong.

What Should Happen?

The fallback thread should correctly detect that the agentic reviewer already finished (via a dedicated completion signal, not queue emptiness) and skip re-running analyze_code_security entirely when the agentic reviewer already succeeded within the delay window.

Error Messages/Logs

None — this is a silent correctness/efficiency bug, not a crash. That's part of what makes it easy to miss: analyze_code_security simply runs twice with no warning.

Steps to Reproduce

This was found via static code review, not live reproduction — no special setup is needed to see the bug, just reading the logic:

1. Open plugins/security-guidance/hooks/security_reminder_hook.py.
2. Locate _agentic_review_with_race. Note q = queue.Queue(maxsize=1).
3. Trace what happens when _agentic finishes before the delay_s timeout: it does q.put_nowait(("agentic", r)), and the main thread's winner, (g, v, m) = q.get() immediately drains that item.
4. Trace _fallback(): after time.sleep(delay_s), it checks if not q.empty(): return. Since the queue was already drained in step 3, this check is always False in this scenario — _fallback proceeds to call analyze_code_security even though _agentic already completed successfully.

To confirm at runtime: add a log line at the start of _fallback()'s analyze_code_security call, make a commit where the agentic reviewer typically finishes well within SG_AGENTIC_RACE_DELAY_S (default 180s), and observe the fallback's expensive path still fires.

Claude Model

None

Is this a regression?

I don't know

Last Working Version

_No response_

Claude Code Version

N/A — found via static code review of the plugin source (plugins/security-guidance/hooks/security_reminder_hook.py on the main branch), not a live reproduction tied to a specific runtime/OS/terminal combination. The bug is in pure Python thread-synchronization logic, independent of platform.

Platform

Anthropic API

Operating System

macOS

Terminal/Shell

PyCharm terminal

Additional Information

Suggested fix — use a dedicated threading.Event instead of queue emptiness as the completion signal:

def _agentic_review_with_race(
repo_root: str,
diff_files: List[Tuple[str, str]],
rel_touched: List[str],
previous_findings: List[Dict[str, Any]],
) -> Tuple[Optional[str], List[Dict[str, Any]], Dict[str, Any]]:
"""Race the agentic reviewer against a delayed single-shot fallback."""
import queue as _queue
import threading as _th
import time as _t

if os.environ.get("SG_AGENTIC_NO_RACE") == "1":
return agentic_review(repo_root, diff_files, rel_touched)

delay_s = int(os.environ.get("SG_AGENTIC_RACE_DELAY_S", "180"))
q: "_queue.Queue[Tuple[str, Any]]" = _queue.Queue(maxsize=1)
fallback_started = _th.Event()
agentic_finished = _th.Event()

def _agentic() -> None:
try:
r = agentic_review(repo_root, diff_files, rel_touched)
except Exception as e:
r = (None, [], {"agentic_fallback": f"race_crash:{type(e).__name__}"})
try:
q.put_nowait(("agentic", r))
except _queue.Full:
pass
finally:
agentic_finished.set()

def _fallback() -> None:
_t.sleep(delay_s)
if agentic_finished.is_set():
return # agentic finished within the delay — never start fallback
fallback_started.set()
try:
g, v = analyze_code_security(
diff_files, is_diff=True, previous_findings=previous_findings
)
except Exception as e:
g, v = None, []
try:
q.put_nowait(("fallback", (g, v, {"agentic": False})))
except _queue.Full:
pass

_th.Thread(target=_agentic, daemon=True).start()
_th.Thread(target=_fallback, daemon=True).start()

winner, (g, v, m) = q.get()
m = dict(m)
m["race_winner"] = 1 if winner == "agentic" else 2
m["race_delay_s"] = delay_s
m["race_started"] = 1 if fallback_started.is_set() else 0
return g, v, m

Confidence: high — directly visible in the thread-synchronization logic; confirmed by tracing queue state through the code, no runtime logs needed.

Found via an AI code-investigation tool (Meeba Brain) while testing its bug-discovery accuracy on real open-source repos — verified by hand against the source before filing.