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