From 7de140aeb7e9a566379592f48247939f08088b49 Mon Sep 17 00:00:00 2001 From: Marcello Alarcon Date: Tue, 23 Jun 2026 13:08:25 -0300 Subject: [PATCH] feat(skills): add int-gemini skill (Google Gemini API offload) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New integration skill to offload cheap/high-volume tasks from the main models to the Google Gemini API (AI Studio), via GEMINI_API_KEY. - Commands: ask (with --allow-sensitive, --fallback, --system-file, --grounding), scan, route, models, smoke, --version. - Use cases: classification, summarization, first drafts, and web research with Google Search grounding — anywhere a large/cheap model fits. - Confidentiality guard: a PARTIAL guard-rail (documented honestly as such) that blocks structured PII by default (CPF/CNPJ/card/CEP/email/BRL/phone/ address/PIX/contract/secret) and best-effort name+context. Free-text names without a role term are a known false-negative, documented + regression- tested; --allow-sensitive is the conscious override. - Robust HTTP: retry on 429/5xx, --fallback (flash-lite 503 -> flash) that never masks a real error, type-safe responses. - 105 tests, network fully blocked in the suite. smoke is exit-0 + JSON {overall, steps[], duration_ms}. Stdlib-only; instance/auth via env. Co-Authored-By: Claude Opus 4.8 --- .claude/skills/int-gemini/SKILL.md | 162 +++++ .../int-gemini/scripts/gemini_client.py | 555 ++++++++++++++++++ .claude/skills/int-gemini/tests/__init__.py | 0 .claude/skills/int-gemini/tests/conftest.py | 76 +++ .claude/skills/int-gemini/tests/helpers.py | 88 +++ .../int-gemini/tests/test_gemini_client.py | 555 ++++++++++++++++++ 6 files changed, 1436 insertions(+) create mode 100644 .claude/skills/int-gemini/SKILL.md create mode 100755 .claude/skills/int-gemini/scripts/gemini_client.py create mode 100644 .claude/skills/int-gemini/tests/__init__.py create mode 100644 .claude/skills/int-gemini/tests/conftest.py create mode 100644 .claude/skills/int-gemini/tests/helpers.py create mode 100644 .claude/skills/int-gemini/tests/test_gemini_client.py diff --git a/.claude/skills/int-gemini/SKILL.md b/.claude/skills/int-gemini/SKILL.md new file mode 100644 index 00000000..f88e983a --- /dev/null +++ b/.claude/skills/int-gemini/SKILL.md @@ -0,0 +1,162 @@ +--- +name: int-gemini +description: Integração com a Google Gemini API (AI Studio) via GEMINI_API_KEY. Use para OFFLOAD de tarefas baratas/volumosas dos modelos Claude/Opus — classificação de transcripts, sumarização, 1º rascunho de conteúdo, e pesquisa web com Google Search grounding. Comandos ask/models/smoke, saída JSON estruturada. Aciona quando o usuário menciona Gemini, "rodar no Gemini", offload de tarefa barata, grounding/pesquisa web via Google, ou distribuir consumo entre modelos. +--- + +# int-gemini — Google Gemini API (AI Studio) + +Camada B (MTA-only). Wrapper CLI stdlib-only para a Generative Language API, pensado +para **distribuir consumo** dos agentes: tirar do Claude/Opus as tarefas baratas e +volumosas e mandar pro Gemini free/low-cost. + +## Setup + +```bash +# .env (raiz do repo) +GEMINI_API_KEY=... # AI Studio (https://aistudio.google.com/apikey). Conta MTA. +GEMINI_MODEL=gemini-2.5-flash # opcional — default do `ask` +GEMINI_API_BASE=... # opcional — override do endpoint +``` + +A chave vem do **AI Studio** (free tier). O **Google One / AI Premium** (app de chat) +**NÃO dá quota de API** — é assinatura do app gemini.google.com, canal separado. + +## Comandos + +```bash +P=.claude/skills/int-gemini/scripts/gemini_client.py + +# Gerar resposta (default gemini-2.5-flash) +python3 $P ask "Classifique o sentimento: 'adorei o atendimento'" + +# Escolher modelo + system instruction + forçar JSON do modelo +python3 $P ask "Extraia {nome, intencao} do texto: ..." \ + --model gemini-2.5-flash-lite --system "Você é um classificador. Responda só JSON." --json + +# Pesquisa web com Google Search grounding (retorna sources[]) +python3 $P ask "Quais as novidades do RD Station em 2026?" --grounding + +# Override do guard de confidencialidade (uso CONSCIENTE — ver seção abaixo) +python3 $P ask "..." --allow-sensitive + +# Fallback p/ flash se o modelo falhar (ex.: 503 do flash-lite) +python3 $P ask "Classifique ..." --model gemini-2.5-flash-lite --fallback + +# System instruction de um arquivo (instruções longas reutilizáveis) +python3 $P ask "..." --system-file path/to/system.txt + +python3 $P --version + +# Qual modelo Gemini usar p/ um tipo de tarefa (helper de roteamento) +python3 $P route classify # → gemini-2.5-flash-lite +python3 $P route summarize # → gemini-2.5-flash + +# Varre um texto pelo guard de confidencialidade (sem chamar a API) +python3 $P scan "honorários de contrato R$ 1.000, CPF 123.456.789-00" + +# Listar modelos acessíveis pela chave +python3 $P models + +# Health-check (sempre exit 0, JSON {overall, steps[], duration_ms}) +python3 $P smoke +``` + +| Comando | O que faz | Exit codes | +|---|---|---| +| `ask ` | Gera resposta. Passa pelo guard de confidencialidade primeiro. | 0=ok, 1=erro API, **2=bloqueado pelo guard** | +| `route ` | Modelo recomendado p/ `classify/triage/extract/summarize/draft/translate/research/reason` | 0 | +| `scan ` | Só o guard (sem rede) — `{safe, confidential_findings[], detected_types[]}` | 0 | +| `models` | Lista modelos `generateContent` da chave | 0 / 1 | +| `smoke` | Health-check 4 steps (env, generate, guard, models) | **sempre 0** | + +**Todos os comandos** emitem JSON em stdout. Saída do `ask`: `{ok, model, text, grounded, +sources[], usage{prompt/candidates/total_tokens}, finish_reason}` — `model` é o modelo que +de fato respondeu (com `--fallback`, pode diferir do `--model` pedido). +Logging estruturado vai pra **stderr**; o JSON de resultado vai pra **stdout** — pipeável. + +**Precedência de modelo:** `--model` (flag) > `GEMINI_MODEL` (env) > default `gemini-2.5-flash`. +O `route` é só um guia — não sobrepõe; o caller passa o resultado pro `--model`. + +## Modelos (verificado ao vivo 2026-06-23 — 37 modelos `generateContent`) + +| Modelo | Janela in | out | Uso recomendado | +|---|---|---|---| +| `gemini-2.5-flash` (default) | 1.048.576 | 65.536 | Sumarização, 1º rascunho, grounding | +| `gemini-2.5-flash-lite` | 1.048.576 | 65.536 | Classificação em volume (mais barato/rápido) | +| `gemini-2.5-pro` | 1.048.576 | 65.536 | Raciocínio (free tier removido abr/2026 → pago) | +| `gemini-3-flash-preview` / `gemini-3.5-flash` | 1.048.576 | 65.536 | Geração mais nova (preview) | + +A chave MTA acessa também a família `gemini-3.x` (preview) e modelos de imagem/TTS. +Rode `models` para a lista viva. + +## Quando usar Gemini vs Claude + +| Tarefa | Modelo | Por quê | +|---|---|---| +| Classificação/triagem em volume (transcripts, leads, e-mails) | **Gemini Flash-Lite** | Barato, rápido, free tier cobre o volume | +| Sumarização de reunião (transcrição de reunião) / texto longo | **Gemini Flash** | Janela 1M tokens, custo baixo | +| 1º rascunho de conteúdo (depois Claude refina) | **Gemini Flash** | Tira o trabalho bruto do Opus | +| Pesquisa web factual | **Gemini Flash --grounding** | Google Search nativo + sources | +| Arquitetura, estratégia, código de produção, decisão de risco | **Claude/Opus** | Qualidade de raciocínio + contexto MTA | +| Qualquer dado **confidencial de cliente** (sem redação) | **Claude** (ou tier no-train) | Free tier do Gemini **treina** com os prompts; o guard é PARCIAL | + +## Confidencialidade (CRÍTICO) — guard-rail PARCIAL, não garantia + +O **free tier do Gemini usa prompts+respostas para treino do modelo**. Por isso o `ask` +roda um guard ANTES de chamar a API e **bloqueia por padrão** (exit 2) ao detectar dado +sensível. **Mas o guard é uma rede de segurança PARCIAL — não confie como garantia.** + +**Pega de forma CONFIÁVEL (dado estruturado):** CPF, CNPJ, cartão, CEP, e-mail, valor em +R$, telefone (incl. nono dígito separado e fixo de 8), endereço com logradouro, PIX, +termos de contrato (cláusula/honorários/NDA), segredo (api_key/token/senha). + +**Pega só em BEST-EFFORT (gera falso-negativo — NÃO confie):** nome de pessoa (só perto +de "lead/cliente/diretor/sócio…"), endereço sem logradouro nomeado, telefone falado por +extenso, primeiro nome isolado. Ex.: `"João Silva ligou ontem"` (nome solto) **passa**. +O corpus de falso-negativo em `tests/` documenta esses limites explicitamente. + +**Por isso:** tarefas que carregam dado de cliente **por natureza** (resumo de transcrição de reunião, +ficha de lead, e-mail de cliente) → **redija/anonimize antes**, use tier no-train, OU +mantenha no Claude. **Não jogue o texto cru contra o guard esperando que ele proteja.** + +- **Bloqueado → o que fazer:** redigir (substituir nome/telefone/endereço por placeholder), + OU `--allow-sensitive` se você conscientemente avaliou que é seguro (ex.: o "R$" é + genérico de marketing, não financeiro de cliente), OU manter no Claude. +- **Falso-positivo é de propósito:** o guard prefere bloquear (um ID de 10 dígitos vira + "telefone") — o `--allow-sensitive` é o override consciente. Use `scan ` para + inspecionar sem chamar a API. + +## Helper de roteamento (qual modelo) + +`route ` centraliza a escolha do modelo Gemini quando o offload já foi decidido: + +| Tipo de tarefa | Modelo | Por quê | +|---|---|---| +| `classify` / `triage` / `extract` | `gemini-2.5-flash-lite` | Volume — mais barato/rápido | +| `summarize` / `draft` / `translate` / `research` | `gemini-2.5-flash` | Janela 1M, custo baixo | +| `reason` | `gemini-2.5-pro` | Raciocínio (pago no free desde abr/2026) | +| (desconhecido) | `gemini-2.5-flash` | Fallback seguro | + +> **Achado do piloto (2026-06-23):** o `gemini-2.5-flash-lite` pode devolver **HTTP 503 +> "high demand"** transitório — tenha `gemini-2.5-flash` como fallback. Triagem em lote +> cabe num **único request** (24 tarefas → 1 chamada, ~6,6k tokens) — economiza RPD. + +## Limites & confidencialidade — ver análise + +Estudo de limites reais + matemática de volume + piloto recomendado em +`workspace/strategy/[C]gemini-offload-analise-2026-06-23.md`. + +## Notas técnicas + +- Stdlib only (urllib) — sem dependências. `.env` loader igual às outras skills int-*. +- Retry: 1x após backoff 20s em 429/5xx/timeout. +- Type-safe: `_request` sempre retorna `dict` (nunca bytes/str crua). +- `smoke` é gracioso sem a chave (FAIL no step `env`, exit 0) e valida o guard offline. +- **Testes:** `python3 -m pytest .claude/skills/int-gemini/tests/` — **105 testes** + 100% mockados (conftest bloqueia rede real em autouse; nenhum teste gasta quota). + Inclui corpus de **falso-negativo documentado** do guard (limites explícitos). +- Piloto de referência (offload real, dry-run): `workspace/_scripts/_misc/gemini-pilot/`. +- **Auto-melhoria (2026-06-23):** `--fallback`, `--system-file`, `--version`, WARN no + `route` desconhecido e doc de precedência vieram de uma auto-crítica rodada pela própria + `int-gemini ask` (Gemini Flash). Rejeitadas conscientemente: invocação nativa (convenção + MTA usa `python3 $P`) e `smoke` exit≠0 (contrato MTA: smoke é SEMPRE exit 0). diff --git a/.claude/skills/int-gemini/scripts/gemini_client.py b/.claude/skills/int-gemini/scripts/gemini_client.py new file mode 100755 index 00000000..d8de7746 --- /dev/null +++ b/.claude/skills/int-gemini/scripts/gemini_client.py @@ -0,0 +1,555 @@ +#!/usr/bin/env python3 +"""int-gemini — wrapper CLI para a Google Gemini API (AI Studio). + +Chama a Generative Language API (https://generativelanguage.googleapis.com) via +`GEMINI_API_KEY`. Pensado para OFFLOAD de tarefas baratas/volumosas dos modelos +Claude/Opus: classificação, sumarização, 1º rascunho, pesquisa web com grounding. + +Comandos: + ask "" [--model ] [--system ""] [--json] [--grounding] + Gera resposta. --json liga responseMimeType application/json. + --grounding liga Google Search grounding (quando suportado). + models Lista modelos acessíveis pela chave (generateContent). + smoke Health-check: exit 0 SEMPRE + JSON {overall, steps[], duration_ms}. + +Modelos default e alternativas (free tier AI Studio): + gemini-2.5-flash (default — bom custo/qualidade) + gemini-2.5-flash-lite (mais barato/rápido — classificação em volume) + gemini-2.5-pro (raciocínio — janela 1M, RPD baixo no free) + +Sem dependências externas — só stdlib (urllib). Chave sempre via env +`GEMINI_API_KEY` (com fallback do .env loader). Nunca hardcode segredo. + +Retry: 1 retry após backoff em 429 / 5xx / timeout (custom-integration-retry). +Logging estruturado em stderr. Saída em stdout sempre JSON estruturado. +""" +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +import time +import urllib.error +import urllib.request +from pathlib import Path + + +# ── env ────────────────────────────────────────────────────────────────────── + +def _load_dotenv() -> None: + env_path = Path(__file__).resolve().parents[4] / ".env" + if not env_path.exists(): + return + with open(env_path) as f: + for line in f: + line = line.strip() + if not line or line.startswith("#") or "=" not in line: + continue + k, _, v = line.partition("=") + if k.strip() and k.strip() not in os.environ: + os.environ[k.strip()] = v.strip() + + +_load_dotenv() + +API_BASE = os.environ.get( + "GEMINI_API_BASE", "https://generativelanguage.googleapis.com" +).rstrip("/") +API_KEY = os.environ.get("GEMINI_API_KEY", "") +DEFAULT_MODEL = os.environ.get("GEMINI_MODEL", "gemini-2.5-flash") +VERSION = "1.1.0" +UA = f"evo-nexus-int-gemini/{VERSION}" +TIMEOUT = 60 +RETRY_AFTER_S = 20 +RETRYABLE_CODES = {429, 500, 502, 503, 504} + + +# ── logging ────────────────────────────────────────────────────────────────── + +def _log(level: str, msg: str) -> None: + print(f"[int-gemini] {level}: {msg}", file=sys.stderr) + + +# ── guard de confidencialidade ───────────────────────────────────────────────── +# O free tier do Gemini TREINA o modelo com prompts+respostas. Antes de mandar +# um prompt, varremos heurísticas e, por padrão, BLOQUEAMOS. `--allow-sensitive` +# é o override consciente do operador. +# +# IMPORTANTE — guard-rail PARCIAL, não garantia: +# • CONFIÁVEL p/ dado ESTRUTURADO: CPF, CNPJ, cartão, CEP, e-mail, R$. +# • BEST-EFFORT (gera falso-negativo): NOME de pessoa, ENDEREÇO, telefone falado, +# valor por extenso ("cinquenta mil"). Heurística não substitui julgamento. +# Tarefas que carregam dado de cliente por natureza (resumo de transcrição de reunião, ficha de +# lead) DEVEM ser redigidas/anonimizadas ou ir p/ tier no-train — NÃO confie só +# no guard. Aceitamos falso-positivo de propósito (o --allow-sensitive resolve); +# o que perseguimos é reduzir o falso-NEGATIVO nos vetores reais. + +# A ordem importa: padrões mais específicos primeiro (cartão/CEP antes de telefone) +# para que o de-dup por sobreposição não rotule um cartão como "telefone". +_CONFIDENTIAL_PATTERNS: list[tuple[str, str]] = [ + # CPF: 000.000.000-00 ou 11 dígitos isolados + ("cpf", r"\b\d{3}\.?\d{3}\.?\d{3}-?\d{2}\b"), + # CNPJ: 00.000.000/0000-00 + ("cnpj", r"\b\d{2}\.?\d{3}\.?\d{3}/?\d{4}-?\d{2}\b"), + # E-mail + ("email", r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"), + # CEP: 00000-000 ou 00000000 (8 dígitos) + ("cep", r"(? mil/milhão/milhões/bilhão" (best-effort) + ("valor_extenso", + r"\b(?:\d{1,3}|um|dois|tr[êe]s|quatro|cinco|seis|sete|oito|nove|dez|vinte|" + r"trinta|quarenta|cinquenta|cem|cento|duzentos|trezentos|quinhentos|mil)\s+" + r"(?:mil|milh[õo]es?|bilh[õo]es?)\b"), + # PIX explícito + ("pix", r"\bpix\b"), + # Termos de contrato/confidencialidade + ("contrato", r"\b(?:contrato|confidencial|sigiloso|cl[áa]usula|nda|honor[áa]rios?)\b"), + # Chave de API / segredo (evita vazar segredo no prompt) + ("segredo", r"\b(?:api[_-]?key|secret|token|password|senha|bearer)\b"), +] + +_CONFIDENTIAL_RE = [(name, re.compile(pat, re.IGNORECASE)) + for name, pat in _CONFIDENTIAL_PATTERNS] + +# Nome de pessoa perto de contexto de cliente/lead. Best-effort: sequência de 2+ +# palavras Capitalizadas (com de/da/dos opcional) a até ~40 chars de um termo de +# papel. Pega "lead João Silva", "Maria Souza, Diretora", "sócio João Silva". +_ROLE_CONTEXT = (r"lead|leads|cliente|clientes|contato|contatos|prospect|prospects|" + r"diretor[ae]?|s[óo]ci[oa]|propriet[áa]ri[oa]|gerente|respons[áa]vel|" + r"corretor[ae]?|comprador[ae]?|titular|sr\.?|sra\.?|dr\.?|dra\.?") +_FULLNAME = r"[A-ZÀ-Ý][a-zà-ÿ]{1,}(?:\s+(?:d[aeo]s?\s+)?[A-ZÀ-Ý][a-zà-ÿ]{1,}){1,3}" +_NAME_CONTEXT_RE = re.compile( + rf"(?:(?:{_ROLE_CONTEXT})[\s:,\-]+(?P{_FULLNAME}))" + rf"|(?:(?P{_FULLNAME})[\s,\-]+(?:{_ROLE_CONTEXT})\b)", + re.IGNORECASE, +) + + +def scan_confidential(text: str) -> list[dict]: + """Varre o texto e devolve achados de confidencialidade [{type, sample}]. + + Heurística — guard-rail PARCIAL (ver nota no topo da seção). Falsos-positivos + são esperados e aceitos (o operador resolve com `--allow-sensitive`); o foco é + minimizar o falso-NEGATIVO. Lista vazia = nada detectado. + + De-dup por sobreposição: um mesmo trecho não vira dois tipos (ex.: um cartão + não é também rotulado "telefone"). O primeiro padrão a casar vence o span. + """ + findings: list[dict] = [] + claimed: list[tuple[int, int]] = [] # spans já reivindicados por outro tipo + + def _overlaps(s: int, e: int) -> bool: + return any(s < ce and cs < e for cs, ce in claimed) + + for name, rx in _CONFIDENTIAL_RE: + for m in rx.finditer(text): + s, e = m.span() + if _overlaps(s, e): + continue + claimed.append((s, e)) + sample = m.group(0) + masked = sample[:3] + "…" if len(sample) > 4 else sample + findings.append({"type": name, "sample": masked}) + break # um achado por tipo basta p/ reportar + + # nome+contexto: padrão composto (grupo n1 OU n2), tratado à parte + nm = _NAME_CONTEXT_RE.search(text) + if nm: + name_str = nm.group("n1") or nm.group("n2") or "" + if name_str: + findings.append({"type": "nome_contexto", "sample": name_str[:3] + "…"}) + + return findings + + +# ── HTTP ───────────────────────────────────────────────────────────────────── + +class GeminiError(Exception): + """Erro de chamada à API Gemini (status + corpo).""" + + +def _request(method: str, path: str, *, body: dict | None = None, _attempt: int = 0) -> dict: + """Faz um request à Gemini API e devolve sempre um dict parseado. + + Levanta GeminiError em falha não-retryável. Type-safe: nunca retorna bytes + nem string crua — sempre dict. + """ + if not API_KEY: + raise GeminiError("GEMINI_API_KEY ausente no ambiente/.env") + + url = f"{API_BASE}{path}" + sep = "&" if "?" in url else "?" + url = f"{url}{sep}key={API_KEY}" + + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"User-Agent": UA, "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=TIMEOUT) as resp: + raw = resp.read().decode("utf-8", errors="replace") + except urllib.error.HTTPError as e: + raw = e.read().decode("utf-8", errors="replace") if e.fp else "" + if e.code in RETRYABLE_CODES and _attempt < 1: + _log("WARN", f"HTTP {e.code} em {path} — retry em {RETRY_AFTER_S}s") + time.sleep(RETRY_AFTER_S) + return _request(method, path, body=body, _attempt=_attempt + 1) + # extrai a mensagem da API se houver + detail = raw[:500] + try: + detail = json.loads(raw).get("error", {}).get("message", detail) + except (ValueError, AttributeError): + pass + raise GeminiError(f"HTTP {e.code} em {path}: {detail}") + except (urllib.error.URLError, TimeoutError) as e: + if _attempt < 1: + _log("WARN", f"conexão falhou em {path} ({e}) — retry em {RETRY_AFTER_S}s") + time.sleep(RETRY_AFTER_S) + return _request(method, path, body=body, _attempt=_attempt + 1) + raise GeminiError(f"conexão falhou em {path}: {e}") + + try: + parsed = json.loads(raw) + except ValueError: + raise GeminiError(f"resposta não-JSON em {path}: {raw[:200]}") + if not isinstance(parsed, dict): + return {"raw": parsed} + return parsed + + +# ── roteamento de modelo por tipo de tarefa ─────────────────────────────────── +# Mapeia um "tipo de tarefa" pro modelo Gemini certo. Centraliza a tabela +# "Quando usar Gemini vs Claude" do SKILL.md em código testável. NÃO decide +# Gemini-vs-Claude (isso é do agente) — só escolhe o modelo Gemini quando o +# offload já foi decidido. + +_ROUTING: dict[str, str] = { + # classificação/triagem em volume → mais barato/rápido + "classify": "gemini-2.5-flash-lite", + "triage": "gemini-2.5-flash-lite", + "extract": "gemini-2.5-flash-lite", + # sumarização / rascunho / texto longo → flash (janela 1M, custo baixo) + "summarize": "gemini-2.5-flash", + "draft": "gemini-2.5-flash", + "translate": "gemini-2.5-flash", + # pesquisa web factual → flash + grounding (grounding ligado pelo caller) + "research": "gemini-2.5-flash", + # raciocínio mais pesado dentro do que o offload cobre → pro (pago no free) + "reason": "gemini-2.5-pro", +} +DEFAULT_ROUTE_MODEL = "gemini-2.5-flash" + + +def route_model(task_type: str) -> str: + """Devolve o modelo Gemini recomendado p/ um tipo de tarefa (fallback flash).""" + return _ROUTING.get(task_type.strip().lower(), DEFAULT_ROUTE_MODEL) + + +# ── parsing de resposta ────────────────────────────────────────────────────── + +def _extract_text(resp: dict) -> str: + """Concatena as partes de texto do primeiro candidate.""" + candidates = resp.get("candidates") or [] + if not candidates: + return "" + parts = candidates[0].get("content", {}).get("parts", []) or [] + return "".join(p.get("text", "") for p in parts if isinstance(p, dict)) + + +def _extract_grounding(resp: dict) -> list[dict]: + """Extrai as fontes do groundingMetadata (quando --grounding usado).""" + candidates = resp.get("candidates") or [] + if not candidates: + return [] + gm = candidates[0].get("groundingMetadata") or {} + chunks = gm.get("groundingChunks") or [] + out = [] + for c in chunks: + web = c.get("web") or {} + if web.get("uri"): + out.append({"title": web.get("title", ""), "uri": web["uri"]}) + return out + + +# ── comandos ───────────────────────────────────────────────────────────────── + +def cmd_ask(args) -> int: + model = args.model or DEFAULT_MODEL + + # system instruction pode vir de --system OU --system-file (auto-melhoria #12) + system = args.system + if getattr(args, "system_file", None): + try: + system = Path(args.system_file).read_text(encoding="utf-8") + except OSError as e: + print(json.dumps({"ok": False, "error": f"system-file: {e}", "model": model})) + return 1 + + # guard de confidencialidade — bloqueia por padrão se detectar dado sensível + if not args.allow_sensitive: + findings = scan_confidential(args.prompt) + if system: + findings += scan_confidential(system) + if findings: + types = sorted({f["type"] for f in findings}) + print(json.dumps({ + "ok": False, + "error": "BLOQUEADO: dado potencialmente confidencial detectado " + "(free tier do Gemini treina com o prompt). Redija/anonimize " + "ou use --allow-sensitive para override consciente.", + "confidential_findings": findings, + "detected_types": types, + "model": model, + }, ensure_ascii=False, indent=2)) + _log("WARN", f"prompt bloqueado por guard: {', '.join(types)}") + return 2 + + payload: dict = { + "contents": [{"role": "user", "parts": [{"text": args.prompt}]}], + } + if system: + payload["systemInstruction"] = {"parts": [{"text": system}]} + gen_config: dict = {} + if args.json: + gen_config["responseMimeType"] = "application/json" + if gen_config: + payload["generationConfig"] = gen_config + if args.grounding: + # Tool de Google Search grounding (Gemini 2.x). + payload["tools"] = [{"google_search": {}}] + + # fallback transparente p/ flash quando o modelo escolhido falha (auto-melhoria #5) + # — útil pro flash-lite, que devolve 503 "high demand" no free tier. + tried = [model] + if getattr(args, "fallback", False) and model != DEFAULT_ROUTE_MODEL: + tried.append(DEFAULT_ROUTE_MODEL) + + resp = None + used_model = model + last_err: GeminiError | None = None + for m in tried: + try: + resp = _request("POST", f"/v1beta/models/{m}:generateContent", body=payload) + used_model = m + if m != model: + _log("WARN", f"fallback: {model} falhou, respondido por {m}") + break + except GeminiError as e: + last_err = e + if resp is None: + print(json.dumps({"ok": False, "error": str(last_err), "model": model}), file=sys.stdout) + _log("ERROR", str(last_err)) + return 1 + + text = _extract_text(resp) + usage = resp.get("usageMetadata", {}) + out = { + "ok": True, + "model": used_model, + "text": text, + "grounded": bool(args.grounding), + "sources": _extract_grounding(resp) if args.grounding else [], + "usage": { + "prompt_tokens": usage.get("promptTokenCount"), + "candidates_tokens": usage.get("candidatesTokenCount"), + "total_tokens": usage.get("totalTokenCount"), + }, + "finish_reason": (resp.get("candidates") or [{}])[0].get("finishReason"), + } + print(json.dumps(out, ensure_ascii=False, indent=2)) + return 0 + + +def cmd_models(_args) -> int: + try: + resp = _request("GET", "/v1beta/models") + except GeminiError as e: + print(json.dumps({"ok": False, "error": str(e)})) + _log("ERROR", str(e)) + return 1 + models = [] + for m in resp.get("models", []): + methods = m.get("supportedGenerationMethods", []) or [] + if "generateContent" in methods: + models.append({ + "name": m.get("name", "").replace("models/", ""), + "input_token_limit": m.get("inputTokenLimit"), + "output_token_limit": m.get("outputTokenLimit"), + }) + print(json.dumps({"ok": True, "count": len(models), "models": models}, indent=2)) + return 0 + + +def cmd_route(args) -> int: + """Imprime o modelo recomendado para um tipo de tarefa.""" + model = route_model(args.task_type) + known = args.task_type.strip().lower() in _ROUTING + if not known: + _log("WARN", f"tipo de tarefa desconhecido '{args.task_type}' — fallback {model}") + print(json.dumps({ + "ok": True, "task_type": args.task_type, "model": model, "known": known, + }, ensure_ascii=False)) + return 0 + + +def cmd_scan(args) -> int: + """Roda só o guard de confidencialidade num texto — sem chamar a API.""" + findings = scan_confidential(args.text) + print(json.dumps({ + "ok": True, "safe": not findings, + "confidential_findings": findings, + "detected_types": sorted({f["type"] for f in findings}), + }, ensure_ascii=False, indent=2)) + return 0 + + +def cmd_smoke(_args) -> int: + """Health-check — sempre exit 0, JSON estruturado. Gracioso sem chave.""" + t0 = time.time() + steps: list[dict] = [] + overall = "PASS" + + # step 1: env — chave presente? + ts = time.time() + if API_KEY: + steps.append({"step": "env", "status": "PASS", "duration_ms": round((time.time() - ts) * 1000)}) + else: + steps.append({ + "step": "env", "status": "FAIL", + "error": "GEMINI_API_KEY ausente", + "duration_ms": round((time.time() - ts) * 1000), + }) + overall = "FAIL" + + # step 2: chamada mínima (só se a chave existir) — não desperdiça quota à toa + ts = time.time() + if not API_KEY: + steps.append({"step": "generate", "status": "SKIP", "duration_ms": 0}) + else: + try: + resp = _request("POST", f"/v1beta/models/{DEFAULT_MODEL}:generateContent", body={ + "contents": [{"role": "user", "parts": [{"text": "ping"}]}], + "generationConfig": {"maxOutputTokens": 5}, + }) + if _extract_text(resp) or resp.get("candidates"): + steps.append({"step": "generate", "status": "PASS", "duration_ms": round((time.time() - ts) * 1000)}) + else: + steps.append({ + "step": "generate", "status": "FAIL", + "error": "resposta sem candidates", + "duration_ms": round((time.time() - ts) * 1000), + }) + overall = "FAIL" + except GeminiError as e: + steps.append({ + "step": "generate", "status": "FAIL", + "error": str(e)[:300], + "duration_ms": round((time.time() - ts) * 1000), + }) + overall = "FAIL" + + # step 3: guard de confidencialidade funcional (sem rede — sempre roda) + ts = time.time() + pos = scan_confidential("meu CPF é 123.456.789-00") + neg = scan_confidential("resumir transcript de reunião interna") + if pos and not neg: + steps.append({"step": "guard", "status": "PASS", "duration_ms": round((time.time() - ts) * 1000)}) + else: + steps.append({ + "step": "guard", "status": "FAIL", + "error": f"guard inconsistente (pos={bool(pos)} neg={bool(neg)})", + "duration_ms": round((time.time() - ts) * 1000), + }) + overall = "FAIL" + + # step 4: modelos esperados acessíveis pela chave (só com chave) + ts = time.time() + if not API_KEY: + steps.append({"step": "models", "status": "SKIP", "duration_ms": 0}) + else: + try: + resp = _request("GET", "/v1beta/models") + names = {m.get("name", "").replace("models/", "") + for m in resp.get("models", []) + if "generateContent" in (m.get("supportedGenerationMethods") or [])} + if DEFAULT_MODEL in names: + steps.append({ + "step": "models", "status": "PASS", + "found": len(names), "default_accessible": True, + "duration_ms": round((time.time() - ts) * 1000), + }) + else: + steps.append({ + "step": "models", "status": "FAIL", + "error": f"modelo default {DEFAULT_MODEL} não acessível pela chave", + "duration_ms": round((time.time() - ts) * 1000), + }) + overall = "FAIL" + except GeminiError as e: + steps.append({ + "step": "models", "status": "FAIL", + "error": str(e)[:300], + "duration_ms": round((time.time() - ts) * 1000), + }) + overall = "FAIL" + + print(json.dumps({"overall": overall, "steps": steps, "duration_ms": round((time.time() - t0) * 1000)}, indent=2)) + return 0 + + +def main() -> int: + p = argparse.ArgumentParser(prog="gemini_client.py", description=__doc__.split("\n")[0]) + p.add_argument("--version", action="version", version=f"int-gemini {VERSION}") + sub = p.add_subparsers(dest="command", required=True) + + a = sub.add_parser("ask", help="Gera resposta a partir de um prompt") + a.add_argument("prompt", help="Texto do prompt") + a.add_argument("--model", default=None, help=f"Modelo (default {DEFAULT_MODEL})") + a.add_argument("--system", default=None, help="System instruction") + a.add_argument("--system-file", default=None, help="System instruction de um arquivo") + a.add_argument("--json", action="store_true", + help="Pede ao MODELO p/ responder em JSON (responseMimeType)") + a.add_argument("--grounding", action="store_true", help="Liga Google Search grounding") + a.add_argument("--fallback", action="store_true", + help="Se o modelo falhar, tenta gemini-2.5-flash (ex.: 503 do flash-lite)") + a.add_argument("--allow-sensitive", action="store_true", + help="Override consciente do guard de confidencialidade") + a.set_defaults(func=cmd_ask) + + r = sub.add_parser("route", help="Modelo recomendado p/ um tipo de tarefa") + r.add_argument("task_type", help="classify|triage|extract|summarize|draft|translate|research|reason") + r.set_defaults(func=cmd_route) + + sc = sub.add_parser("scan", help="Roda só o guard de confidencialidade (sem rede)") + sc.add_argument("text", help="Texto a varrer") + sc.set_defaults(func=cmd_scan) + + sub.add_parser("models", help="Lista modelos acessíveis").set_defaults(func=cmd_models) + sub.add_parser("smoke", help="Health-check — sempre exit 0, JSON").set_defaults(func=cmd_smoke) + + args = p.parse_args() + return args.func(args) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.claude/skills/int-gemini/tests/__init__.py b/.claude/skills/int-gemini/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/.claude/skills/int-gemini/tests/conftest.py b/.claude/skills/int-gemini/tests/conftest.py new file mode 100644 index 00000000..9b857ac0 --- /dev/null +++ b/.claude/skills/int-gemini/tests/conftest.py @@ -0,0 +1,76 @@ +"""conftest.py — pytest config para int-gemini. + +Stuba urllib.request.urlopen em autouse para que NENHUM teste toque a rede real. +Padrão idêntico ao int-woocommerce / int-wordpress. +""" +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +from typing import Sequence + +import pytest + +_TESTS_DIR = Path(__file__).resolve().parent +if str(_TESTS_DIR) not in sys.path: + sys.path.insert(0, str(_TESTS_DIR)) + +_SKILL_ROOT = _TESTS_DIR.parent +_SCRIPT_PATH = _SKILL_ROOT / "scripts" / "gemini_client.py" + +# Garante que a chave existe no import (cmd não falha por env ausente nos testes +# que não testam o caminho "sem chave"). Os testes que querem testar a ausência +# manipulam gemini_client.API_KEY diretamente via monkeypatch. +import os # noqa: E402 + +os.environ.setdefault("GEMINI_API_KEY", "test-key-not-real") + +_spec = importlib.util.spec_from_file_location("gemini_client", _SCRIPT_PATH) +_gemini = importlib.util.module_from_spec(_spec) +_spec.loader.exec_module(_gemini) + +sys.modules["gemini_client"] = _gemini + + +@pytest.fixture(autouse=True) +def block_network(monkeypatch): + """Bloqueia toda rede real — qualquer tentativa levanta RuntimeError.""" + def _deny(*args, **kwargs): + raise RuntimeError( + "PROIBIDO: teste tentou abrir conexão de rede real. " + "Use o fixture mock_urlopen para configurar respostas stub." + ) + + monkeypatch.setattr("urllib.request.urlopen", _deny) + monkeypatch.setattr(_gemini.urllib.request, "urlopen", _deny) + # neutraliza o sleep do retry pra teste não travar 20s + monkeypatch.setattr(_gemini.time, "sleep", lambda *_a, **_k: None) + yield + + +@pytest.fixture() +def mock_urlopen(monkeypatch): + """Substitui urlopen por uma fila de respostas (ou exceções). + + Uso: + mock_urlopen([make_response({...}, 200)]) + """ + responses: list = [] + + def _fake_urlopen(req, timeout=60): + if not responses: + raise RuntimeError("mock_urlopen: fila de respostas esgotada") + item = responses.pop(0) + if isinstance(item, BaseException): + raise item + return item + + monkeypatch.setattr("urllib.request.urlopen", _fake_urlopen) + monkeypatch.setattr(_gemini.urllib.request, "urlopen", _fake_urlopen) + + def _setup(resp_list: Sequence): + responses.clear() + responses.extend(resp_list) + + return _setup diff --git a/.claude/skills/int-gemini/tests/helpers.py b/.claude/skills/int-gemini/tests/helpers.py new file mode 100644 index 00000000..15b98648 --- /dev/null +++ b/.claude/skills/int-gemini/tests/helpers.py @@ -0,0 +1,88 @@ +"""Helpers de teste para int-gemini — constroem respostas HTTP mock.""" +from __future__ import annotations + +import json +import urllib.error +from typing import Union + + +class MockHTTPResponse: + """Response simulada compatível com urllib.request.urlopen context manager.""" + + def __init__(self, body: Union[dict, list, str], status: int = 200): + if isinstance(body, str): + self._body = body.encode() + else: + self._body = json.dumps(body).encode() + self.status = status + + def read(self, n: int = -1) -> bytes: + return self._body + + def __enter__(self): + return self + + def __exit__(self, *args): + pass + + +def make_response(body: Union[dict, list, str], status: int = 200) -> MockHTTPResponse: + return MockHTTPResponse(body, status) + + +def make_generate_response(text: str, *, prompt_tokens: int = 10, + candidates_tokens: int = 5, total_tokens: int = 15, + finish: str = "STOP", + grounding_chunks: list | None = None) -> MockHTTPResponse: + """Resposta no formato generateContent da Gemini API.""" + candidate: dict = { + "content": {"parts": [{"text": text}]}, + "finishReason": finish, + } + if grounding_chunks is not None: + candidate["groundingMetadata"] = {"groundingChunks": grounding_chunks} + return make_response({ + "candidates": [candidate], + "usageMetadata": { + "promptTokenCount": prompt_tokens, + "candidatesTokenCount": candidates_tokens, + "totalTokenCount": total_tokens, + }, + }) + + +def make_models_response(names: list[str]) -> MockHTTPResponse: + return make_response({ + "models": [ + { + "name": f"models/{n}", + "inputTokenLimit": 1048576, + "outputTokenLimit": 65536, + "supportedGenerationMethods": ["generateContent"], + } + for n in names + ] + }) + + +class _MockBody: + """fp truthy com .read() — o _request usa `e.fp` pra decidir se lê o corpo.""" + + def __init__(self, raw: bytes): + self._raw = raw + + def read(self, *_a, **_k) -> bytes: + return self._raw + + def close(self) -> None: + pass + + +def make_http_error(status: int, message: str = "boom") -> urllib.error.HTTPError: + """HTTPError simulado para casos de falha (429/5xx/etc.).""" + body = json.dumps({"error": {"message": message}}).encode() + err = urllib.error.HTTPError( + url="http://test/x", code=status, msg=message, + hdrs={}, fp=_MockBody(body), + ) + return err diff --git a/.claude/skills/int-gemini/tests/test_gemini_client.py b/.claude/skills/int-gemini/tests/test_gemini_client.py new file mode 100644 index 00000000..76538eba --- /dev/null +++ b/.claude/skills/int-gemini/tests/test_gemini_client.py @@ -0,0 +1,555 @@ +"""Suíte de testes do int-gemini — 100% mockada (sem rede real). + +Cobre: HTTP/_request, parsing, guard de confidencialidade, roteamento e os +comandos da CLI (ask/models/route/scan/smoke). conftest bloqueia rede em autouse. +""" +from __future__ import annotations + +import argparse +import json +import urllib.error + +import pytest + +import gemini_client as gc +from helpers import ( + make_generate_response, + make_http_error, + make_models_response, + make_response, +) + + +# ── _request: HTTP layer ─────────────────────────────────────────────────────── + +def test_request_success(mock_urlopen): + mock_urlopen([make_response({"hello": "world"})]) + out = gc._request("GET", "/v1beta/models") + assert out == {"hello": "world"} + + +def test_request_missing_key_raises(monkeypatch): + monkeypatch.setattr(gc, "API_KEY", "") + with pytest.raises(gc.GeminiError, match="GEMINI_API_KEY ausente"): + gc._request("GET", "/v1beta/models") + + +def test_request_retries_on_429_then_succeeds(mock_urlopen): + mock_urlopen([make_http_error(429), make_response({"ok": 1})]) + out = gc._request("GET", "/x") + assert out == {"ok": 1} + + +@pytest.mark.parametrize("code", [500, 502, 503, 504]) +def test_request_retries_on_5xx(mock_urlopen, code): + mock_urlopen([make_http_error(code), make_response({"ok": 1})]) + assert gc._request("GET", "/x") == {"ok": 1} + + +def test_request_gives_up_after_one_retry(mock_urlopen): + mock_urlopen([make_http_error(503), make_http_error(503)]) + with pytest.raises(gc.GeminiError, match="HTTP 503"): + gc._request("GET", "/x") + + +def test_request_non_retryable_4xx_raises_immediately(mock_urlopen): + mock_urlopen([make_http_error(400, "bad request")]) + with pytest.raises(gc.GeminiError, match="bad request"): + gc._request("GET", "/x") + + +def test_request_extracts_api_error_message(mock_urlopen): + mock_urlopen([make_http_error(403, "permission denied")]) + with pytest.raises(gc.GeminiError, match="permission denied"): + gc._request("GET", "/x") + + +def test_request_url_error_retries_then_raises(mock_urlopen): + err = urllib.error.URLError("conn refused") + mock_urlopen([err, err]) + with pytest.raises(gc.GeminiError, match="conexão falhou"): + gc._request("GET", "/x") + + +def test_request_non_json_raises(mock_urlopen): + mock_urlopen([make_response("not json at all")]) + with pytest.raises(gc.GeminiError, match="não-JSON"): + gc._request("GET", "/x") + + +def test_request_non_dict_json_wrapped(mock_urlopen): + mock_urlopen([make_response([1, 2, 3])]) + out = gc._request("GET", "/x") + assert out == {"raw": [1, 2, 3]} + + +def test_request_always_returns_dict(mock_urlopen): + """Type-safety: _request nunca devolve bytes/str/list crua.""" + mock_urlopen([make_response("42")]) # JSON válido, mas é int + out = gc._request("GET", "/x") + assert isinstance(out, dict) + + +# ── parsing ───────────────────────────────────────────────────────────────────── + +def test_extract_text_concatenates_parts(): + resp = {"candidates": [{"content": {"parts": [{"text": "ab"}, {"text": "cd"}]}}]} + assert gc._extract_text(resp) == "abcd" + + +def test_extract_text_empty_when_no_candidates(): + assert gc._extract_text({}) == "" + assert gc._extract_text({"candidates": []}) == "" + + +def test_extract_text_skips_non_dict_parts(): + resp = {"candidates": [{"content": {"parts": [{"text": "x"}, "junk"]}}]} + assert gc._extract_text(resp) == "x" + + +def test_extract_grounding_returns_sources(): + resp = {"candidates": [{"groundingMetadata": {"groundingChunks": [ + {"web": {"title": "T1", "uri": "http://a"}}, + {"web": {"title": "T2", "uri": "http://b"}}, + ]}}]} + out = gc._extract_grounding(resp) + assert out == [{"title": "T1", "uri": "http://a"}, {"title": "T2", "uri": "http://b"}] + + +def test_extract_grounding_skips_chunks_without_uri(): + resp = {"candidates": [{"groundingMetadata": {"groundingChunks": [ + {"web": {"title": "no uri"}}, + {"retrievedContext": {}}, + ]}}]} + assert gc._extract_grounding(resp) == [] + + +def test_extract_grounding_empty_when_no_metadata(): + assert gc._extract_grounding({"candidates": [{}]}) == [] + + +# ── guard de confidencialidade ────────────────────────────────────────────────── + +@pytest.mark.parametrize("text,expected_type", [ + ("meu CPF é 123.456.789-00", "cpf"), + ("CPF 12345678900", "cpf"), + ("CNPJ 12.345.678/0001-90", "cnpj"), + ("ligue (11) 99999-8888", "telefone"), + ("contato +55 11 98888-7777", "telefone"), + ("email joao@cliente.com.br", "email"), + ("valor R$ 1.097,00", "valor_financeiro"), + ("paga via PIX", "pix"), + ("a cláusula 4 do contrato", "contrato"), + ("os honorários acordados", "contrato"), + ("isto é confidencial", "contrato"), + ("a api_key vazou", "segredo"), + ("o token bearer", "segredo"), + ("minha senha é", "segredo"), +]) +def test_scan_detects_sensitive(text, expected_type): + findings = gc.scan_confidential(text) + assert expected_type in {f["type"] for f in findings} + + +@pytest.mark.parametrize("text", [ + "resumir transcript de reunião interna", + "gerar calendário editorial do mês seguinte", + "classificar sentimento de feedback público", + "rascunho de post sobre marketing digital", + "", +]) +def test_scan_safe_returns_empty(text): + assert gc.scan_confidential(text) == [] + + +def test_scan_masks_sample(): + findings = gc.scan_confidential("CPF 123.456.789-00") + cpf = next(f for f in findings if f["type"] == "cpf") + assert cpf["sample"].endswith("…") + assert "456" not in cpf["sample"] # não ecoa o número inteiro + + +def test_scan_short_sample_not_masked(): + findings = gc.scan_confidential("PIX agora") + pix = next(f for f in findings if f["type"] == "pix") + assert pix["sample"].lower() == "pix" + + +def test_scan_multiple_findings(): + findings = gc.scan_confidential("R$ 500 via PIX, CPF 111.222.333-44") + types = {f["type"] for f in findings} + assert {"valor_financeiro", "pix", "cpf"} <= types + + +# ── guard: vetores reais reforçados (achados Opus #1, #2) ─────────────────────── + +@pytest.mark.parametrize("text,expected", [ + # CEP (#1) + ("CEP 01310-100", "cep"), + ("cep 01310100", "cep"), + # endereço (#1) + ("Rua das Flores, 123", "endereco"), + ("av. Paulista 1500", "endereco"), + ("Avenida Brasil 200", "endereco"), + ("Alameda Santos, 45", "endereco"), + # nome + contexto de cliente/lead (#1) — a PII mais comum + ("lead João Silva", "nome_contexto"), + ("Maria Souza, Diretora Financeira", "nome_contexto"), + ("sócio João da Silva Santos", "nome_contexto"), + ("contato Maria Aparecida", "nome_contexto"), + # valor por extenso (#1) + ("negócio fechado em 50 mil", "valor_extenso"), + ("cinquenta mil reais", "valor_extenso"), + # telefone — nono dígito separado, com 0, fixo 8 dígitos (#2) + ("(11) 9 9999-9999", "telefone"), + ("11 9 9999 9999", "telefone"), + ("011 99999-8888", "telefone"), + ("ligue 9999-8888", "telefone"), + ("+55 11 99999-8888", "telefone"), +]) +def test_scan_detects_reinforced_vectors(text, expected): + assert expected in {f["type"] for f in gc.scan_confidential(text)} + + +def test_scan_card_not_labeled_phone(): + """#4 — sequência longa de dígitos é rotulada cartão, não telefone.""" + types = {f["type"] for f in gc.scan_confidential("cartão 4111 1111 1111 1111")} + assert "cartao" in types + assert "telefone" not in types # de-dup por sobreposição + + +def test_scan_phone_with_55_not_labeled_card(): + types = {f["type"] for f in gc.scan_confidential("ligar +55 11 99999-8888")} + assert "telefone" in types + assert "cartao" not in types + + +def test_scan_dedup_no_double_label_same_span(): + """Um mesmo trecho (16 dígitos colados) vira exatamente um tipo.""" + findings = gc.scan_confidential("4111111111111111") + assert len(findings) == 1 + + +# ── guard: corpus de FALSO-NEGATIVO documentado (achado Opus #3) ──────────────── +# Estes casos DOCUMENTAM o limite do guard: é guard-rail PARCIAL, não garantia. +# Cada assert registra o que o guard NÃO pega e POR QUÊ — a regressão avisa se o +# comportamento mudar. + +@pytest.mark.parametrize("text,why", [ + ("João Silva ligou ontem", "nome solto sem termo de papel próximo"), + ("entrega no 123 do bloco B", "sem palavra de logradouro (rua/av)"), + ("meu número é nove nove nove nove", "dígitos por extenso não são capturados"), + ("falar com a Ana", "primeiro nome isolado é ambíguo demais p/ heurística"), +]) +def test_scan_known_false_negatives(text, why): + """Guard NÃO pega estes vetores — limite conhecido e aceito (ver SKILL.md). + + Se algum passar a ser detectado no futuro, ótimo — mas HOJE não é, então + tarefas com dado de cliente exigem redação/tier no-train, não confiança cega. + """ + assert gc.scan_confidential(text) == [], why + + +def test_scan_does_not_flag_normal_operational_text(): + """Texto operacional típico (sem PII) não dispara o guard — anti-ruído.""" + assert gc.scan_confidential("Migrar runtime EvoNexus para Contabo") == [] + assert gc.scan_confidential("gerar calendário editorial do mês seguinte") == [] + + +# ── roteamento ────────────────────────────────────────────────────────────────── + +@pytest.mark.parametrize("task,model", [ + ("classify", "gemini-2.5-flash-lite"), + ("triage", "gemini-2.5-flash-lite"), + ("extract", "gemini-2.5-flash-lite"), + ("summarize", "gemini-2.5-flash"), + ("draft", "gemini-2.5-flash"), + ("translate", "gemini-2.5-flash"), + ("research", "gemini-2.5-flash"), + ("reason", "gemini-2.5-pro"), +]) +def test_route_model_known(task, model): + assert gc.route_model(task) == model + + +def test_route_model_case_insensitive(): + assert gc.route_model("CLASSIFY") == "gemini-2.5-flash-lite" + assert gc.route_model(" Summarize ") == "gemini-2.5-flash" + + +def test_route_model_unknown_falls_back_to_flash(): + assert gc.route_model("nonsense") == gc.DEFAULT_ROUTE_MODEL == "gemini-2.5-flash" + + +# ── cmd_ask ───────────────────────────────────────────────────────────────────── + +def _ask_args(prompt="oi", model=None, system=None, json_=False, + grounding=False, allow_sensitive=False, system_file=None, + fallback=False): + return argparse.Namespace( + prompt=prompt, model=model, system=system, json=json_, + grounding=grounding, allow_sensitive=allow_sensitive, + system_file=system_file, fallback=fallback, + ) + + +def test_cmd_ask_success(mock_urlopen, capsys): + mock_urlopen([make_generate_response("resposta!", total_tokens=42)]) + rc = gc.cmd_ask(_ask_args(prompt="resumir isto")) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is True + assert out["text"] == "resposta!" + assert out["usage"]["total_tokens"] == 42 + assert out["finish_reason"] == "STOP" + + +def test_cmd_ask_uses_default_model(mock_urlopen, capsys): + mock_urlopen([make_generate_response("x")]) + gc.cmd_ask(_ask_args()) + out = json.loads(capsys.readouterr().out) + assert out["model"] == gc.DEFAULT_MODEL + + +def test_cmd_ask_honors_explicit_model(mock_urlopen, capsys): + mock_urlopen([make_generate_response("x")]) + gc.cmd_ask(_ask_args(model="gemini-2.5-flash-lite")) + out = json.loads(capsys.readouterr().out) + assert out["model"] == "gemini-2.5-flash-lite" + + +def test_cmd_ask_grounding_returns_sources(mock_urlopen, capsys): + mock_urlopen([make_generate_response( + "factual", grounding_chunks=[{"web": {"title": "src", "uri": "http://s"}}])]) + gc.cmd_ask(_ask_args(prompt="pesquisar novidades", grounding=True)) + out = json.loads(capsys.readouterr().out) + assert out["grounded"] is True + assert out["sources"] == [{"title": "src", "uri": "http://s"}] + + +def test_cmd_ask_no_grounding_no_sources(mock_urlopen, capsys): + mock_urlopen([make_generate_response( + "x", grounding_chunks=[{"web": {"title": "src", "uri": "http://s"}}])]) + gc.cmd_ask(_ask_args(grounding=False)) + out = json.loads(capsys.readouterr().out) + assert out["sources"] == [] + + +def test_cmd_ask_api_error_returns_1(mock_urlopen, capsys): + mock_urlopen([make_http_error(400, "invalid")]) + rc = gc.cmd_ask(_ask_args()) + assert rc == 1 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False + assert "invalid" in out["error"] + + +# ── guard integrado ao cmd_ask ────────────────────────────────────────────────── + +def test_cmd_ask_blocks_sensitive_prompt(capsys): + """Guard bloqueia ANTES de qualquer rede (rede está bloqueada no conftest).""" + rc = gc.cmd_ask(_ask_args(prompt="cliente CPF 123.456.789-00")) + assert rc == 2 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False + assert "cpf" in out["detected_types"] + + +def test_cmd_ask_blocks_sensitive_system(capsys): + rc = gc.cmd_ask(_ask_args(prompt="ok", system="honorários R$ 5.000")) + assert rc == 2 + out = json.loads(capsys.readouterr().out) + assert {"valor_financeiro", "contrato"} & set(out["detected_types"]) + + +def test_cmd_ask_allow_sensitive_bypasses_guard(mock_urlopen, capsys): + mock_urlopen([make_generate_response("processado")]) + rc = gc.cmd_ask(_ask_args(prompt="CPF 111.222.333-44", allow_sensitive=True)) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is True + + +def test_cmd_ask_safe_prompt_passes_guard(mock_urlopen, capsys): + mock_urlopen([make_generate_response("ok")]) + rc = gc.cmd_ask(_ask_args(prompt="resumir reunião interna")) + assert rc == 0 + + +# ── cmd_ask: auto-melhorias (fallback / system-file) ─────────────────────────── + +def test_cmd_ask_fallback_on_model_error(mock_urlopen, capsys): + """flash-lite falha (503), --fallback usa flash e responde.""" + from helpers import make_http_error + mock_urlopen([ + make_http_error(503), make_http_error(503), # flash-lite + retry falham + make_generate_response("via flash"), # flash responde + ]) + rc = gc.cmd_ask(_ask_args(model="gemini-2.5-flash-lite", fallback=True)) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is True + assert out["model"] == "gemini-2.5-flash" # reporta o modelo que de fato respondeu + + +def test_cmd_ask_no_fallback_propagates_error(mock_urlopen, capsys): + mock_urlopen([make_http_error(503), make_http_error(503)]) + rc = gc.cmd_ask(_ask_args(model="gemini-2.5-flash-lite", fallback=False)) + assert rc == 1 + out = json.loads(capsys.readouterr().out) + assert out["ok"] is False + assert out["model"] == "gemini-2.5-flash-lite" + + +def test_cmd_ask_fallback_both_fail(mock_urlopen, capsys): + from helpers import make_http_error + mock_urlopen([make_http_error(503), make_http_error(503), # flash-lite + make_http_error(503), make_http_error(503)]) # flash fallback + rc = gc.cmd_ask(_ask_args(model="gemini-2.5-flash-lite", fallback=True)) + assert rc == 1 + + +def test_cmd_ask_system_file(mock_urlopen, capsys, tmp_path): + sysf = tmp_path / "sys.txt" + sysf.write_text("Você é um classificador.", encoding="utf-8") + mock_urlopen([make_generate_response("ok")]) + rc = gc.cmd_ask(_ask_args(prompt="classificar", system_file=str(sysf))) + assert rc == 0 + + +def test_cmd_ask_system_file_missing(capsys): + rc = gc.cmd_ask(_ask_args(system_file="/nao/existe.txt")) + assert rc == 1 + assert "system-file" in json.loads(capsys.readouterr().out)["error"] + + +def test_cmd_ask_system_file_scanned_by_guard(capsys, tmp_path): + """system-file também passa pelo guard de confidencialidade.""" + sysf = tmp_path / "sys.txt" + sysf.write_text("processar CPF 123.456.789-00", encoding="utf-8") + rc = gc.cmd_ask(_ask_args(prompt="ok", system_file=str(sysf))) + assert rc == 2 # bloqueado + + +# ── cmd_models ────────────────────────────────────────────────────────────────── + +def test_cmd_models_filters_generate_content(mock_urlopen, capsys): + mock_urlopen([make_models_response(["gemini-2.5-flash", "gemini-2.5-pro"])]) + rc = gc.cmd_models(argparse.Namespace()) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["count"] == 2 + assert {m["name"] for m in out["models"]} == {"gemini-2.5-flash", "gemini-2.5-pro"} + + +def test_cmd_models_error_returns_1(mock_urlopen, capsys): + mock_urlopen([make_http_error(403, "denied")]) + rc = gc.cmd_models(argparse.Namespace()) + assert rc == 1 + assert json.loads(capsys.readouterr().out)["ok"] is False + + +# ── cmd_route / cmd_scan ──────────────────────────────────────────────────────── + +def test_cmd_route(capsys): + rc = gc.cmd_route(argparse.Namespace(task_type="classify")) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["model"] == "gemini-2.5-flash-lite" + assert out["known"] is True + + +def test_cmd_route_unknown(capsys): + gc.cmd_route(argparse.Namespace(task_type="zzz")) + out = json.loads(capsys.readouterr().out) + assert out["known"] is False + assert out["model"] == "gemini-2.5-flash" + + +def test_cmd_scan_sensitive(capsys): + rc = gc.cmd_scan(argparse.Namespace(text="CPF 123.456.789-00")) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["safe"] is False + assert "cpf" in out["detected_types"] + + +def test_cmd_scan_safe(capsys): + gc.cmd_scan(argparse.Namespace(text="texto público inofensivo")) + out = json.loads(capsys.readouterr().out) + assert out["safe"] is True + + +# ── cmd_smoke ─────────────────────────────────────────────────────────────────── + +def test_smoke_all_pass(mock_urlopen, capsys): + # generate (1 call) + models (1 call) + mock_urlopen([ + make_generate_response("pong"), + make_models_response([gc.DEFAULT_MODEL, "gemini-2.5-flash-lite"]), + ]) + rc = gc.cmd_smoke(argparse.Namespace()) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["overall"] == "PASS" + steps = {s["step"]: s["status"] for s in out["steps"]} + assert steps == {"env": "PASS", "generate": "PASS", "guard": "PASS", "models": "PASS"} + + +def test_smoke_exit_zero_even_when_no_key(monkeypatch, capsys): + monkeypatch.setattr(gc, "API_KEY", "") + rc = gc.cmd_smoke(argparse.Namespace()) + assert rc == 0 # smoke SEMPRE exit 0 + out = json.loads(capsys.readouterr().out) + assert out["overall"] == "FAIL" + steps = {s["step"]: s["status"] for s in out["steps"]} + assert steps["env"] == "FAIL" + assert steps["generate"] == "SKIP" + assert steps["models"] == "SKIP" + assert steps["guard"] == "PASS" # guard não depende de rede + + +def test_smoke_fails_when_default_model_missing(mock_urlopen, capsys): + mock_urlopen([ + make_generate_response("pong"), + make_models_response(["algum-outro-modelo"]), + ]) + rc = gc.cmd_smoke(argparse.Namespace()) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["overall"] == "FAIL" + steps = {s["step"]: s["status"] for s in out["steps"]} + assert steps["models"] == "FAIL" + + +def test_smoke_fails_on_generate_error(mock_urlopen, capsys): + mock_urlopen([ + make_http_error(500, "down"), + make_http_error(500, "down"), # retry também falha + make_models_response([gc.DEFAULT_MODEL]), + ]) + rc = gc.cmd_smoke(argparse.Namespace()) + assert rc == 0 + out = json.loads(capsys.readouterr().out) + assert out["overall"] == "FAIL" + steps = {s["step"]: s["status"] for s in out["steps"]} + assert steps["generate"] == "FAIL" + + +def test_smoke_has_duration(mock_urlopen, capsys): + mock_urlopen([ + make_generate_response("pong"), + make_models_response([gc.DEFAULT_MODEL]), + ]) + gc.cmd_smoke(argparse.Namespace()) + out = json.loads(capsys.readouterr().out) + assert "duration_ms" in out + assert all("duration_ms" in s for s in out["steps"]) + + +# ── output contract ───────────────────────────────────────────────────────────── + +def test_ask_output_is_valid_json(mock_urlopen, capsys): + mock_urlopen([make_generate_response("x")]) + gc.cmd_ask(_ask_args()) + json.loads(capsys.readouterr().out) # não levanta = contrato ok