ed
1577cca568
The previous exclusion list had 'gemini_native' which is
NOT a real function name in src/ai_client.py. The actual
function is _send_gemini_cli (already migrated to
run_with_tool_loop via send_func + on_pre_dispatch in
commit 4748d134).
The current deferred vendors are now correctly:
- anthropic (uses anthropic SDK)
- gemini (uses google-genai streaming)
- deepseek (uses requests.post)
These will be addressed in Phase 5 t5_6/7/8. When those
ship, the DEFERRED_VENDORS frozenset should be emptied
so the audit gates the migration.
Verified: script still passes; gemini_cli's run_with_tool_loop
usage is detected correctly.
49 lines
1.9 KiB
Python
49 lines
1.9 KiB
Python
"""Audit: fail if any _send_<vendor> in src/ai_client.py contains an inline
|
|
tool-call loop (i.e., a for loop with MAX_TOOL_ROUNDS in it).
|
|
|
|
The follow-up track's invariant: all tool loops should go through
|
|
run_with_tool_loop. Inline loops are forbidden EXCEPT for the 3
|
|
vendored-call-path vendors (anthropic, gemini, deepseek) which use
|
|
their own SDKs and are tracked as deferred work (Phase 5 t5_6/7/8
|
|
in state.toml).
|
|
|
|
Note: gemini_cli was migrated to run_with_tool_loop via send_func
|
|
in commit 4748d134. The previous exclusion list incorrectly
|
|
included 'gemini_native' (a non-existent function name); that was
|
|
removed on 2026-06-11.
|
|
|
|
Usage: uv run python scripts/audit_no_inline_tool_loops.py
|
|
Exit code: 0 = pass; 1 = violations found.
|
|
"""
|
|
import re
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
TARGET = Path("src/ai_client.py")
|
|
DEFERRED_VENDORS = frozenset(["anthropic", "gemini", "deepseek"])
|
|
|
|
def main() -> int:
|
|
text = TARGET.read_text(encoding="utf-8")
|
|
violations: list[str] = []
|
|
for match in re.finditer(r"^def (_send_\w+)\(", text, re.MULTILINE):
|
|
func_name: str = match.group(1)
|
|
vendor = func_name[len("_send_"):]
|
|
if vendor in DEFERRED_VENDORS:
|
|
continue
|
|
func_start = match.start()
|
|
next_def = re.search(r"\n(?:def|async def) _send_\w+\(", text[func_start + 1:])
|
|
func_end = func_start + 1 + (next_def.start() if next_def else len(text) - func_start - 1)
|
|
func_body = text[func_start:func_end]
|
|
if "for _round_idx in range(MAX_TOOL_ROUNDS" in func_body or "for round_idx in range(MAX_TOOL_ROUNDS" in func_body:
|
|
if "run_with_tool_loop" not in func_body:
|
|
violations.append(vendor)
|
|
if violations:
|
|
print(f"FAIL: {len(violations)} vendor(s) have inline tool loops: {violations}")
|
|
print("Use src.ai_client.run_with_tool_loop instead.")
|
|
return 1
|
|
print("OK: all _send_<vendor> functions use run_with_tool_loop (deferred vendors excluded)")
|
|
return 0
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|