Add optional persist parameter to workspace and retries configuration tools

Author: LukeA12

MaxKernel/hitl_agent/prompts/interactive_prompt.py

@@ -42,9 +42,11 @@
 
     * **If the request is to CHANGE/SET the workspace or working directory**:
         * **Action**: Call `set_working_directory` tool with the absolute path.
+        * **Note**: Set `persist=True` ONLY if the user explicitly asks to persist, save, or make the change permanent. Otherwise, leave it as `False` (default).
 
     * **If the request is to CHANGE/SET the maximum compilation retries**:
         * **Action**: Call `set_max_compilation_retries` tool with the integer value.
+        * **Note**: Set `persist=True` ONLY if the user explicitly asks to persist, save, or make the change permanent. Otherwise, leave it as `False` (default).
 
     * **If the request is simple explanation** (like "What is a TPU?"):
         * **Action**: Delegate to `ExplanationAgent`.

MaxKernel/hitl_agent/tools/retries_tool.py

@@ -5,15 +5,18 @@
 from google.adk.tools import FunctionTool, ToolContext
 
 
-async def set_max_compilation_retries_fn(retries: int, tool_context: ToolContext) -> str:
+async def set_max_compilation_retries_fn(
+    retries: int, tool_context: ToolContext, persist: bool = False
+) -> str:
   """Set the maximum number of compilation validation and auto-fixing attempts.
 
   Args:
       retries: The maximum number of attempts (must be positive).
+      persist: Whether to permanently save this change to the configuration file (default is False).
 
   Returns:
       A string containing the operation result description:
-      - If successful: "Successfully updated maximum compilation retries to: <retries>"
+      - If successful: "Successfully updated maximum compilation retries to: <retries> (persisted: <persist>)"
       - If failed: "Error: The number of retries must be a positive integer."
   """
   if retries <= 0:
@@ -24,24 +27,25 @@ async def set_max_compilation_retries_fn(retries: int, tool_context: ToolContext
   hitl_cfg.MAX_COMPILATION_RETRIES = retries
   os.environ["MAX_COMPILATION_RETRIES"] = str(retries)
 
-  # Update .env file
-  try:
-    env_path = ".env"
-    if os.path.exists(env_path):
-      with open(env_path, "r") as f:
-        lines = f.readlines()
-      with open(env_path, "w") as f:
-        updated = False
-        for line in lines:
-          if line.strip().startswith("MAX_COMPILATION_RETRIES="):
+  # Update .env file if persist is requested
+  if persist:
+    try:
+      env_path = ".env"
+      if os.path.exists(env_path):
+        with open(env_path, "r") as f:
+          lines = f.readlines()
+        with open(env_path, "w") as f:
+          updated = False
+          for line in lines:
+            if line.strip().startswith("MAX_COMPILATION_RETRIES="):
+              f.write(f"MAX_COMPILATION_RETRIES={retries}\n")
+              updated = True
+            else:
+              f.write(line)
+          if not updated:
             f.write(f"MAX_COMPILATION_RETRIES={retries}\n")
-            updated = True
-          else:
-            f.write(line)
-        if not updated:
-          f.write(f"MAX_COMPILATION_RETRIES={retries}\n")
-  except Exception as e:
-    logging.error(f"Failed to update .env: {e}")
+    except Exception as e:
+      logging.error(f"Failed to update .env: {e}")
 
   # Update the active validation loops in-memory
   try:
@@ -50,7 +54,7 @@ async def set_max_compilation_retries_fn(retries: int, tool_context: ToolContext
   except Exception as e:
     logging.warning(f"Could not update hitl_loop max_retries: {e}")
 
-  return f"Successfully updated maximum compilation retries to: {retries}"
+  return f"Successfully updated maximum compilation retries to: {retries} (persisted: {persist})"
 
 
 set_max_compilation_retries = FunctionTool(set_max_compilation_retries_fn)

MaxKernel/hitl_agent/tools/workspace_tool.py

@@ -9,18 +9,21 @@
 from hitl_agent.tools.filesystem_tools import filesystem_tool_r, filesystem_tool_rw
 
 
-async def set_working_directory_fn(path: str, tool_context: ToolContext) -> str:
+async def set_working_directory_fn(
+    path: str, tool_context: ToolContext, persist: bool = False
+) -> str:
   """Set the active workspace/working directory path for the session.
 
   All filesystem tools (list_directory, read_file, write_file) will use
   the new path.
 
   Args:
       path: The absolute path of the directory.
+      persist: Whether to permanently save this change to the configuration file (default is False).
 
   Returns:
       A string containing the operation result description:
-      - If successful: "Successfully switched working directory to: <abs_path>"
+      - If successful: "Successfully switched working directory to: <abs_path> (persisted: <persist>)"
       - If failed: "Error: Path '<path>' does not exist." or "Error: Path '<path>' is not a directory."
   """
   if not os.path.exists(path):
@@ -34,24 +37,25 @@ async def set_working_directory_fn(path: str, tool_context: ToolContext) -> str:
   config.WORKDIR = abs_path
   os.environ["WORKDIR"] = abs_path
 
-  # Update .env file
-  try:
-    env_path = ".env"
-    if os.path.exists(env_path):
-      with open(env_path, "r") as f:
-        lines = f.readlines()
-      with open(env_path, "w") as f:
-        updated = False
-        for line in lines:
-          if line.strip().startswith("WORKDIR="):
+  # Update .env file if persist is requested
+  if persist:
+    try:
+      env_path = ".env"
+      if os.path.exists(env_path):
+        with open(env_path, "r") as f:
+          lines = f.readlines()
+        with open(env_path, "w") as f:
+          updated = False
+          for line in lines:
+            if line.strip().startswith("WORKDIR="):
+              f.write(f'WORKDIR="{abs_path}"\n')
+              updated = True
+            else:
+              f.write(line)
+          if not updated:
             f.write(f'WORKDIR="{abs_path}"\n')
-            updated = True
-          else:
-            f.write(line)
-        if not updated:
-          f.write(f'WORKDIR="{abs_path}"\n')
-  except Exception as e:
-    logging.error(f"Failed to update .env: {e}")
+    except Exception as e:
+      logging.error(f"Failed to update .env: {e}")
 
   # Update active filesystem tools
   # Update read-only filesystem tool
@@ -80,7 +84,7 @@ async def set_working_directory_fn(path: str, tool_context: ToolContext) -> str:
         sampling_capabilities=filesystem_tool_rw._sampling_capabilities,
     )
 
-  return f"Successfully switched working directory to: {abs_path}"
+  return f"Successfully switched working directory to: {abs_path} (persisted: {persist})"
 
 
 set_working_directory = FunctionTool(set_working_directory_fn)