diff --git a/README.md b/README.md index 198c76a..22da9bc 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,7 @@ FLASH (**F**ast **L**ocal **A**gent **SH**ell) CLI is an AI-powered command-line - AI can use a `shell` tool to execute commands and see their output. - Manually execute shell commands using the `!` prefix. - **`flash://` Links**: Open Flash from a browser or another app with a prompt ready to go (`flash://?prompt=What+is+Python`). +- **Image Recognition**: Send a local image to a vision-capable model with `/image [prompt]`. - **Context Management**: Automatic history trimming to stay within token limits. - **Markdown Support**: Rich formatting for AI responses in the terminal. @@ -128,10 +129,28 @@ python run.py - `/help` or `/?`: Display the help message. - `/model`: Show the currently active model and Ollama host. - `/clear`: Clear the conversation history. +- `/image [prompt]`: Send a local image to the model. - `/version`: Show the current version and check GitHub for updates. - `/update`: Update Flash to the latest version (pipx installs only). - `/bye`: Exit the application. +### Image Recognition + +`/image [prompt]` attaches a local image (`.png`, `.jpg`, `.jpeg`, +`.webp`, `.gif`, `.bmp`) to your next message and sends both to the model. +If you leave off the prompt, Flash asks it to describe the image. This +requires a vision-capable model — text-only models will ignore the image +or error. Pull one and switch to it first, e.g.: + +```bash +ollama pull llama3.2-vision +``` + +``` +/model llama3.2-vision +/image ~/Pictures/screenshot.png What's going on in this UI? +``` + ### Updates Flash checks `main` on GitHub for a newer version on startup and shows it diff --git a/flash/ai.py b/flash/ai.py index f8da8ed..1299b78 100644 --- a/flash/ai.py +++ b/flash/ai.py @@ -3,6 +3,7 @@ import json import os import re +import shlex import shutil import sys import threading @@ -14,6 +15,7 @@ from dotenv import load_dotenv from ollama import ResponseError from rich.console import Console, Group +from rich.live import Live from rich.markdown import Markdown from rich.panel import Panel from rich.text import Text @@ -23,11 +25,12 @@ from .memory import forget_memory, list_memory from .notify import notify_reply_ready from .paths import ENV_PATH -from .repl_input import COMMANDS, read_line +from .repl_input import COMMANDS, IMAGE_EXTENSIONS, read_line from .theme import ( ACCENT, ACCENT_ANSI, CHEVRON, + CURSOR, DIM, DIM_ANSI, ELLIPSIS, @@ -59,6 +62,7 @@ OLLAMA_HOST_DEFAULT = "http://localhost:11434" ENV_NAME_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") +DEFAULT_IMAGE_PROMPT = "Describe this image in detail." load_dotenv(dotenv_path=ENV_PATH) @@ -189,8 +193,15 @@ def banner( print() -def _message(role: str, text: str) -> dict: - return {"role": role, "content": text} +def _message( + role: str, + text: str, + images: Union[list[str], None] = None, # noqa: UP007, RUF100 +) -> dict: + message: dict = {"role": role, "content": text} + if images: + message["images"] = images + return message def _message_text(message: dict) -> str: @@ -292,22 +303,30 @@ def _chat(client: "ollama.Client", messages: list, tools_arg=None): ) -def _load_thinking_states() -> list[str]: +def _load_states(key: str, fallback: list[str]) -> list[str]: try: p = Path(__file__).parent / "thinking_states.json" data = json.loads(p.read_text(encoding="utf-8")) - states = list(data.get("states", [])) + states = list(data.get(key, [])) if not states: raise ValueError("no states") return states except ValueError: - return [ - "Thinking", - "Pondering", - "Analyzing", - "Considering", - "Reflecting", - ] + return fallback + + +def _load_thinking_states() -> list[str]: + return _load_states( + "states", + ["Thinking", "Pondering", "Analyzing", "Considering", "Reflecting"], + ) + + +def _load_image_thinking_states() -> list[str]: + return _load_states( + "image_states", + ["Examining the image", "Analyzing the image", "Looking closely"], + ) _thinking_state_index = 0 @@ -353,9 +372,16 @@ def _chat_with_retries( def _try_chat( - client: "ollama.Client", messages: list, status, tools_arg=None + client: "ollama.Client", + messages: list, + status, + tools_arg=None, + *, + is_image: bool = False, ) -> tuple[Union[object, None], Union[str, None]]: # noqa: UP007, RUF100 - state = _next_thinking_state(_load_thinking_states()) + states = _load_image_thinking_states() if is_image \ + else _load_thinking_states() + state = _next_thinking_state(states) word = f"{state}{ELLIPSIS}" period = len(word) + 2 * GLIMMER_SPREAD stop_event = threading.Event() @@ -390,27 +416,58 @@ def _chat_with_status( client: "ollama.Client", messages: list, tools_arg=None, + *, + is_image: bool = False, ) -> tuple[Union[object, None], Union[str, None]]: # noqa: UP007, RUF100 with console.status( f"[bold {ACCENT}]Thinking{ELLIPSIS}", spinner="dots", spinner_style=ACCENT ) as status: - return _try_chat(client, messages, status, tools_arg) + return _try_chat( + client, messages, status, tools_arg, is_image=is_image + ) def _print_backend_error(detail: str) -> None: show_error(f"Ollama backend error: {detail}") +STREAM_CPS = 200.0 # simulated characters-per-second reveal rate +STREAM_MIN_DURATION = 0.25 +STREAM_MAX_DURATION = 2.0 +STREAM_FRAME_SECONDS = 0.04 + + def _render_markdown(console: Console, text: str, *, end: str = "\n") -> None: - console.print( - Markdown( - text, - code_theme="monokai", - hyperlinks=True - ), - end=end + """Render `text` as Markdown, revealing it progressively with a + trailing cursor dot -- the full reply already arrived in one shot, so + this is a paced typewriter effect rather than real token streaming.""" + + def render(body: str) -> Markdown: + return Markdown(body, code_theme="monokai", hyperlinks=True) + + if not text.strip() or not console.is_terminal: + console.print(render(text), end=end) + return + + duration = max( + STREAM_MIN_DURATION, min(STREAM_MAX_DURATION, len(text) / STREAM_CPS) ) + steps = max(1, int(duration / STREAM_FRAME_SECONDS)) + chunk = max(1, (len(text) + steps - 1) // steps) + + with Live( + render(CURSOR), console=console, + refresh_per_second=int(1 / STREAM_FRAME_SECONDS), transient=True, + ) as live: + cut = 0 + while cut < len(text): + cut = min(len(text), cut + chunk) + partial = text[:cut] + (f" {CURSOR}" if cut < len(text) else "") + live.update(render(partial)) + time.sleep(STREAM_FRAME_SECONDS) + + console.print(render(text), end=end) def _handle_scheme_flags(args) -> None: @@ -551,6 +608,10 @@ def main() -> None: while True: try: + pending_images: Union[ # noqa: UP007, RUF100 + list[str], None + ] = None + if pending: uin = pending.pop(0) if not _confirm_url_prompt(uin): @@ -698,6 +759,37 @@ def main() -> None: _run_update() continue + if uin == "/image" or uin.startswith("/image "): + arg = uin[len("/image"):].strip() + if not arg: + warn("Usage: /image [prompt]") + continue + try: + parts = shlex.split(arg) + except ValueError as exc: + warn(f"Could not parse path: {exc}") + continue + if not parts: + warn("Usage: /image [prompt]") + continue + + image_path = Path(parts[0]).expanduser() + if not image_path.is_file(): + show_error(f"Image not found: {image_path}") + continue + if image_path.suffix.lower() not in IMAGE_EXTENSIONS: + show_error( + f"Unsupported image type '{image_path.suffix}'. " + "Supported: " + + ", ".join(sorted(IMAGE_EXTENSIONS)) + ) + continue + + # Fall through to the normal send path below with UIN + # replaced by the prompt and PENDING_IMAGES attached. + uin = " ".join(parts[1:]).strip() or DEFAULT_IMAGE_PROMPT + pending_images = [str(image_path)] + direct_command = _direct_shell_command(uin) if direct_command: print( @@ -731,11 +823,12 @@ def main() -> None: ) continue - messages.append(_message("user", uin)) + messages.append(_message("user", uin, pending_images)) _trim_history(messages) res, err = _chat_with_status( - console, client, [system_message] + messages, tools + console, client, [system_message] + messages, tools, + is_image=bool(pending_images), ) if err: _print_backend_error(err) diff --git a/flash/repl_input.py b/flash/repl_input.py index 4502524..7794557 100644 --- a/flash/repl_input.py +++ b/flash/repl_input.py @@ -1,13 +1,21 @@ """REPL input with a dropdown menu of slash-command suggestions.""" +import json +import os +import time +from pathlib import Path from typing import Union from prompt_toolkit import PromptSession -from prompt_toolkit.completion import Completer, Completion -from prompt_toolkit.formatted_text import ANSI +from prompt_toolkit.completion import Completer, Completion, PathCompleter +from prompt_toolkit.document import Document +from prompt_toolkit.formatted_text import ANSI, StyleAndTextTuples from .memory import MEMORY_PATH from .paths import ENV_PATH +from .theme import SPARKLE, ptk_sweep_reveal + +IMAGE_EXTENSIONS = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".bmp"} # Single source of truth for both the completion dropdown and /help. COMMANDS = [ @@ -19,6 +27,7 @@ ("/memory", f"show saved memory, numbered ({MEMORY_PATH})"), ("/forget", "delete one memory by its 1-based index (/forget N)"), ("/clear", "clear saved context"), + ("/image", "send an image to the model (/image [prompt])"), ("/version", "show the current version and check for updates"), ("/update", "update Flash to the latest version (pipx installs)"), ("/help", "show this help (alias: /?)"), @@ -26,11 +35,77 @@ ] +def _is_image_path(path: str) -> bool: + """Passed to PathCompleter: always show directories (to navigate into), + and files whose extension is a supported image type.""" + + if os.path.isdir(path): + return True + return os.path.splitext(path)[1].lower() in IMAGE_EXTENSIONS + + +_image_path_completer = PathCompleter( + expanduser=True, file_filter=_is_image_path +) + + +def _parse_image_path_arg( + remainder: str, +) -> Union[tuple[str, bool], None]: # noqa: UP007 + """Track quoting while scanning the /image path argument typed so far. + + Returns `(literal_path, in_quote)`: `literal_path` is the path with any + quote marks stripped out (what's actually on disk), and `in_quote` is + True if the text currently ends inside a quote the user opened + themselves. Returns None once an unquoted space ends the path argument + (the start of the optional trailing prompt). + """ + + literal_chars = [] + quote: Union[str, None] = None # noqa: UP007 + for ch in remainder: + if quote: + if ch == quote: + quote = None + else: + literal_chars.append(ch) + elif ch in "\"'": + quote = ch + elif ch == " ": + return None + else: + literal_chars.append(ch) + return "".join(literal_chars), quote is not None + + class SlashCommandCompleter(Completer): - """Suggests / commands as the line is typed.""" + """Suggests / commands as the line is typed, and image file paths as + the argument to /image (auto-quoting suggestions that contain spaces).""" def get_completions(self, document, complete_event): text = document.text_before_cursor + + if text.startswith("/image "): + remainder = text[len("/image "):] + parsed = _parse_image_path_arg(remainder) + if parsed is None: + return # past the path, now typing the optional prompt + literal_path, in_quote = parsed + + sub_document = Document( + literal_path, cursor_position=len(literal_path) + ) + for completion in _image_path_completer.get_completions( + sub_document, complete_event + ): + suffix = completion.text + if not in_quote and " " in suffix: + suffix = f'"{suffix}"' + yield Completion( + suffix, start_position=0, display=completion.display + ) + return + if not text.startswith("/") or " " in text: return @@ -44,16 +119,79 @@ def get_completions(self, document, complete_event): ) +def _load_suggestions() -> list[str]: + try: + p = Path(__file__).parent / "suggestions.json" + data = json.loads(p.read_text(encoding="utf-8")) + items = list(data.get("suggestions", [])) + if items: + return items + except (OSError, ValueError): + pass + return ["Try /help to see every command"] + + +_SUGGESTIONS = _load_suggestions() + +# Timing for one suggestion's reveal-sweep -> hold -> conceal-sweep -> gap. +_SWEEP_BAND = 1.6 +_REVEAL = 0.9 +_HOLD = 5.0 +_CONCEAL = 0.9 +_GAP = 1.2 +_CYCLE = _REVEAL + _HOLD + _CONCEAL + _GAP +PLACEHOLDER_REFRESH_SECONDS = 0.08 + + +def _sweep_edge(progress: float, label_len: int) -> float: + """Map 0..1 sweep progress to an `edge` spanning the full label.""" + + span = label_len + 2 * _SWEEP_BAND + return -_SWEEP_BAND + progress * span + + +def _suggestion_placeholder() -> StyleAndTextTuples: + """Current animation frame: one suggestion's letters materializing in a + coral sweep, holding, then erased by another sweep, cycling through + `_SUGGESTIONS` over time.""" + + total = _CYCLE * len(_SUGGESTIONS) + pos = time.monotonic() % total + idx = int(pos // _CYCLE) + t = pos - idx * _CYCLE + + label = f"{SPARKLE} {_SUGGESTIONS[idx]}" + + if t < _REVEAL: + edge = _sweep_edge(t / _REVEAL, len(label)) + revealing = True + elif t < _REVEAL + _HOLD: + edge = len(label) + _SWEEP_BAND + revealing = True + elif t < _REVEAL + _HOLD + _CONCEAL: + edge = _sweep_edge((t - _REVEAL - _HOLD) / _CONCEAL, len(label)) + revealing = False + else: + return [] + + return ptk_sweep_reveal( + label, edge, revealing=revealing, band=_SWEEP_BAND + ) + + _session: Union[PromptSession, None] = None # noqa: UP007 def read_line(prompt_ansi: str) -> str: - """Read one line; suggests / commands in a dropdown while typing one.""" + """Read one line; suggests / commands in a dropdown while typing one, + and animates a rotating hint at the cursor while the line is empty.""" global _session if _session is None: _session = PromptSession( completer=SlashCommandCompleter(), complete_while_typing=True, + placeholder=_suggestion_placeholder, + refresh_interval=PLACEHOLDER_REFRESH_SECONDS, ) return _session.prompt(ANSI(prompt_ansi)) diff --git a/flash/suggestions.json b/flash/suggestions.json new file mode 100644 index 0000000..be2e367 --- /dev/null +++ b/flash/suggestions.json @@ -0,0 +1,28 @@ +{ + "suggestions": [ + "Try \"explain the changes you just made\"", + "Try \"!ls -la\" to run a shell command directly", + "Try \"/model\" to see or switch the active model", + "Try \"/image screenshot.png what's in this?\"", + "Try \"/memory\" to see what I remember", + "Try \"/auto on\" to let commands run without confirmation", + "Try \"find the bug in this file\"", + "Try \"/help\" to see every command", + "Try \"summarize the last commit\"", + "Try \"/set MODEL llama3.1\" to change config", + "Try \"find every TODO in this repo\"", + "Try \"search for how errors are logged here\"", + "Try \"grep for that function across the codebase\"", + "Try \"what's today's date?\"", + "Try \"look up the latest release notes for this\"", + "Try \"remember that I prefer tabs over spaces\"", + "Try \"/forget 2\" to delete a saved memory", + "Try \"/clear\" to start a fresh context", + "Try \"/version\" to check for updates", + "Try \"/refresh\" after editing your env file", + "Try \"what OS am I running?\"", + "Try \"walk through your reasoning on that\"", + "Try \"/unset MODEL\" to reset a config value", + "Try \"/bye\" when you're done" + ] +} diff --git a/flash/theme.py b/flash/theme.py index f1ba823..a7ede12 100644 --- a/flash/theme.py +++ b/flash/theme.py @@ -7,6 +7,7 @@ import sys +from prompt_toolkit.formatted_text import StyleAndTextTuples from rich.console import Console from rich.markdown import Markdown from rich.text import Text @@ -32,13 +33,14 @@ def _can_encode(text: str) -> bool: # report a non-UTF8 stdout encoding and raise UnicodeEncodeError on these # glyphs instead of substituting a fallback, crashing the whole process. # Fall back to plain ASCII there rather than risk that. -_UNICODE_OK = _can_encode("✻⏺⎿❯…") +_UNICODE_OK = _can_encode("✻⏺⎿❯…●") SPARKLE = "✻" if _UNICODE_OK else "*" # ✻ BULLET = "⏺" if _UNICODE_OK else "*" # ⏺ BRANCH = "⎿" if _UNICODE_OK else "L" # ⎿ CHEVRON = "❯" if _UNICODE_OK else ">" # ❯ ELLIPSIS = "…" if _UNICODE_OK else "..." # … +CURSOR = "●" if _UNICODE_OK else "." # ● # Raw ANSI escapes for text fed straight into input()/print(), where rich # markup can't reach (e.g. the interactive prompt string). @@ -102,3 +104,42 @@ def glimmer(text: str, offset: float, spread: float = 2.5) -> str: parts.append(f"[#{''.join(f'{c:02x}' for c in rgb)}]{ch}[/]") return "".join(parts) + + +_REST_RGB = (148, 148, 148) # ~ grey62, matches DIM + + +def ptk_sweep_reveal( + text: str, + edge: float, + *, + revealing: bool, + band: float = 1.6, +) -> StyleAndTextTuples: + """prompt_toolkit style fragments for `text`, where a moving `edge` + sweeps letters into or out of existence with a coral glow riding the + boundary between them. + + `revealing=True` materializes characters left of `edge`, leaving + everything to its right blank (not yet appeared). `revealing=False` + erases characters left of `edge`, leaving everything to its right + intact (not yet erased). Sweep `edge` from `-band` to + `len(text) + band` for a full pass in either mode. + """ + + fragments: StyleAndTextTuples = [] + for i, ch in enumerate(text): + d = (edge - i) if revealing else (i - edge) + shown = max(0.0, min(1.0, (d + band) / (2 * band))) + if shown <= 0.0: + fragments.append(("", " ")) + continue + + glow = max(0.0, 1.0 - (d / band) ** 2) if abs(d) < band else 0.0 + rgb = tuple( + round(base + (accent - base) * glow) + for base, accent in zip(_REST_RGB, _ACCENT_RGB) + ) + fragments.append((f"fg:#{''.join(f'{c:02x}' for c in rgb)}", ch)) + + return fragments diff --git a/flash/thinking_states.json b/flash/thinking_states.json index fee18a9..81b1c07 100644 --- a/flash/thinking_states.json +++ b/flash/thinking_states.json @@ -30,5 +30,22 @@ "Refining", "Double-checking", "FLASHing through possibilities" + ], + "image_states": [ + "Examining the image", + "Analyzing the image", + "Looking closely", + "Inspecting", + "Studying the image", + "Scanning the image", + "Peering closer", + "Taking a closer look", + "Reading the image", + "Parsing the visuals", + "Making out details", + "Piecing together the scene", + "Focusing in", + "Interpreting the image", + "Assessing the image" ] } \ No newline at end of file diff --git a/flash/tools.py b/flash/tools.py index 4cf65a3..ce2665f 100644 --- a/flash/tools.py +++ b/flash/tools.py @@ -1,10 +1,14 @@ """AI Tool System""" +import fnmatch import os import platform +import re import subprocess # nosec B404 from datetime import datetime +from pathlib import Path from tempfile import mkdtemp +from typing import Union from ddgs import DDGS from rich.text import Text @@ -20,6 +24,11 @@ === Tool System Prompt === Answer concisely. Use shell only when command output is needed. When using shell, call the tool without extra text first. +When searching a codebase or directory for files by name or pattern, use + the glob tool instead of shell find/ls. When searching file contents for + a pattern, use the grep tool instead of shell grep/rg. Both are read-only, + faster, and work the same on every platform, so prefer them over shell + for search whenever they cover the need. When searching for recent information, use the web_search tool. When you need to know the user's operating system, use the get_os tool. To think or plan mid-task without ending your turn, use the reason tool. @@ -162,6 +171,145 @@ def shell_tool(command: str, timeout=None, is_user=False) -> str: return final +# Directories that are rarely what a codebase search is looking for and +# can be huge (dependency trees, VCS internals, caches) -- pruned while +# walking so grep/glob stay fast and relevant. +_SEARCH_EXCLUDE_DIRS = { + ".git", ".hg", ".svn", "__pycache__", "node_modules", ".venv", "venv", + ".tox", ".mypy_cache", ".pytest_cache", ".ruff_cache", "dist", "build", + ".idea", ".vscode", +} +MAX_SEARCH_FILES = 5000 +MAX_GREP_MATCHES = 200 +MAX_GLOB_RESULTS = 500 +MAX_MATCH_LINE_LENGTH = 300 + + +def _should_skip_dir(name: str) -> bool: + return name in _SEARCH_EXCLUDE_DIRS or name.endswith(".egg-info") + + +def _iter_files(root: Path): + if root.is_file(): + yield root + return + + count = 0 + for dirpath, dirnames, filenames in os.walk(root): + dirnames[:] = [d for d in dirnames if not _should_skip_dir(d)] + for filename in filenames: + count += 1 + if count > MAX_SEARCH_FILES: + return + yield Path(dirpath) / filename + + +def _relative_path(file_path: Path, root: Path) -> str: + base = root if root.is_dir() else root.parent + try: + return file_path.relative_to(base).as_posix() + except ValueError: + return file_path.as_posix() + + +def glob_tool(pattern: str, path: str = ".") -> str: + """Tool to find files by name pattern.""" + + label = f"Glob({pattern})" + (f" in {path}" if path != "." else "") + tool_line(label) + + root = Path(path).expanduser() + if not root.exists(): + result = f"Error: path not found: {root}" + tool_result(result, style=ERROR) + return result + + matches = [] + for file_path in _iter_files(root): + rel = _relative_path(file_path, root) + if fnmatch.fnmatch(rel, pattern) or fnmatch.fnmatch( + file_path.name, pattern + ): + matches.append(rel) + if len(matches) >= MAX_GLOB_RESULTS: + break + + matches.sort() + result = "\n".join(matches) if matches else "No files matched." + if len(matches) >= MAX_GLOB_RESULTS: + result += f"\n... truncated at {MAX_GLOB_RESULTS} matches" + + tool_result( + f"{len(matches)} match{'es' if len(matches) != 1 else ''}" + if matches else "No matches." + ) + return result + + +def grep_tool( + pattern: str, + path: str = ".", + glob_filter: Union[str, None] = None, # noqa: UP007, RUF100 + case_insensitive: bool = False, +) -> str: + """Tool to search file contents by regex.""" + + label = f"Grep({pattern})" + (f" in {path}" if path != "." else "") + tool_line(label) + + root = Path(path).expanduser() + if not root.exists(): + result = f"Error: path not found: {root}" + tool_result(result, style=ERROR) + return result + + try: + regex = re.compile(pattern, re.IGNORECASE if case_insensitive else 0) + except re.error as exc: + result = f"Error: invalid regex: {exc}" + tool_result(result, style=ERROR) + return result + + matches = [] + files_matched = set() + for file_path in _iter_files(root): + rel = _relative_path(file_path, root) + if glob_filter and not ( + fnmatch.fnmatch(rel, glob_filter) + or fnmatch.fnmatch(file_path.name, glob_filter) + ): + continue + + try: + text = file_path.read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError): + continue + + for lineno, line in enumerate(text.splitlines(), start=1): + if not regex.search(line): + continue + snippet = line.strip() + if len(snippet) > MAX_MATCH_LINE_LENGTH: + snippet = snippet[:MAX_MATCH_LINE_LENGTH] + "..." + matches.append(f"{rel}:{lineno}: {snippet}") + files_matched.add(rel) + if len(matches) >= MAX_GREP_MATCHES: + break + if len(matches) >= MAX_GREP_MATCHES: + break + + result = "\n".join(matches) if matches else "No matches." + if len(matches) >= MAX_GREP_MATCHES: + result += f"\n... truncated at {MAX_GREP_MATCHES} matches" + + tool_result( + f"{len(matches)} match{'es' if len(matches) != 1 else ''} in " + f"{len(files_matched)} file{'s' if len(files_matched) != 1 else ''}" + if matches else "No matches." + ) + return result + + def web_search(query: str, max_results: int) -> str: """Search the web and return the top DuckDuckGo results.""" @@ -285,6 +433,76 @@ def get_date() -> str: }, }, }, + { + "type": "function", + "function": { + "name": "glob", + "description": ( + "Find files by name pattern (e.g. '*.py', '**/test_*.py'). " + "Read-only and fast; prefer this over shell find/ls when " + "searching a directory for files." + ), + "parameters": { + "type": "object", + "properties": { + "pattern": { + "type": "string", + "description": ( + "Glob pattern to match against each file's " + "path, e.g. '*.py' or 'flash/**/*.py'." + ), + }, + "path": { + "type": "string", + "description": ( + "Directory to search. Defaults to the current " + "directory." + ), + }, + }, + "required": ["pattern"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "grep", + "description": ( + "Search file contents for a regex pattern, returning each " + "match as 'path:line: text'. Read-only and fast; prefer " + "this over shell grep/rg when searching file contents." + ), + "parameters": { + "type": "object", + "properties": { + "pattern": { + "type": "string", + "description": "Regular expression to search for.", + }, + "path": { + "type": "string", + "description": ( + "File or directory to search. Defaults to the " + "current directory." + ), + }, + "glob_filter": { + "type": "string", + "description": ( + "Optional glob pattern to only search matching " + "files, e.g. '*.py'." + ), + }, + "case_insensitive": { + "type": "boolean", + "description": "Match case-insensitively.", + }, + }, + "required": ["pattern"], + }, + }, + }, { "type": "function", "function": { @@ -428,6 +646,8 @@ def get_date() -> str: FUNCTIONS = { "shell": shell_tool, + "glob": glob_tool, + "grep": grep_tool, "web_search": web_search, "get_os": get_os, "reason": reason, diff --git a/flash/version.py b/flash/version.py index 3965994..d451f5b 100644 --- a/flash/version.py +++ b/flash/version.py @@ -1,4 +1,4 @@ -__version__ = "0.1.0" +__version__ = "0.2.0" REPO = "Natuworkguy/Flash" REPO_URL = f"https://github.com/{REPO}" diff --git a/pyproject.toml b/pyproject.toml index 0fa13c3..643156d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,7 +28,7 @@ where = ["."] include = ["flash*"] [tool.setuptools.package-data] -flash = ["system_prompt.txt", "thinking_states.json"] +flash = ["system_prompt.txt", "thinking_states.json", "suggestions.json"] [tool.pyright] # flash/urlscheme.py imports winreg only on Windows (os.name == "nt") to diff --git a/tests/test_ai.py b/tests/test_ai.py index 7c158d3..1b28f76 100644 --- a/tests/test_ai.py +++ b/tests/test_ai.py @@ -4,6 +4,7 @@ Config, _direct_shell_command, _int_env, + _message, _run_update, _trim_history, _trim_tool_output, @@ -108,3 +109,15 @@ def test_run_update_failure_propagates(monkeypatch): "flash.ai.perform_update", lambda: (False, "pipx not found.") ) assert _run_update() is False # nosec B101 + + +def test_message_without_images(): + message = _message("user", "hello") + assert message == {"role": "user", "content": "hello"} # nosec B101 + assert "images" not in message # nosec B101 + + +def test_message_with_images(): + message = _message("user", "what is this", ["photo.png"]) + assert message["images"] == ["photo.png"] # nosec B101 + assert message["content"] == "what is this" # nosec B101 diff --git a/tests/test_tools.py b/tests/test_tools.py index 75b9d2d..1bb49d1 100644 --- a/tests/test_tools.py +++ b/tests/test_tools.py @@ -2,7 +2,7 @@ import subprocess # nosec B404 -from flash.tools import shell_tool +from flash.tools import glob_tool, grep_tool, shell_tool def test_shell_tool_timeout(monkeypatch): @@ -31,3 +31,57 @@ def mock_run(*_, **__): result = shell_tool("echo success") assert result == "success" # nosec B101 + + +def test_glob_tool_finds_matching_files(tmp_path): + (tmp_path / "a.py").write_text("print(1)") + (tmp_path / "b.txt").write_text("not python") + sub = tmp_path / "sub" + sub.mkdir() + (sub / "c.py").write_text("print(2)") + + result = glob_tool("*.py", str(tmp_path)) + assert "a.py" in result # nosec B101 + assert "sub/c.py" in result # nosec B101 + assert "b.txt" not in result # nosec B101 + + +def test_glob_tool_missing_path(): + result = glob_tool("*.py", "/no/such/directory") + assert "not found" in result # nosec B101 + + +def test_grep_tool_finds_matches(tmp_path): + (tmp_path / "a.py").write_text("def foo():\n return 1\n") + (tmp_path / "b.py").write_text("def bar():\n return 2\n") + + result = grep_tool("def foo", str(tmp_path)) + assert "a.py:1:" in result # nosec B101 + assert "b.py" not in result # nosec B101 + + +def test_grep_tool_glob_filter(tmp_path): + (tmp_path / "a.py").write_text("target\n") + (tmp_path / "a.txt").write_text("target\n") + + result = grep_tool("target", str(tmp_path), glob_filter="*.py") + assert "a.py:1:" in result # nosec B101 + assert "a.txt" not in result # nosec B101 + + +def test_grep_tool_case_insensitive(tmp_path): + (tmp_path / "a.py").write_text("HELLO world\n") + + assert "No matches" in grep_tool("hello", str(tmp_path)) # nosec B101 + result = grep_tool("hello", str(tmp_path), case_insensitive=True) + assert "HELLO world" in result # nosec B101 + + +def test_grep_tool_invalid_regex(): + result = grep_tool("(", ".") + assert "invalid regex" in result # nosec B101 + + +def test_grep_tool_missing_path(): + result = grep_tool("x", "/no/such/directory") + assert "not found" in result # nosec B101