diff --git a/docs/06_caching.md b/docs/06_caching.md index 264eab7c..084e4231 100644 --- a/docs/06_caching.md +++ b/docs/06_caching.md @@ -29,9 +29,7 @@ from askui.models.shared.settings import ( caching_settings = CachingSettings( strategy="record", # One of: "execute", "record", "auto", or None cache_dir=".askui_cache", # Directory to store cache files - writing_settings=CacheWritingSettings( - filename="my_test.json" # Filename for the cache file (optional) - ), + filename="my_test.json", # Name of the trajectory for this test case execution_settings=CacheExecutionSettings( delay_time_between_actions=1.0 # Delay in seconds between each cached action ), @@ -42,6 +40,7 @@ caching_settings = CachingSettings( - **`strategy`**: The caching strategy to use (`"execute"`, `"record"`, `"auto"`, or `None`). - **`cache_dir`**: Directory where cache files are stored. Defaults to `".askui_cache"`. +- **`filename`**: Name of the trajectory/cache file for this test case (the `.json` suffix is optional). It is the lookup key in `"execute"`/`"auto"` modes and the target filename in `"record"`/`"auto"` modes. If empty, no trajectory is auto-detected and recordings receive an auto-generated filename. - **`writing_settings`**: Configuration for cache recording (optional). See [Writing Settings](#writing-settings) below. - **`execution_settings`**: Configuration for cache playback (optional). See [Execution Settings](#execution-settings) below. @@ -59,7 +58,7 @@ writing_settings = CacheWritingSettings( #### Parameters -- **`filename`**: Name of the cache file to write. If not specified, a timestamped filename will be generated automatically (format: `cached_trajectory_YYYYMMDDHHMMSSffffff.json`). +- **`filename`**: Name of the cache file to write. Prefer setting `filename` directly on `CachingSettings` (the top-level `filename` takes precedence and is also used for trajectory lookup in `execute`/`auto` modes). If neither is specified, a timestamped filename will be generated automatically (format: `cached_trajectory_YYYYMMDDHHMMSSffffff.json`). ### Execution Settings @@ -89,16 +88,14 @@ Record agent actions to a cache file for later replay: ```python from askui import ComputerAgent -from askui.models.shared.settings import CachingSettings, CacheWritingSettings +from askui.models.shared.settings import CachingSettings with ComputerAgent() as agent: agent.act( goal="Fill out the login form with username 'admin' and password 'secret123'", caching_settings=CachingSettings( strategy="record", # you could also use "auto" here - writing_settings=CacheWritingSettings( - filename="login_test.json" - ), + filename="login_test.json", ) ) ``` @@ -107,7 +104,7 @@ After execution, a cache file will be created at `.askui_cache/login_test.json` ### Executing from Cache (Replaying) -Provide the agent with access to previously recorded trajectories: +Set `strategy="execute"` (or `"auto"`) and give the trajectory's `filename`. The SDK automatically looks up `/`: ```python from askui import ComputerAgent @@ -118,116 +115,24 @@ with ComputerAgent() as agent: goal="Fill out the login form", caching_settings=CachingSettings( strategy="execute", # you could also use "auto" here + filename="login_test.json", ) ) ``` -When using `strategy="execute"`, the agent receives two additional tools: - -1. **`retrieve_available_trajectories_tool`**: Lists all available cache files in the cache directory -2. **`execute_cached_executions_tool`**: Executes a specific cached trajectory - -The agent will automatically check if a relevant cached trajectory exists and use it if appropriate. After executing a cached trajectory, the agent will verify the results and make corrections if needed. +If a usable trajectory with that name exists, the SDK surfaces its details (path +and required parameters) to the agent automatically in the first message, and the +agent replays it via the `CacheExecutor` before doing anything else — you no +longer need to describe available cache files in your goal prompt, and there is +no separate "list trajectories" tool. After replay, the agent verifies the +results (via the `verify_cache_execution` tool) and makes corrections if needed. -### Referencing Cache Files in Goal Prompts - -When using `strategy="execute"` or `strategy="auto"`, **you need to inform the agent about which cache files are available and when to use them**. This is done by including cache file information directly in your goal prompt. - -#### Explicit Cache File References - -For specific tasks, mention the cache file name and what it accomplishes: - -```python -from askui import ComputerAgent -from askui.models.shared.settings import CachingSettings - -with ComputerAgent() as agent: - agent.act( - goal="""Open the website in Google Chrome. - - If the cache file "open_website_in_chrome.json" is available, please use it - for this execution. It will open a new window in Chrome and navigate to the website.""", - caching_settings=CachingSettings( - strategy="execute", - cache_dir=".cache" - ) - ) -``` - -#### Pattern-Based Cache File References - -For test suites or repetitive workflows, you can establish naming conventions: - -```python -from askui import ComputerAgent -from askui.models.shared.settings import CachingSettings - -test_id = "TEST_001" - -with ComputerAgent() as agent: - agent.act( - goal=f"""Execute test {test_id} according to the test definition. - - Check if a cache file named "{test_id}.json" exists. If it does, use it to - replay the test actions, then verify the results.""", - caching_settings=CachingSettings( - strategy="execute", - cache_dir="test_cache" - ) - ) -``` +Behavior when a trajectory is **not** found: -#### General Rules for Cache Selection - -You can also provide general instructions for the agent to identify applicable cache files: - -```python -from askui import ComputerAgent -from askui.models.shared.settings import CachingSettings - -with ComputerAgent() as agent: - agent.act( - goal="""Fill out the user registration form. - - Look for cache files that match the pattern "user_registration_*.json". - Choose the most recent one if multiple are available, as it likely contains - the most up-to-date interaction sequence.""", - caching_settings=CachingSettings( - strategy="execute", - cache_dir=".cache" - ) - ) -``` - -#### Multiple Cache Files - -For complex workflows, you can reference multiple cache files: - -```python -from askui import ComputerAgent -from askui.models.shared.settings import CachingSettings - -with ComputerAgent() as agent: - agent.act( - goal="""Complete the full checkout process: - - 1. If "login.json" exists, use it to log in - 2. If "add_to_cart.json" exists, use it to add items to cart - 3. If "checkout.json" exists, use it to complete the checkout - - After each cached execution, verify the step completed successfully before proceeding.""", - caching_settings=CachingSettings( - strategy="execute", - cache_dir=".cache" - ) - ) -``` - -**Best Practices:** -- Be specific about what the cache file does to help the agent decide if it's applicable -- Include verification instructions after cached execution -- Use consistent naming conventions for easier cache file management -- Mention any prerequisites or expected UI state for the cached trajectory +- `strategy="execute"`: the agent performs the task normally (nothing is recorded). +- `strategy="auto"`: the agent is told no cache exists and performs the task + normally, while recording the run to `filename` for next time. An existing but + invalidated cache is re-recorded (self-healing) rather than replayed. ### Using Custom Execution Settings @@ -260,24 +165,21 @@ Enable both reading and writing simultaneously: ```python from askui import ComputerAgent -from askui.models.shared.settings import CachingSettings, CacheWritingSettings +from askui.models.shared.settings import CachingSettings with ComputerAgent() as agent: agent.act( goal="Complete the checkout process", caching_settings=CachingSettings( strategy="auto", - writing_settings=CacheWritingSettings( - filename="checkout_test.json" - ), + filename="checkout_test.json", ) ) ``` In this mode: -- The agent can use existing cached trajectories to speed up execution -- New actions will be recorded to the specified cache file -- If a cached execution is used, no new cache file will be written (to avoid duplicates) +- If a usable trajectory named `checkout_test.json` exists, it is replayed and no new cache file is written (to avoid overwriting the existing one) +- Otherwise, the agent performs the task normally and records the run to `checkout_test.json` ## Cache File Format @@ -330,13 +232,12 @@ In write mode, the `CacheManager`: In read mode: -1. Two caching tools are added to the agent's toolbox -2. A special system prompt (`CACHE_USE_PROMPT`) is appended to instruct the agent on how to use trajectories -3. The agent can call `retrieve_available_trajectories_tool` to see available cache files -4. The agent can call `execute_cached_executions_tool` with a trajectory file path to replay it -5. During replay, each tool use block is executed sequentially with a configurable delay between actions (default: 1.0 seconds) -6. Screenshot and trajectory retrieval tools are skipped during replay -7. The agent is instructed to verify results after replay and make corrections if needed +1. The SDK checks whether a trajectory named `filename` exists in `cache_dir` +2. If a usable trajectory is found, its details (path and required parameters) are injected into the first user message, a special system prompt (`CACHE_USE_PROMPT`) is appended, and the `CacheExecutor` speaker plus the `verify_cache_execution` tool are wired up +3. The agent hands off to the `CacheExecutor` via the `switch_speaker` tool to replay the trajectory +4. During replay, each tool use block is executed sequentially with a configurable delay between actions (default: 1.0 seconds) +5. Screenshot and non-cacheable tools are skipped/paused during replay; if a non-cacheable step is encountered the agent executes it manually and (unless it was the last step) resumes replay +6. The agent is instructed to verify results after replay (via `verify_cache_execution`) and make corrections if needed; reporting failure invalidates the cache The delay between actions can be customized using `CacheExecutionSettings` to accommodate different application response times. @@ -354,7 +255,6 @@ from askui import ComputerAgent from askui.models.shared.settings import ( CachingSettings, CacheExecutionSettings, - CacheWritingSettings, ) # Step 1: Record a successful login flow @@ -365,9 +265,7 @@ with ComputerAgent() as agent: caching_settings=CachingSettings( strategy="record", cache_dir="test_cache", - writing_settings=CacheWritingSettings( - filename="user_login.json" - ), + filename="user_login.json", ) ) @@ -375,14 +273,11 @@ with ComputerAgent() as agent: print("\nReplaying login flow for regression test...") with ComputerAgent() as agent: agent.act( - goal="""Log in to the application. - - If the cache file "user_login.json" is available, please use it to replay - the login sequence. It contains the steps to navigate to the login page and - authenticate with the test credentials.""", + goal="Log in to the application.", caching_settings=CachingSettings( strategy="execute", cache_dir="test_cache", + filename="user_login.json", execution_settings=CacheExecutionSettings( delay_time_between_actions=2.0 ), diff --git a/src/askui/agent_base.py b/src/askui/agent_base.py index 6a1915c6..a0564c83 100644 --- a/src/askui/agent_base.py +++ b/src/askui/agent_base.py @@ -2,7 +2,7 @@ import time import types from pathlib import Path -from typing import Annotated, Literal, Optional, Type, overload +from typing import Annotated, Any, Literal, Optional, Type, overload from dotenv import load_dotenv from PIL import Image as PILImage @@ -13,11 +13,12 @@ from askui.callbacks import ConversationCallback, ConversationStatisticsCallback from askui.container import telemetry from askui.locators.locators import Locator -from askui.models.shared.agent_message_param import MessageParam +from askui.models.shared.agent_message_param import MessageParam, TextBlockParam from askui.models.shared.conversation import Conversation, Speakers from askui.models.shared.secrets import Secret, SecretVault from askui.models.shared.settings import ( ActSettings, + CacheFile, CacheWritingSettings, CachingSettings, GetSettings, @@ -31,7 +32,6 @@ from askui.tools.android.agent_os import AndroidAgentOs from askui.tools.caching_tools import ( InspectCacheMetadata, - RetrieveCachedTestExecutions, VerifyCacheExecution, ) from askui.tools.get_tool import GetTool @@ -269,13 +269,18 @@ def act( # Make the vault available for substitution (tools) and redaction (history). # The Conversation propagates it to the ToolCollection. self._conversation.secret_vault = active_vault - _act_settings = act_settings or self.act_settings + # Deep-copy so caching-related mutations (e.g. injecting the CACHE_USE + # prompt) do not accumulate on the Agent's persistent, reused settings + # object and leak into subsequent act() calls. + _act_settings = (act_settings or self.act_settings).model_copy(deep=True) _caching_settings: CachingSettings = caching_settings or self.caching_settings - tools, cache_manager = self._patch_act_with_cache( + tools, cache_manager, cache_hint = self._patch_act_with_cache( _caching_settings, _act_settings, tools, goal_str ) + if cache_hint: + messages = self._inject_cache_hint(messages, cache_hint) _tools = self._build_tools(tools) # setup opentelemetry for tracing @@ -297,7 +302,13 @@ def act( ) def _build_tools(self, tools: list[Tool] | ToolCollection | None) -> ToolCollection: - tool_collection = self.act_tool_collection + # Build a fresh per-call collection copied from the agent's base tools so + # that per-call additions (caching tools, switch_speaker, per-call tools) + # do not accumulate on the persistent `act_tool_collection` across calls. + # Otherwise a run-specific `VerifyCacheExecution` (wired to that run's + # CacheExecutor/CacheManager) would linger and could persist a later, + # unrelated run's result to the previous run's trajectory file. + tool_collection = self.act_tool_collection + ToolCollection() if isinstance(tools, list): tool_collection.append_tool(*tools) if isinstance(tools, ToolCollection): @@ -324,9 +335,18 @@ def _patch_act_with_cache( settings: ActSettings, tools: list[Tool] | ToolCollection | None, goal: str, - ) -> tuple[list[Tool] | ToolCollection, CacheManager | None]: + ) -> tuple[list[Tool] | ToolCollection, CacheManager | None, str | None]: """Patch act settings and tools with caching functionality. + In ``execute``/``auto`` modes the trajectory for the current test case is + auto-detected from ``caching_settings.filename`` (no separate discovery + tool call is required): if a usable trajectory exists, its details are + returned as a ``cache_hint`` to be surfaced to the agent, and the + ``CacheExecutor`` speaker plus verification tooling are wired up. In + ``record``/``auto`` modes a cache manager is set up to record the run + (in ``auto`` only when no usable trajectory was found, so an existing + cache is never overwritten by a fresh recording). + Args: caching_settings: The caching settings to apply settings: The act settings to modify @@ -334,29 +354,71 @@ def _patch_act_with_cache( goal: The goal string for cache recording Returns: - A tuple of (modified_tools, cache_manager) + A tuple of ``(modified_tools, cache_manager, cache_hint)`` where + ``cache_hint`` is an optional instruction to inject into the first + user message. """ caching_tools: list[Tool] = [] cache_manager: CacheManager | None = None + cache_hint: str | None = None + + # Remove any CacheExecutor registered by a previous act() call so it does + # not leak (and get advertised via switch_speaker) into this run. + self._conversation.speakers.remove_speaker("CacheExecutor") + + strategy = caching_settings.strategy + filename = self._resolve_cache_filename(caching_settings) + + # Detect an existing trajectory for execute/auto modes. + cache_file: CacheFile | None = None + trajectory_path: Path | None = None + if strategy in ("execute", "auto") and filename: + trajectory_path = self._resolve_trajectory_path( + caching_settings.cache_dir, filename + ) + cache_file = self._read_trajectory_if_present(trajectory_path) + + # Decide whether to replay the detected trajectory. In auto mode an + # invalid cache is re-recorded (self-heal) rather than replayed. + execute_trajectory = cache_file is not None and ( + cache_file.metadata.is_valid or strategy == "execute" + ) + should_record = strategy == "record" or ( + strategy == "auto" and not execute_trajectory + ) + + if execute_trajectory or should_record: + cache_manager = CacheManager() - # Setup execute mode: add caching tools and modify system prompt - if caching_settings.strategy in ["execute", "auto"]: - # Create CacheExecutor with execution settings and add to speakers + # Setup execute mode: wire the CacheExecutor and verification tooling and + # tell the agent (via the hint) exactly which trajectory to replay. + if ( + execute_trajectory + and cache_file is not None + and trajectory_path is not None + ): cache_executor = CacheExecutor(caching_settings.execution_settings) self._conversation.speakers.add_speaker(cache_executor) - # Add caching tools (switch_speaker tool is added automatically - # by Conversation._setup_speaker_handoff) + # switch_speaker tool is added automatically by + # Conversation._setup_speaker_handoff caching_tools.extend( [ - RetrieveCachedTestExecutions(caching_settings.cache_dir), - VerifyCacheExecution(), + VerifyCacheExecution( + cache_executor=cache_executor, + cache_manager=cache_manager, + ), InspectCacheMetadata(), ] ) if settings.messages.system is None: settings.messages.system = create_default_prompt() settings.messages.system.cache_use = CACHE_USE_PROMPT + cache_hint = self._build_cache_execution_hint(trajectory_path, cache_file) + elif strategy == "auto": + # Auto mode with nothing usable to replay: let the agent know a new + # trajectory is being recorded for next time. + cache_hint = self._build_no_cache_hint() # Add caching tools to the tools list if isinstance(tools, list): @@ -366,14 +428,11 @@ def _patch_act_with_cache( else: tools = caching_tools - # Setup record mode: create cache manager for recording - if caching_settings.strategy in ["record", "auto"]: + # Setup record mode: start recording the trajectory. + if should_record and cache_manager is not None: cache_writer_settings = ( caching_settings.writing_settings or CacheWritingSettings() ) - filename = cache_writer_settings.filename or "" - - cache_manager = CacheManager() cache_manager.start_recording( cache_dir=caching_settings.cache_dir, file_name=filename, @@ -382,7 +441,129 @@ def _patch_act_with_cache( vlm_provider=self._vlm_provider, ) - return tools, cache_manager + return tools, cache_manager, cache_hint + + @staticmethod + def _resolve_cache_filename(caching_settings: CachingSettings) -> str: + """Resolve the trajectory filename, preferring the top-level setting.""" + if caching_settings.filename: + return caching_settings.filename + if ( + caching_settings.writing_settings + and caching_settings.writing_settings.filename + ): + return caching_settings.writing_settings.filename + return "" + + @staticmethod + def _resolve_trajectory_path(cache_dir: str, filename: str) -> Path: + """Build the full trajectory path, ensuring a ``.json`` suffix.""" + name = filename if filename.endswith(".json") else f"{filename}.json" + return Path(cache_dir) / name + + @staticmethod + def _read_trajectory_if_present(trajectory_path: Path) -> "CacheFile | None": + """Read a trajectory file if it exists and is readable, else ``None``.""" + if not trajectory_path.is_file(): + return None + try: + return CacheManager.read_cache_file(trajectory_path) + except Exception: + logger.exception( + "Found trajectory %s but failed to read it; ignoring cache", + trajectory_path, + ) + return None + + @staticmethod + def _build_cache_execution_hint( + trajectory_path: Path, cache_file: CacheFile + ) -> str: + """Build the first-message hint describing an available cached trajectory.""" + path_str = str(trajectory_path) + parameters = cache_file.cache_parameters + if parameters: + param_lines = "\n".join( + f" - {name}: {description}" for name, description in parameters.items() + ) + param_block = ( + "This trajectory requires the following parameters (provide " + f"values for ALL of them):\n{param_lines}" + ) + example_params = ", ".join(f"'{name}': ''" for name in parameters) + switch_example = ( + "switch_speaker(speaker_name='CacheExecutor', speaker_context={" + f"'trajectory_file': '{path_str}', " + f"'parameter_values': {{{example_params}}}}})" + ) + else: + param_block = "This trajectory requires no parameters." + switch_example = ( + "switch_speaker(speaker_name='CacheExecutor', speaker_context={" + f"'trajectory_file': '{path_str}'}})" + ) + + validity_note = "" + if not cache_file.metadata.is_valid: + validity_note = ( + "\nNOTE: This cached trajectory is currently marked INVALID " + f"(reason: {cache_file.metadata.invalidation_reason}). It may not " + "replay correctly; execute with caution and verify the result " + "carefully." + ) + + return ( + "\n" + "A cached trajectory for this test case is available and should be " + "used to fast-forward execution instead of performing the steps " + "manually.\n" + f"- trajectory_file: {path_str}\n" + f"{param_block}\n" + "Before taking any other action, switch to the CacheExecutor speaker " + "using the switch_speaker tool, for example:\n" + f"{switch_example}" + f"{validity_note}\n" + "" + ) + + @staticmethod + def _build_no_cache_hint() -> str: + """Build the first-message hint used in auto mode when no cache exists.""" + return ( + "\n" + "No cached trajectory exists for this test case yet, so there is " + "nothing to replay. Accomplish the goal normally; your actions are " + "being recorded so they can be replayed on future runs.\n" + "" + ) + + @staticmethod + def _inject_cache_hint( + messages: list[MessageParam], cache_hint: str + ) -> list[MessageParam]: + """Append the cache hint to the first user message. + + The hint is appended to (not inserted before) the first user message to + avoid introducing consecutive same-role messages at the start of the + history. If no user message exists (unusual), the messages are returned + unchanged. + """ + index = next( + (i for i, m in enumerate(messages) if m.role == "user"), + None, + ) + if index is None: + return messages + target = messages[index] + if isinstance(target.content, str): + new_content: str | list[Any] = f"{target.content}\n\n{cache_hint}" + else: + new_content = [ + *target.content, + TextBlockParam(type="text", text=cache_hint), + ] + messages[index] = target.model_copy(update={"content": new_content}) + return messages @overload def get( diff --git a/src/askui/models/shared/conversation.py b/src/askui/models/shared/conversation.py index 1a74b097..97d1056e 100644 --- a/src/askui/models/shared/conversation.py +++ b/src/askui/models/shared/conversation.py @@ -236,9 +236,14 @@ def _is_max_steps_reached(self) -> bool: @tracer.start_as_current_span("_teardown_control_loop") def _teardown_control_loop(self) -> None: - # Finish recording if cache_manager is active and not executing from cache + # Finish recording if cache_manager is active and not executing from cache. + # This runs in the conversation's `finally`, so any error here must not + # mask the real control-loop outcome - log and swallow instead. if self.cache_manager is not None and not self._executed_from_cache: - self.cache_manager.finish_recording(self.get_messages()) + try: + self.cache_manager.finish_recording(self.get_messages()) + except Exception: + logger.exception("Failed to finish cache recording") def _setup_speaker_handoff(self) -> None: """Set up speaker handoff infrastructure. diff --git a/src/askui/models/shared/settings.py b/src/askui/models/shared/settings.py index 293d7fb2..31489672 100644 --- a/src/askui/models/shared/settings.py +++ b/src/askui/models/shared/settings.py @@ -202,7 +202,7 @@ class CacheMetadata(BaseModel): visual_validation: Visual validation configuration """ - version: str = "0.2" + version: str = "0.3" created_at: datetime goal: str | None = None last_executed_at: datetime | None = None @@ -234,7 +234,6 @@ class CacheWritingSettings(BaseModel): Args: filename: Name for the cache file (auto-generated if empty) parameter_identification_strategy: How to identify parameters("llm" or "preset") - llm_parameter_id_api_provider: API provider for LLM parameter identification visual_verification_method: Visual hash method ("phash", "ahash", or "none") visual_validation_region_size: Size of region to hash around coordinates """ @@ -273,11 +272,19 @@ class CachingSettings(BaseModel): - "auto": Execute from cache if available, otherwise record cache_dir (str): Directory path for storing cache files. Default: ".askui_cache". + filename (str): Name of the trajectory/cache file for this test case + (the ".json" suffix is optional). It is used as the lookup key in + "execute"/"auto" modes (the SDK checks whether + `/` exists and, if so, feeds its details to the + agent automatically) and as the target filename in "record"/"auto" + modes. If empty, no trajectory is auto-detected and recordings get an + auto-generated filename. writing_settings: Settings for cache recording (used in "record"/"auto" modes) execution_settings: Settings for cache playback (used in "execute"/"auto" modes) """ strategy: CACHING_STRATEGY | None = None cache_dir: str = ".askui_cache" + filename: str = "" writing_settings: CacheWritingSettings | None = None execution_settings: CacheExecutionSettings | None = None diff --git a/src/askui/prompts/act_prompts.py b/src/askui/prompts/act_prompts.py index 64ce8d08..ed72db07 100644 --- a/src/askui/prompts/act_prompts.py +++ b/src/askui/prompts/act_prompts.py @@ -432,20 +432,20 @@ CACHE_USE_PROMPT = ( "\n" - "CRITICAL: Before taking ANY action, you MUST first call the" - " retrieve_available_trajectories_tool to check for cached trajectories. If the" - " name of an available cached trajectory matches the one specified by the user," - " you MUST switch to the CacheExecutor speaker using the switch_speaker tool" - " before calling any other tools!\n" - " You are only allowed to use cache files with the exact names the user allowed you" - " to use. NEVER use cache files with other names without permission, even if the" - " names are very similar!" + "CRITICAL: When a cached trajectory is available for this task, its details are" + " provided directly in the conversation inside a " + " block (trajectory_file path and required parameters). Before taking ANY other" + " action, you MUST switch to the CacheExecutor speaker using the switch_speaker" + " tool with exactly that trajectory_file.\n" + " Only use the trajectory_file provided in the " + " block. NEVER invent or guess other trajectory paths.\n" + " If instead you see a block (or no block at all), no" + " trajectory is available - proceed with manual execution.\n" "\n" "WORKFLOW:\n" - "1. ALWAYS start by calling retrieve_available_trajectories_tool\n" - "2. If a matching cached trajectory exists, switch to CacheExecutor using" - " the switch_speaker tool with speaker_context containing the trajectory details\n" - "3. Only proceed with manual execution if no matching trajectory is available\n" + "1. If a block is present, immediately switch to" + " CacheExecutor using the switch_speaker tool with the provided trajectory_file\n" + "2. Otherwise, proceed with manual execution\n" "\n" "EXECUTING TRAJECTORIES:\n" "- Use switch_speaker(speaker_name='CacheExecutor', speaker_context={" @@ -457,6 +457,7 @@ "\n" "DYNAMIC PARAMETERS:\n" "- Trajectories may require parameters like {{current_date}} or {{user_name}}\n" + "- The required parameters are listed in the block\n" "- Provide values via parameter_values in the speaker_context\n" "- Example: switch_speaker(speaker_name='CacheExecutor', speaker_context={" "'trajectory_file': 'test.json', 'parameter_values': {" @@ -469,11 +470,13 @@ "- Trajectory pauses at non-cacheable steps, returning NEEDS_AGENT status with" " current step index\n" "- Execute the non-cacheable step manually\n" - "- Resume by switching to CacheExecutor again with start_from_step_index" - " in the speaker_context\n" + "- The pause message tells you the exact start_from_step_index to resume with," + " or states that it was the final step (in which case do NOT resume - verify" + " instead)\n" "\n" "CONTINUING TRAJECTORIES:\n" - "- Resume after non-cacheable steps: switch_speaker(speaker_name='CacheExecutor'," + "- Resume after non-cacheable steps only when the pause message provides a" + " start_from_step_index: switch_speaker(speaker_name='CacheExecutor'," " speaker_context={'trajectory_file': 'test.json'," " 'start_from_step_index': 5, 'parameter_values': {...}})\n" "\n" diff --git a/src/askui/speaker/cache_executor.py b/src/askui/speaker/cache_executor.py index 404c0b21..fd8a96d0 100644 --- a/src/askui/speaker/cache_executor.py +++ b/src/askui/speaker/cache_executor.py @@ -126,6 +126,16 @@ def __init__( # Activation context received via on_activate() self._activation_context: dict[str, Any] = {} + @property + def current_cache_file(self) -> "CacheFile | None": + """The cache file of the most recently activated trajectory, if any.""" + return self._cache_file + + @property + def current_cache_file_path(self) -> str | None: + """Path of the most recently activated trajectory, if any.""" + return self._cache_file_path + @override def can_handle(self, conversation: "Conversation") -> bool: # noqa: ARG002 """Check if cache execution is active or should be activated. @@ -206,9 +216,11 @@ def handle_step( if self._current_step_index < len(self._trajectory): time.sleep(self._delay_time_between_actions) - # Check if we have a trajectory - if not self._trajectory or not self._toolbox: - logger.error("Cache executor called but no trajectory or toolbox available") + # Require a toolbox to execute. An empty trajectory is allowed: it flows + # into `_get_next_step()`'s COMPLETED path (which requests verification) + # rather than silently bouncing back to the agent. + if self._toolbox is None: + logger.error("Cache executor called but no toolbox available") return SpeakerResult( status="switch_speaker", next_speaker="AgentSpeaker", @@ -276,6 +288,24 @@ def _handle_needs_agent(self, result: ExecutionResult) -> SpeakerResult: tool_to_execute = result.tool_result if tool_to_execute: + resume_index = result.step_index + 1 + more_steps_remain = self._has_executable_steps_from(resume_index) + + if more_steps_remain: + resume_instruction = ( + "Execute this tool with the necessary parameters. To replay the " + "remaining cached steps afterwards, switch back to the " + "CacheExecutor with " + f"start_from_step_index={resume_index}." + ) + else: + resume_instruction = ( + "This is the FINAL step of the trajectory. Execute this tool " + "with the necessary parameters, then verify the outcome with the " + "verify_cache_execution tool. Do NOT switch back to the " + "CacheExecutor - there are no further cached steps to replay." + ) + instruction_message = MessageParam( role="user", content=[ @@ -286,8 +316,7 @@ def _handle_needs_agent(self, result: ExecutionResult) -> SpeakerResult: "The previous steps were executed successfully " f"from cache. The next step requires the " f"'{tool_to_execute.name}' tool, which cannot be " - "executed from cache. Please execute this tool with " - "the necessary parameters." + f"executed from cache. {resume_instruction}" ), ) ], @@ -415,14 +444,20 @@ def _activate_from_context( if not self._cache_file: self._cache_file = cache_manager.read_cache_file(Path(trajectory_file)) - # Validate step index - if start_from_step_index < 0 or start_from_step_index >= len( - self._cache_file.trajectory - ): + # Validate step index. `start_from_step_index == len(trajectory)` is + # allowed and means "there is nothing left to replay" - this happens when + # the agent resumes after handling the trajectory's last step (e.g. a + # non-cacheable final step). It is treated as an immediate completion by + # `_get_next_step()` instead of being rejected as out of range. + trajectory_len = len(self._cache_file.trajectory) + if start_from_step_index < 0 or start_from_step_index > trajectory_len: + valid_range = ( + f"0-{trajectory_len}" if trajectory_len > 0 else "0 (empty trajectory)" + ) error_msg = ( f"Invalid start_from_step_index: {start_from_step_index}. " - f"Trajectory has {len(self._cache_file.trajectory)} steps " - f"(valid indices: 0-{len(self._cache_file.trajectory) - 1})." + f"Trajectory has {trajectory_len} steps " + f"(valid indices: {valid_range})." ) raise ValueError(error_msg) @@ -520,7 +555,7 @@ def _get_next_step( if self._current_step_index >= len(self._trajectory): return ExecutionResult( status="COMPLETED", - step_index=self._current_step_index - 1, + step_index=max(self._current_step_index - 1, 0), message_history=self._message_history, ) @@ -584,6 +619,20 @@ def _get_next_step( message_history=[assistant_message], ) + def _has_executable_steps_from(self, index: int) -> bool: + """Return whether any step at or after `index` would still be replayed. + + Skippable steps (e.g. `switch_speaker`, verbosity tools) are ignored. + Non-cacheable steps count as executable because resuming would replay up + to them and pause again. Used to decide whether the agent should resume + cache execution after handling a non-cacheable step, or whether that step + was the trajectory's last and no resume is needed. + """ + return any( + not self._should_skip_step(self._trajectory[i]) + for i in range(index, len(self._trajectory)) + ) + def _should_pause_for_agent(self, step: ToolUseBlockParam) -> bool: """Check if execution should pause for agent intervention.""" if not self._toolbox: diff --git a/src/askui/speaker/speaker.py b/src/askui/speaker/speaker.py index 53cf6c8b..b370431f 100644 --- a/src/askui/speaker/speaker.py +++ b/src/askui/speaker/speaker.py @@ -127,6 +127,16 @@ def add_speaker(self, speaker: Speaker) -> None: """Add a speaker to the collection.""" self.speakers[speaker.name] = speaker + def remove_speaker(self, name: str) -> None: + """Remove a speaker by name if present (the default speaker is kept). + + Used to avoid a speaker registered for one ``act()`` call (e.g. a + ``CacheExecutor``) leaking into subsequent calls that do not need it. + """ + if name == self.default_speaker: + return + self.speakers.pop(name, None) + def get_names(self) -> list[str]: """Get list of all speaker names.""" return list(self.speakers.keys()) diff --git a/src/askui/tools/caching_tools.py b/src/askui/tools/caching_tools.py index f5379e38..9e16f928 100644 --- a/src/askui/tools/caching_tools.py +++ b/src/askui/tools/caching_tools.py @@ -1,5 +1,6 @@ import logging from pathlib import Path +from typing import TYPE_CHECKING from pydantic import validate_call from typing_extensions import override @@ -7,131 +8,32 @@ from ..models.shared.tools import Tool from ..utils.caching.cache_manager import CacheManager -logger = logging.getLogger(__name__) - - -class RetrieveCachedTestExecutions(Tool): - """ - List all available trajectory files that can be used for fast-forward execution - """ - - def __init__(self, cache_dir: str, trajectories_format: str = ".json") -> None: - super().__init__( - name="retrieve_available_trajectories_tool", - description=( - "Use this tool to list all available pre-recorded trajectory " - "files in the trajectories directory. These trajectories " - "represent successful UI interaction sequences that can be " - "replayed using the execute_trajectory_tool. Call this tool " - "first to see which trajectories are available before " - "executing one. The tool returns a list of file paths to " - "available trajectory files.\n\n" - "By default, only valid (non-invalidated) caches are returned. " - "Set include_invalid=True to see all caches including those " - "marked as invalid due to repeated failures." - ), - input_schema={ - "type": "object", - "properties": { - "include_invalid": { - "type": "boolean", - "description": ( - "Whether to include invalid/invalidated caches in " - "the results. Default is False (only show valid " - "caches)." - ), - "default": False, - }, - }, - "required": [], - }, - ) - self._cache_dir = Path(cache_dir) - self._trajectories_format = trajectories_format - self.is_cacheable = True - - @override - @validate_call - def __call__(self, include_invalid: bool = False) -> list[str]: # type: ignore - """Retrieve available cached trajectories. - - Args: - include_invalid: Whether to include invalid caches - - Returns: - List of strings with filename and parameters info. - """ - logger.info( - "Retrieving cached trajectories from %s (include_invalid=%s)", - self._cache_dir, - include_invalid, - ) +if TYPE_CHECKING: + from ..speaker.cache_executor import CacheExecutor - if not Path.is_dir(self._cache_dir): - error_msg = f"Trajectories directory not found: {self._cache_dir}" - logger.error(error_msg) - raise FileNotFoundError(error_msg) - - all_files = [ - f - for f in self._cache_dir.iterdir() - if str(f).endswith(self._trajectories_format) - ] - logger.debug("Found %d total cache files", len(all_files)) - - available: list[str] = [] - invalid_count = 0 - unreadable_count = 0 - - for f in all_files: - try: - cache_file = CacheManager.read_cache_file(f) - - # Check if we should include this cache - if not include_invalid and not cache_file.metadata.is_valid: - invalid_count += 1 - logger.debug( - "Excluding invalid cache: %s (reason: %s)", - f.name, - cache_file.metadata.invalidation_reason, - ) - continue - - # Add cache info with filename and parameters - available.append( - f"filename: {f!s} (parameters: {cache_file.cache_parameters})" - ) - - except Exception: # noqa: PERF203 - unreadable_count += 1 - logger.exception("Failed to read cache file %s", f.name) - continue - - logger.info( - "Found %d cache(s), excluded %d invalid, %d unreadable", - len(available), - invalid_count, - unreadable_count, - ) - - if not available: - if include_invalid: - warning_msg = f"Warning: No trajectory files found in {self._cache_dir}" - else: - warning_msg = ( - f"Warning: No valid trajectory files found in " - f"{self._cache_dir}. " - "Try include_invalid=True to see all caches." - ) - logger.warning(warning_msg) - - return available +logger = logging.getLogger(__name__) class VerifyCacheExecution(Tool): - """Tool for agent to explicitly report cache execution verification results.""" + """Tool for the agent to report cache execution verification results. + + When wired with the active `CacheExecutor` and `CacheManager`, this tool + also persists the outcome to the trajectory's metadata: a successful + verification records the execution attempt, while an unsuccessful one + additionally invalidates the cache so it is not reused. + + Args: + cache_executor: The active `CacheExecutor`, used to resolve which + trajectory was replayed. If `None`, the tool only reports the result. + cache_manager: The active `CacheManager`, used to persist metadata. If + `None`, the tool only reports the result. + """ - def __init__(self) -> None: + def __init__( + self, + cache_executor: "CacheExecutor | None" = None, + cache_manager: "CacheManager | None" = None, + ) -> None: super().__init__( name="verify_cache_execution", description=( @@ -147,7 +49,9 @@ def __init__(self) -> None: "Set success=False if:\n" "- The execution did not achieve the target state\n" "- You had to make corrections or perform additional actions\n" - "- The final state is incorrect or incomplete" + "- The final state is incorrect or incomplete\n\n" + "Reporting success=False invalidates the cache so it is not " + "reused until it is re-recorded." ), input_schema={ "type": "object", @@ -173,12 +77,14 @@ def __init__(self) -> None: "required": ["success", "verification_notes"], }, ) + self._cache_executor = cache_executor + self._cache_manager = cache_manager self.is_cacheable = False # Verification is not cacheable @override @validate_call def __call__(self, success: bool, verification_notes: str) -> str: - """Record cache verification result. + """Record cache verification result and persist it to metadata. Args: success: Whether cache execution achieved target state @@ -197,8 +103,36 @@ def __call__(self, success: bool, verification_notes: str) -> str: logger.warning("Cache verification failed!") logger.debug("Cache verification notes: %s", verification_notes) + self._persist_verification(success, verification_notes) return message + def _persist_verification(self, success: bool, verification_notes: str) -> None: + """Persist the verification outcome to the trajectory metadata, if wired.""" + if self._cache_executor is None or self._cache_manager is None: + return + + cache_file = self._cache_executor.current_cache_file + cache_file_path = self._cache_executor.current_cache_file_path + if cache_file is None or cache_file_path is None: + logger.debug("No active cache execution to persist verification result for") + return + + if success: + self._cache_manager.update_metadata_on_completion( + cache_file=cache_file, + cache_file_path=cache_file_path, + success=True, + ) + else: + reason = ( + f"Agent reported unsuccessful cache execution: {verification_notes}" + ) + self._cache_manager.mark_execution_unsuccessful( + cache_file=cache_file, + cache_file_path=cache_file_path, + reason=reason, + ) + class InspectCacheMetadata(Tool): """ @@ -224,11 +158,7 @@ def __init__(self) -> None: "properties": { "trajectory_file": { "type": "string", - "description": ( - "Full path to the trajectory file to inspect. " - "Use retrieve_available_trajectories_tool to " - "find available files." - ), + "description": ("Full path to the trajectory file to inspect."), }, }, "required": ["trajectory_file"], @@ -249,10 +179,7 @@ def __call__(self, trajectory_file: str) -> str: logger.info("Inspecting cache metadata: %s", Path(trajectory_file).name) if not Path(trajectory_file).is_file(): - error_msg = ( - f"Trajectory file not found: {trajectory_file}\n" - "Use retrieve_available_trajectories_tool to see available files." - ) + error_msg = f"Trajectory file not found: {trajectory_file}" logger.error(error_msg) return error_msg diff --git a/src/askui/utils/caching/cache_manager.py b/src/askui/utils/caching/cache_manager.py index d59b1821..e831d958 100644 --- a/src/askui/utils/caching/cache_manager.py +++ b/src/askui/utils/caching/cache_manager.py @@ -258,6 +258,34 @@ def update_metadata_on_completion( except Exception: logger.exception("Failed to update cache metadata") + def mark_execution_unsuccessful( + self, + cache_file: CacheFile, + cache_file_path: str, + reason: str, + ) -> None: + """Record a failed execution attempt, invalidate the cache, and persist. + + Used when the agent explicitly reports (via `verify_cache_execution`) + that a replayed trajectory did not achieve the target state, so the cache + should not be trusted for future runs. + + Args: + cache_file: The cache file to update + cache_file_path: Path to write the updated cache file + reason: Human-readable reason for invalidation + """ + try: + self.record_execution_attempt(cache_file, success=False) + self.invalidate_cache(cache_file, reason=reason) + self._write_cache_file(cache_file, cache_file_path) + logger.info( + "Invalidated cache after unsuccessful execution: %s", + Path(cache_file_path).name, + ) + except Exception: + logger.exception("Failed to invalidate cache metadata") + def _write_cache_file(self, cache_file: CacheFile, cache_file_path: str) -> None: """Write cache file to disk. @@ -344,7 +372,7 @@ def start_recording( else f"{file_name}.json" ) self._goal = goal - self._toolbox = toolbox + self._toolbox = toolbox or self._toolbox self._accumulated_usage = UsageParam() self._was_cached_execution = False self._cache_writer_settings = cache_writer_settings or CacheWritingSettings() @@ -377,6 +405,14 @@ def finish_recording(self, messages: list[MessageParam]) -> str: self._reset_recording_state() return "Skipped writing cache (was cached execution)" + # Do not write (or overwrite) a cache that has nothing to replay. A + # trajectory with no cacheable steps would be a silent no-op "cache hit" + # on execute/auto and could clobber a previously good cache. + if not self._has_cacheable_steps(self._tool_blocks): + logger.info("No cacheable steps recorded; skipping cache write") + self._reset_recording_state() + return "Skipped writing cache (no cacheable steps)" + # Blank non-cacheable tool inputs BEFORE parameterization # (so they don't get sent to LLM for parameter identification) if self._toolbox is not None: @@ -452,6 +488,23 @@ def _parameterize_trajectory( vlm_provider=self._vlm_provider, ) + def _has_cacheable_steps(self, trajectory: list[ToolUseBlockParam]) -> bool: + """Whether the trajectory contains at least one cacheable tool step. + + Without a toolbox we cannot tell which tools are cacheable, so we + conservatively treat a non-empty trajectory as cacheable. + """ + if not trajectory: + return False + if self._toolbox is None: + return True + tools = self._toolbox.tool_map + for tool_block in trajectory: + tool = tools.get(tool_block.name) + if tool is None or tool.is_cacheable: + return True + return False + def _blank_non_cacheable_tool_inputs( self, trajectory: list[ToolUseBlockParam] ) -> list[ToolUseBlockParam]: @@ -645,7 +698,7 @@ def _generate_cache_file( cache_file = CacheFile( metadata=CacheMetadata( - version="0.2", + version="0.3", created_at=datetime.now(tz=timezone.utc), goal=goal_to_save, token_usage=self._accumulated_usage, diff --git a/src/askui/utils/caching/cache_parameter_handler.py b/src/askui/utils/caching/cache_parameter_handler.py index 249d7064..f1090a8e 100644 --- a/src/askui/utils/caching/cache_parameter_handler.py +++ b/src/askui/utils/caching/cache_parameter_handler.py @@ -25,6 +25,8 @@ # Regex pattern for matching parameters: {{parameter_name}} # Allows alphanumeric characters and underscores, must start with letter/underscore CACHE_PARAMETER_PATTERN = r"\{\{([a-zA-Z_][a-zA-Z0-9_]*)\}\}" +# Pattern a parameter *name* must fully match to be usable with the {{...}} syntax. +CACHE_PARAMETER_NAME_PATTERN = r"[a-zA-Z_][a-zA-Z0-9_]*" class CacheParameterDefinition: @@ -199,13 +201,11 @@ def _identify_parameters_with_llm( len(parameter_data.get("parameters", [])), ) - # Convert to our data structures - parameter_definitions = [ - CacheParameterDefinition( - name=p["name"], value=p["value"], description=p["description"] - ) - for p in parameter_data.get("parameters", []) - ] + # Convert to our data structures, dropping entries that would corrupt + # the trajectory (invalid names / empty values). + parameter_definitions = CacheParameterHandler._build_parameter_definitions( + parameter_data.get("parameters", []) + ) parameters_dict = {p.name: p.description for p in parameter_definitions} @@ -241,6 +241,47 @@ def _identify_parameters_with_llm( else: return parameters_dict, parameter_definitions + @staticmethod + def _build_parameter_definitions( + raw_parameters: Any, + ) -> list[CacheParameterDefinition]: + """Validate LLM-identified parameters, dropping corrupting entries. + + Drops parameters whose name is not a valid `{{param}}` identifier (they + would never be detected by `extract_parameters`, so validation would + wrongly pass and substitution would never happen) and parameters with an + empty value (an empty replacement key matches everywhere and would shred + every string in the trajectory). + """ + definitions: list[CacheParameterDefinition] = [] + if not isinstance(raw_parameters, list): + return definitions + for p in raw_parameters: + if not isinstance(p, dict): + continue + name = p.get("name") + value = p.get("value") + if not isinstance(name, str) or not re.fullmatch( + CACHE_PARAMETER_NAME_PATTERN, name + ): + logger.warning( + "Skipping identified parameter with invalid name: %r", name + ) + continue + if value is None or not str(value).strip(): + logger.warning( + "Skipping identified parameter %r with empty value", name + ) + continue + definitions.append( + CacheParameterDefinition( + name=name, + value=value, + description=str(p.get("description", "")), + ) + ) + return definitions + @staticmethod def _replace_values_with_parameters( trajectory: list[ToolUseBlockParam], diff --git a/tests/e2e/agent/test_act_caching.py b/tests/e2e/agent/test_act_caching.py index 711b4caa..3c1918f4 100644 --- a/tests/e2e/agent/test_act_caching.py +++ b/tests/e2e/agent/test_act_caching.py @@ -9,9 +9,10 @@ def test_act_with_caching_strategy_execute(vision_agent: ComputerAgent) -> None: - """Test that caching_strategy='execute' adds retrieve and execute tools.""" + """Test that caching_strategy='execute' with a detected trajectory runs.""" with tempfile.TemporaryDirectory() as temp_dir: - # Create a dummy cache file + # Create a dummy cache file and reference it by name so it is + # auto-detected and surfaced to the agent. cache_dir = Path(temp_dir) cache_file = cache_dir / "test_cache.json" cache_file.write_text("[]", encoding="utf-8") @@ -22,6 +23,7 @@ def test_act_with_caching_strategy_execute(vision_agent: ComputerAgent) -> None: caching_settings=CachingSettings( strategy="execute", cache_dir=str(cache_dir), + filename="test_cache.json", ), ) assert True @@ -165,6 +167,7 @@ def test_act_with_custom_cached_execution_tool_settings( caching_settings=CachingSettings( strategy="execute", cache_dir=str(cache_dir), + filename="test_cache.json", execution_settings=custom_settings, ), ) diff --git a/tests/unit/speaker/test_cache_executor.py b/tests/unit/speaker/test_cache_executor.py new file mode 100644 index 00000000..38499869 --- /dev/null +++ b/tests/unit/speaker/test_cache_executor.py @@ -0,0 +1,149 @@ +"""Unit tests for the CacheExecutor speaker.""" + +import json +import tempfile +from pathlib import Path + +import pytest + +from askui.models.shared.agent_message_param import ( + MessageParam, + TextBlockParam, + ToolUseBlockParam, +) +from askui.models.shared.tools import ToolCollection +from askui.speaker.cache_executor import CacheExecutor, ExecutionResult +from askui.utils.caching.cache_manager import CacheManager + + +def _write_trajectory(path: Path, step_names: list[str]) -> None: + """Write a cache file whose trajectory has one tool_use per given name.""" + cache_data = { + "metadata": { + "version": "0.3", + "created_at": "2025-01-01T00:00:00Z", + "is_valid": True, + "execution_attempts": 0, + "failures": [], + }, + "trajectory": [ + {"id": str(i), "name": name, "input": {}, "type": "tool_use"} + for i, name in enumerate(step_names) + ], + "cache_parameters": {}, + } + path.write_text(json.dumps(cache_data), encoding="utf-8") + + +def _first_text_block(message: MessageParam) -> str: + """Return the text of the first text block in a message's content.""" + assert isinstance(message.content, list) + block = message.content[0] + assert isinstance(block, TextBlockParam) + return block.text + + +def _context(path: Path, start_from_step_index: int) -> dict: + return { + "trajectory_file": str(path), + "start_from_step_index": start_from_step_index, + "parameter_values": {}, + "toolbox": ToolCollection(), + } + + +class TestStartIndexValidation: + def test_resume_at_end_does_not_raise(self) -> None: + """start_from_step_index == len(trajectory) means 'already complete'.""" + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "t.json" + _write_trajectory(path, ["click_a", "click_b", "click_c"]) + + executor = CacheExecutor() + # 3 steps -> index 3 is the "just past the end" resume index. + executor._activate_from_context(_context(path, 3), CacheManager()) + + assert executor._current_step_index == 3 + result = executor._get_next_step() + assert result.status == "COMPLETED" + + def test_index_beyond_end_raises(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "t.json" + _write_trajectory(path, ["click_a", "click_b"]) + + executor = CacheExecutor() + with pytest.raises(ValueError, match="Invalid start_from_step_index"): + executor._activate_from_context(_context(path, 3), CacheManager()) + + def test_negative_index_raises(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "t.json" + _write_trajectory(path, ["click_a"]) + + executor = CacheExecutor() + with pytest.raises(ValueError, match="Invalid start_from_step_index"): + executor._activate_from_context(_context(path, -1), CacheManager()) + + def test_empty_trajectory_resume_at_zero_completes(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "t.json" + _write_trajectory(path, []) + + executor = CacheExecutor() + executor._activate_from_context(_context(path, 0), CacheManager()) + result = executor._get_next_step() + assert result.status == "COMPLETED" + + +class TestHasExecutableStepsFrom: + def test_detects_remaining_executable_steps(self) -> None: + executor = CacheExecutor() + executor._trajectory = [ + ToolUseBlockParam(id="0", name="click_a", input={}), + ToolUseBlockParam(id="1", name="click_b", input={}), + ] + assert executor._has_executable_steps_from(1) is True + assert executor._has_executable_steps_from(2) is False + + def test_skippable_trailing_steps_are_ignored(self) -> None: + executor = CacheExecutor() + executor._trajectory = [ + ToolUseBlockParam(id="0", name="click_a", input={}), + ToolUseBlockParam(id="1", name="switch_speaker_abc", input={}), + ] + # Only a skippable step remains after index 0 -> nothing executable. + assert executor._has_executable_steps_from(1) is False + + +class TestNeedsAgentMessage: + def test_last_step_message_tells_agent_not_to_resume(self) -> None: + executor = CacheExecutor() + executor._trajectory = [ + ToolUseBlockParam(id="0", name="click_a", input={}), + ToolUseBlockParam(id="1", name="human_decision", input={}), + ] + result = ExecutionResult( + status="NEEDS_AGENT", + step_index=1, + tool_result=executor._trajectory[1], + ) + speaker_result = executor._handle_needs_agent(result) + text = _first_text_block(speaker_result.messages_to_add[0]) + assert "FINAL step" in text + assert "start_from_step_index" not in text + + def test_intermediate_step_message_provides_resume_index(self) -> None: + executor = CacheExecutor() + executor._trajectory = [ + ToolUseBlockParam(id="0", name="human_decision", input={}), + ToolUseBlockParam(id="1", name="click_b", input={}), + ] + result = ExecutionResult( + status="NEEDS_AGENT", + step_index=0, + tool_result=executor._trajectory[0], + ) + speaker_result = executor._handle_needs_agent(result) + text = _first_text_block(speaker_result.messages_to_add[0]) + assert "start_from_step_index=1" in text diff --git a/tests/unit/test_caching_agent_helpers.py b/tests/unit/test_caching_agent_helpers.py new file mode 100644 index 00000000..ac677d1b --- /dev/null +++ b/tests/unit/test_caching_agent_helpers.py @@ -0,0 +1,150 @@ +"""Unit tests for the caching helper logic on the Agent base class.""" + +import json +import tempfile +from datetime import datetime, timezone +from pathlib import Path + +from askui.agent_base import Agent +from askui.models.shared.agent_message_param import MessageParam, TextBlockParam +from askui.models.shared.settings import ( + CacheFile, + CacheMetadata, + CacheWritingSettings, + CachingSettings, +) + + +def _cache_file(is_valid: bool = True, parameters: dict | None = None) -> CacheFile: + return CacheFile( + metadata=CacheMetadata( + created_at=datetime.now(tz=timezone.utc), + is_valid=is_valid, + invalidation_reason=None if is_valid else "too many failures", + ), + trajectory=[], + cache_parameters=parameters or {}, + ) + + +class TestResolveCacheFilename: + def test_prefers_top_level_filename(self) -> None: + settings = CachingSettings( + filename="top.json", + writing_settings=CacheWritingSettings(filename="nested.json"), + ) + assert Agent._resolve_cache_filename(settings) == "top.json" + + def test_falls_back_to_writing_settings(self) -> None: + settings = CachingSettings( + writing_settings=CacheWritingSettings(filename="nested.json") + ) + assert Agent._resolve_cache_filename(settings) == "nested.json" + + def test_empty_when_neither_set(self) -> None: + assert Agent._resolve_cache_filename(CachingSettings()) == "" + + +class TestResolveTrajectoryPath: + def test_adds_json_suffix(self) -> None: + assert Agent._resolve_trajectory_path("dir", "login") == Path("dir/login.json") + + def test_keeps_existing_json_suffix(self) -> None: + assert Agent._resolve_trajectory_path("dir", "login.json") == Path( + "dir/login.json" + ) + + +class TestReadTrajectoryIfPresent: + def test_missing_file_returns_none(self) -> None: + assert Agent._read_trajectory_if_present(Path("/nope/x.json")) is None + + def test_reads_existing_file(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "c.json" + path.write_text( + json.dumps( + { + "metadata": { + "version": "0.3", + "created_at": "2025-01-01T00:00:00Z", + "is_valid": True, + "execution_attempts": 0, + "failures": [], + }, + "trajectory": [], + "cache_parameters": {}, + } + ), + encoding="utf-8", + ) + result = Agent._read_trajectory_if_present(path) + assert result is not None + assert result.metadata.version == "0.3" + + def test_unreadable_file_returns_none(self) -> None: + with tempfile.TemporaryDirectory() as temp_dir: + path = Path(temp_dir) / "bad.json" + path.write_text("{ this is not valid json", encoding="utf-8") + assert Agent._read_trajectory_if_present(path) is None + + +class TestBuildCacheExecutionHint: + def test_includes_path_and_switch_instruction(self) -> None: + hint = Agent._build_cache_execution_hint(Path("dir/login.json"), _cache_file()) + assert "" in hint + assert "dir/login.json" in hint + assert "switch_speaker(speaker_name='CacheExecutor'" in hint + assert "no parameters" in hint + + def test_lists_parameters(self) -> None: + hint = Agent._build_cache_execution_hint( + Path("dir/login.json"), + _cache_file(parameters={"username": "the login name"}), + ) + assert "username: the login name" in hint + assert "'username': ''" in hint + + def test_invalid_cache_is_flagged(self) -> None: + hint = Agent._build_cache_execution_hint( + Path("dir/login.json"), _cache_file(is_valid=False) + ) + assert "INVALID" in hint + assert "too many failures" in hint + + +class TestInjectCacheHint: + def test_appends_to_string_content(self) -> None: + messages = [MessageParam(role="user", content="do the thing")] + result = Agent._inject_cache_hint(messages, "HINT") + assert result[0].content == "do the thing\n\nHINT" + + def test_appends_block_to_list_content(self) -> None: + messages = [ + MessageParam( + role="user", + content=[TextBlockParam(type="text", text="do the thing")], + ) + ] + result = Agent._inject_cache_hint(messages, "HINT") + assert isinstance(result[0].content, list) + last_block = result[0].content[-1] + assert isinstance(last_block, TextBlockParam) + assert last_block.text == "HINT" + + def test_empty_messages_is_noop(self) -> None: + assert Agent._inject_cache_hint([], "HINT") == [] + + def test_targets_first_user_message_not_index_zero(self) -> None: + messages = [ + MessageParam(role="assistant", content="prior assistant turn"), + MessageParam(role="user", content="the goal"), + ] + result = Agent._inject_cache_hint(messages, "HINT") + assert result[0].content == "prior assistant turn" + assert result[1].content == "the goal\n\nHINT" + + def test_no_user_message_is_noop(self) -> None: + messages = [MessageParam(role="assistant", content="only assistant")] + result = Agent._inject_cache_hint(messages, "HINT") + assert result[0].content == "only assistant" diff --git a/tests/unit/tools/test_caching_tools.py b/tests/unit/tools/test_caching_tools.py index 4f162c86..c5465145 100644 --- a/tests/unit/tools/test_caching_tools.py +++ b/tests/unit/tools/test_caching_tools.py @@ -2,22 +2,23 @@ import json import tempfile +from datetime import datetime, timezone from pathlib import Path -import pytest - +from askui.models.shared.settings import CacheFile, CacheMetadata +from askui.speaker.cache_executor import CacheExecutor from askui.tools.caching_tools import ( InspectCacheMetadata, - RetrieveCachedTestExecutions, VerifyCacheExecution, ) +from askui.utils.caching.cache_manager import CacheManager -def _create_valid_cache_file(path: Path, is_valid: bool = True) -> None: +def _write_cache_file(path: Path, is_valid: bool = True) -> None: """Create a valid cache file with required metadata structure.""" cache_data = { "metadata": { - "version": "1.0", + "version": "0.3", "created_at": "2025-01-01T00:00:00Z", "is_valid": is_valid, "execution_attempts": 0, @@ -29,130 +30,12 @@ def _create_valid_cache_file(path: Path, is_valid: bool = True) -> None: path.write_text(json.dumps(cache_data), encoding="utf-8") -def test_retrieve_cached_test_executions_lists_json_files() -> None: - """Test that RetrieveCachedTestExecutions lists all JSON files in cache dir.""" - with tempfile.TemporaryDirectory() as temp_dir: - cache_dir = Path(temp_dir) - - # Create valid cache files - _create_valid_cache_file(cache_dir / "cache1.json") - _create_valid_cache_file(cache_dir / "cache2.json") - (cache_dir / "not_cache.txt").write_text("text", encoding="utf-8") - - tool = RetrieveCachedTestExecutions(cache_dir=str(cache_dir)) - result = tool() - - assert len(result) == 2 - assert any("cache1.json" in path for path in result) - assert any("cache2.json" in path for path in result) - assert not any("not_cache.txt" in path for path in result) - - -def test_retrieve_cached_test_executions_returns_empty_list_when_no_files() -> None: - """Test that RetrieveCachedTestExecutions returns empty list when no files exist.""" - with tempfile.TemporaryDirectory() as temp_dir: - cache_dir = Path(temp_dir) - - tool = RetrieveCachedTestExecutions(cache_dir=str(cache_dir)) - result = tool() - - assert result == [] - - -def test_retrieve_cached_test_executions_raises_error_when_dir_not_found() -> None: - """Test that RetrieveCachedTestExecutions raises error if directory doesn't exist""" - tool = RetrieveCachedTestExecutions(cache_dir="/non/existent/directory") - - with pytest.raises(FileNotFoundError, match="Trajectories directory not found"): - tool() - - -def test_retrieve_cached_test_executions_respects_custom_format() -> None: - """Test that RetrieveCachedTestExecutions respects custom file format.""" - with tempfile.TemporaryDirectory() as temp_dir: - cache_dir = Path(temp_dir) - - # Create files with different extensions - _create_valid_cache_file(cache_dir / "cache1.json") - _create_valid_cache_file(cache_dir / "cache2.traj") - - # Default format (.json) - tool_json = RetrieveCachedTestExecutions( - cache_dir=str(cache_dir), trajectories_format=".json" - ) - result_json = tool_json() - assert len(result_json) == 1 - assert "cache1.json" in result_json[0] - - # Custom format (.traj) - tool_traj = RetrieveCachedTestExecutions( - cache_dir=str(cache_dir), trajectories_format=".traj" - ) - result_traj = tool_traj() - assert len(result_traj) == 1 - assert "cache2.traj" in result_traj[0] - - -def test_retrieve_cached_test_executions_filters_invalid_by_default() -> None: - """Test that invalid caches are filtered out by default.""" - with tempfile.TemporaryDirectory() as temp_dir: - cache_dir = Path(temp_dir) - - # Create valid and invalid cache files - _create_valid_cache_file(cache_dir / "valid.json", is_valid=True) - _create_valid_cache_file(cache_dir / "invalid.json", is_valid=False) - - tool = RetrieveCachedTestExecutions(cache_dir=str(cache_dir)) - result = tool(include_invalid=False) - - assert len(result) == 1 - assert any("valid.json" in path for path in result) - assert not any("invalid.json" in path for path in result) - - -def test_retrieve_cached_test_executions_includes_invalid_when_requested() -> None: - """Test that invalid caches are included when include_invalid=True.""" - with tempfile.TemporaryDirectory() as temp_dir: - cache_dir = Path(temp_dir) - - # Create valid and invalid cache files - _create_valid_cache_file(cache_dir / "valid.json", is_valid=True) - _create_valid_cache_file(cache_dir / "invalid.json", is_valid=False) - - tool = RetrieveCachedTestExecutions(cache_dir=str(cache_dir)) - result = tool(include_invalid=True) - - assert len(result) == 2 - assert any("valid.json" in path for path in result) - assert any("invalid.json" in path for path in result) - - -def test_retrieve_cached_test_executions_returns_parameter_info() -> None: - """Test that cache parameter info is included in the result.""" - with tempfile.TemporaryDirectory() as temp_dir: - cache_dir = Path(temp_dir) - - # Create cache file with parameters - cache_data = { - "metadata": { - "version": "1.0", - "created_at": "2025-01-01T00:00:00Z", - "is_valid": True, - "execution_attempts": 0, - "failures": [], - }, - "trajectory": [], - "cache_parameters": {"target_url": "placeholder", "user_id": "123"}, - } - cache_file = cache_dir / "with_params.json" - cache_file.write_text(json.dumps(cache_data), encoding="utf-8") - - tool = RetrieveCachedTestExecutions(cache_dir=str(cache_dir)) - result = tool() - - assert len(result) == 1 - assert "parameters:" in result[0] - assert "target_url" in result[0] +def _activated_cache_executor(cache_file_path: Path) -> CacheExecutor: + """Return a CacheExecutor with an activated (loaded) cache file.""" + executor = CacheExecutor() + executor._cache_file = CacheManager.read_cache_file(cache_file_path) + executor._cache_file_path = str(cache_file_path) + return executor def test_verify_cache_execution_initializes_correctly() -> None: @@ -182,6 +65,50 @@ def test_verify_cache_execution_reports_failure() -> None: assert "Button was not clicked" in result +def test_verify_cache_execution_success_updates_metadata() -> None: + """A successful verification records the execution attempt on disk.""" + with tempfile.TemporaryDirectory() as temp_dir: + cache_path = Path(temp_dir) / "trajectory.json" + _write_cache_file(cache_path, is_valid=True) + + executor = _activated_cache_executor(cache_path) + tool = VerifyCacheExecution( + cache_executor=executor, cache_manager=CacheManager() + ) + tool(success=True, verification_notes="all good") + + persisted = CacheManager.read_cache_file(cache_path) + assert persisted.metadata.is_valid is True + assert persisted.metadata.execution_attempts == 1 + assert persisted.metadata.last_executed_at is not None + + +def test_verify_cache_execution_failure_invalidates_cache() -> None: + """An unsuccessful verification invalidates the cache on disk.""" + with tempfile.TemporaryDirectory() as temp_dir: + cache_path = Path(temp_dir) / "trajectory.json" + _write_cache_file(cache_path, is_valid=True) + + executor = _activated_cache_executor(cache_path) + tool = VerifyCacheExecution( + cache_executor=executor, cache_manager=CacheManager() + ) + tool(success=False, verification_notes="needed manual corrections") + + persisted = CacheManager.read_cache_file(cache_path) + assert persisted.metadata.is_valid is False + assert persisted.metadata.invalidation_reason is not None + assert "needed manual corrections" in persisted.metadata.invalidation_reason + + +def test_verify_cache_execution_without_wiring_is_noop() -> None: + """Without a wired executor/manager the tool only reports (no crash).""" + tool = VerifyCacheExecution() + # Should not raise even though there is nothing to persist. + result = tool(success=False, verification_notes="no active execution") + assert "success=False" in result + + def test_inspect_cache_metadata_initializes_correctly() -> None: """Test that InspectCacheMetadata initializes correctly.""" tool = InspectCacheMetadata() @@ -204,7 +131,7 @@ def test_inspect_cache_metadata_returns_metadata() -> None: cache_file = Path(temp_dir) / "test_cache.json" cache_data = { "metadata": { - "version": "1.0", + "version": "0.3", "created_at": "2025-01-01T00:00:00Z", "is_valid": True, "execution_attempts": 5, @@ -221,8 +148,19 @@ def test_inspect_cache_metadata_returns_metadata() -> None: result = tool(trajectory_file=str(cache_file)) assert "=== Cache Metadata ===" in result - assert "Version: 1.0" in result + assert "Version: 0.3" in result assert "Is Valid: True" in result assert "Total Execution Attempts: 5" in result assert "Total Steps: 1" in result assert "url" in result + + +def test_cache_manager_generates_version_0_3() -> None: + """New cache files are written with the current 0.3 metadata version.""" + assert CacheMetadata(created_at=datetime.now(tz=timezone.utc)).version == "0.3" + # Sanity: CacheFile round-trips with the new version. + cache_file = CacheFile( + metadata=CacheMetadata(created_at=datetime.now(tz=timezone.utc)), + trajectory=[], + ) + assert cache_file.metadata.version == "0.3" diff --git a/tests/unit/utils/caching/__init__.py b/tests/unit/utils/caching/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unit/utils/caching/test_cache_manager.py b/tests/unit/utils/caching/test_cache_manager.py new file mode 100644 index 00000000..631ae8b3 --- /dev/null +++ b/tests/unit/utils/caching/test_cache_manager.py @@ -0,0 +1,88 @@ +"""Tests for CacheManager recording write/skip behavior.""" + +import tempfile +from pathlib import Path +from typing import Any + +from askui.models.shared.agent_message_param import MessageParam, ToolUseBlockParam +from askui.models.shared.settings import CacheWritingSettings +from askui.models.shared.tools import Tool, ToolCollection +from askui.utils.caching.cache_manager import CacheManager + + +class _CacheableTool(Tool): + def __init__(self, cacheable: bool) -> None: + super().__init__(name="mini_tool", description="mini") + self.is_cacheable = cacheable + + def __call__(self, **_: Any) -> str: + return "ok" + + +def _assistant_tool_use(tool_name: str) -> MessageParam: + return MessageParam( + role="assistant", + content=[ToolUseBlockParam(id="0", name=tool_name, input={"x": 1})], + ) + + +def test_finish_recording_skips_when_no_cacheable_steps() -> None: + """A run with no cacheable steps must not write (or overwrite) a cache file.""" + with tempfile.TemporaryDirectory() as temp_dir: + manager = CacheManager() + manager.start_recording( + cache_dir=temp_dir, + file_name="out.json", + cache_writer_settings=CacheWritingSettings( + visual_verification_method="none" + ), + ) + # No assistant tool_use messages -> empty trajectory. + result = manager.finish_recording([MessageParam(role="user", content="hi")]) + + assert "no cacheable steps" in result.lower() + assert not (Path(temp_dir) / "out.json").exists() + + +def test_finish_recording_skips_when_only_non_cacheable_steps() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + toolbox = ToolCollection(tools=[_CacheableTool(cacheable=False)]) + tool_name = next(iter(toolbox.tool_map.keys())) + + manager = CacheManager() + manager.start_recording( + cache_dir=temp_dir, + file_name="out.json", + toolbox=toolbox, + cache_writer_settings=CacheWritingSettings( + visual_verification_method="none" + ), + ) + result = manager.finish_recording([_assistant_tool_use(tool_name)]) + + assert "no cacheable steps" in result.lower() + assert not (Path(temp_dir) / "out.json").exists() + + +def test_finish_recording_writes_when_cacheable_step_present() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + toolbox = ToolCollection(tools=[_CacheableTool(cacheable=True)]) + tool_name = next(iter(toolbox.tool_map.keys())) + + manager = CacheManager() + manager.start_recording( + cache_dir=temp_dir, + file_name="out.json", + toolbox=toolbox, + cache_writer_settings=CacheWritingSettings( + visual_verification_method="none" + ), + ) + result = manager.finish_recording([_assistant_tool_use(tool_name)]) + + out = Path(temp_dir) / "out.json" + assert out.exists() + assert "Cache file written" in result + written = CacheManager.read_cache_file(out) + assert written.metadata.version == "0.3" + assert len(written.trajectory) == 1 diff --git a/tests/unit/utils/caching/test_cache_parameter_handler.py b/tests/unit/utils/caching/test_cache_parameter_handler.py new file mode 100644 index 00000000..1a9cf0b5 --- /dev/null +++ b/tests/unit/utils/caching/test_cache_parameter_handler.py @@ -0,0 +1,113 @@ +"""Tests for LLM parameter identification robustness in the recording path.""" + +from typing import Any + +from askui.models.shared.agent_message_param import ( + MessageParam, + TextBlockParam, + ToolUseBlockParam, +) +from askui.utils.caching.cache_parameter_handler import CacheParameterHandler + + +class _FakeResponse: + def __init__(self, text: str) -> None: + self.content = [TextBlockParam(type="text", text=text)] + + +class _FakeVlmProvider: + """Minimal VlmProvider stand-in returning a canned JSON response.""" + + model_id = "fake-model" + + def __init__(self, response_text: str) -> None: + self._response_text = response_text + + def create_message(self, **_: Any) -> _FakeResponse: + return _FakeResponse(self._response_text) + + +def _trajectory(value: str) -> list[ToolUseBlockParam]: + return [ToolUseBlockParam(id="0", name="type_tool", input={"text": value})] + + +def _parameterize( + response_text: str, value: str = "admin" +) -> tuple[str | None, list[ToolUseBlockParam], dict[str, str]]: + provider = _FakeVlmProvider(response_text) + return CacheParameterHandler.identify_and_parameterize( + trajectory=_trajectory(value), + goal=f"log in as {value}", + identification_strategy="llm", + vlm_provider=provider, # type: ignore[arg-type] + ) + + +class TestParameterIdentificationRobustness: + def test_empty_value_parameter_is_dropped_and_trajectory_intact(self) -> None: + """An empty parameter value must not shred every string in the trajectory.""" + response = ( + '{"parameters": [{"name": "username", "value": "", ' + '"description": "the user"}]}' + ) + goal, trajectory, params = _parameterize(response, value="Submit") + assert params == {} + # The input must be untouched (no '{{...}}' corruption between chars). + assert trajectory[0].input == {"text": "Submit"} + assert goal == "log in as Submit" + + def test_invalid_parameter_name_is_dropped(self) -> None: + """A name that is not a valid {{identifier}} would break validation.""" + response = ( + '{"parameters": [{"name": "user name", "value": "admin", ' + '"description": "the user"}]}' + ) + _, trajectory, params = _parameterize(response) + assert params == {} + assert trajectory[0].input == {"text": "admin"} + + def test_valid_parameter_is_applied(self) -> None: + response = ( + '{"parameters": [{"name": "username", "value": "admin", ' + '"description": "the user"}]}' + ) + goal, trajectory, params = _parameterize(response) + assert params == {"username": "the user"} + assert trajectory[0].input == {"text": "{{username}}"} + assert goal == "log in as {{username}}" + + def test_malformed_response_falls_back_to_no_parameters(self) -> None: + _, trajectory, params = _parameterize("not json at all") + assert params == {} + assert trajectory[0].input == {"text": "admin"} + + +class TestValidateParameters: + def test_reports_missing_parameters(self) -> None: + trajectory = [ + ToolUseBlockParam(id="0", name="type_tool", input={"text": "{{token}}"}) + ] + is_valid, missing = CacheParameterHandler.validate_parameters(trajectory, {}) + assert is_valid is False + assert missing == ["token"] + + def test_all_present(self) -> None: + trajectory = [ + ToolUseBlockParam(id="0", name="type_tool", input={"text": "{{token}}"}) + ] + is_valid, missing = CacheParameterHandler.validate_parameters( + trajectory, {"token": "abc"} + ) + assert is_valid is True + assert missing == [] + + +def test_substitute_parameters_replaces_placeholder() -> None: + block = ToolUseBlockParam(id="0", name="type_tool", input={"text": "{{token}}"}) + result = CacheParameterHandler.substitute_parameters(block, {"token": "secret"}) + assert result.input == {"text": "secret"} + + +def test_message_param_import_is_available() -> None: + # Guard that MessageParam remains importable for this module's provider stub. + assert MessageParam is not None