diff --git a/docs/06_caching.md b/docs/06_caching.md index 264eab7c..37279a7e 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,19 @@ 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), resolved **relative to `cache_dir`**. 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. + + The filename may include **subdirectories**, which lets you mirror your test tree. The SDK does not derive it automatically from the running test file — you (or your test harness) supply it — but the same value is used for both lookup and recording, so save and load always agree. For example, for a test at `tests/mytests_1/test_something.py` you can set `filename="mytests_1/test_something"` (nested directories are created on record): + + ```python + caching_settings = CachingSettings( + strategy="auto", + cache_dir=".askui_cache", + filename="mytests_1/test_something", # -> .askui_cache/mytests_1/test_something.json + ) + ``` + + A pytest harness can derive this from the test path, e.g. `filename=str(Path(request.node.path).relative_to(rootdir).with_suffix(""))`. - **`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 +70,19 @@ 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`). +- **`parameter_identification_strategy`**: How dynamic values are turned into `{{parameters}}` when recording (`"llm"`, the default, or `"preset"`). See [Dynamic Parameters](#dynamic-parameters). + +### Dynamic Parameters + +While recording, some entered values (e.g. today's date, a one-time code) must be supplied fresh on each replay rather than replayed literally. These are turned into `{{parameter}}` placeholders and requested from the agent on execution. + +With the default `"llm"` strategy, identification is deliberately **conservative and precision-first**: +- Only user-entered **free text** is considered (values typed into fields). Coordinates, key names, action/enum values, counts and tool names are never eligible, so they are never mis-parameterized. +- The model is instructed to parameterize a value **only** when replaying the recorded literal would clearly be wrong on a later run (dates relative to "now", generated IDs/tokens/OTPs, intentionally per-run identities). When in doubt, the value is left as a literal — recording zero parameters is normal and expected. +- Identified values are validated against the recorded candidates, so hallucinated or reformatted values are dropped. + +If a value you expected to be parameterized was left literal (or vice-versa), the caching logs/report show what was recorded (see [Observability](#observability)); you can also switch to `"preset"` and template values yourself with the `{{name}}` syntax. ### Execution Settings @@ -75,12 +98,23 @@ execution_settings = CacheExecutionSettings( #### Parameters -- **`delay_time_between_actions`**: The time to wait (in seconds) between executing consecutive cached actions. This delay helps ensure UI elements can materialize before the next action is executed. Defaults to `1.0` seconds. +- **`delay_time_between_actions`**: The time to wait (in seconds) between executing consecutive cached actions during replay. This delay helps ensure UI elements can materialize before the next action is executed. Defaults to `1.0` seconds. You can adjust this value based on your application's responsiveness: - For faster applications or quick interactions, you might use a smaller delay (e.g., `0.2` or `0.5` seconds) - For slower applications or complex UI updates, you might need a longer delay (e.g., `2.0` or `3.0` seconds) +> **Important:** the delay is a *playback* option, so it lives on `execution_settings`, **not** directly on `CachingSettings`: +> +> ```python +> caching_settings = CachingSettings( +> strategy="execute", +> execution_settings=CacheExecutionSettings(delay_time_between_actions=3.0), +> ) +> ``` +> +> Passing `delay_time_between_actions` directly to `CachingSettings(...)` is a mistake and now raises a validation error (previously it was silently ignored and the default of `1.0`s was used). + ## Usage Examples ### Recording a Cache @@ -89,16 +123,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 +139,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 +150,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: +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. -1. **`retrieve_available_trajectories_tool`**: Lists all available cache files in the cache directory -2. **`execute_cached_executions_tool`**: Executes a specific cached trajectory +Behavior when a trajectory is **not** found: -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. - -### 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" - ) - ) -``` - -#### 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 +200,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,16 +267,32 @@ 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. +## Observability + +Caching explains what it is doing and why, via both standard logs and the +attached reporter(s) (so the information also appears in e.g. the HTML report, +not only on stderr). Reporter messages use the source/role `Cache`. You will see +events such as: + +- **Cache hit**: `Cache hit: replaying 'login.json' (12 steps, 1 parameter(s), valid).` +- **Cache miss**: `No usable cached trajectory for 'login.json'; running normally and recording this run for next time.` +- **Pause on a non-cacheable step**: `Paused replay at step 4: the 'get_file_tool' tool cannot be replayed from cache; the agent will perform this step.` +- **Completion**: `Finished replaying 12 cached step(s); asking the agent to verify the result.` +- **Verification outcome (with the reason)**: `Cache verification FAILED for 'login.json' - the replay did not achieve the expected result. Agent's reason: the submit button was missing. The cache will be invalidated so it is not reused.` +- **Invalidation / recording**: `Cache invalidated and will not be reused: ...` and `Recorded trajectory to 'login.json' (12 steps, 1 parameter(s): current_date).` + +In particular, verification failures now include the agent's explanation and the +affected cache file, instead of an unexplained `Cache verification failed!`. + ## Limitations - **UI State Sensitivity**: Cached trajectories assume the UI is in the same state as when they were recorded. If the UI has changed, the replay may fail or produce incorrect results. @@ -354,7 +307,6 @@ from askui import ComputerAgent from askui.models.shared.settings import ( CachingSettings, CacheExecutionSettings, - CacheWritingSettings, ) # Step 1: Record a successful login flow @@ -365,9 +317,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 +325,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..84df7198 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,13 +32,16 @@ from askui.tools.android.agent_os import AndroidAgentOs from askui.tools.caching_tools import ( InspectCacheMetadata, - RetrieveCachedTestExecutions, VerifyCacheExecution, ) from askui.tools.get_tool import GetTool from askui.tools.locate_tool import LocateTool from askui.utils.annotation_writer import AnnotationWriter -from askui.utils.caching.cache_manager import CacheManager +from askui.utils.caching.cache_manager import ( + CacheManager, + ensure_relative_cache_filename, +) +from askui.utils.caching.reporting_utils import report_cache_event from askui.utils.image_utils import ImageSource from askui.utils.source_utils import InputSource, load_image_source, load_source @@ -269,13 +273,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 +306,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 +339,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 +358,81 @@ 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) - # 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 + # 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(reporter=self._reporter) + + # 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, + reporter=self._reporter, + ), 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) + + validity = ( + "valid" if cache_file.metadata.is_valid else "INVALID (will try anyway)" + ) + report_cache_event( + self._reporter, + f"Cache hit: replaying '{trajectory_path.name}' " + f"({len(cache_file.trajectory)} steps, " + f"{len(cache_file.cache_parameters)} parameter(s), {validity}).", + log=logger, + ) + else: + cache_hint = self._report_cache_miss(strategy, filename) # Add caching tools to the tools list if isinstance(tools, list): @@ -366,14 +442,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 +455,158 @@ def _patch_act_with_cache( vlm_provider=self._vlm_provider, ) - return tools, cache_manager + return tools, cache_manager, cache_hint + + def _report_cache_miss(self, strategy: str | None, filename: str) -> str | None: + """Report that no usable trajectory was found and return the miss hint. + + Returns the "no cached trajectory" hint in auto mode (so the agent knows + it is recording for next time) and ``None`` otherwise. + """ + if strategy == "auto": + if filename: + report_cache_event( + self._reporter, + f"No usable cached trajectory for '{filename}'; running " + "normally and recording this run for next time.", + log=logger, + ) + return self._build_no_cache_hint() + if strategy == "execute" and filename: + report_cache_event( + self._reporter, + f"No usable cached trajectory for '{filename}'; running normally " + "(execute mode does not record).", + log=logger, + ) + return None + + @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. + + The filename may include subdirectories but must stay within + ``cache_dir`` (see `ensure_relative_cache_filename`). + """ + ensure_relative_cache_filename(filename) + 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..24463a26 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,11 +234,14 @@ 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 """ + # Reject unknown fields so a misplaced/misspelled setting raises instead of + # being silently ignored. + model_config = ConfigDict(extra="forbid") + filename: str = "" parameter_identification_strategy: CACHE_PARAMETER_IDENTIFICATION_STRATEGY = "llm" visual_verification_method: CACHING_VISUAL_VERIFICATION_METHOD = "phash" @@ -254,6 +257,10 @@ class CacheExecutionSettings(BaseModel): visual_validation_threshold: Max Hamming distance for validation """ + # Reject unknown fields so a misplaced/misspelled setting raises instead of + # being silently ignored. + model_config = ConfigDict(extra="forbid") + delay_time_between_actions: float = 1.0 # keep >1s to give UI time to materialize skip_visual_validation: bool = False visual_validation_threshold: int = 10 @@ -273,11 +280,31 @@ 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) + + Note: + Playback options such as the delay between replayed actions live on + `execution_settings` (a `CacheExecutionSettings`), e.g. + `CachingSettings(execution_settings=CacheExecutionSettings( + delay_time_between_actions=3.0))`. Unknown top-level fields are rejected + so a misplaced option raises instead of being silently ignored. """ + # Reject unknown fields so a misplaced/misspelled setting (e.g. passing + # delay_time_between_actions here instead of on execution_settings) raises + # instead of being silently ignored. + model_config = ConfigDict(extra="forbid") + 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/prompts/caching.py b/src/askui/prompts/caching.py index 26de3deb..29ae1b97 100644 --- a/src/askui/prompts/caching.py +++ b/src/askui/prompts/caching.py @@ -1,36 +1,43 @@ -CACHING_PARAMETER_IDENTIFIER_SYSTEM_PROMPT = """You are analyzing UI automation \ -trajectories to identify values that should be parameterized as parameters. +CACHING_PARAMETER_IDENTIFIER_SYSTEM_PROMPT = """You are reviewing a short list \ +of text values that a user or agent entered during a UI automation recording. \ +The recording will be REPLAYED VERBATIM on future runs. Your job is to decide \ +which of these values MUST be supplied fresh on each run because replaying the \ +recorded literal would be wrong. -Identify values that are likely to change between executions, such as: -- Dates and timestamps (e.g., "2025-12-11", "10:30 AM", "2025-12-11T14:30:00Z") -- Usernames, emails, names (e.g., "john.doe", "test@example.com", "John Smith") -- Session IDs, tokens, UUIDs, API keys -- Dynamic text that references current state or time-sensitive information -- File paths with user-specific or time-specific components -- Temporary or generated identifiers +Be conservative and precise. Parameterize a value ONLY if replaying the exact \ +recorded literal on a later run would clearly produce a wrong, stale, or invalid \ +result. When in doubt, DO NOT parameterize: a value left alone is replayed \ +literally, which is the desired default. Returning an empty list is a normal and \ +common answer. -DO NOT mark as parameters: -- UI element coordinates (x, y positions) -- Fixed button labels or static UI text -- Configuration values that don't change (e.g., timeouts, retry counts) -- Generic action names like "click", "type", "scroll" -- Tool names -- Boolean values or common constants +Parameterize (these are genuinely run-specific / dynamic): +- Values relative to "now": today's date, a current timestamp, "tomorrow", etc. +- One-time or generated values: session IDs, tokens, OTPs, UUIDs, verification \ +codes, or order/reference numbers created during this run +- A per-run identity the caller clearly varies on purpose (e.g. the specific \ +username/email being tested) - and only when it is clearly run-specific -For each parameter, provide: -1. A descriptive name in snake_case (e.g., "current_date", "user_email") -2. The actual value found in the trajectory -3. A brief description of what it represents +Do NOT parameterize (replay the literal): +- Stable text that is part of the test itself: search terms, fixed field \ +contents, button/label text, messages, URLs that never change +- Values that would be identical every time this test runs +- Numbers, quantities, or short tokens unless they are clearly generated/dynamic +- Anything you are unsure about -Return your analysis as a JSON object with this structure: +For each value you DO parameterize, return: +- name: a snake_case identifier (e.g. "current_date", "otp_code") +- value: the value EXACTLY as it appears in the provided list, verbatim +- description: what it represents AND why it must change on each run + +Return your analysis as a JSON object with this exact structure: { "parameters": [ { "name": "current_date", "value": "2025-12-11", - "description": "Current date in YYYY-MM-DD format" + "description": "Today's date; must be the run date, not the recorded date" } ] } -If no parameters are found, return an empty parameters array.""" +If nothing qualifies, return {"parameters": []}.""" diff --git a/src/askui/speaker/cache_executor.py b/src/askui/speaker/cache_executor.py index 404c0b21..b630c5fb 100644 --- a/src/askui/speaker/cache_executor.py +++ b/src/askui/speaker/cache_executor.py @@ -17,6 +17,7 @@ from askui.models.shared.settings import CacheExecutionSettings from askui.utils.caching.cache_manager import CacheManager from askui.utils.caching.cache_parameter_handler import CacheParameterHandler +from askui.utils.caching.reporting_utils import report_cache_event from askui.utils.visual_validation import ( compute_ahash, compute_hamming_distance, @@ -126,6 +127,19 @@ def __init__( # Activation context received via on_activate() self._activation_context: dict[str, Any] = {} + # Reporter for surfacing replay progress (set on activation). + self._reporter: "Reporter | None" = None + + @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 +220,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", @@ -266,16 +282,36 @@ def _handle_success(self, result: ExecutionResult) -> SpeakerResult: def _handle_needs_agent(self, result: ExecutionResult) -> SpeakerResult: """Handle cache execution pausing for non-cacheable tool.""" - logger.info( - "Paused cache execution at step %d " - "(non-cacheable tool - agent will handle this step)", - result.step_index, + tool_name = getattr(result.tool_result, "name", "unknown") + report_cache_event( + self._reporter, + f"Paused replay at step {result.step_index}: the '{tool_name}' tool " + "cannot be replayed from cache; the agent will perform this step.", + log=logger, ) self._executing_from_cache = False 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 +322,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}" ), ) ], @@ -309,8 +344,11 @@ def _handle_completed( result: ExecutionResult, # noqa: ARG002 ) -> SpeakerResult: """Handle cache execution completion.""" - logger.info( - "Cache trajectory execution completed - requesting agent verification" + report_cache_event( + self._reporter, + f"Finished replaying {len(self._trajectory)} cached step(s); " + "asking the agent to verify the result.", + log=logger, ) self._executing_from_cache = False self._cache_verification_pending = True @@ -344,10 +382,13 @@ def _handle_failed( self, cache_manager: CacheManager, result: ExecutionResult ) -> SpeakerResult: """Handle cache execution failure.""" - logger.error( - "Cache execution failed at step %d: %s", - result.step_index, - result.error_message, + report_cache_event( + self._reporter, + f"Cache replay failed at step {result.step_index}: " + f"{result.error_message or 'unknown error'}. " + "The agent will complete the task manually.", + log=logger, + level=logging.ERROR, ) self._executing_from_cache = False @@ -415,14 +456,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) @@ -475,15 +522,22 @@ def _activate_from_context( self._visual_validation_enabled = False logger.debug("Visual validation disabled or not configured") - logger.info( - "Cache execution activated: %s (%d steps, starting from step %d)", - Path(trajectory_file).name, - len(self._cache_file.trajectory), - start_from_step_index, + # Reporter for surfacing replay progress/outcomes to the user. + reporter: Reporter | None = context.get("reporter") + self._reporter = reporter + + step_count = len(self._cache_file.trajectory) + from_suffix = ( + "" if start_from_step_index == 0 else f" from step {start_from_step_index}" + ) + report_cache_event( + self._reporter, + f"Replaying cached trajectory '{Path(trajectory_file).name}' " + f"({step_count} steps){from_suffix}.", + log=logger, ) # Report cache execution statistics to the reporter - reporter: Reporter | None = context.get("reporter") if reporter and self._cache_file.metadata.token_usage: reporter.add_cache_execution_statistics( self._cache_file.metadata.token_usage.model_dump() @@ -501,6 +555,7 @@ def reset_state(self) -> None: self._current_step_index = 0 self._message_history = [] self._activation_context = {} + self._reporter = None def _get_next_step( self, conversation_messages: list[MessageParam] | None = None @@ -520,7 +575,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 +639,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..6aeb6932 100644 --- a/src/askui/tools/caching_tools.py +++ b/src/askui/tools/caching_tools.py @@ -1,137 +1,44 @@ import logging from pathlib import Path +from typing import TYPE_CHECKING from pydantic import validate_call from typing_extensions import override from ..models.shared.tools import Tool from ..utils.caching.cache_manager import CacheManager +from ..utils.caching.reporting_utils import report_cache_event -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 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) +if TYPE_CHECKING: + from ..reporting import Reporter + from ..speaker.cache_executor import CacheExecutor - 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. + reporter: Optional reporter used to surface the verification outcome + (and its reason) to the user in addition to the logs. + """ - def __init__(self) -> None: + def __init__( + self, + cache_executor: "CacheExecutor | None" = None, + cache_manager: "CacheManager | None" = None, + reporter: "Reporter | None" = None, + ) -> None: super().__init__( name="verify_cache_execution", description=( @@ -147,7 +54,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 +82,15 @@ def __init__(self) -> None: "required": ["success", "verification_notes"], }, ) + self._cache_executor = cache_executor + self._cache_manager = cache_manager + self._reporter = reporter 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 @@ -187,17 +99,64 @@ def __call__(self, success: bool, verification_notes: str) -> str: Returns: Confirmation message """ - message = ( + cache_name = self._current_cache_name() + if success: + report_cache_event( + self._reporter, + f"Cache verification PASSED for {cache_name}: {verification_notes}", + log=logger, + level=logging.INFO, + ) + else: + report_cache_event( + self._reporter, + f"Cache verification FAILED for {cache_name} - the replay did not " + f"achieve the expected result. Agent's reason: {verification_notes}. " + "The cache will be invalidated so it is not reused.", + log=logger, + level=logging.WARNING, + ) + + self._persist_verification(success, verification_notes) + return ( f"Cache verification reported: success={success}, " f"notes={verification_notes}" ) + + def _current_cache_name(self) -> str: + """Human-readable name of the trajectory being verified.""" + if self._cache_executor is not None: + path = self._cache_executor.current_cache_file_path + if path: + return f"'{Path(path).name}'" + return "the cached trajectory" + + 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: - logger.info("Cache verified successfully") + self._cache_manager.update_metadata_on_completion( + cache_file=cache_file, + cache_file_path=cache_file_path, + success=True, + ) else: - logger.warning("Cache verification failed!") - logger.debug("Cache verification notes: %s", verification_notes) - - return message + 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 +183,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 +204,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..cc06c847 100644 --- a/src/askui/utils/caching/cache_manager.py +++ b/src/askui/utils/caching/cache_manager.py @@ -3,7 +3,7 @@ import json import logging from datetime import datetime, timezone -from pathlib import Path +from pathlib import Path, PurePath from typing import TYPE_CHECKING, Any from PIL import Image @@ -29,6 +29,7 @@ StepFailureCountValidator, TotalFailureRateValidator, ) +from askui.utils.caching.reporting_utils import report_cache_event from askui.utils.visual_validation import ( compute_ahash, compute_phash, @@ -39,10 +40,33 @@ if TYPE_CHECKING: from askui.model_providers.vlm_provider import VlmProvider + from askui.reporting import Reporter logger = logging.getLogger(__name__) +def ensure_relative_cache_filename(filename: str) -> None: + """Validate that a cache filename stays within ``cache_dir``. + + The filename may include subdirectories, but must be relative and must not + traverse upwards, so it cannot escape ``cache_dir`` (e.g. an absolute path or + ``../..`` would otherwise be written/read outside the cache directory). + + Args: + filename: The cache filename to validate (may include subdirectories). + + Raises: + ValueError: If the filename is absolute or contains a ``..`` component. + """ + pure = PurePath(filename) + if pure.is_absolute() or ".." in pure.parts: + error_msg = ( + "Cache filename must be relative to cache_dir and must not contain " + f"'..' or be absolute. Got: {filename!r}" + ) + raise ValueError(error_msg) + + class CacheManager: """Manages cache metadata, validation, updates, and recording. @@ -56,13 +80,21 @@ class CacheManager: - Updating metadata on disk """ - def __init__(self, validators: list[CacheValidator] | None = None) -> None: + def __init__( + self, + validators: list[CacheValidator] | None = None, + reporter: "Reporter | None" = None, + ) -> None: """Initialize cache manager. Args: validators: Optional list of cache validators. If None, uses default validators (StepFailureCount, TotalFailureRate, StaleCache). + reporter: Optional reporter used to surface recording/invalidation + activity to the user in addition to the logs. """ + self._reporter = reporter + # Validation if validators is None: # Use default validators @@ -167,7 +199,14 @@ def invalidate_cache(self, cache_file: CacheFile, reason: str) -> None: """ cache_file.metadata.is_valid = False cache_file.metadata.invalidation_reason = reason - logger.warning("Cache invalidated: %s", reason) + # Concise headline at WARNING; the full (possibly long) reason at INFO. + report_cache_event( + self._reporter, + "Cache invalidated and will not be reused.", + log=logger, + level=logging.WARNING, + detail=f"Invalidation reason: {reason}", + ) def mark_cache_valid(self, cache_file: CacheFile) -> None: """Mark a cache file as valid. @@ -258,6 +297,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. @@ -266,6 +333,7 @@ def _write_cache_file(self, cache_file: CacheFile, cache_file_path: str) -> None cache_file_path: Path to write the cache file """ cache_path = Path(cache_file_path) + cache_path.parent.mkdir(parents=True, exist_ok=True) with cache_path.open("w", encoding="utf-8") as f: json.dump( cache_file.model_dump(mode="json"), @@ -336,15 +404,20 @@ def start_recording( """ self._recording = True self._tool_blocks = [] + if file_name: + ensure_relative_cache_filename(file_name) self._cache_dir = Path(cache_dir) - self._cache_dir.mkdir(exist_ok=True) + # `parents=True` so a nested cache_dir (e.g. ".askui_cache/mytests_1") + # is created. The per-file parent for nested filenames is created at + # write time in `_generate_cache_file` / `_write_cache_file`. + self._cache_dir.mkdir(parents=True, exist_ok=True) self._file_name = ( file_name if file_name.endswith(".json") or not file_name 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 +450,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: @@ -407,6 +488,19 @@ def finish_recording(self, messages: list[MessageParam]) -> str: goal_to_save, trajectory_to_save, parameters_dict, cache_file_path ) + if parameters_dict: + param_summary = f"{len(parameters_dict)} parameter(s): " + ", ".join( + parameters_dict.keys() + ) + else: + param_summary = "no parameters" + report_cache_event( + self._reporter, + f"Recorded trajectory to '{cache_file_path.name}' " + f"({len(trajectory_to_save)} steps, {param_summary}).", + log=logger, + ) + # Reset recording state self._reset_recording_state() @@ -452,6 +546,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 +756,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, @@ -655,6 +766,7 @@ def _generate_cache_file( cache_parameters=parameters_dict, ) + cache_file_path.parent.mkdir(parents=True, exist_ok=True) with cache_file_path.open("w", encoding="utf-8") as f: json.dump(cache_file.model_dump(mode="json"), f, indent=4) logger.info("Cache file successfully written: %s", cache_file_path) diff --git a/src/askui/utils/caching/cache_parameter_handler.py b/src/askui/utils/caching/cache_parameter_handler.py index 249d7064..4c5b2f6b 100644 --- a/src/askui/utils/caching/cache_parameter_handler.py +++ b/src/askui/utils/caching/cache_parameter_handler.py @@ -25,6 +25,40 @@ # 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_]*" + +# Tool-input keys that never hold user-entered free text (coordinates, enums, +# key names, counts, ...). Values under these keys are never offered to the LLM +# as parameter candidates, which removes the bulk of false positives (clicks' +# coordinates, action names like "left_click", key names, tool names, etc.). +# +# Only *structural* keys are listed here. Content-capable keys that a user might +# legitimately type a dynamic value into (e.g. "id" for a reference/order id, +# "name" for a username, "amount" for a typed monetary value) are intentionally +# NOT excluded - they are offered to the LLM, and the conservative system prompt +# decides whether they are truly run-specific. +CACHE_INPUT_CONTROL_KEYS = frozenset( + { + "action", + "button", + "clicks", + "coordinate", + "coordinates", + "count", + "direction", + "duration", + "key", + "keys", + "scroll_amount", + "scroll_direction", + "start_coordinate", + "tool", + "type", + "x", + "y", + } +) class CacheParameterDefinition: @@ -135,23 +169,35 @@ def _identify_parameters_with_llm( logger.debug("Empty trajectory provided, skipping parameter identification") return {}, [] + # Only user-entered free-text values are eligible to become parameters. + # This excludes coordinates, key names, action/enum values and tool + # names up front, so the model is never even asked about them. + candidate_values = CacheParameterHandler._collect_candidate_values(trajectory) + if not candidate_values: + logger.info( + "No free-text values in trajectory; skipping parameter identification" + ) + return {}, [] + logger.info( - "Starting parameter identification for trajectory with %s steps", + "Evaluating %s candidate value(s) for parameterization " + "(trajectory has %s steps)", + len(candidate_values), len(trajectory), ) - # Convert trajectory to serializable format for analysis - trajectory_data = [tool.model_dump(mode="json") for tool in trajectory] - logger.debug("Converted %s tool blocks to JSON format", len(trajectory_data)) - + candidate_list = "\n".join( + f"{i}. {value!r}" for i, value in enumerate(candidate_values, 1) + ) user_message = ( - "Analyze this UI automation trajectory and identify all values that " - "should be parameters:\n\n" - f"```json\n{json.dumps(trajectory_data, indent=2)}\n```\n\n" - "Return only the JSON object with identified parameters. " - "Be thorough but conservative - only mark values that are clearly " - "dynamic or time-sensitive." + "The following text values were entered during the recording. For " + "each, decide whether it MUST be supplied fresh on every run (see the " + "instructions). Only choose values from this list, verbatim; when in " + "doubt, leave a value out.\n\n" + f"Candidate values:\n{candidate_list}\n\n" + "Return only the JSON object with the parameters that qualify." ) + allowed_values = set(candidate_values) response_text = "" # Initialize for error logging try: @@ -168,44 +214,23 @@ def _identify_parameters_with_llm( ) logger.debug("Received response from LLM") - # Extract text from response - if isinstance(response.content, list): - response_text = next( - ( - block.text - for block in response.content - if hasattr(block, "text") - ), - "", - ) - else: - response_text = str(response.content) - - # Parse the JSON response + response_text = CacheParameterHandler._extract_response_text(response) logger.debug("Parsing LLM response to extract parameter definitions") - # Handle markdown code blocks if present - if "```json" in response_text: - logger.debug("Removing JSON markdown code block wrapper from response") - response_text = ( - response_text.split("```json")[1].split("```")[0].strip() - ) - elif "```" in response_text: - logger.debug("Removing code block wrapper from response") - response_text = response_text.split("```")[1].split("```")[0].strip() - - parameter_data = json.loads(response_text) + parameter_data = json.loads( + CacheParameterHandler._strip_json_code_fence(response_text) + ) logger.debug( "Successfully parsed JSON response with %s parameters", 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) or that the model + # invented (values not among the offered candidates). + parameter_definitions = CacheParameterHandler._build_parameter_definitions( + parameter_data.get("parameters", []), + allowed_values=allowed_values, + ) parameters_dict = {p.name: p.description for p in parameter_definitions} @@ -241,6 +266,128 @@ def _identify_parameters_with_llm( else: return parameters_dict, parameter_definitions + @staticmethod + def _extract_response_text(response: Any) -> str: + """Extract the text payload from a VLM response message.""" + if isinstance(response.content, list): + return next( + (block.text for block in response.content if hasattr(block, "text")), + "", + ) + return str(response.content) + + @staticmethod + def _strip_json_code_fence(response_text: str) -> str: + """Strip a ```json ...``` (or ``` ...```) fence around the JSON payload.""" + if "```json" in response_text: + return response_text.split("```json")[1].split("```")[0].strip() + if "```" in response_text: + return response_text.split("```")[1].split("```")[0].strip() + return response_text + + @staticmethod + def _build_parameter_definitions( + raw_parameters: Any, + allowed_values: set[str] | None = None, + ) -> 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), parameters with an + empty value (an empty replacement key matches everywhere and would shred + every string in the trajectory), and - when `allowed_values` is given - + parameters whose value was not one of the candidate values offered to the + model (i.e. hallucinated or reformatted values that would not match). + """ + 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 + if allowed_values is not None and str(value) not in allowed_values: + logger.warning( + "Skipping identified parameter %r: value %r was not a candidate", + name, + value, + ) + continue + definitions.append( + CacheParameterDefinition( + name=name, + value=value, + description=str(p.get("description", "")), + ) + ) + return definitions + + @staticmethod + def _collect_candidate_values( + trajectory: list[ToolUseBlockParam], + ) -> list[str]: + """Collect user-entered free-text values eligible for parameterization. + + Walks each tool input and collects string leaf values, skipping values + under control keys (see `CACHE_INPUT_CONTROL_KEYS`) such as coordinates, + key names, action/enum values and counts. Values are de-duplicated while + preserving first-seen order. This is the set of values the model is + allowed to choose from. + """ + candidates: list[str] = [] + seen: set[str] = set() + for block in trajectory: + CacheParameterHandler._collect_from_input( + block.input, candidates, seen, key_allowed=True + ) + return candidates + + @staticmethod + def _collect_from_input( + value: Any, + candidates: list[str], + seen: set[str], + key_allowed: bool, + ) -> None: + """Recursively gather candidate strings from a tool input value.""" + if isinstance(value, str): + stripped = value.strip() + if ( + key_allowed + and len(stripped) >= 2 + and "{{" not in value + and value not in seen + ): + seen.add(value) + candidates.append(value) + elif isinstance(value, dict): + for key, sub_value in value.items(): + sub_allowed = ( + key_allowed and str(key).lower() not in CACHE_INPUT_CONTROL_KEYS + ) + CacheParameterHandler._collect_from_input( + sub_value, candidates, seen, sub_allowed + ) + elif isinstance(value, list): + for item in value: + CacheParameterHandler._collect_from_input( + item, candidates, seen, key_allowed + ) + @staticmethod def _replace_values_with_parameters( trajectory: list[ToolUseBlockParam], diff --git a/src/askui/utils/caching/reporting_utils.py b/src/askui/utils/caching/reporting_utils.py new file mode 100644 index 00000000..b6e783d9 --- /dev/null +++ b/src/askui/utils/caching/reporting_utils.py @@ -0,0 +1,49 @@ +"""Helpers for surfacing caching activity to both logs and the reporter. + +Caching decisions (cache hit/miss, replay progress, pauses, verification +outcomes, invalidation, recording) are relevant to users trying to understand +"what and why" the cache is doing. These helpers emit a single message to the +standard logger AND, when a reporter is available, to the reporter so the +information also shows up in the HTML report / attached reporters instead of only +on stderr. +""" + +import logging +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from askui.reporting import Reporter + +# Source/role used for caching messages in reporters. +CACHE_REPORTER_SOURCE = "Cache" + + +def report_cache_event( + reporter: "Reporter | None", + message: str, + *, + log: logging.Logger, + level: int = logging.INFO, + detail: str | None = None, +) -> None: + """Emit a caching event to the logger and (if present) the reporter. + + Args: + reporter: The reporter to forward the message to, or ``None``. + message: Concise headline of what/why the cache is doing. + log: The module logger to write to. + level: Logging level for the headline record (default ``logging.INFO``). + detail: Optional longer description. It is always logged at ``INFO`` so a + verbose reason does not bloat a higher-level (e.g. WARNING) headline, + and it is appended to the reporter message so the full context is + still surfaced there. + """ + log.log(level, message) + if detail: + log.info(detail) + if reporter is not None: + reporter_message = message if not detail else f"{message} {detail}" + try: + reporter.add_message(CACHE_REPORTER_SOURCE, reporter_message) + except Exception: # noqa: BLE001 - reporting must never break caching + log.debug("Failed to forward cache event to reporter", exc_info=True) 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..edbea5b9 --- /dev/null +++ b/tests/unit/test_caching_agent_helpers.py @@ -0,0 +1,179 @@ +"""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 + +import pytest +from pydantic import ValidationError + +from askui.agent_base import Agent +from askui.models.shared.agent_message_param import MessageParam, TextBlockParam +from askui.models.shared.settings import ( + CacheExecutionSettings, + CacheFile, + CacheMetadata, + CacheWritingSettings, + CachingSettings, +) + + +class TestCachingSettingsRejectUnknownFields: + def test_misplaced_delay_raises_instead_of_being_ignored(self) -> None: + # delay_time_between_actions belongs on execution_settings, not here. + with pytest.raises(ValidationError): + CachingSettings(strategy="execute", delay_time_between_actions=3.0) # type: ignore[call-arg] + + def test_delay_on_execution_settings_is_applied(self) -> None: + settings = CachingSettings( + strategy="execute", + execution_settings=CacheExecutionSettings(delay_time_between_actions=3.0), + ) + assert settings.execution_settings is not None + assert settings.execution_settings.delay_time_between_actions == 3.0 + + +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" + ) + + def test_preserves_nested_subdirectories(self) -> None: + assert Agent._resolve_trajectory_path( + ".askui_cache", "mytests_1/test_something" + ) == Path(".askui_cache/mytests_1/test_something.json") + + def test_rejects_filename_escaping_cache_dir(self) -> None: + for bad in ("/abs/name", "../escape", "a/../../b"): + with pytest.raises(ValueError, match="relative to cache_dir"): + Agent._resolve_trajectory_path(".askui_cache", bad) + + +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..b1c8609a --- /dev/null +++ b/tests/unit/utils/caching/test_cache_manager.py @@ -0,0 +1,156 @@ +"""Tests for CacheManager recording write/skip behavior.""" + +import tempfile +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +import pytest + +from askui.models.shared.agent_message_param import MessageParam, ToolUseBlockParam +from askui.models.shared.settings import ( + CacheFile, + CacheMetadata, + CacheWritingSettings, +) +from askui.models.shared.tools import Tool, ToolCollection +from askui.utils.caching.cache_manager import ( + CacheManager, + ensure_relative_cache_filename, +) + + +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 + + +def test_finish_recording_creates_nested_directories() -> None: + """A filename with subdirectories is saved at the matching nested path.""" + 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="mytests_1/test_something.json", + toolbox=toolbox, + cache_writer_settings=CacheWritingSettings( + visual_verification_method="none" + ), + ) + result = manager.finish_recording([_assistant_tool_use(tool_name)]) + + nested = Path(temp_dir) / "mytests_1" / "test_something.json" + assert nested.exists() + assert "Cache file written" in result + + +class TestEnsureRelativeCacheFilename: + def test_accepts_nested_relative(self) -> None: + # Should not raise. + ensure_relative_cache_filename("mytests_1/test_something.json") + + @pytest.mark.parametrize( + "bad", + ["/tmp/evil.json", "../../foo.json", "a/../../b.json"], + ) + def test_rejects_absolute_and_traversal(self, bad: str) -> None: + with pytest.raises(ValueError, match="relative to cache_dir"): + ensure_relative_cache_filename(bad) + + +def test_start_recording_rejects_unsafe_filename() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + manager = CacheManager() + with pytest.raises(ValueError, match="relative to cache_dir"): + manager.start_recording(cache_dir=temp_dir, file_name="../escape.json") + + +def test_update_metadata_on_completion_creates_nested_directories() -> None: + """Metadata writes also create nested parents (never crash on missing dir).""" + with tempfile.TemporaryDirectory() as temp_dir: + nested = Path(temp_dir) / "suite" / "case.json" + cache_file = CacheFile( + metadata=CacheMetadata(created_at=datetime.now(tz=timezone.utc)), + trajectory=[], + ) + manager = CacheManager() + manager.update_metadata_on_completion(cache_file, str(nested), success=True) + + assert nested.exists() + assert CacheManager.read_cache_file(nested).metadata.execution_attempts == 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..930ab056 --- /dev/null +++ b/tests/unit/utils/caching/test_cache_parameter_handler.py @@ -0,0 +1,199 @@ +"""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 + + +class _RaisingVlmProvider: + """Provider that fails if the LLM is called - used to assert it is skipped.""" + + model_id = "should-not-be-called" + + def create_message(self, **_: Any) -> _FakeResponse: + msg = "LLM should not be called when there are no candidates" + raise AssertionError(msg) + + +class TestCandidateCollection: + def test_excludes_coordinates_actions_keys_and_tool_names(self) -> None: + trajectory = [ + ToolUseBlockParam( + id="0", + name="computer_tool", + input={"action": "left_click", "coordinate": [100, 200]}, + ), + ToolUseBlockParam( + id="1", name="keyboard_tool", input={"action": "key", "key": "Return"} + ), + ToolUseBlockParam( + id="2", + name="computer_tool", + input={"action": "type", "text": "hello world"}, + ), + ] + candidates = CacheParameterHandler._collect_candidate_values(trajectory) + # Only the typed free text is a candidate. + assert candidates == ["hello world"] + + def test_content_capable_keys_are_offered_as_candidates(self) -> None: + # id / name / amount can hold user-entered dynamic values, so they are + # NOT excluded (only truly structural keys are). + trajectory = [ + ToolUseBlockParam( + id="0", + name="form_tool", + input={"id": "ORDER-12345", "name": "Jane Doe", "amount": "100.00"}, + ) + ] + candidates = CacheParameterHandler._collect_candidate_values(trajectory) + assert candidates == ["ORDER-12345", "Jane Doe", "100.00"] + + def test_dedupes_and_skips_short_and_templated_values(self) -> None: + trajectory = [ + ToolUseBlockParam(id="0", name="t", input={"text": "repeat"}), + ToolUseBlockParam(id="1", name="t", input={"text": "repeat"}), + ToolUseBlockParam(id="2", name="t", input={"note": "x"}), # too short + ToolUseBlockParam(id="3", name="t", input={"note": "{{already}}"}), + ] + candidates = CacheParameterHandler._collect_candidate_values(trajectory) + assert candidates == ["repeat"] + + def test_no_candidates_skips_llm_call(self) -> None: + """A trajectory of only clicks must not trigger an LLM call.""" + trajectory = [ + ToolUseBlockParam( + id="0", + name="computer_tool", + input={"action": "left_click", "coordinate": [10, 20]}, + ) + ] + goal, out_traj, params = CacheParameterHandler.identify_and_parameterize( + trajectory=trajectory, + goal="click the button", + identification_strategy="llm", + vlm_provider=_RaisingVlmProvider(), # type: ignore[arg-type] + ) + assert params == {} + assert out_traj == trajectory + assert goal == "click the button" + + +class TestHallucinatedValueRejection: + def test_value_not_in_candidates_is_dropped(self) -> None: + # The model returns a value that was never a candidate -> reject it. + response = ( + '{"parameters": [{"name": "made_up", "value": "not-a-candidate", ' + '"description": "hallucinated"}]}' + ) + _, trajectory, params = _parameterize(response, value="admin") + assert params == {} + assert trajectory[0].input == {"text": "admin"} diff --git a/tests/unit/utils/caching/test_cache_reporting.py b/tests/unit/utils/caching/test_cache_reporting.py new file mode 100644 index 00000000..ef2a0485 --- /dev/null +++ b/tests/unit/utils/caching/test_cache_reporting.py @@ -0,0 +1,106 @@ +"""Tests for caching observability (logs + reporter events).""" + +import logging +import tempfile +from pathlib import Path +from typing import Any + +from askui.speaker.cache_executor import CacheExecutor, ExecutionResult +from askui.tools.caching_tools import VerifyCacheExecution +from askui.utils.caching.cache_manager import CacheManager +from askui.utils.caching.reporting_utils import ( + CACHE_REPORTER_SOURCE, + report_cache_event, +) + +logger = logging.getLogger(__name__) + + +class _CapturingReporter: + """Minimal reporter capturing (role, content) pairs.""" + + def __init__(self) -> None: + self.messages: list[tuple[str, Any]] = [] + + def add_message(self, role: str, content: Any, _image: Any = None) -> None: + self.messages.append((role, content)) + + +def test_report_cache_event_logs_and_reports(caplog: Any) -> None: + reporter = _CapturingReporter() + with caplog.at_level(logging.INFO): + report_cache_event(reporter, "something happened", log=logger) # type: ignore[arg-type] + assert ("Cache", "something happened") in [(r, c) for r, c in reporter.messages] + assert any("something happened" in rec.message for rec in caplog.records) + assert reporter.messages[0][0] == CACHE_REPORTER_SOURCE + + +def test_report_cache_event_without_reporter_only_logs(caplog: Any) -> None: + with caplog.at_level(logging.WARNING): + report_cache_event(None, "no reporter here", log=logger, level=logging.WARNING) + assert any("no reporter here" in rec.message for rec in caplog.records) + + +def test_report_cache_event_detail_headline_warning_reason_info(caplog: Any) -> None: + reporter = _CapturingReporter() + with caplog.at_level(logging.INFO): + report_cache_event( + reporter, # type: ignore[arg-type] + "Cache invalidated and will not be reused.", + log=logger, + level=logging.WARNING, + detail="Invalidation reason: something very long", + ) + # Headline is a WARNING; the full reason is an INFO record. + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + infos = [r for r in caplog.records if r.levelno == logging.INFO] + assert any("Cache invalidated" in r.message for r in warnings) + assert any("something very long" in r.message for r in infos) + # The reporter receives the combined message. + assert "something very long" in reporter.messages[0][1] + assert "Cache invalidated" in reporter.messages[0][1] + + +def _write_valid_cache(path: Path) -> None: + path.write_text( + '{"metadata": {"version": "0.3", "created_at": "2025-01-01T00:00:00Z", ' + '"is_valid": true, "execution_attempts": 0, "failures": []}, ' + '"trajectory": [], "cache_parameters": {}}', + encoding="utf-8", + ) + + +def test_verify_failure_reports_reason_and_cache_name() -> None: + with tempfile.TemporaryDirectory() as temp_dir: + cache_path = Path(temp_dir) / "login.json" + _write_valid_cache(cache_path) + + executor = CacheExecutor() + executor._cache_file = CacheManager.read_cache_file(cache_path) + executor._cache_file_path = str(cache_path) + + reporter = _CapturingReporter() + tool = VerifyCacheExecution( + cache_executor=executor, + cache_manager=CacheManager(), + reporter=reporter, # type: ignore[arg-type] + ) + tool(success=False, verification_notes="button was missing") + + reported = " ".join(str(c) for _, c in reporter.messages) + assert "login.json" in reported + assert "button was missing" in reported + assert "FAILED" in reported + + +def test_replay_failure_message_handles_missing_error_message() -> None: + executor = CacheExecutor() + reporter = _CapturingReporter() + executor._reporter = reporter # type: ignore[assignment] + + result = ExecutionResult(status="FAILED", step_index=2, error_message=None) + executor._handle_failed(CacheManager(), result) + + reported = " ".join(str(c) for _, c in reporter.messages) + assert "unknown error" in reported + assert "None" not in reported