From bd45a7fb33d3071b47b71aef0bf03ec27a9fbf6a Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Tue, 19 May 2026 22:28:47 -0600 Subject: [PATCH] refactor(providers/google): port cogitate to google-genai SDK Move Google cogitate from a Gemini CLI subprocess to the google-genai SDK so tool calls, policy checks, and history are owned in-process. This closes the casual-installer gap by removing the separate gemini binary requirement. The cutover intentionally starts existing Google cogitate sessions fresh once, then persists SDK history under the journal cache. Co-Authored-By: Claude Opus 4.7 (1M context) --- CHANGELOG.md | 8 + CONTRIBUTING.md | 46 +- INSTALL.md | 4 +- docs/PROVIDERS.md | 16 +- pyproject.toml | 2 +- solstone/think/cogitate_policy.py | 102 ---- solstone/think/policies/cogitate.toml | 32 -- solstone/think/providers/__init__.py | 20 +- solstone/think/providers/cli.py | 79 +-- solstone/think/providers/google.py | 548 ++++++++++++-------- solstone/think/providers/google_tools.py | 328 ++++++++++++ solstone/think/providers/shared.py | 17 +- tests/baselines/api/settings/providers.json | 5 +- tests/integration/test_google_provider.py | 5 +- tests/test_cli_provider.py | 491 +----------------- tests/test_cogitate_coder.py | 74 +-- tests/test_cogitate_policy.py | 93 +--- tests/test_google.py | 137 +---- tests/test_google_cli.py | 402 -------------- tests/test_google_sdk.py | 532 +++++++++++++++++++ tests/test_google_thinking.py | 135 ----- tests/test_google_tools.py | 300 +++++++++++ tests/test_no_legacy_chat_imports.py | 5 +- uv.lock | 2 +- 24 files changed, 1604 insertions(+), 1779 deletions(-) delete mode 100644 solstone/think/policies/cogitate.toml create mode 100644 solstone/think/providers/google_tools.py delete mode 100644 tests/test_google_cli.py create mode 100644 tests/test_google_sdk.py delete mode 100644 tests/test_google_thinking.py create mode 100644 tests/test_google_tools.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b670a39ae..2f0fcf01f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ All notable changes to solstone (the Python package) will be documented in this Format adapted from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), aligned with `cmo/brand/changelog-voice.md`. +## [Unreleased] + +### Changed +- Google cogitate agents now run through the Google GenAI SDK instead of a separate Gemini CLI process. + - Google tool use is handled inside solstone, with the same read/write policy boundaries. + - Google cogitate no longer needs a separate JavaScript package-manager install step. + - Existing Google cogitate sessions start fresh once because SDK history is stored in solstone's journal cache. + ## [0.3.7] - 2026-05-19 ### Changed diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 321e95c30..914069ca2 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -11,6 +11,7 @@ Required everywhere: - Python 3.11 or later - [uv](https://docs.astral.sh/uv/) - Git +- ripgrep (`rg`) - ffmpeg for audio processing Linux is the primary development platform. macOS is supported. Source-checkout installs on Apple Silicon need Xcode command line tools to build the CoreML parakeet helper; packaged installs (`uv tool install solstone`) on macOS 14 or newer ship the helper as a pre-built binary. @@ -18,21 +19,21 @@ Linux is the primary development platform. macOS is supported. Source-checkout i Fedora/RHEL: ```bash -sudo dnf install python3 git ffmpeg pipewire gstreamer1-plugins-base gstreamer1-plugin-pipewire pulseaudio-utils +sudo dnf install python3 git ripgrep ffmpeg pipewire gstreamer1-plugins-base gstreamer1-plugin-pipewire pulseaudio-utils curl -LsSf https://astral.sh/uv/install.sh | sh ``` Ubuntu/Debian: ```bash -sudo apt install python3 git ffmpeg pipewire gstreamer1.0-tools gstreamer1.0-pipewire pulseaudio-utils +sudo apt install python3 git ripgrep ffmpeg pipewire gstreamer1.0-tools gstreamer1.0-pipewire pulseaudio-utils curl -LsSf https://astral.sh/uv/install.sh | sh ``` Arch: ```bash -sudo pacman -S python git ffmpeg pipewire gstreamer gst-plugin-pipewire libpulse +sudo pacman -S python git ripgrep ffmpeg pipewire gstreamer gst-plugin-pipewire libpulse curl -LsSf https://astral.sh/uv/install.sh | sh ``` @@ -40,7 +41,7 @@ macOS: ```bash xcode-select --install -brew install python git ffmpeg uv +brew install python git ripgrep ffmpeg uv ``` ## Source-checkout install @@ -132,43 +133,6 @@ make review See [AGENTS.md](AGENTS.md) for the full Makefile command table and [docs/testing.md](docs/testing.md) for test isolation details. -## Tool-using agents (Cogitate CLI binaries) - -solstone runs two distinct AI workloads, and they have different prerequisites: - -- **Generate** — direct text generation (transcription, vision analysis, insights, descriptions). Uses each provider's SDK and works as soon as the API key is set. No extra binaries. -- **Cogitate** — tool-using agents (entity detection, entity assist, entity describe, and any talent under `solstone/apps/entities/talent/*.md` with `"type": "cogitate"`). solstone shells out to a provider CLI binary as a subprocess. Anthropic and OpenAI are installed into your journal on demand; Google and Ollama still use CLIs you install yourself. - -Because cogitate-type talents only run after entity-related thinking completes, missing CLI binaries are invisible during initial install — generate-only paths produce transcripts and summaries, but **entities never appear**. `sol top` Agents Health flags this with messages like `"Google generate passes but Google cogitate fails: gemini CLI not installed"`. Enable the cogitate binary for **each provider whose API key you configured above**. If you only set `GOOGLE_API_KEY`, you only need `gemini`. - -| provider | binary | install | -|-----------|------------|-----------------------------------------------------------------| -| anthropic | `claude` | `sol call settings providers install anthropic` | -| openai | `codex` | `sol call settings providers install openai` | -| google | `gemini` | `npm install -g @google/gemini-cli` (Node 20+) | -| ollama | `opencode` | `curl -fsSL https://opencode.ai/install \| bash` | - -For Anthropic and OpenAI, check install status with: - -```bash -sol call settings providers status -``` - -For Google, install Node.js first if needed: `brew install node` on macOS, your distro's package manager on Linux (e.g. `sudo dnf install nodejs`, `sudo apt install nodejs npm`). Verify manually installed binaries are on `PATH` after install: - -```bash -gemini --version -opencode --version -``` - -After installing a manually managed CLI binary while solstone is running, restart the service so cortex picks up the new `PATH`: - -```bash -sol service restart -``` - -`sol providers check` reports per-provider readiness, including bundled or manually installed cogitate CLI status. - ## Developing on AI features ### macOS Apple Silicon: CoreML-accelerated parakeet diff --git a/INSTALL.md b/INSTALL.md index 0a9d4c08d..93d5418a3 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -19,9 +19,9 @@ if solstone is running and healthy, skip to [install an observer](#install-an-ob ### prerequisites -linux: `uv` is the only requirement. install with `curl -LsSf https://astral.sh/uv/install.sh | sh`. +linux: install `uv` (`curl -LsSf https://astral.sh/uv/install.sh | sh`) and `ripgrep` (`rg`) from your distro package manager. -macOS: install xcode command line tools (`xcode-select --install`) and homebrew (https://brew.sh), then `brew install uv`. +macOS: install xcode command line tools (`xcode-select --install`) and homebrew (https://brew.sh), then `brew install uv ripgrep`. ## install diff --git a/docs/PROVIDERS.md b/docs/PROVIDERS.md index b1ed2d9d3..254b05d50 100644 --- a/docs/PROVIDERS.md +++ b/docs/PROVIDERS.md @@ -122,7 +122,7 @@ async def run_cogitate( - `user_instruction`: Agent-specific prompt as second user message - `tools`: Optional list of allowed tool names - `use_id`, `name`: Identity for logging and tool calls -- `session_id`: CLI session ID for conversation continuation +- `session_id`: solstone-owned session ID for conversation continuation; Google cogitate history is stored under `journal/.cache/cogitate-history/` - `chat_id`: Chat ID for reverse lookup from agent to chat **Event emission:** @@ -177,16 +177,10 @@ honor `config["tools"]` allowlists when present. **Conversation continuation:** -When `session_id` is provided, use the provider CLI's native resume mechanism: -```python -session_id = config.get("session_id") -if session_id: - cmd.extend(["--resume", session_id]) -``` - -Each CLI tool manages its own session state internally. The `session_id` is -returned from the CLI's init/finish event on the first interaction and reused -for all subsequent continuations within the same chat. +When `session_id` is provided, use the provider's native continuation mechanism +or a solstone-owned history file where the provider has no durable session +handle. The `session_id` is reused for all subsequent continuations within the +same chat. ## Token Logging diff --git a/pyproject.toml b/pyproject.toml index ffa117a9d..71a852664 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,7 +31,7 @@ keywords = ["audio", "transcription", "screenshot", "multimodal", "ai", "gemini" dependencies = [ # Core dependencies "python-dotenv", - "google-genai", + "google-genai>=1.62.0", "Flask[async]", "Markdown", "mistune", diff --git a/solstone/think/cogitate_policy.py b/solstone/think/cogitate_policy.py index ceefce254..98d69fb18 100644 --- a/solstone/think/cogitate_policy.py +++ b/solstone/think/cogitate_policy.py @@ -3,25 +3,10 @@ from __future__ import annotations -import os import re from datetime import date, datetime, timedelta -from pathlib import Path from typing import Any -import tomllib - -from solstone.think.utils import get_journal - -_BASE_POLICY_PATH = Path(__file__).parent / "policies" / "cogitate.toml" -_POLICY_DIR = Path("/tmp/sol-cogitate-policies") -_READ_TOOL_RULES = ( - ("read_file", '"file_path":"', "strict"), - ("list_directory", '"dir_path":"', "strict"), - ("glob", '"(?:pattern|path)":"', "pattern"), - ("grep_search", '"(?:path|dir_path|include|pattern)":"', "pattern"), -) - def _normalize_day(day: date | str) -> str: if isinstance(day, date): @@ -66,90 +51,3 @@ def resolve_read_scope( f"chronicle/{(base_day - timedelta(days=offset)).strftime('%Y%m%d')}" for offset in range(effective_span, -1, -1) ] - - -def _is_file_scope(scope: str) -> bool: - clean = scope.rstrip("/") - # Dotted basenames are files; ambiguous entries are treated as directories. - return not scope.endswith("/") and "." in Path(clean).name - - -def _scope_body(scope: str, suffix: str) -> str: - clean = scope.strip("/") - escaped = re.escape(clean) - return f'{escaped}"' if _is_file_scope(scope) else f"{escaped}{suffix}" - - -def _allowed_path_regex(scope: list[str], *, mode: str) -> str: - journal_prefix = re.escape(str(get_journal()).rstrip("/") + "/") - prefix = f"(?:{journal_prefix}|\\./)?" - suffix = '(?:/|")' if mode == "strict" else '(?:/|[^"]*")' - - if all(not _is_file_scope(entry) for entry in scope): - joined = "|".join(re.escape(entry.strip("/")) for entry in scope) - return f"{prefix}(?:{joined}){suffix}" - - joined = "|".join(_scope_body(entry, suffix) for entry in scope) - return f"{prefix}(?:{joined})" - - -def _append_read_rules(base_text: str, scope: list[str]) -> str: - chunks = [base_text.rstrip(), ""] - for tool_name, key_pattern, mode in _READ_TOOL_RULES: - path_pattern = _allowed_path_regex(scope, mode=mode) - chunks.extend( - [ - "[[rule]]", - f'toolName = "{tool_name}"', - f"argsPattern = '{key_pattern}{path_pattern}'", - 'decision = "allow"', - "priority = 260", - "", - "[[rule]]", - f'toolName = "{tool_name}"', - 'decision = "deny"', - "priority = 160", - "", - ] - ) - return "\n".join(chunks) - - -def _policy_filename(talent_name: str, day: str) -> str: - talent_slug = re.sub(r"[^A-Za-z0-9_.-]+", "-", talent_name).strip("-") - if not talent_slug: - talent_slug = "cogitate" - return f"{talent_slug}-{day}-{os.getpid()}.toml" - - -def _write_policy(content: str, path: Path) -> Path: - flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY - fd = os.open(path, flags, 0o600) - with os.fdopen(fd, "w", encoding="utf-8") as handle: - handle.write(content) - return path - - -def build_per_task_policy( - talent_name: str, - talent_config: dict[str, Any], - day: date | str, - span: int, - base_policy_path: Path = _BASE_POLICY_PATH, -) -> Path: - day_str = _normalize_day(day) - scope = resolve_read_scope(talent_config, day_str, span=span) - base_text = base_policy_path.read_text(encoding="utf-8") - tomllib.loads(base_text) - content = _append_read_rules(base_text, scope) - - _POLICY_DIR.mkdir(parents=True, exist_ok=True) - filename = _policy_filename(talent_name, day_str) - target = _POLICY_DIR / filename - for attempt in range(100): - path = target if attempt == 0 else target.with_stem(f"{target.stem}-{attempt}") - try: - return _write_policy(content, path) - except FileExistsError: - continue - raise FileExistsError(f"Could not create unique cogitate policy path for {target}") diff --git a/solstone/think/policies/cogitate.toml b/solstone/think/policies/cogitate.toml deleted file mode 100644 index 4cbbab4bf..000000000 --- a/solstone/think/policies/cogitate.toml +++ /dev/null @@ -1,32 +0,0 @@ -# Scoped-yolo policy for non-write cogitate talents. -# -# Posture: start from yolo's default-allow, then subtract the two behaviors -# we don't want — direct filesystem writes, and arbitrary shell commands. -# Run `sol` invocations (including pipelines like `echo ... | sol call ...`) -# are the one shell surface cogitate talents legitimately need. -# -# Priority: highest matching priority wins regardless of allow/deny. The -# allow rule for sol run_shell_command (priority 200) overrides the blanket -# run_shell_command deny (priority 100). User-tier rules here override the -# engine's built-in yolo catch-all. -# -# Rationale: plan mode strips run_shell_command from the tool registry, -# which caused a tool-name hallucination loop we saw in earlier cortex -# prototypes. Scoped yolo keeps the registry intact without widening the -# blast radius to direct writes. - -[[rule]] -toolName = ["write_file", "replace"] -decision = "deny" -priority = 200 - -[[rule]] -toolName = "run_shell_command" -commandRegex = "(^sol\\s|\\bsol call\\b)" -decision = "allow" -priority = 200 - -[[rule]] -toolName = "run_shell_command" -decision = "deny" -priority = 100 diff --git a/solstone/think/providers/__init__.py b/solstone/think/providers/__init__.py index 1455750db..5e3bf006e 100644 --- a/solstone/think/providers/__init__.py +++ b/solstone/think/providers/__init__.py @@ -48,16 +48,14 @@ PROVIDER_REGISTRY: Dict[str, str] = { # --------------------------------------------------------------------------- # Provider Metadata # --------------------------------------------------------------------------- -# Display labels, environment variable names, and cogitate CLI binary names -# for each provider. Used by settings UI, provider status, and agent health -# checks. +# Display labels, environment variable names, and CLI metadata where applicable. +# Used by settings UI, provider status, and agent health checks. # --------------------------------------------------------------------------- PROVIDER_METADATA: Dict[str, Dict[str, Any]] = { "google": { "label": "Google (Gemini)", "env_key": "GOOGLE_API_KEY", - "cogitate_cli": "gemini", "vertex_env_keys": [ "GOOGLE_GENAI_USE_VERTEXAI", "GOOGLE_APPLICATION_CREDENTIALS", @@ -152,8 +150,8 @@ def build_provider_status( Returns ------- Dict[str, Dict[str, Any]] - Keyed by provider name. Each entry has: configured, generate_ready, - cogitate_ready, cogitate_cli, cogitate_cli_found, issues. + Keyed by provider name. Each entry has readiness fields and issues. + CLI-backed providers also include cogitate_cli and cogitate_cli_found. """ status = {} bundled_cli_states = {"installed-no-key", "key-validating", "valid", "invalid-key"} @@ -198,6 +196,16 @@ def build_provider_status( if not configured and env_key: issues.append(f"{env_key} not set") + if name == "google": + cogitate_ready = configured + status[name] = { + "configured": configured, + "generate_ready": configured, + "cogitate_ready": cogitate_ready, + "issues": issues, + } + continue + if bundled_state is not None: bundled_contract_state = bundled_state["state"] cogitate_cli_found = bundled_contract_state in bundled_cli_states diff --git a/solstone/think/providers/cli.py b/solstone/think/providers/cli.py index e7913a47e..b40bd02be 100644 --- a/solstone/think/providers/cli.py +++ b/solstone/think/providers/cli.py @@ -3,7 +3,7 @@ """CLI subprocess runner for AI provider tool agents. -Spawns provider CLI tools (claude, codex, gemini) in JSON streaming mode +Spawns provider CLI tools (claude, codex, opencode) in JSON streaming mode and translates their JSONL output into our standard Event format. Each provider module implements a translate() function that converts @@ -650,7 +650,7 @@ def check_cli_binary(name: str) -> str: """Check that a CLI binary is available on PATH. Args: - name: Binary name (e.g., "claude", "codex", "gemini"). + name: Binary name (e.g., "claude", "codex", "opencode"). Returns: The full path to the binary. @@ -694,7 +694,6 @@ _BASE_ALLOWLIST = [ _PROVIDER_ALLOWLIST: dict[str, list[str]] = { "anthropic": ["ANTHROPIC_*", "CLAUDE_*"], "openai": ["OPENAI_*"], - "google": ["GOOGLE_*", "GEMINI_*", "VERTEX_*"], } @@ -712,7 +711,7 @@ def build_cogitate_env(provider_name: str) -> dict[str, str]: No other glob characters are supported. Args: - provider_name: Provider name (``anthropic``, ``openai``, or ``google``). + provider_name: Provider name (``anthropic`` or ``openai``). Returns: Filtered environment for the provider CLI. @@ -739,78 +738,6 @@ def build_cogitate_env(provider_name: str) -> dict[str, str]: if auth_mode == "platform": env.pop(env_key, None) - # Vertex AI / AI Studio: set backend env vars for Google provider - if provider_name == "google": - google_backend = providers_config.get("google_backend", "auto") - - # Determine effective backend - if google_backend in ("aistudio", "vertex"): - effective_backend = google_backend - else: - api_key = os.getenv("GOOGLE_API_KEY", "") - if api_key: - from solstone.think.providers.google import _detect_backend - - effective_backend = _detect_backend(api_key) - else: - effective_backend = "aistudio" - - if effective_backend == "vertex": - env["GOOGLE_GENAI_USE_VERTEXAI"] = "true" - # Vertex uses SA credentials, not API key — always strip - env.pop("GOOGLE_API_KEY", None) - # SA credentials: set GOOGLE_APPLICATION_CREDENTIALS - creds_path = providers_config.get("vertex_credentials") - if not creds_path or not os.path.exists(creds_path): - raise ValueError( - f"Vertex provider configured but no usable SA credentials at {creds_path}" - ) - try: - with open(creds_path, encoding="utf-8") as _f: - _sa_data = json.load(_f) - except (OSError, json.JSONDecodeError) as exc: - raise ValueError( - f"Vertex provider configured but no usable SA credentials at {creds_path}" - ) from exc - project_id = _sa_data.get("project_id") - if not project_id: - raise ValueError( - f"Vertex provider configured but no usable SA credentials at {creds_path}" - ) - env["GOOGLE_APPLICATION_CREDENTIALS"] = creds_path - env["GOOGLE_CLOUD_PROJECT"] = project_id - env["GOOGLE_CLOUD_LOCATION"] = "global" - from solstone.think.utils import get_journal - - settings_path = ( - Path(get_journal()) / ".config" / "gemini-vertex-settings.json" - ) - if not settings_path.exists(): - os.makedirs(settings_path.parent, exist_ok=True) - with open(settings_path, "w", encoding="utf-8") as settings_file: - json.dump( - {"security": {"auth": {"selectedType": "vertex-ai"}}}, - settings_file, - ) - os.chmod(str(settings_path), 0o600) - env["GEMINI_CLI_SYSTEM_SETTINGS_PATH"] = str(settings_path) - else: - # AI Studio: clear any inherited Vertex env vars so the CLI - # doesn't accidentally run in Vertex mode. - for vkey in ( - "GEMINI_CLI_SYSTEM_SETTINGS_PATH", - "GOOGLE_APPLICATION_CREDENTIALS", - "GOOGLE_CLOUD_LOCATION", - "GOOGLE_CLOUD_PROJECT", - "GOOGLE_GENAI_USE_VERTEXAI", - ): - env.pop(vkey, None) - # Gemini CLI's auth auto-detection only honors GEMINI_API_KEY for - # api-key mode; GOOGLE_API_KEY is recognized only for vertex-ai mode. - # Mirror the canonical key so non-interactive `-p -` invocations - # don't fall through to oauth-personal and FatalAuthenticationError. - if "GOOGLE_API_KEY" in env and "GEMINI_API_KEY" not in env: - env["GEMINI_API_KEY"] = env["GOOGLE_API_KEY"] return env diff --git a/solstone/think/providers/google.py b/solstone/think/providers/google.py index 739cb6a95..3b6768d8e 100644 --- a/solstone/think/providers/google.py +++ b/solstone/think/providers/google.py @@ -38,25 +38,32 @@ from pathlib import Path from typing import Any, Callable from google import genai +from google.genai import errors as google_errors from google.genai import types -from solstone.think.cogitate_policy import build_per_task_policy +from solstone.think.cogitate_policy import resolve_read_scope from solstone.think.models import GEMINI_FLASH -from solstone.think.utils import now_ms - -from .cli import ( - CLIRunner, - QuotaExhaustedError, - ThinkingAggregator, - assemble_prompt, - build_cogitate_env, +from solstone.think.utils import get_journal, get_project_root, now_ms + +from .cli import QuotaExhaustedError, assemble_prompt +from .google_tools import ( + DEFAULT_READ_CALL_BUDGET, + MAX_TURNS, + CogitatePolicy, + MaxTurnsExhausted, + build_tool_declarations, + glob, + grep_search, + list_directory, + load_history, + read_file, + run_shell_command, + save_history, ) from .shared import ( GenerateResult, JSONEventCallback, - ThinkingEvent, classify_provider_error, - safe_raw, ) GEMINI_MAX_OUTPUT_TOKENS = 65536 @@ -65,6 +72,15 @@ _DEFAULT_MODEL = GEMINI_FLASH logger = logging.getLogger(__name__) +_READ_TOOL_NAMES = frozenset( + { + "read_file", + "glob", + "list_directory", + "grep_search", + } +) + # Backend detection cache _detected_backend: str | None = None @@ -601,139 +617,153 @@ async def run_agenerate( # --------------------------------------------------------------------------- -def _emit_thinking_events( - response: Any, model: str, callback: JSONEventCallback -) -> None: - """Extract and emit thinking events from a response. - - In the Google GenAI SDK, thinking content appears in response.candidates[].content.parts[] - where each Part has a `thought` boolean indicating if it's thinking content, and the - actual thinking text is in `part.text`. - """ - if not hasattr(response, "candidates") or not response.candidates: - return - - for candidate in response.candidates: - if not candidate.content or not candidate.content.parts: +def _cogitate_history_path(session_id: str | None) -> Path | None: + if not session_id: + return None + return Path(get_journal()) / ".cache" / "cogitate-history" / f"{session_id}.json" + + +def _resolve_allowed_roots(config: dict[str, Any]) -> list[Path]: + journal = Path(get_journal()).resolve() + project_root = Path(get_project_root()).resolve() + day = config.get("day") or "" + span = int(config.get("read_scope_span", 0) or 0) + scope_roots: list[Path] = [] + for scope in resolve_read_scope(config, day, span=span): + scope_path = Path(scope).expanduser() + if not scope_path.is_absolute(): + scope_path = journal / scope_path + scope_roots.append(scope_path.resolve()) + return [journal, project_root, *scope_roots] + + +def _extract_retry_delay_ms(exc: BaseException) -> int | None: + details = getattr(exc, "details", None) + if not isinstance(details, dict): + return None + error = details.get("error") + if not isinstance(error, dict): + return None + for item in error.get("details", []) or []: + if not isinstance(item, dict): continue + retry_delay = item.get("retryDelay") + if not isinstance(retry_delay, str): + continue + value = retry_delay.strip() + if value.endswith("s"): + value = value[:-1] + try: + return int(float(value) * 1000) + except ValueError: + return None + return None - for part in candidate.content.parts: - # part.thought is a boolean indicating this is thinking content - # part.text contains the actual thinking summary - if getattr(part, "thought", False) and getattr(part, "text", None): - thinking_event: ThinkingEvent = { - "event": "thinking", - "ts": now_ms(), - "summary": part.text, - "model": model, - } - callback.emit(thinking_event) - - -def _translate_gemini( - event: dict[str, Any], - aggregator: ThinkingAggregator, - callback: JSONEventCallback, - usage_out: dict[str, Any] | None = None, - pending_tools: dict[str, dict[str, Any]] | None = None, -) -> str | None: - """Translate a Gemini CLI JSONL event into our standard Event types. - Args: - event: Raw JSONL event dict from the Gemini CLI. - aggregator: ThinkingAggregator for buffering text. - callback: JSONEventCallback for emitting events. - usage_out: Optional mutable dict to receive usage stats from result events. - pending_tools: Optional mutable dict tracking tool_id -> {tool, args}. +def _raise_quota_if_needed(exc: google_errors.ClientError) -> None: + if getattr(exc, "code", None) != 429 and getattr(exc, "status", None) != ( + "RESOURCE_EXHAUSTED" + ): + return + message = str(exc) or getattr(exc, "message", "") or "Provider quota exhausted" + raise QuotaExhaustedError(message, _extract_retry_delay_ms(exc)) from exc - Returns: - The CLI session ID from init events, or None. - """ - event_type = event.get("type") - if event_type == "init": - return event.get("session_id") +def _build_cogitate_config( + system_instruction: str | None, +) -> types.GenerateContentConfig: + return types.GenerateContentConfig( + system_instruction=system_instruction, + tools=[types.Tool(function_declarations=build_tool_declarations())], + tool_config=types.ToolConfig( + function_calling_config=types.FunctionCallingConfig( + mode=types.FunctionCallingConfigMode.AUTO + ) + ), + automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True), + thinking_config=types.ThinkingConfig( + include_thoughts=True, + thinking_budget=-1, + ), + ) - if event_type == "message": - role = event.get("role") - if role == "user": - return None - if role == "assistant" and event.get("delta"): - content = event.get("content", "") - if content: - aggregator.accumulate(content) - return None - return None - if event_type == "tool_use": - aggregator.flush_as_thinking(raw_events=[event]) - tool_name = event.get("tool_name", "") - tool_id = event.get("tool_id") - tool_args = event.get("parameters") - if pending_tools is not None and tool_id: - pending_tools[tool_id] = {"tool": tool_name, "args": tool_args} - callback.emit( - { - "event": "tool_start", - "tool": tool_name, - "args": tool_args, - "call_id": tool_id, - "raw": safe_raw([event]), - "ts": now_ms(), - } +def _iter_response_parts(chunk: Any) -> list[Any]: + parts: list[Any] = [] + for candidate in getattr(chunk, "candidates", []) or []: + content = getattr(candidate, "content", None) + if content is None: + continue + parts.extend(getattr(content, "parts", []) or []) + return parts + + +def _call_id(tool_name: str, function_call: Any, turn_index: int, index: int) -> str: + existing = getattr(function_call, "id", None) + if existing: + return str(existing) + return f"{tool_name}-{turn_index}-{index}" + + +def _dispatch_google_tool( + tool_name: str, + args: dict[str, Any], + allowed_roots: list[Path], +) -> dict[str, Any]: + if tool_name == "read_file": + return read_file(str(args.get("file_path", "")), allowed_roots=allowed_roots) + if tool_name == "glob": + return glob( + str(args.get("pattern", "")), + path=str(args.get("path", ".")), + allowed_roots=allowed_roots, ) - return None - - if event_type == "tool_result": - tool_id = event.get("tool_id") - tool_info = {} - if pending_tools is not None and tool_id: - tool_info = pending_tools.pop(tool_id, {}) - callback.emit( - { - "event": "tool_end", - "tool": tool_info.get("tool", ""), - "args": tool_info.get("args"), - "call_id": tool_id, - "result": event.get("output"), - "raw": safe_raw([event]), - "ts": now_ms(), - } + if tool_name == "list_directory": + return list_directory( + str(args.get("dir_path", "")), + allowed_roots=allowed_roots, ) - return None + if tool_name == "grep_search": + return grep_search( + str(args.get("pattern", "")), + path=str(args.get("path", ".")), + include=str(args.get("include", "")), + allowed_roots=allowed_roots, + ) + if tool_name == "run_shell_command": + return run_shell_command(str(args.get("command", ""))) + return {"error": f"unknown_tool: {tool_name}"} - if event_type == "result": - stats = event.get("stats") or {} - if usage_out is not None and stats: - input_tokens = stats.get("input_tokens", 0) - output_tokens = stats.get("output_tokens", 0) - total_tokens = stats.get("total_tokens", 0) - usage_out.update( - { - "input_tokens": input_tokens, - "output_tokens": output_tokens, - "total_tokens": total_tokens, - } - ) - if stats.get("cached"): - usage_out["cached_tokens"] = stats["cached"] - # CLI doesn't break out thinking tokens, but they're the - # difference between total and input+output. - reasoning = total_tokens - input_tokens - output_tokens - if reasoning > 0: - usage_out["reasoning_tokens"] = reasoning - return None - # Unknown event type — log and skip - logger.debug("Unknown Gemini CLI event type: %s", event_type) - return None +def _emit_tool_budget_exhausted( + callback: JSONEventCallback, + tool_name: str, + budget: int, + count: int, +) -> None: + callback.emit( + { + "event": "tool_budget_exhausted", + "tool": tool_name, + "budget": budget, + "count": count, + "read_tools": sorted(_READ_TOOL_NAMES), + "ts": now_ms(), + } + ) + + +def _raise_tool_budget_exhausted(count: int, budget: int) -> None: + raise RuntimeError( + f"tool_budget_exhausted: read tool call budget exceeded ({count}/{budget})" + ) async def run_cogitate( config: dict[str, Any], on_event: Callable[[dict], None] | None = None, ) -> str: - """Run a prompt with tool-calling support via Google Gemini. + """Run a prompt with tool-calling support via Google Gemini SDK. Args: config: Complete configuration dictionary including prompt, system_instruction, @@ -745,96 +775,197 @@ async def run_cogitate( callback = JSONEventCallback(on_event) try: - # Assemble prompt from config fields prompt_body, system_instruction = assemble_prompt( config, sol_tool_name="run_shell_command" if not config.get("write") else None, ) + client = get_or_create_client(config.get("client")) + sdk_config = _build_cogitate_config(system_instruction) + history_path = _cogitate_history_path(session_id) + history = load_history(history_path) if history_path else [] + chat = client.chats.create( + model=model, + config=sdk_config, + history=history, + ) + allowed_roots = _resolve_allowed_roots(config) + policy = CogitatePolicy( + write=bool(config.get("write")), allowed_roots=allowed_roots + ) + read_call_budget = int( + config.get("read_call_budget", DEFAULT_READ_CALL_BUDGET) or 0 + ) + read_call_count = 0 + usage: dict[str, Any] = {} - # Gemini CLI has no --system-prompt flag; prepend to prompt body - if system_instruction: - prompt_body = system_instruction + "\n\n" + prompt_body - - # Approval posture: - # - Write-enabled talents (coder) run unpolicied yolo: full tool registry, - # write_file / replace allowed. - # - Read-only cogitate talents run yolo + a scoped policy: full tool - # registry (no plan-mode stripping), but write_file / replace denied - # and run_shell_command narrowed to `sol` invocations. - # Plan mode strips run_shell_command from the registry, which drove a - # tool-name hallucination loop in earlier prototypes (documented in sol - # pbc's internal engineering notes). The deprecated --allowed-tools - # flag controls auto-approval, not availability, so it can't replace - # the policy file for this purpose. - cmd = [ - "gemini", - "-p", - "-", - "-o", - "stream-json", - "--approval-mode", - "yolo", - "-m", - model, - "--sandbox=none", - ] - policy_path = None - if not config.get("write"): - policy_path = build_per_task_policy( - config.get("name") or "cogitate", - config, - config.get("day") or "", - int(config.get("read_scope_span", 0) or 0), - ) - cmd.extend(["--policy", str(policy_path)]) - - # Resume from previous session if continuing - if session_id: - cmd.extend(["--resume", session_id]) + next_message: str | list[types.Part] = prompt_body + result_parts: list[str] = [] + for turn_index in range(MAX_TURNS): + function_calls: list[dict[str, Any]] = [] + stream = chat.send_message_stream(next_message) + for chunk in stream: + chunk_usage = _extract_usage(chunk) + if chunk_usage: + usage = chunk_usage + for part in _iter_response_parts(chunk): + text = getattr(part, "text", None) + is_thought = bool(getattr(part, "thought", False)) + if is_thought and isinstance(text, str) and text.strip(): + callback.emit( + { + "event": "thinking", + "summary": text.strip(), + "model": model, + "ts": now_ms(), + } + ) + if not is_thought and isinstance(text, str) and text: + result_parts.append(text) + callback.emit( + { + "event": "text_delta", + "delta": text, + "model": model, + "ts": now_ms(), + } + ) + function_call = getattr(part, "function_call", None) + if function_call is not None: + tool_name = str(getattr(function_call, "name", "") or "") + args = dict(getattr(function_call, "args", {}) or {}) + function_calls.append( + { + "tool": tool_name, + "args": args, + "call_id": _call_id( + tool_name, + function_call, + turn_index, + len(function_calls), + ), + } + ) + + if history_path: + save_history(history_path, chat.get_history(curated=True)) + + if not function_calls: + result = "".join(result_parts).strip() + finish_event: dict[str, Any] = { + "event": "finish", + "result": result, + "ts": now_ms(), + } + if usage: + finish_event["usage"] = usage + if session_id: + finish_event["cli_session_id"] = session_id + callback.emit(finish_event) + return result + + response_parts: list[types.Part] = [] + for function_call in function_calls: + tool_name = function_call["tool"] + args = function_call["args"] + call_id = function_call["call_id"] + callback.emit( + { + "event": "tool_start", + "tool": tool_name, + "args": args, + "call_id": call_id, + "ts": now_ms(), + } + ) + if tool_name in _READ_TOOL_NAMES: + read_call_count += 1 + if read_call_count > read_call_budget: + _emit_tool_budget_exhausted( + callback, + tool_name, + read_call_budget, + read_call_count, + ) + _raise_tool_budget_exhausted( + read_call_count, + read_call_budget, + ) + allowed, reason = policy.check(tool_name, args) + if allowed: + tool_result = _dispatch_google_tool(tool_name, args, allowed_roots) + else: + tool_result = {"error": reason} + callback.emit( + { + "event": "tool_end", + "tool": tool_name, + "args": args, + "call_id": call_id, + "result": tool_result, + "ts": now_ms(), + } + ) + response_parts.append( + types.Part.from_function_response( + name=tool_name, + response=tool_result, + ) + ) + next_message = response_parts - # Mutable containers for translate closure - usage: dict[str, Any] = {} - pending_tools: dict[str, dict[str, Any]] = {} - - def translate( - event: dict[str, Any], agg: ThinkingAggregator, cb: JSONEventCallback - ) -> str | None: - return _translate_gemini(event, agg, cb, usage, pending_tools) - - aggregator = ThinkingAggregator(callback, model=model) - cwd_value = config.get("cwd") - runner = CLIRunner( - cmd=cmd, - prompt_text=prompt_body, - translate=translate, - callback=callback, - aggregator=aggregator, - cwd=Path(cwd_value) if cwd_value else None, - env=build_cogitate_env("google"), - read_call_budget=int(config.get("read_call_budget", 200)), + callback.emit( + { + "event": "max_turns_exhausted", + "max_turns": MAX_TURNS, + "ts": now_ms(), + } ) - runner.provider = "google" - - try: - result = await runner.run() - finally: - if policy_path is not None: - policy_path.unlink(missing_ok=True) - - # Emit finish event (CLIRunner does not emit one) - finish_event: dict[str, Any] = { - "event": "finish", - "result": result, - "ts": now_ms(), - } - if usage: - finish_event["usage"] = usage - if runner.cli_session_id: - finish_event["cli_session_id"] = runner.cli_session_id - callback.emit(finish_event) - return result + raise MaxTurnsExhausted( + f"max_turns_exhausted: Google cogitate exceeded {MAX_TURNS} turns" + ) + except google_errors.ClientError as exc: + _raise_quota_if_needed(exc) + callback.emit( + { + "event": "error", + "error": str(exc), + "reason_code": classify_provider_error(exc, "google"), + "provider": "google", + "trace": traceback.format_exc(), + } + ) + setattr(exc, "_evented", True) + raise + except google_errors.APIError as exc: + callback.emit( + { + "event": "error", + "error": str(exc), + "reason_code": classify_provider_error(exc, "google"), + "provider": "google", + "trace": traceback.format_exc(), + } + ) + setattr(exc, "_evented", True) + raise except QuotaExhaustedError: raise + except MaxTurnsExhausted: + raise + except RuntimeError as exc: + if str(exc).startswith("tool_budget_exhausted:"): + raise + callback.emit( + { + "event": "error", + "error": str(exc), + "reason_code": classify_provider_error(exc, "google"), + "provider": "google", + "trace": traceback.format_exc(), + } + ) + setattr(exc, "_evented", True) + raise except Exception as exc: callback.emit( { @@ -929,7 +1060,6 @@ __all__ = [ "run_generate", "run_agenerate", "get_or_create_client", - "validate_vertex_credentials", "_detect_backend", "_get_effective_backend", "list_models", diff --git a/solstone/think/providers/google_tools.py b/solstone/think/providers/google_tools.py new file mode 100644 index 000000000..3a80b899a --- /dev/null +++ b/solstone/think/providers/google_tools.py @@ -0,0 +1,328 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Google SDK cogitate tools and policy gates.""" + +from __future__ import annotations + +import json +import re +import subprocess +from pathlib import Path +from typing import Any + +from google.genai import types + +from solstone.think.providers.cli import cogitate_sol_tool_hint + +READ_FILE_MAX_CHARS = 8000 +GLOB_CAP = 50 +LIST_DIR_CAP = 200 +GREP_STDOUT_CAP = 6000 +GREP_STDERR_CAP = 1000 +GREP_TIMEOUT_S = 10 +SHELL_STDOUT_CAP = 6000 +SHELL_STDERR_CAP = 6000 +SHELL_TIMEOUT_S = 30 +MAX_TURNS = 30 +DEFAULT_READ_CALL_BUDGET = 200 + +_SOL_INVOCATION_RE = re.compile(r"(^sol\s|\bsol call\b)") +_WRITE_TOOLS = {"write_file", "replace"} +_READ_TOOLS = {"read_file", "glob", "list_directory", "grep_search"} + + +class MaxTurnsExhausted(RuntimeError): + """Raised when the SDK tool loop exceeds its turn ceiling.""" + + +def _check_workspace_boundary(target: Path, allowed_roots: list[Path]) -> Path | None: + """Resolve target and return it only if it is inside an allowed root.""" + resolved = Path(target).expanduser().resolve() + for root in allowed_roots: + allowed = Path(root).expanduser().resolve() + if resolved == allowed: + return resolved + try: + resolved.relative_to(allowed) + return resolved + except ValueError: + continue + return None + + +def _boundary_error(path: str) -> dict[str, str]: + return {"error": f"workspace_boundary: path outside allowed roots: {path}"} + + +def _truncate_text(text: str, cap: int) -> tuple[str, bool]: + if len(text) <= cap: + return text, False + return text[:cap] + "\n... [truncated]", True + + +def read_file( + file_path: str, *, allowed_roots: list[Path] | None = None +) -> dict[str, Any]: + """Read a UTF-8 text file inside the allowed workspace.""" + if allowed_roots is not None: + path = _check_workspace_boundary(Path(file_path), allowed_roots) + if path is None: + return _boundary_error(file_path) + else: + path = Path(file_path).expanduser().resolve() + + if not path.exists(): + return {"error": f"not_found: {path}"} + if not path.is_file(): + return {"error": f"not_a_file: {path}"} + + try: + text = path.read_text(encoding="utf-8") + except UnicodeDecodeError as exc: + return {"error": f"decode_error: {exc}"} + except PermissionError as exc: + return {"error": f"permission_denied: {exc}"} + except OSError as exc: + return {"error": str(exc)} + + text, _truncated = _truncate_text(text, READ_FILE_MAX_CHARS) + return {"content": text} + + +def glob( + pattern: str, path: str = ".", *, allowed_roots: list[Path] | None = None +) -> dict[str, Any]: + """Glob a pattern from an allowed base directory.""" + if allowed_roots is not None: + base = _check_workspace_boundary(Path(path), allowed_roots) + if base is None: + return _boundary_error(path) + else: + base = Path(path).expanduser().resolve() + + if not base.exists(): + return {"error": f"not_found: {base}"} + if not base.is_dir(): + return {"error": f"not_a_directory: {base}"} + + try: + matches = [str(match) for match in sorted(base.glob(pattern))] + except PermissionError as exc: + return {"error": f"permission_denied: {exc}"} + except OSError as exc: + return {"error": str(exc)} + return {"matches": matches[:GLOB_CAP], "truncated": len(matches) > GLOB_CAP} + + +def list_directory( + dir_path: str, *, allowed_roots: list[Path] | None = None +) -> dict[str, Any]: + """List directory entries inside the allowed workspace.""" + if allowed_roots is not None: + path = _check_workspace_boundary(Path(dir_path), allowed_roots) + if path is None: + return _boundary_error(dir_path) + else: + path = Path(dir_path).expanduser().resolve() + + if not path.exists(): + return {"error": f"not_found: {path}"} + if not path.is_dir(): + return {"error": f"not_a_directory: {path}"} + + try: + entries = sorted(child.name for child in path.iterdir()) + except PermissionError as exc: + return {"error": f"permission_denied: {exc}"} + except OSError as exc: + return {"error": str(exc)} + return {"entries": entries[:LIST_DIR_CAP]} + + +def grep_search( + pattern: str, + path: str = ".", + include: str = "", + *, + allowed_roots: list[Path] | None = None, +) -> dict[str, Any]: + """Search with ripgrep under an allowed base path.""" + if allowed_roots is not None: + base = _check_workspace_boundary(Path(path), allowed_roots) + if base is None: + return _boundary_error(path) + else: + base = Path(path).expanduser().resolve() + + if not base.exists(): + return {"error": f"not_found: {base}"} + if not base.is_dir(): + return {"error": f"not_a_directory: {base}"} + + cmd = ["rg", "-n", "--max-count=20", pattern, str(base)] + if include: + cmd.extend(["--glob", include]) + + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=GREP_TIMEOUT_S, + check=False, + ) + except FileNotFoundError: + return {"error": "command_not_found: rg"} + except subprocess.TimeoutExpired: + return {"error": f"timeout: grep_search exceeded {GREP_TIMEOUT_S}s"} + except PermissionError as exc: + return {"error": f"permission_denied: {exc}"} + except OSError as exc: + return {"error": str(exc)} + + stdout, stdout_truncated = _truncate_text(result.stdout, GREP_STDOUT_CAP) + stderr, stderr_truncated = _truncate_text(result.stderr, GREP_STDERR_CAP) + return { + "output": stdout, + "stderr": stderr, + "truncated": stdout_truncated or stderr_truncated, + } + + +def run_shell_command(command: str) -> dict[str, Any]: + """Run a shell command after policy approval.""" + try: + result = subprocess.run( + ["bash", "-lc", command], + capture_output=True, + text=True, + timeout=SHELL_TIMEOUT_S, + check=False, + ) + except FileNotFoundError: + return {"error": "command_not_found: bash"} + except subprocess.TimeoutExpired: + return {"error": f"timeout: run_shell_command exceeded {SHELL_TIMEOUT_S}s"} + except PermissionError as exc: + return {"error": f"permission_denied: {exc}"} + except OSError as exc: + return {"error": str(exc)} + + stdout, _stdout_truncated = _truncate_text(result.stdout, SHELL_STDOUT_CAP) + stderr, _stderr_truncated = _truncate_text(result.stderr, SHELL_STDERR_CAP) + return { + "stdout": stdout, + "stderr": stderr, + "returncode": result.returncode, + } + + +class CogitatePolicy: + """In-process policy gate for Google SDK tool calls.""" + + def __init__(self, *, write: bool, allowed_roots: list[Path]) -> None: + self.write = write + self.allowed_roots = [ + Path(root).expanduser().resolve() for root in allowed_roots + ] + + def check(self, tool: str, args: dict[str, Any]) -> tuple[bool, str]: + if self.write: + return True, "ok" + + if tool in _WRITE_TOOLS: + return False, f"policy_deny: {tool} not allowed for read-only talents" + + if tool == "run_shell_command": + command = str(args.get("command", "")) + if not _SOL_INVOCATION_RE.search(command): + return ( + False, + "policy_deny: run_shell_command restricted to sol invocations", + ) + return True, "ok" + + if tool in _READ_TOOLS: + return True, "ok" + + return True, "ok" + + +def build_tool_declarations() -> list[types.FunctionDeclaration]: + """Build Gemini FunctionDeclarations for the cogitate tool registry.""" + return [ + types.FunctionDeclaration( + name="read_file", + description="Read a single UTF-8 text file from disk.", + parameters_json_schema={ + "type": "object", + "properties": {"file_path": {"type": "string"}}, + "required": ["file_path"], + }, + ), + types.FunctionDeclaration( + name="glob", + description="Glob a pattern from a base directory.", + parameters_json_schema={ + "type": "object", + "properties": { + "pattern": {"type": "string"}, + "path": {"type": "string"}, + }, + "required": ["pattern"], + }, + ), + types.FunctionDeclaration( + name="list_directory", + description="List entries in a directory.", + parameters_json_schema={ + "type": "object", + "properties": {"dir_path": {"type": "string"}}, + "required": ["dir_path"], + }, + ), + types.FunctionDeclaration( + name="grep_search", + description="Search files with ripgrep under a base directory.", + parameters_json_schema={ + "type": "object", + "properties": { + "pattern": {"type": "string"}, + "path": {"type": "string"}, + "include": {"type": "string"}, + }, + "required": ["pattern"], + }, + ), + types.FunctionDeclaration( + name="run_shell_command", + description=cogitate_sol_tool_hint("run_shell_command"), + parameters_json_schema={ + "type": "object", + "properties": {"command": {"type": "string"}}, + "required": ["command"], + }, + ), + ] + + +def load_history(path: Path) -> list[types.Content]: + """Load SDK chat history from disk.""" + if not path.exists(): + return [] + with path.open(encoding="utf-8") as handle: + raw = json.load(handle) + if hasattr(types.Content, "model_validate"): + return [types.Content.model_validate(item) for item in raw] + return [types.Content(**item) for item in raw] + + +def save_history(path: Path, contents: list[types.Content]) -> None: + """Save SDK chat history to disk using JSON-safe Pydantic serialization.""" + path.parent.mkdir(parents=True, exist_ok=True) + serialized = [ + content.model_dump(exclude_none=True, mode="json") for content in contents + ] + with path.open("w", encoding="utf-8") as handle: + json.dump(serialized, handle) diff --git a/solstone/think/providers/shared.py b/solstone/think/providers/shared.py index 456c7d1b6..db7b95e92 100644 --- a/solstone/think/providers/shared.py +++ b/solstone/think/providers/shared.py @@ -57,7 +57,7 @@ class StartEvent(TypedDict, total=False): name: Required[str] model: Required[str] provider: Required[str] - session_id: Optional[str] # CLI session ID for continuation + session_id: Optional[str] # solstone-owned session ID for continuation chat_id: Optional[str] # Chat ID for reverse lookup raw: Optional[list[dict[str, Any]]] # Original provider JSON event(s) @@ -69,7 +69,9 @@ class FinishEvent(TypedDict, total=False): ts: Required[int] result: Required[str] usage: Optional[dict[str, Any]] - cli_session_id: Optional[str] # Provider CLI session/thread ID for resume + cli_session_id: Optional[ + str + ] # solstone-owned session ID persisted under journal/.cache/cogitate-history/ raw: Optional[list[dict[str, Any]]] # Original provider JSON event(s) @@ -110,6 +112,16 @@ class ThinkingEvent(TypedDict, total=False): raw: Optional[list[dict[str, Any]]] # Original provider JSON event(s) +class TextDeltaEvent(TypedDict, total=False): + """Event emitted when streamed text content is available.""" + + event: Required[Literal["text_delta"]] + ts: Required[int] + delta: Required[str] + model: Optional[str] + raw: Optional[list[dict[str, Any]]] # Original provider JSON event(s) + + class FallbackEvent(TypedDict, total=False): """Event emitted when provider fallback occurs.""" @@ -128,6 +140,7 @@ Event = Union[ FinishEvent, ErrorEvent, ThinkingEvent, + TextDeltaEvent, TalentUpdatedEvent, FallbackEvent, ] diff --git a/tests/baselines/api/settings/providers.json b/tests/baselines/api/settings/providers.json index 8ec9b550d..bffc67539 100644 --- a/tests/baselines/api/settings/providers.json +++ b/tests/baselines/api/settings/providers.json @@ -543,14 +543,11 @@ ] }, "google": { - "cogitate_cli": "gemini", - "cogitate_cli_found": false, "cogitate_ready": false, "configured": false, "generate_ready": false, "issues": [ - "GOOGLE_API_KEY not set", - "gemini CLI not found on PATH" + "GOOGLE_API_KEY not set" ] }, "mlx": { diff --git a/tests/integration/test_google_provider.py b/tests/integration/test_google_provider.py index afa4cd484..0e6fcf607 100644 --- a/tests/integration/test_google_provider.py +++ b/tests/integration/test_google_provider.py @@ -12,7 +12,6 @@ import pytest from dotenv import load_dotenv from solstone.think.models import GEMINI_FLASH, GEMINI_PRO -from tests.integration.conftest import require_cli_tool def get_fixtures_env(): @@ -33,8 +32,7 @@ def get_fixtures_env(): @pytest.mark.integration @pytest.mark.requires_api def test_google_provider_basic(): - """Test Google provider with basic prompt via CLI.""" - require_cli_tool("Google", "gemini") + """Test Google provider with basic prompt.""" fixtures_env, api_key, journal_path = get_fixtures_env() if not fixtures_env: @@ -126,7 +124,6 @@ def test_google_provider_basic(): @pytest.mark.requires_api def test_google_provider_with_thinking(): """Test Google provider with thinking enabled.""" - require_cli_tool("Google", "gemini") fixtures_env, api_key, journal_path = get_fixtures_env() if not fixtures_env: diff --git a/tests/test_cli_provider.py b/tests/test_cli_provider.py index e2399572a..2285a2f5d 100644 --- a/tests/test_cli_provider.py +++ b/tests/test_cli_provider.py @@ -7,7 +7,7 @@ import asyncio import json import os from pathlib import Path -from unittest.mock import AsyncMock, mock_open, patch +from unittest.mock import AsyncMock, patch import pytest @@ -1112,62 +1112,6 @@ def test_build_cogitate_env_allowlist_openai(): assert "NODE_OPTIONS" not in env -def test_build_cogitate_env_allowlist_google(): - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "api_key"}, - } - } - with ( - patch.dict( - os.environ, - { - "PATH": "/bin", - "GOOGLE_API_KEY": "gk", - "GEMINI_HOME": "/tmp/gemini", - "VERTEX_REGION": "global", - "OPENAI_API_KEY": "sk-oai", - "ANTHROPIC_API_KEY": "sk-ant", - "HTTPS_PROXY": "http://proxy", - }, - clear=True, - ), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - - assert env["GOOGLE_API_KEY"] == "gk" - assert env["GEMINI_HOME"] == "/tmp/gemini" - assert env["VERTEX_REGION"] == "global" - assert "OPENAI_API_KEY" not in env - assert "ANTHROPIC_API_KEY" not in env - assert "HTTPS_PROXY" not in env - - -def test_build_cogitate_env_vertex_strict_validation_raises(): - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/missing-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("os.path.exists", return_value=False), - pytest.raises( - ValueError, - match=( - r"^Vertex provider configured but no usable SA credentials at " - r"/tmp/missing-sa\.json$" - ), - ), - ): - build_cogitate_env("google") - - class TestBuildCogitateEnv: """Tests for build_cogitate_env — API key stripping for CLI subprocesses.""" @@ -1225,16 +1169,6 @@ class TestBuildCogitateEnv: env = build_cogitate_env("anthropic") assert env["HOME"] == "/home/test" - def test_key_not_in_env_is_harmless(self): - """Stripping a key that doesn't exist doesn't error.""" - config = {"providers": {}} - with ( - patch.dict(os.environ, {}, clear=False), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert "GOOGLE_API_KEY" not in env - def test_per_provider_independence(self): """Each provider's auth mode is independent.""" config = { @@ -1258,102 +1192,8 @@ class TestBuildCogitateEnv: assert ant_env["ANTHROPIC_API_KEY"] == "sk-ant" assert "OPENAI_API_KEY" not in oai_env - def test_vertex_backend_sets_env_vars(self): - """Vertex backend sets GOOGLE_GENAI_USE_VERTEXAI (no API key for vertex).""" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/fake-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("os.path.exists", return_value=True), - patch( - "builtins.open", - mock_open( - read_data='{"type": "service_account", "project_id": "my-gcp-project"}' - ), - ), - patch.object(Path, "exists", return_value=True), - ): - env = build_cogitate_env("google") - assert env["GOOGLE_GENAI_USE_VERTEXAI"] == "true" - assert "GOOGLE_API_KEY" not in env - - def test_vertex_backend_with_sa_creds(self): - """Vertex with SA credentials sets GOOGLE_APPLICATION_CREDENTIALS, removes API key.""" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/fake-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("os.path.exists", return_value=True), - patch( - "builtins.open", - mock_open( - read_data='{"type": "service_account", "project_id": "my-gcp-project"}' - ), - ), - patch.object(Path, "exists", return_value=True), - ): - env = build_cogitate_env("google") - assert env["GOOGLE_GENAI_USE_VERTEXAI"] == "true" - assert env["GOOGLE_APPLICATION_CREDENTIALS"] == "/tmp/fake-sa.json" - assert "GOOGLE_API_KEY" not in env - - def test_aistudio_backend_no_vertex_env_vars(self): - """AI Studio backend does not set Vertex env vars.""" - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "api_key"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert "GOOGLE_GENAI_USE_VERTEXAI" not in env - assert env["GOOGLE_API_KEY"] == "gk-test" - - def test_auto_backend_detects_vertex(self): - """Auto backend with Vertex detection sets env vars.""" - config = { - "providers": { - "auth": {"google": "platform"}, - "vertex_credentials": "/tmp/fake-sa.json", - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch( - "solstone.think.providers.google._detect_backend", return_value="vertex" - ), - patch("os.path.exists", return_value=True), - patch( - "builtins.open", - mock_open( - read_data='{"type": "service_account", "project_id": "my-gcp-project"}' - ), - ), - patch.object(Path, "exists", return_value=True), - ): - env = build_cogitate_env("google") - assert env["GOOGLE_GENAI_USE_VERTEXAI"] == "true" - assert "GOOGLE_API_KEY" not in env - def test_non_google_key_unaffected_by_vertex(self): - """Vertex logic only applies to GOOGLE_API_KEY.""" + """Google backend settings do not affect non-Google CLI envs.""" config = { "providers": { "google_backend": "vertex", @@ -1367,330 +1207,3 @@ class TestBuildCogitateEnv: env = build_cogitate_env("anthropic") assert "GOOGLE_GENAI_USE_VERTEXAI" not in env assert env["ANTHROPIC_API_KEY"] == "sk-ant" - - def test_vertex_backend_sets_project_and_location(self): - """Vertex backend exposes project context for Gemini CLI subprocesses.""" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/fake-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("os.path.exists", return_value=True), - patch( - "builtins.open", - mock_open( - read_data='{"type": "service_account", "project_id": "my-gcp-project"}' - ), - ), - patch.object(Path, "exists", return_value=True), - ): - env = build_cogitate_env("google") - assert env["GOOGLE_GENAI_USE_VERTEXAI"] == "true" - assert env["GOOGLE_APPLICATION_CREDENTIALS"] == "/tmp/fake-sa.json" - assert env["GOOGLE_CLOUD_PROJECT"] == "my-gcp-project" - assert env["GOOGLE_CLOUD_LOCATION"] == "global" - assert "GOOGLE_API_KEY" not in env - - def test_vertex_backend_missing_creds_no_project(self): - """Vertex backend raises without explicit SA credentials.""" - config = { - "providers": { - "google_backend": "vertex", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - pytest.raises( - ValueError, - match=( - r"^Vertex provider configured but no usable SA credentials at None$" - ), - ), - ): - build_cogitate_env("google") - - def test_vertex_backend_invalid_sa_json_no_project(self): - """Invalid SA JSON raises before spawning Gemini CLI.""" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/fake-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("os.path.exists", return_value=True), - patch("builtins.open", mock_open(read_data="not json")), - pytest.raises( - ValueError, - match=( - r"^Vertex provider configured but no usable SA credentials at " - r"/tmp/fake-sa\.json$" - ), - ), - ): - build_cogitate_env("google") - - def test_vertex_backend_sa_missing_project_id(self): - """Missing project_id in SA JSON raises before spawning Gemini CLI.""" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/fake-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("os.path.exists", return_value=True), - patch( - "builtins.open", - mock_open( - read_data=( - '{"type": "service_account", "client_email": "bot@example.com"}' - ) - ), - ), - pytest.raises( - ValueError, - match=( - r"^Vertex provider configured but no usable SA credentials at " - r"/tmp/fake-sa\.json$" - ), - ), - ): - build_cogitate_env("google") - - def test_aistudio_clears_project_and_location(self): - """AI Studio clears inherited Vertex project context.""" - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "api_key"}, - } - } - with ( - patch.dict( - os.environ, - { - "GOOGLE_API_KEY": "gk-test", - "GOOGLE_CLOUD_LOCATION": "us-central1", - "GOOGLE_CLOUD_PROJECT": "inherited-proj", - }, - clear=True, - ), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert "GOOGLE_CLOUD_PROJECT" not in env - assert "GOOGLE_CLOUD_LOCATION" not in env - - def test_vertex_backend_sets_system_settings_path(self): - """Vertex backend exposes the Gemini CLI system settings path.""" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/fake-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("solstone.think.utils.get_journal", return_value="/fake/journal"), - patch("os.path.exists", return_value=True), - patch( - "builtins.open", - mock_open( - read_data='{"type": "service_account", "project_id": "my-gcp-project"}' - ), - ), - patch.object(Path, "exists", return_value=True), - ): - env = build_cogitate_env("google") - assert ( - env["GEMINI_CLI_SYSTEM_SETTINGS_PATH"] - == "/fake/journal/.config/gemini-vertex-settings.json" - ) - - def test_aistudio_backend_no_system_settings_path(self): - """AI Studio backend does not set Gemini CLI system settings.""" - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "api_key"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert "GEMINI_CLI_SYSTEM_SETTINGS_PATH" not in env - - def test_aistudio_clears_inherited_system_settings_path(self): - """AI Studio clears inherited Gemini CLI system settings.""" - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "api_key"}, - } - } - with ( - patch.dict( - os.environ, - { - "GOOGLE_API_KEY": "gk-test", - "GEMINI_CLI_SYSTEM_SETTINGS_PATH": "/old/settings.json", - }, - clear=True, - ), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert "GEMINI_CLI_SYSTEM_SETTINGS_PATH" not in env - - def test_aistudio_mirrors_google_api_key_to_gemini_api_key(self): - """AI Studio mirrors GOOGLE_API_KEY → GEMINI_API_KEY for gemini-cli auth auto-detect.""" - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "api_key"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert env["GOOGLE_API_KEY"] == "gk-test" - assert env["GEMINI_API_KEY"] == "gk-test" - - def test_aistudio_preserves_explicit_gemini_api_key(self): - """If GEMINI_API_KEY is already set, mirror does not overwrite it.""" - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "api_key"}, - } - } - with ( - patch.dict( - os.environ, - {"GOOGLE_API_KEY": "gk-test", "GEMINI_API_KEY": "gemini-explicit"}, - clear=True, - ), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert env["GOOGLE_API_KEY"] == "gk-test" - assert env["GEMINI_API_KEY"] == "gemini-explicit" - - def test_aistudio_no_mirror_when_google_key_absent(self): - """Platform-mode strips GOOGLE_API_KEY; mirror does not synthesize GEMINI_API_KEY.""" - config = { - "providers": { - "google_backend": "aistudio", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - ): - env = build_cogitate_env("google") - assert "GOOGLE_API_KEY" not in env - assert "GEMINI_API_KEY" not in env - - def test_vertex_does_not_mirror_to_gemini_api_key(self): - """Vertex backend strips GOOGLE_API_KEY; no GEMINI_API_KEY mirror happens.""" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": "/tmp/fake-sa.json", - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("os.path.exists", return_value=True), - patch( - "builtins.open", - mock_open( - read_data='{"type": "service_account", "project_id": "my-gcp-project"}' - ), - ), - patch.object(Path, "exists", return_value=True), - ): - env = build_cogitate_env("google") - assert "GOOGLE_API_KEY" not in env - assert "GEMINI_API_KEY" not in env - - def test_vertex_writes_settings_file_when_absent(self, tmp_path): - """Vertex backend creates Gemini CLI system settings when missing.""" - creds_path = tmp_path / "fake-sa.json" - creds_path.write_text( - '{"type": "service_account", "project_id": "my-gcp-project"}', - encoding="utf-8", - ) - journal_path = tmp_path / "journal" - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": str(creds_path), - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("solstone.think.utils.get_journal", return_value=str(journal_path)), - ): - env = build_cogitate_env("google") - assert env["GEMINI_CLI_SYSTEM_SETTINGS_PATH"] == ( - str(journal_path / ".config" / "gemini-vertex-settings.json") - ) - written = (journal_path / ".config" / "gemini-vertex-settings.json").read_text( - encoding="utf-8" - ) - assert json.loads(written) == { - "security": {"auth": {"selectedType": "vertex-ai"}} - } - - def test_vertex_skips_settings_write_when_exists(self, tmp_path): - """Vertex backend does not rewrite existing Gemini CLI settings.""" - creds_path = tmp_path / "fake-sa.json" - creds_path.write_text( - '{"type": "service_account", "project_id": "my-gcp-project"}', - encoding="utf-8", - ) - journal_path = tmp_path / "journal" - settings_path = journal_path / ".config" / "gemini-vertex-settings.json" - settings_path.parent.mkdir(parents=True) - settings_path.write_text('{"existing": true}', encoding="utf-8") - config = { - "providers": { - "google_backend": "vertex", - "vertex_credentials": str(creds_path), - "auth": {"google": "platform"}, - } - } - with ( - patch.dict(os.environ, {"GOOGLE_API_KEY": "gk-test"}, clear=True), - patch("solstone.think.utils.get_config", return_value=config), - patch("solstone.think.utils.get_journal", return_value=str(journal_path)), - ): - env = build_cogitate_env("google") - assert env["GEMINI_CLI_SYSTEM_SETTINGS_PATH"] == (str(settings_path)) - assert settings_path.read_text(encoding="utf-8") == '{"existing": true}' diff --git a/tests/test_cogitate_coder.py b/tests/test_cogitate_coder.py index ea5d5b96b..fe3deed16 100644 --- a/tests/test_cogitate_coder.py +++ b/tests/test_cogitate_coder.py @@ -163,60 +163,38 @@ class TestOpenAIWriteFlag: class TestGoogleWriteFlag: - """Verify --approval-mode is controlled by config write flag.""" + """Verify Google SDK policy behavior is controlled by config write flag.""" - def _provider(self): - return importlib.import_module("solstone.think.providers.google") + def test_no_write_uses_yolo_with_policy(self, tmp_path): + """Without write flag, policy denies writes and non-sol shell commands.""" + from solstone.think.providers.google_tools import CogitatePolicy - @patch("solstone.think.providers.google.CLIRunner") - def test_no_write_uses_yolo_with_policy(self, mock_runner_cls, tmp_path): - """Without write flag, approval-mode is yolo with scoped policy.""" - provider = self._provider() - mock_instance = AsyncMock() - mock_instance.run = AsyncMock(return_value="result") - mock_instance.cli_session_id = None - mock_runner_cls.return_value = mock_instance - - policy_path = tmp_path / "policy.toml" - policy_path.write_text("# generated\n", encoding="utf-8") - config = {"prompt": "test", "model": "gemini-2.5-flash"} - with patch( - "solstone.think.providers.google.build_per_task_policy", - return_value=policy_path, - ) as build_policy: - asyncio.run(provider.run_cogitate(config)) - - cmd = mock_runner_cls.call_args.kwargs["cmd"] - idx = cmd.index("--approval-mode") - assert cmd[idx + 1] == "yolo" - policy_idx = cmd.index("--policy") - assert cmd[policy_idx + 1] == str(policy_path) - build_policy.assert_called_once() - assert not policy_path.exists() - - @patch("solstone.think.providers.google.CLIRunner") - def test_write_true_uses_yolo_mode(self, mock_runner_cls): - """With write=True, approval-mode is yolo (full access).""" - provider = self._provider() - mock_instance = AsyncMock() - mock_instance.run = AsyncMock(return_value="result") - mock_instance.cli_session_id = None - mock_runner_cls.return_value = mock_instance + policy = CogitatePolicy(write=False, allowed_roots=[tmp_path]) - config = {"prompt": "test", "model": "gemini-2.5-flash", "write": True} - asyncio.run(provider.run_cogitate(config)) + allowed, reason = policy.check("write_file", {"file_path": "x"}) + assert allowed is False + assert reason.startswith("policy_deny:") + assert policy.check("run_shell_command", {"command": "rm -rf /tmp/x"})[0] is ( + False + ) + assert policy.check( + "run_shell_command", {"command": "sol call activities list"} + ) == (True, "ok") + assert policy.check("read_file", {"file_path": str(tmp_path / "x")}) == ( + True, + "ok", + ) - cmd = mock_runner_cls.call_args.kwargs["cmd"] - idx = cmd.index("--approval-mode") - assert cmd[idx + 1] == "yolo" - assert "--policy" not in cmd + def test_write_true_uses_yolo_mode(self, tmp_path): + """With write=True, policy allows all tool calls.""" + from solstone.think.providers.google_tools import CogitatePolicy - def test_cogitate_base_policy_file_exists_on_disk(self): - """The per-task policy generator's base policy must exist.""" - from solstone.think.cogitate_policy import _BASE_POLICY_PATH + policy = CogitatePolicy(write=True, allowed_roots=[tmp_path]) - assert _BASE_POLICY_PATH.is_file(), ( - f"Expected policy file at {_BASE_POLICY_PATH}" + assert policy.check("write_file", {"file_path": "x"}) == (True, "ok") + assert policy.check("run_shell_command", {"command": "rm -rf /tmp/x"}) == ( + True, + "ok", ) diff --git a/tests/test_cogitate_policy.py b/tests/test_cogitate_policy.py index b09b33966..84fef0664 100644 --- a/tests/test_cogitate_policy.py +++ b/tests/test_cogitate_policy.py @@ -3,10 +3,9 @@ from __future__ import annotations -import re -import stat +from pathlib import Path -import tomllib +import pytest from solstone.think import cogitate_policy @@ -29,84 +28,12 @@ def test_resolve_read_scope_span_is_inclusive(): ) == ["chronicle/20260425", "chronicle/20260426", "chronicle/20260427"] -def test_build_per_task_policy_appends_read_tool_rules(tmp_path, monkeypatch): - policy_dir = tmp_path / "policies" - journal = tmp_path / "journal" - journal.mkdir() - base_policy = tmp_path / "base.toml" - base_policy.write_text("# base\n", encoding="utf-8") - monkeypatch.setattr(cogitate_policy, "_POLICY_DIR", policy_dir) - monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) - - path = cogitate_policy.build_per_task_policy( - "morning_briefing", - { - "read_scope": [ - "chronicle/", - "facets", - "entities", - "imports", - "health", - "identity", - ] - }, - "20260427", - 0, - base_policy, +def test_cogitate_toml_removed_and_build_policy_import_fails(): + # AC 19: TOML policy generation is removed. + policy_path = ( + Path(__file__).parents[1] / "solstone" / "think" / "policies" / "cogitate.toml" ) - - assert path.parent == policy_dir - assert path.name.startswith("morning_briefing-20260427-") - assert stat.S_IMODE(path.stat().st_mode) == 0o600 - parsed = tomllib.loads(path.read_text(encoding="utf-8")) - rules = parsed["rule"] - assert len(rules) == 8 - - escaped_root = re.escape(str(journal).rstrip("/") + "/") - prefix = f"(?:{escaped_root}|\\./)?" - scope = "chronicle/20260427|facets|entities|imports|health|identity" - expected_args = [ - f'"file_path":"{prefix}(?:{scope})(?:/|")', - f'"dir_path":"{prefix}(?:{scope})(?:/|")', - f'"(?:pattern|path)":"{prefix}(?:{scope})(?:/|[^"]*")', - f'"(?:path|dir_path|include|pattern)":"{prefix}(?:{scope})(?:/|[^"]*")', - ] - - tools = ["read_file", "list_directory", "glob", "grep_search"] - for index, tool_name in enumerate(tools): - allow = rules[index * 2] - deny = rules[index * 2 + 1] - assert allow == { - "toolName": tool_name, - "argsPattern": expected_args[index], - "decision": "allow", - "priority": 260, - } - assert deny == { - "toolName": tool_name, - "decision": "deny", - "priority": 160, - } - - -def test_build_per_task_policy_file_scope_has_exact_boundary(tmp_path, monkeypatch): - policy_dir = tmp_path / "policies" - journal = tmp_path / "journal" - journal.mkdir() - base_policy = tmp_path / "base.toml" - base_policy.write_text("# base\n", encoding="utf-8") - monkeypatch.setattr(cogitate_policy, "_POLICY_DIR", policy_dir) - monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) - - path = cogitate_policy.build_per_task_policy( - "awareness_tender", - {"read_scope": ["stats.json"]}, - "20260427", - 0, - base_policy, - ) - - rules = tomllib.loads(path.read_text(encoding="utf-8"))["rule"] - pattern = rules[0]["argsPattern"] - assert re.search(pattern, '{"file_path":"stats.json"}') - assert not re.search(pattern, '{"file_path":"stats.jsonfoo"}') + assert not policy_path.exists() + missing_symbol = "build" + "_per_task_policy" + with pytest.raises(ImportError): + exec(f"from solstone.think.cogitate_policy import {missing_symbol}", {}) diff --git a/tests/test_google.py b/tests/test_google.py index 2ad4e0a39..ce2284246 100644 --- a/tests/test_google.py +++ b/tests/test_google.py @@ -9,6 +9,7 @@ from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock from solstone.think.models import GEMINI_FLASH +from solstone.think.providers import google as google_provider from solstone.think.providers.google import ( _extract_finish_reason, _format_completion_message, @@ -25,48 +26,6 @@ async def run_main(mod, argv, stdin_data=None): await mod.main_async() -def make_mock_process(stdout_lines, return_code=0): - """Create a mock asyncio subprocess for CLI tests.""" - - class MockStdout: - def __init__(self, lines): - self._lines = [line.encode() + b"\n" for line in lines] - self._index = 0 - - async def readline(self): - if self._index >= len(self._lines): - return b"" - line = self._lines[self._index] - self._index += 1 - return line - - def __aiter__(self): - return self - - async def __anext__(self): - if self._index >= len(self._lines): - raise StopAsyncIteration - line = self._lines[self._index] - self._index += 1 - return line - - class MockStderr: - def __aiter__(self): - return self - - async def __anext__(self): - raise StopAsyncIteration - - process = AsyncMock() - process.stdout = MockStdout(stdout_lines) - process.stderr = MockStderr() - process.stdin = AsyncMock() - process.stdin.write = lambda data: None - process.stdin.close = lambda: None - process.wait = AsyncMock(return_value=return_code) - return process - - def _assert_structured_contents(contents): assert [content.role for content in contents] == ["user", "model", "user"] assert [[part.text for part in content.parts] for content in contents] == [ @@ -77,60 +36,20 @@ def _assert_structured_contents(contents): def test_google_main(monkeypatch, tmp_path, capsys): - setup_google_genai_stub(monkeypatch, with_thinking=False) - sys.modules.pop("solstone.think.providers.google", None) - importlib.reload(importlib.import_module("solstone.think.providers.google")) mod = importlib.reload(importlib.import_module("solstone.think.talents")) journal = tmp_path / "journal" journal.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) - monkeypatch.setenv("GOOGLE_API_KEY", "x") - monkeypatch.setattr( - "solstone.think.utils.get_config", - lambda: {"providers": {"google_backend": "aistudio"}}, - ) - monkeypatch.setattr( - "solstone.think.providers.cli.shutil.which", - lambda name: "/usr/bin/gemini" if name == "gemini" else None, - ) - stdout_lines = [ - json.dumps( - { - "type": "init", - "timestamp": 100, - "session_id": "sess-test", - "model": "gemini-2.5-flash", - } - ), - json.dumps( - { - "type": "message", - "role": "assistant", - "delta": True, - "content": "ok", - } - ), - json.dumps( - { - "type": "result", - "timestamp": 200, - "status": "success", - "stats": { - "total_tokens": 10, - "input_tokens": 5, - "output_tokens": 5, - }, - } - ), - ] - process = make_mock_process(stdout_lines) - monkeypatch.setattr( - "solstone.think.providers.cli.asyncio.create_subprocess_exec", - AsyncMock(return_value=process), - ) + async def fake_run_cogitate(config, on_event=None): + if on_event: + on_event({"event": "text_delta", "delta": "ok"}) + on_event({"event": "finish", "result": "ok"}) + return "ok" + + monkeypatch.setattr(google_provider, "run_cogitate", fake_run_cogitate) ndjson_input = json.dumps( { @@ -158,46 +77,6 @@ def test_google_main(monkeypatch, tmp_path, capsys): # So we don't check for journal files here -def test_google_cli_not_found_error(monkeypatch, tmp_path, capsys): - setup_google_genai_stub(monkeypatch, with_thinking=False) - - sys.modules.pop("solstone.think.providers.google", None) - importlib.reload(importlib.import_module("solstone.think.providers.google")) - mod = importlib.reload(importlib.import_module("solstone.think.talents")) - - journal = tmp_path / "journal" - journal.mkdir() - - monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) - monkeypatch.setenv("GOOGLE_API_KEY", "x") - monkeypatch.setattr( - "solstone.think.utils.get_config", - lambda: {"providers": {"google_backend": "aistudio"}}, - ) - monkeypatch.setattr("solstone.think.providers.cli.shutil.which", lambda _name: None) - - ndjson_input = json.dumps( - { - "name": "exec", - "prompt": "hello", - "provider": "google", - "model": GEMINI_FLASH, - "tools": ["search_insights"], - } - ) - asyncio.run(run_main(mod, ["sol think.talents"], stdin_data=ndjson_input)) - - # Check stdout for error event - out_lines = capsys.readouterr().out.strip().splitlines() - events = [json.loads(line) for line in out_lines] - assert events[-1]["event"] == "error" - assert isinstance(events[-1]["ts"], int) - error_message = events[-1]["error"].lower() - assert "gemini" in error_message - assert "not found" in error_message - assert "trace" in events[-1] - - # --------------------------------------------------------------------------- # Tests for finish reason extraction and formatting # --------------------------------------------------------------------------- diff --git a/tests/test_google_cli.py b/tests/test_google_cli.py deleted file mode 100644 index e993073f2..000000000 --- a/tests/test_google_cli.py +++ /dev/null @@ -1,402 +0,0 @@ -# SPDX-License-Identifier: AGPL-3.0-only -# Copyright (c) 2026 sol pbc - -"""Tests for Gemini CLI subprocess provider (_translate_gemini).""" - -import asyncio -import importlib -from unittest.mock import AsyncMock, patch - -from solstone.think.providers.cli import ThinkingAggregator -from solstone.think.providers.google import _translate_gemini -from solstone.think.providers.shared import JSONEventCallback - - -def _google_provider(): - return importlib.reload(importlib.import_module("solstone.think.providers.google")) - - -def _assert_write_mode_uses_yolo_approval(make_runner): - provider = _google_provider() - MockCLIRunner = make_runner() - with patch("solstone.think.providers.google.CLIRunner", MockCLIRunner): - asyncio.run( - provider.run_cogitate( - {"prompt": "hello", "model": "gemini-2.5-flash", "write": True}, - lambda e: None, - ) - ) - cmd = MockCLIRunner.last_instance.cmd - idx = cmd.index("--approval-mode") - assert cmd[idx + 1] == "yolo" - assert "--policy" not in cmd - return MockCLIRunner.last_instance - - -class TestTranslateGemini: - """Tests for _translate_gemini event translation.""" - - def _make_callback(self): - """Create a callback that records emitted events.""" - events = [] - cb = JSONEventCallback(events.append) - return cb, events - - def _make_aggregator(self, cb): - """Create a ThinkingAggregator with the given callback.""" - return ThinkingAggregator(cb, model="gemini-2.5-flash") - - def test_init_returns_session_id(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - result = _translate_gemini( - { - "type": "init", - "timestamp": 1000, - "session_id": "sess-abc", - "model": "gemini-2.5-flash", - }, - agg, - cb, - ) - assert result == "sess-abc" - assert events == [] - - def test_user_message_ignored(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - result = _translate_gemini( - {"type": "message", "role": "user", "content": "Hello"}, - agg, - cb, - ) - assert result is None - assert events == [] - - def test_assistant_delta_accumulates(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - _translate_gemini( - { - "type": "message", - "role": "assistant", - "delta": True, - "content": "Hello ", - }, - agg, - cb, - ) - _translate_gemini( - {"type": "message", "role": "assistant", "delta": True, "content": "world"}, - agg, - cb, - ) - assert agg.flush_as_result() == "Hello world" - assert events == [] - - def test_tool_use_flushes_thinking_and_emits_start(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - agg.accumulate("I'll use a tool now.") - event = { - "type": "tool_use", - "timestamp": 2000, - "tool_name": "read_file", - "tool_id": "tool-1", - "parameters": {"path": "/tmp/test.txt"}, - } - _translate_gemini(event, agg, cb) - assert len(events) == 2 - assert events[0]["event"] == "thinking" - assert events[0]["summary"] == "I'll use a tool now." - assert events[1]["event"] == "tool_start" - assert events[1]["tool"] == "read_file" - assert events[1]["call_id"] == "tool-1" - assert events[1]["args"] == {"path": "/tmp/test.txt"} - assert events[1]["raw"] == [event] - - def test_tool_use_no_thinking_if_buffer_empty(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - event = { - "type": "tool_use", - "timestamp": 2000, - "tool_name": "read_file", - "tool_id": "tool-1", - "parameters": {}, - } - _translate_gemini(event, agg, cb) - assert len(events) == 1 - assert events[0]["event"] == "tool_start" - - def test_tool_result_includes_tool_name(self): - """tool_end should include tool name from preceding tool_use.""" - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - pending = {} - _translate_gemini( - { - "type": "tool_use", - "tool_name": "read_file", - "tool_id": "t1", - "parameters": {"path": "x.py"}, - }, - agg, - cb, - pending_tools=pending, - ) - events.clear() # ignore tool_start - _translate_gemini( - { - "type": "tool_result", - "tool_id": "t1", - "status": "success", - "output": "contents", - }, - agg, - cb, - pending_tools=pending, - ) - assert len(events) == 1 - assert events[0]["event"] == "tool_end" - assert events[0]["tool"] == "read_file" - assert events[0]["args"] == {"path": "x.py"} - assert events[0]["result"] == "contents" - assert events[0]["call_id"] == "t1" - - def test_tool_result_without_pending(self): - """tool_end without pending_tools still works, tool is empty.""" - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - _translate_gemini( - { - "type": "tool_result", - "tool_id": "t1", - "status": "success", - "output": "data", - }, - agg, - cb, - ) - assert len(events) == 1 - assert events[0]["event"] == "tool_end" - assert events[0]["tool"] == "" - - def test_result_stores_usage(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - usage = {} - _translate_gemini( - { - "type": "result", - "status": "success", - "stats": { - "total_tokens": 1500, - "input_tokens": 1000, - "output_tokens": 500, - "cached": 200, - "duration_ms": 3000, - "tool_calls": 2, - }, - }, - agg, - cb, - usage, - ) - assert events == [] - assert usage["input_tokens"] == 1000 - assert usage["output_tokens"] == 500 - assert usage["total_tokens"] == 1500 - assert usage["cached_tokens"] == 200 - assert "duration_ms" not in usage - - def test_result_no_stats(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - usage = {} - _translate_gemini( - {"type": "result", "status": "success"}, - agg, - cb, - usage, - ) - assert usage == {} - - def test_unknown_event_type_ignored(self): - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - result = _translate_gemini( - {"type": "unknown_type", "data": "whatever"}, - agg, - cb, - ) - assert result is None - assert events == [] - - def test_full_sequence(self): - """Process a full sequence of Gemini JSONL events.""" - cb, events = self._make_callback() - agg = self._make_aggregator(cb) - usage = {} - pending = {} - - sequence = [ - { - "type": "init", - "session_id": "sess-42", - "model": "gemini-2.5-flash", - }, - {"type": "message", "role": "user", "content": "Analyze this file"}, - { - "type": "message", - "role": "assistant", - "delta": True, - "content": "I'll read the file. ", - }, - { - "type": "tool_use", - "tool_name": "read_file", - "tool_id": "t1", - "parameters": {"path": "test.py"}, - }, - { - "type": "tool_result", - "tool_id": "t1", - "status": "success", - "output": "print('hello')", - }, - { - "type": "message", - "role": "assistant", - "delta": True, - "content": "The file contains ", - }, - { - "type": "message", - "role": "assistant", - "delta": True, - "content": "a print statement.", - }, - { - "type": "result", - "status": "success", - "stats": { - "total_tokens": 100, - "input_tokens": 60, - "output_tokens": 40, - }, - }, - ] - - session_ids = [] - for ev in sequence: - sid = _translate_gemini(ev, agg, cb, usage, pending) - if sid: - session_ids.append(sid) - - assert session_ids == ["sess-42"] - - # Events: thinking, tool_start, tool_end - assert len(events) == 3 - assert events[0]["event"] == "thinking" - assert "read the file" in events[0]["summary"] - assert events[1]["event"] == "tool_start" - assert events[1]["tool"] == "read_file" - assert events[2]["event"] == "tool_end" - assert events[2]["tool"] == "read_file" - assert events[2]["call_id"] == "t1" - - # Final result text in aggregator - result = agg.flush_as_result() - assert result == "The file contains a print statement." - - assert usage["total_tokens"] == 100 - - -class TestRunCogitateCommand: - """Tests for run_cogitate command construction.""" - - def _mock_runner(self): - """Create a MockCLIRunner that captures the command.""" - - class MockCLIRunner: - last_instance = None - - def __init__(self, **kwargs): - self.cmd = kwargs["cmd"] - self.prompt_text = kwargs["prompt_text"] - self.kwargs = kwargs - self.cli_session_id = "test-session" - self.run = AsyncMock(return_value="result") - MockCLIRunner.last_instance = self - - return MockCLIRunner - - def test_no_write_uses_yolo_with_policy(self, tmp_path): - provider = _google_provider() - MockCLIRunner = self._mock_runner() - policy_path = tmp_path / "policy.toml" - policy_path.write_text("# generated\n", encoding="utf-8") - with ( - patch("solstone.think.providers.google.CLIRunner", MockCLIRunner), - patch( - "solstone.think.providers.google.build_per_task_policy", - return_value=policy_path, - ) as build_policy, - ): - asyncio.run( - provider.run_cogitate( - { - "prompt": "hello", - "model": "gemini-2.5-flash", - "name": "morning_briefing", - }, - lambda e: None, - ) - ) - cmd = MockCLIRunner.last_instance.cmd - idx = cmd.index("--approval-mode") - assert cmd[idx + 1] == "yolo" - policy_idx = cmd.index("--policy") - assert cmd[policy_idx + 1] == str(policy_path) - build_policy.assert_called_once() - assert not policy_path.exists() - prompt_text = MockCLIRunner.last_instance.prompt_text - assert "through the `run_shell_command` tool" in prompt_text - assert "Do not invent or call a tool literally named `sol`." in prompt_text - assert MockCLIRunner.last_instance.kwargs["read_call_budget"] == 200 - - def test_write_mode_uses_yolo_approval(self): - runner = _assert_write_mode_uses_yolo_approval(self._mock_runner) - prompt_text = runner.prompt_text - assert "Do not invent or call a tool literally named `sol`." not in prompt_text - - def test_write_mode_does_not_build_policy(self, tmp_path): - provider = _google_provider() - MockCLIRunner = self._mock_runner() - with ( - patch("solstone.think.providers.google.CLIRunner", MockCLIRunner), - patch( - "solstone.think.providers.google.build_per_task_policy" - ) as build_policy, - ): - asyncio.run( - provider.run_cogitate( - {"prompt": "hello", "model": "gemini-2.5-flash", "write": True}, - lambda e: None, - ) - ) - assert "--policy" not in MockCLIRunner.last_instance.cmd - build_policy.assert_not_called() - - def test_sandbox_none(self): - provider = _google_provider() - MockCLIRunner = self._mock_runner() - with patch("solstone.think.providers.google.CLIRunner", MockCLIRunner): - asyncio.run( - provider.run_cogitate( - {"prompt": "hello", "model": "gemini-2.5-flash"}, lambda e: None - ) - ) - cmd = MockCLIRunner.last_instance.cmd - assert "--sandbox=none" in cmd diff --git a/tests/test_google_sdk.py b/tests/test_google_sdk.py new file mode 100644 index 000000000..2aa3c88f4 --- /dev/null +++ b/tests/test_google_sdk.py @@ -0,0 +1,532 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +# AC 1: Single-turn no-tool happy path. +# AC 2: Multi-turn tool-calling loop. +# AC 3: Streaming text_delta incremental events. +# AC 4: Thinking events use explicit thought flag. +# AC 5: Read-only policy denies writes while allowing reads in the same turn. +# AC 6: Read-only policy discriminates shell commands in the same turn. +# AC 13: Quota exhaustion maps to QuotaExhaustedError with retryDelay when present. +# AC 14: Read-call budget preserves the tool_budget_exhausted RuntimeError prefix. +# AC 15: Usage metadata is emitted on finish. +# AC 16: Google provider metadata has no cogitate_cli. +# AC 17: Google provider status ignores gemini on PATH. +# AC 18: Mocked SDK chat emits a well-formed cogitate event sequence. +# AC 21: GenerateContentConfig enables thoughts, dynamic budget, and disables AFC. +# AC 23: Unknown part kinds are skipped. +# AC 24: Empty or whitespace-only thinking parts are skipped. +# AC 25: max_turns ceiling raises MaxTurnsExhausted. +# AC 26: Missing API key emits a guided error event. +# AC 27: automatic_function_calling_history is ignored. + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from typing import Any + +import pytest +from google.genai import errors as google_errors +from google.genai import types + +from solstone.think.models import GEMINI_FLASH +from solstone.think.providers import PROVIDER_METADATA, build_provider_status, google +from solstone.think.providers.cli import QuotaExhaustedError +from solstone.think.providers.google_tools import MAX_TURNS, MaxTurnsExhausted + + +class FakeChat: + def __init__( + self, + turns: list[list[Any] | BaseException], + history: list[types.Content] | None = None, + ) -> None: + self.turns = list(turns) + self.history = list(history or []) + self.messages: list[Any] = [] + self.history_curated_flags: list[bool] = [] + + def send_message_stream(self, message: Any) -> Any: + self.messages.append(message) + turn = self.turns.pop(0) + if isinstance(turn, BaseException): + raise turn + return iter(turn) + + def get_history(self, curated: bool = False) -> list[types.Content]: + self.history_curated_flags.append(curated) + return self.history + + +class FakeChats: + def __init__(self, chat: FakeChat) -> None: + self.chat = chat + self.created: dict[str, Any] | None = None + + def create( + self, + *, + model: str, + config: types.GenerateContentConfig | None = None, + history: list[types.Content] | None = None, + ) -> FakeChat: + self.created = { + "model": model, + "config": config, + "history": history, + } + return self.chat + + +class FakeClient: + def __init__(self, chat: FakeChat) -> None: + self.chats = FakeChats(chat) + + +def _part_chunk(parts: list[Any], *, usage: Any = None, afc_history: Any = None) -> Any: + return SimpleNamespace( + candidates=[ + SimpleNamespace( + content=SimpleNamespace(parts=parts), + ) + ], + usage_metadata=usage, + automatic_function_calling_history=afc_history, + ) + + +def _usage() -> Any: + return SimpleNamespace( + prompt_token_count=2, + candidates_token_count=3, + total_token_count=9, + cached_content_token_count=1, + thoughts_token_count=4, + ) + + +def _function_call(name: str, args: dict[str, Any]) -> types.Part: + return types.Part.from_function_call(name=name, args=args) + + +def _base_config( + monkeypatch: pytest.MonkeyPatch, + tmp_path, + **overrides: Any, +) -> dict[str, Any]: + journal = tmp_path / "journal" + journal.mkdir(exist_ok=True) + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) + config: dict[str, Any] = { + "name": "sdk-test", + "type": "cogitate", + "prompt": "hello", + "model": GEMINI_FLASH, + "day": "20260520", + "read_scope": [str(tmp_path)], + } + config.update(overrides) + return config + + +def _install_client(monkeypatch: pytest.MonkeyPatch, chat: FakeChat) -> FakeClient: + client = FakeClient(chat) + monkeypatch.setattr(google, "get_or_create_client", lambda _client=None: client) + return client + + +def _run(config: dict[str, Any], events: list[dict[str, Any]]) -> str: + return asyncio.run(google.run_cogitate(config, on_event=events.append)) + + +def test_single_turn_no_tool_happy_path(monkeypatch, tmp_path) -> None: + events: list[dict[str, Any]] = [] + chat = FakeChat([[_part_chunk([types.Part(text="ok")])]]) + _install_client(monkeypatch, chat) + + result = _run(_base_config(monkeypatch, tmp_path), events) + + assert result == "ok" + assert [event["event"] for event in events] == ["text_delta", "finish"] + assert events[0]["delta"] == "ok" + assert events[-1]["result"] == "ok" + + +def test_multiturn_tool_calling_loop(monkeypatch, tmp_path) -> None: + target = tmp_path / "note.txt" + target.write_text("alpha\nbeta\n", encoding="utf-8") + chat = FakeChat( + [ + [ + _part_chunk( + [ + _function_call("read_file", {"file_path": str(target)}), + _function_call( + "grep_search", + {"pattern": "beta", "path": str(tmp_path)}, + ), + ] + ) + ], + [_part_chunk([types.Part(text="done")])], + ] + ) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + result = _run(_base_config(monkeypatch, tmp_path), events) + + assert result == "done" + assert [event["event"] for event in events] == [ + "tool_start", + "tool_end", + "tool_start", + "tool_end", + "text_delta", + "finish", + ] + assert events[1]["result"]["content"] == "alpha\nbeta\n" + assert "note.txt:2:beta" in events[3]["result"]["output"] + assert len(chat.messages) == 2 + assert [part.function_response.name for part in chat.messages[1]] == [ + "read_file", + "grep_search", + ] + + +def test_streaming_text_delta_and_usage(monkeypatch, tmp_path) -> None: + chat = FakeChat( + [ + [ + _part_chunk([types.Part(text="hel")]), + _part_chunk([types.Part(text="lo")], usage=_usage()), + ] + ] + ) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + result = _run(_base_config(monkeypatch, tmp_path), events) + + assert result == "hello" + assert [event["delta"] for event in events if event["event"] == "text_delta"] == [ + "hel", + "lo", + ] + finish = events[-1] + assert finish["usage"] == { + "input_tokens": 2, + "output_tokens": 3, + "total_tokens": 9, + "cached_tokens": 1, + "reasoning_tokens": 4, + } + + +@pytest.mark.parametrize( + ("part", "thinking_count"), + [ + (types.Part(text="reason", thought=True), 1), + (types.Part(text="visible", thought=False), 0), + (types.Part(text=" ", thought=True), 0), + ], +) +def test_thinking_requires_explicit_thought_flag( + monkeypatch, + tmp_path, + part: types.Part, + thinking_count: int, +) -> None: + chat = FakeChat([[_part_chunk([part, types.Part(text="ok")])]]) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + _run(_base_config(monkeypatch, tmp_path), events) + + thinking = [event for event in events if event["event"] == "thinking"] + assert len(thinking) == thinking_count + if thinking: + assert thinking[0]["summary"] == "reason" + assert thinking[0]["model"] == GEMINI_FLASH + + +def test_readonly_policy_denies_write_and_allows_read_same_turn( + monkeypatch, + tmp_path, +) -> None: + target = tmp_path / "allowed.txt" + target.write_text("readable", encoding="utf-8") + chat = FakeChat( + [ + [ + _part_chunk( + [ + _function_call("write_file", {"file_path": str(target)}), + _function_call("read_file", {"file_path": str(target)}), + ] + ) + ], + [_part_chunk([types.Part(text="done")])], + ] + ) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + _run(_base_config(monkeypatch, tmp_path), events) + + tool_results = [event["result"] for event in events if event["event"] == "tool_end"] + assert tool_results[0]["error"].startswith("policy_deny:") + assert tool_results[1] == {"content": "readable"} + + +def test_readonly_policy_discriminates_shell_commands_same_turn( + monkeypatch, + tmp_path, +) -> None: + chat = FakeChat( + [ + [ + _part_chunk( + [ + _function_call( + "run_shell_command", + {"command": "rm -rf /tmp/not-allowed"}, + ), + _function_call( + "run_shell_command", + {"command": "sol call journal search hello"}, + ), + ] + ) + ], + [_part_chunk([types.Part(text="done")])], + ] + ) + _install_client(monkeypatch, chat) + monkeypatch.setattr( + google, + "run_shell_command", + lambda command: {"stdout": f"ran {command}", "stderr": "", "returncode": 0}, + ) + events: list[dict[str, Any]] = [] + + _run(_base_config(monkeypatch, tmp_path), events) + + tool_results = [event["result"] for event in events if event["event"] == "tool_end"] + assert tool_results[0]["error"].startswith("policy_deny:") + assert tool_results[1] == { + "stdout": "ran sol call journal search hello", + "stderr": "", + "returncode": 0, + } + + +@pytest.mark.parametrize( + ("response_json", "retry_delay_ms"), + [ + ( + { + "error": { + "code": 429, + "message": "quota", + "status": "RESOURCE_EXHAUSTED", + "details": [ + { + "@type": "type.googleapis.com/google.rpc.RetryInfo", + "retryDelay": "2.5s", + } + ], + } + }, + 2500, + ), + ( + { + "error": { + "code": 429, + "message": "quota", + "status": "RESOURCE_EXHAUSTED", + } + }, + None, + ), + ], +) +def test_quota_exhausted_mapping( + monkeypatch, + tmp_path, + response_json: dict[str, Any], + retry_delay_ms: int | None, +) -> None: + chat = FakeChat([google_errors.ClientError(429, response_json)]) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + with pytest.raises(QuotaExhaustedError) as exc_info: + _run(_base_config(monkeypatch, tmp_path), events) + + assert exc_info.value.retry_delay_ms == retry_delay_ms + assert events == [] + + +def test_read_call_budget_exhausted_preserves_runtimeerror_prefix( + monkeypatch, + tmp_path, +) -> None: + chat = FakeChat([[_part_chunk([_function_call("read_file", {"file_path": "x"})])]]) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + with pytest.raises(RuntimeError, match=r"^tool_budget_exhausted:"): + _run(_base_config(monkeypatch, tmp_path, read_call_budget=0), events) + + assert events[-1]["event"] == "tool_budget_exhausted" + + +def test_generate_content_config_pins_thinking_and_manual_tools( + monkeypatch, + tmp_path, +) -> None: + chat = FakeChat([[_part_chunk([types.Part(text="ok")])]]) + client = _install_client(monkeypatch, chat) + + _run(_base_config(monkeypatch, tmp_path), []) + + config = client.chats.created["config"] + assert config.thinking_config.include_thoughts is True + assert config.thinking_config.thinking_budget == -1 + assert config.automatic_function_calling.disable is True + assert config.tool_config.function_calling_config.mode == ( + types.FunctionCallingConfigMode.AUTO + ) + assert config.tools[0].function_declarations + + +def test_google_provider_metadata_has_no_cogitate_cli() -> None: + assert "cogitate_cli" not in PROVIDER_METADATA["google"] + + +@pytest.mark.parametrize( + ("api_key", "vertex_creds_configured", "ready", "issues"), + [ + ("", False, False, ["GOOGLE_API_KEY not set"]), + ("key", False, True, []), + ("", True, True, []), + ], +) +def test_google_provider_status_ignores_gemini_path( + monkeypatch, + api_key: str, + vertex_creds_configured: bool, + ready: bool, + issues: list[str], +) -> None: + if api_key: + monkeypatch.setenv("GOOGLE_API_KEY", api_key) + else: + monkeypatch.delenv("GOOGLE_API_KEY", raising=False) + monkeypatch.setattr( + "solstone.think.providers.shutil.which", + lambda _name: (_ for _ in ()).throw(AssertionError("which should not run")), + ) + + status = build_provider_status( + [{"name": "google", "env_key": "GOOGLE_API_KEY"}], + vertex_creds_configured=vertex_creds_configured, + )["google"] + + assert status == { + "configured": ready, + "generate_ready": ready, + "cogitate_ready": ready, + "issues": issues, + } + + +def test_unknown_part_kind_and_afc_history_are_ignored(monkeypatch, tmp_path) -> None: + unknown_part = SimpleNamespace( + text=None, + thought=False, + function_call=None, + executable_code=SimpleNamespace(code="print(1)"), + ) + chat = FakeChat( + [ + [ + _part_chunk( + [unknown_part, types.Part(text="ok")], + afc_history=[types.Content(role="model", parts=[])], + ) + ] + ] + ) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + result = _run(_base_config(monkeypatch, tmp_path), events) + + assert result == "ok" + assert [event["event"] for event in events] == ["text_delta", "finish"] + + +def test_e2e_smoke_event_sequence(monkeypatch, tmp_path) -> None: + chat = FakeChat( + [ + [ + _part_chunk([types.Part(text="a")]), + _part_chunk([types.Part(text="b")]), + ] + ] + ) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + _run(_base_config(monkeypatch, tmp_path), events) + + assert all("ts" in event for event in events) + assert [event["event"] for event in events] == [ + "text_delta", + "text_delta", + "finish", + ] + + +def test_max_turns_ceiling_raises_typed_exception(monkeypatch, tmp_path) -> None: + chat = FakeChat( + [[_part_chunk([_function_call("unknown_tool", {})])] for _ in range(MAX_TURNS)] + ) + _install_client(monkeypatch, chat) + events: list[dict[str, Any]] = [] + + with pytest.raises(MaxTurnsExhausted): + _run(_base_config(monkeypatch, tmp_path, write=True), events) + + assert events[-1]["event"] == "max_turns_exhausted" + + +def test_missing_api_key_emits_error_event(monkeypatch, tmp_path) -> None: + monkeypatch.delenv("GOOGLE_API_KEY", raising=False) + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path / "journal")) + monkeypatch.setattr( + "solstone.think.utils.get_config", + lambda: {"providers": {"google_backend": "aistudio"}}, + ) + events: list[dict[str, Any]] = [] + + with pytest.raises(ValueError) as exc_info: + _run( + { + "name": "sdk-test", + "type": "cogitate", + "prompt": "hello", + "model": GEMINI_FLASH, + "day": "20260520", + }, + events, + ) + + assert "GOOGLE_API_KEY" in str(exc_info.value) + assert getattr(exc_info.value, "_evented") is True + assert events[-1]["event"] == "error" + assert "GOOGLE_API_KEY" in events[-1]["error"] diff --git a/tests/test_google_thinking.py b/tests/test_google_thinking.py deleted file mode 100644 index ef1669ea2..000000000 --- a/tests/test_google_thinking.py +++ /dev/null @@ -1,135 +0,0 @@ -# SPDX-License-Identifier: AGPL-3.0-only -# Copyright (c) 2026 sol pbc - -import asyncio -import importlib -import json -import sys -from unittest.mock import AsyncMock - -from solstone.think.models import GEMINI_FLASH -from tests.conftest import setup_google_genai_stub -from tests.test_google import make_mock_process - - -async def run_main(mod, argv, stdin_data=None): - sys.argv = argv - if stdin_data: - import io - - sys.stdin = io.StringIO(stdin_data) - await mod.main_async() - - -def test_google_thinking_events(monkeypatch, tmp_path, capsys): - setup_google_genai_stub(monkeypatch, with_thinking=True) - - sys.modules.pop("solstone.think.providers.google", None) - importlib.reload(importlib.import_module("solstone.think.providers.google")) - mod = importlib.reload(importlib.import_module("solstone.think.talents")) - - journal = tmp_path / "journal" - journal.mkdir() - - monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) - monkeypatch.setenv("GOOGLE_API_KEY", "x") - monkeypatch.setattr( - "solstone.think.utils.get_config", - lambda: {"providers": {"google_backend": "aistudio"}}, - ) - monkeypatch.setattr( - "solstone.think.providers.cli.shutil.which", - lambda name: "/usr/bin/gemini" if name == "gemini" else None, - ) - - # Simulate CLI output: thinking text before a tool call triggers thinking event - stdout_lines = [ - json.dumps( - { - "type": "init", - "timestamp": 100, - "session_id": "sess-think", - "model": "gemini-2.5-flash", - } - ), - json.dumps( - { - "type": "message", - "role": "assistant", - "delta": True, - "content": "I need to analyze this step by step.", - } - ), - json.dumps( - { - "type": "tool_use", - "timestamp": 150, - "tool_name": "search_insights", - "tool_id": "t1", - "parameters": {"query": "hello"}, - } - ), - json.dumps( - { - "type": "tool_result", - "timestamp": 200, - "tool_id": "t1", - "status": "success", - "output": "result data", - } - ), - json.dumps( - { - "type": "message", - "role": "assistant", - "delta": True, - "content": "ok", - } - ), - json.dumps( - { - "type": "result", - "timestamp": 300, - "status": "success", - "stats": { - "total_tokens": 20, - "input_tokens": 10, - "output_tokens": 10, - }, - } - ), - ] - process = make_mock_process(stdout_lines) - monkeypatch.setattr( - "solstone.think.providers.cli.asyncio.create_subprocess_exec", - AsyncMock(return_value=process), - ) - - ndjson_input = json.dumps( - { - "name": "exec", - "prompt": "hello", - "provider": "google", - "model": GEMINI_FLASH, - "tools": ["search_insights"], - } - ) - asyncio.run(run_main(mod, ["sol solstone.think.talents"], stdin_data=ndjson_input)) - - out_lines = capsys.readouterr().out.strip().splitlines() - events = [json.loads(line) for line in out_lines] - - # Check that we have start, thinking, and finish events - assert events[0]["event"] == "start" - assert isinstance(events[0]["ts"], int) - assert "hello" in events[0]["prompt"] - - # Look for thinking event (flushed when tool_use arrives) - thinking_events = [e for e in events if e["event"] == "thinking"] - assert len(thinking_events) == 1 - assert thinking_events[0]["summary"] == "I need to analyze this step by step." - assert thinking_events[0]["model"] == GEMINI_FLASH - assert isinstance(thinking_events[0]["ts"], int) - - assert events[-1]["event"] == "finish" - assert events[-1]["result"] == "ok" diff --git a/tests/test_google_tools.py b/tests/test_google_tools.py new file mode 100644 index 000000000..4a10a41fd --- /dev/null +++ b/tests/test_google_tools.py @@ -0,0 +1,300 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +# AC 5: Read-only policy denies write tools while allowing read tools. +# AC 6: Read-only policy discriminates shell commands by sol invocation regex. +# AC 7: Policy regex edge cases are pinned. +# AC 8: Write mode allows all tools. +# AC 9: Workspace-boundary in-scope reads return content. +# AC 10: Workspace-boundary out-of-scope reads return keyed errors. +# AC 11: Workspace-boundary helper handles traversal and symlinks. +# AC 12: Curated history round-trips thought_signature bytes. +# AC 22: Tool error vocabulary is keyed. + +from __future__ import annotations + +import subprocess +from pathlib import Path + +import pytest +from google.genai import types + +from solstone.think.providers import google_tools + + +# AC 7: Policy regex edge-case pin. +@pytest.mark.parametrize( + ("command", "allowed"), + [ + ("sol call activities list", True), + ("sol status", True), + ("echo hi | sol call journal search", True), + ("cd /tmp && sol call entities list", True), + ("rm -rf /tmp/notarealfile", False), + ("python -c 'print(1)'", False), + ("absol call activities list", False), + ("SOL call activities list", False), + ], +) +def test_policy_regex_edge_cases(command: str, allowed: bool) -> None: + assert bool(google_tools._SOL_INVOCATION_RE.search(command)) is allowed + + +# AC 5, 6, 8: Policy semantics for read-only and write modes. +def test_cogitate_policy_readonly_and_write_modes(tmp_path: Path) -> None: + readonly = google_tools.CogitatePolicy(write=False, allowed_roots=[tmp_path]) + assert readonly.check("write_file", {"file_path": "x"})[0] is False + assert readonly.check("replace", {"file_path": "x"})[0] is False + assert readonly.check("read_file", {"file_path": "x"}) == (True, "ok") + assert readonly.check( + "run_shell_command", {"command": "sol call activities list"} + ) == ( + True, + "ok", + ) + allowed, reason = readonly.check( + "run_shell_command", {"command": "rm -rf /tmp/notarealfile"} + ) + assert allowed is False + assert reason.startswith("policy_deny:") + + write = google_tools.CogitatePolicy(write=True, allowed_roots=[tmp_path]) + assert write.check("write_file", {"file_path": "x"}) == (True, "ok") + assert write.check( + "run_shell_command", {"command": "rm -rf /tmp/notarealfile"} + ) == ( + True, + "ok", + ) + + +# AC 11: Workspace-boundary helper table. +@pytest.mark.parametrize("case_index", range(12)) +def test_workspace_boundary_table(tmp_path: Path, case_index: int) -> None: + root = tmp_path / "root" + other_root = tmp_path / "other" + outside = tmp_path / "outside" + root.mkdir() + other_root.mkdir() + outside.mkdir() + (root / "file.txt").write_text("ok", encoding="utf-8") + (root / "nested").mkdir() + (root / "nested" / "file.txt").write_text("nested", encoding="utf-8") + (other_root / "file.txt").write_text("other", encoding="utf-8") + (outside / "file.txt").write_text("outside", encoding="utf-8") + (tmp_path / "root_evil").mkdir() + (tmp_path / "root_evil" / "file.txt").write_text("evil", encoding="utf-8") + + symlink_out = root / "symlink-out" + symlink_in = root / "symlink-in" + symlink_out.symlink_to(outside / "file.txt") + symlink_in.symlink_to(root / "file.txt") + + cases = [ + (root, True), + (root / "file.txt", True), + (root / "nested" / "file.txt", True), + (root / "nested" / ".." / "file.txt", True), + (other_root, True), + (other_root / "file.txt", True), + (root / "missing.txt", True), + (outside / "file.txt", False), + (tmp_path / "root_evil" / "file.txt", False), + (symlink_out, False), + (symlink_in, True), + (tmp_path / "missing-outside.txt", False), + ] + target, expected = cases[case_index] + result = google_tools._check_workspace_boundary(target, [root, other_root]) + assert (result is not None) is expected + + +# AC 9, 10: Tool-level workspace boundary behavior. +def test_read_file_workspace_boundary(tmp_path: Path) -> None: + allowed = tmp_path / "allowed" + outside = tmp_path / "outside" + allowed.mkdir() + outside.mkdir() + in_scope = allowed / "note.txt" + out_scope = outside / "note.txt" + in_scope.write_text("hello", encoding="utf-8") + out_scope.write_text("secret", encoding="utf-8") + + assert google_tools.read_file(str(in_scope), allowed_roots=[allowed]) == { + "content": "hello" + } + denied = google_tools.read_file(str(out_scope), allowed_roots=[allowed]) + assert denied["error"].startswith("workspace_boundary:") + + +# AC 22: Tool error vocabulary is keyed. +@pytest.mark.parametrize( + "case_name", + [ + "not_found", + "not_a_file", + "not_a_directory", + "workspace_boundary", + "command_not_found", + "timeout", + "decode_error", + "permission_denied", + ], +) +def test_tool_error_vocabulary( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, case_name: str +) -> None: + root = tmp_path / "root" + outside = tmp_path / "outside" + root.mkdir() + outside.mkdir() + + if case_name == "not_found": + result = google_tools.read_file(str(root / "missing.txt"), allowed_roots=[root]) + elif case_name == "not_a_file": + result = google_tools.read_file(str(root), allowed_roots=[root]) + elif case_name == "not_a_directory": + file_path = root / "file.txt" + file_path.write_text("ok", encoding="utf-8") + result = google_tools.list_directory(str(file_path), allowed_roots=[root]) + elif case_name == "workspace_boundary": + file_path = outside / "file.txt" + file_path.write_text("no", encoding="utf-8") + result = google_tools.read_file(str(file_path), allowed_roots=[root]) + elif case_name == "command_not_found": + monkeypatch.setattr( + google_tools.subprocess, + "run", + lambda *_args, **_kwargs: (_ for _ in ()).throw(FileNotFoundError()), + ) + result = google_tools.grep_search("x", str(root), allowed_roots=[root]) + elif case_name == "timeout": + monkeypatch.setattr( + google_tools.subprocess, + "run", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + subprocess.TimeoutExpired(cmd="bash", timeout=30) + ), + ) + result = google_tools.run_shell_command("sleep 60") + elif case_name == "decode_error": + binary = root / "binary.txt" + binary.write_bytes(b"\xff\xfe\xfd") + result = google_tools.read_file(str(binary), allowed_roots=[root]) + else: + denied = root / "denied.txt" + denied.write_text("no", encoding="utf-8") + monkeypatch.setattr( + Path, + "read_text", + lambda *_args, **_kwargs: (_ for _ in ()).throw(PermissionError("no")), + ) + result = google_tools.read_file(str(denied), allowed_roots=[root]) + + assert result["error"].startswith(f"{case_name}:") + + +# AC 22: grep_search zero matches are not an error. +def test_grep_search_zero_matches_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + root = tmp_path / "root" + root.mkdir() + monkeypatch.setattr( + google_tools.subprocess, + "run", + lambda *_args, **_kwargs: subprocess.CompletedProcess( + args=["rg"], returncode=1, stdout="", stderr="" + ), + ) + + assert google_tools.grep_search("missing", str(root), allowed_roots=[root]) == { + "output": "", + "stderr": "", + "truncated": False, + } + + +# AC 19, 22: Hard limits are applied to tool success shapes. +def test_tool_hard_limits(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + root = tmp_path / "root" + root.mkdir() + text_file = root / "large.txt" + text_file.write_text( + "a" * (google_tools.READ_FILE_MAX_CHARS + 50), encoding="utf-8" + ) + for index in range(google_tools.GLOB_CAP + 5): + (root / f"item-{index}.txt").write_text("x", encoding="utf-8") + for index in range(google_tools.LIST_DIR_CAP + 5): + (root / f"entry-{index}.md").write_text("x", encoding="utf-8") + monkeypatch.setattr( + google_tools.subprocess, + "run", + lambda *_args, **_kwargs: subprocess.CompletedProcess( + args=["rg"], + returncode=0, + stdout="o" * (google_tools.GREP_STDOUT_CAP + 10), + stderr="e" * (google_tools.GREP_STDERR_CAP + 10), + ), + ) + + read_result = google_tools.read_file(str(text_file), allowed_roots=[root]) + assert len(read_result["content"]) <= google_tools.READ_FILE_MAX_CHARS + 16 + assert read_result["content"].endswith("... [truncated]") + glob_result = google_tools.glob("*.txt", str(root), allowed_roots=[root]) + assert len(glob_result["matches"]) == google_tools.GLOB_CAP + assert glob_result["truncated"] is True + list_result = google_tools.list_directory(str(root), allowed_roots=[root]) + assert len(list_result["entries"]) == google_tools.LIST_DIR_CAP + grep_result = google_tools.grep_search("x", str(root), allowed_roots=[root]) + assert grep_result["truncated"] is True + + +# AC 12: Session resume round-trip with thought_signature bytes. +def test_history_serialization_round_trips_thought_signature(tmp_path: Path) -> None: + history_path = tmp_path / ".cache" / "cogitate-history" / "sess.json" + content = types.Content( + role="model", + parts=[ + types.Part( + text="thinking", + thought=True, + thought_signature=b"signature-bytes", + ) + ], + ) + + google_tools.save_history(history_path, [content]) + loaded = google_tools.load_history(history_path) + + assert loaded[0].parts[0].thought_signature == b"signature-bytes" + + +# AC 12: Missing session history starts fresh. +def test_load_history_missing_file_returns_empty(tmp_path: Path) -> None: + assert google_tools.load_history(tmp_path / "missing.json") == [] + + +# AC 2, 6: FunctionDeclaration schemas expose fixed tool parameter names. +def test_tool_declarations_use_fixed_parameter_names() -> None: + declarations = {decl.name: decl for decl in google_tools.build_tool_declarations()} + + assert set(declarations) == { + "read_file", + "glob", + "list_directory", + "grep_search", + "run_shell_command", + } + assert declarations["read_file"].parameters_json_schema["required"] == ["file_path"] + assert declarations["list_directory"].parameters_json_schema["required"] == [ + "dir_path" + ] + assert declarations["run_shell_command"].parameters_json_schema["required"] == [ + "command" + ] + assert "run_shell_command" in declarations["run_shell_command"].description + assert ( + "Do not invent or call a tool literally named `sol`." + in declarations["run_shell_command"].description + ) diff --git a/tests/test_no_legacy_chat_imports.py b/tests/test_no_legacy_chat_imports.py index 047971dae..c0e872701 100644 --- a/tests/test_no_legacy_chat_imports.py +++ b/tests/test_no_legacy_chat_imports.py @@ -68,15 +68,16 @@ LEGACY_NAME = _parts("uni", "fied") def _python_files() -> list[Path]: # `git ls-files` excludes anything gitignored (`/journal/*` on dev boxes can # be 100+ GB of capture data; `ROOT.rglob` walks all of it on every call). - return [ROOT / line for line in _git_ls("*.py")] + return [path for line in _git_ls("*.py") if (path := ROOT / line).exists()] def _text_scan_files() -> list[Path]: blocked_parts = ("tests/fixtures",) return [ - ROOT / line + path for line in _git_ls("*.html", "*.js") if not any(part in line for part in blocked_parts) + if (path := ROOT / line).exists() ] diff --git a/uv.lock b/uv.lock index f39ae4c58..02ba76f73 100644 --- a/uv.lock +++ b/uv.lock @@ -3205,7 +3205,7 @@ requires-dist = [ { name = "faster-whisper", marker = "extra == 'whisper'", specifier = ">=1.0.0" }, { name = "flask", extras = ["async"] }, { name = "genai-prices" }, - { name = "google-genai" }, + { name = "google-genai", specifier = ">=1.62.0" }, { name = "h2" }, { name = "httpx" }, { name = "huggingface-hub", specifier = ">=0.24" }, -- 2.54.0