From 0ec61be07f9b9c99f5d43549e092e754612a57c6 Mon Sep 17 00:00:00 2001 From: philipph-askui Date: Wed, 26 Aug 2026 22:58:32 -0400 Subject: [PATCH] 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