Two-stage Agent-SDK review: investigate (Read/Grep/Glob over the repo) then a self-refute filter pass. Returns (guidance_or_None, vulns, metrics). On SDK unavailability returns (None, [], {"agentic_fallback": reason}) so the caller can fall back to the single-shot path.
(
repo_dir: str, diff_files: List[Tuple[str, str]], touched_paths: List[str],
)
| 1057 | |
| 1058 | |
| 1059 | def agentic_review( |
| 1060 | repo_dir: str, diff_files: List[Tuple[str, str]], touched_paths: List[str], |
| 1061 | ) -> Tuple[Optional[str], List[Dict[str, Any]], Dict[str, Any]]: |
| 1062 | """Two-stage Agent-SDK review: investigate (Read/Grep/Glob over the repo) |
| 1063 | then a self-refute filter pass. Returns (guidance_or_None, vulns, |
| 1064 | metrics). On SDK unavailability returns (None, [], {"agentic_fallback": |
| 1065 | reason}) so the caller can fall back to the single-shot path.""" |
| 1066 | import time as _t |
| 1067 | |
| 1068 | # Note: do NOT pop ANTHROPIC_AUTH_TOKEN from os.environ here. The race |
| 1069 | # wrapper runs agentic_review() in a thread alongside the single-shot |
| 1070 | # fallback, and os.environ is process-global; mutating it from one thread |
| 1071 | # is a footgun for any future call-time reader. The OAuth-token leak into |
| 1072 | # the SDK spawn is handled per-spawn via opts.env={"ANTHROPIC_AUTH_TOKEN": |
| 1073 | # ""} in _arun() — the SDK applies opts.env after os.environ, so the empty |
| 1074 | # value wins without touching process-global state. |
| 1075 | |
| 1076 | metrics: Dict[str, Any] = {"agentic": True} |
| 1077 | try: |
| 1078 | import asyncio as _asyncio |
| 1079 | |
| 1080 | from claude_agent_sdk import ( |
| 1081 | AssistantMessage, |
| 1082 | ClaudeAgentOptions, |
| 1083 | ResultMessage, |
| 1084 | query, |
| 1085 | ) |
| 1086 | except Exception: |
| 1087 | # Some users don't have claude_agent_sdk in their system python. |
| 1088 | # The SessionStart hook (ensure_agent_sdk.py) creates a venv under |
| 1089 | # ~/.claude/security/ with the SDK installed; try that as a fallback |
| 1090 | # before giving up. The system import is attempted first so users |
| 1091 | # who DO have it never touch the venv. |
| 1092 | _venv_tried = False |
| 1093 | _state_dir = os.environ.get( |
| 1094 | "SECURITY_WARNINGS_STATE_DIR", |
| 1095 | os.path.expanduser("~/.claude/security"), |
| 1096 | ) |
| 1097 | for _sp in glob.glob( |
| 1098 | os.path.join(_state_dir, "agent-sdk-venv", "lib", |
| 1099 | "python*", "site-packages") |
| 1100 | ): |
| 1101 | if os.path.isdir(_sp) and _sp not in sys.path: |
| 1102 | sys.path.insert(0, _sp) |
| 1103 | _venv_tried = True |
| 1104 | try: |
| 1105 | import asyncio as _asyncio # noqa: F811 |
| 1106 | |
| 1107 | from claude_agent_sdk import ( # noqa: F811 |
| 1108 | AssistantMessage, |
| 1109 | ClaudeAgentOptions, |
| 1110 | ResultMessage, |
| 1111 | query, |
| 1112 | ) |
| 1113 | if _venv_tried: |
| 1114 | metrics["sdk_from_venv"] = True |
| 1115 | except Exception as e: # ImportError or transitive failure |
| 1116 | debug_log(f"agentic_review: SDK unavailable ({e}); falling back") |
no test coverage detected