Validate positive/non-negative integer args in workspace tools

- Add positive_int_arg and non_negative_int_arg helpers that raise
  ValueError on invalid input
- Apply validation to limit, timeout_seconds, and max_bytes across
  read_file, list_files, search_text, and shell tools

Author: johnteee

teaagent/workspace_tools.py

@@ -182,7 +182,7 @@ def object_schema(properties: dict[str, Union[str, dict[str, Any]]], *, required
 
 def read_file(config: WorkspaceToolConfig, args: dict[str, Any]) -> dict[str, Any]:
     path = resolve_workspace_path(config, args["path"])
-    max_bytes = args.get("max_bytes", config.max_read_bytes)
+    max_bytes = non_negative_int_arg(args, "max_bytes", default=config.max_read_bytes)
     data = path.read_bytes()
     truncated = len(data) > max_bytes
     return {
@@ -241,7 +241,7 @@ def edit_at_hash(config: WorkspaceToolConfig, args: dict[str, Any]) -> dict[str,
 
 def list_files(config: WorkspaceToolConfig, args: dict[str, Any]) -> dict[str, Any]:
     pattern = args["pattern"]
-    limit = args.get("limit", 200)
+    limit = positive_int_arg(args, "limit", default=200)
     files = []
     for path in sorted(config.root.rglob("*")):
         if path.is_file() and ".git" not in path.parts and fnmatch(relative_path(config, path), pattern):
@@ -254,7 +254,7 @@ def list_files(config: WorkspaceToolConfig, args: dict[str, Any]) -> dict[str, A
 def search_text(config: WorkspaceToolConfig, args: dict[str, Any]) -> dict[str, Any]:
     regex = re.compile(args["pattern"])
     include = args.get("include", "*")
-    limit = args.get("limit", 200)
+    limit = positive_int_arg(args, "limit", default=200)
     matches = []
     for path in sorted(config.root.rglob("*")):
         if not path.is_file() or ".git" in path.parts or not fnmatch(relative_path(config, path), include):
@@ -283,7 +283,7 @@ def git_status(config: WorkspaceToolConfig) -> dict[str, Any]:
 
 
 def run_shell(config: WorkspaceToolConfig, args: dict[str, Any]) -> dict[str, Any]:
-    timeout = args.get("timeout_seconds", config.command_timeout_seconds)
+    timeout = positive_int_arg(args, "timeout_seconds", default=config.command_timeout_seconds)
     result = subprocess.run(
         args["command"],
         cwd=str(config.root),
@@ -319,10 +319,24 @@ def run_shell_inspect(config: WorkspaceToolConfig, args: dict[str, Any]) -> dict
     policy = classify_shell_command_policy(args["command"])
     if policy != "inspect":
         raise ValueError("command is not inspect-safe; retry with workspace_run_shell_mutate")
-    timeout = args.get("timeout_seconds", config.command_timeout_seconds)
+    timeout = positive_int_arg(args, "timeout_seconds", default=config.command_timeout_seconds)
     return run_shell_argv(config, shlex.split(args["command"]), timeout_seconds=timeout)
 
 
+def positive_int_arg(args: dict[str, Any], name: str, *, default: int) -> int:
+    value = args.get(name, default)
+    if value < 1:
+        raise ValueError(f"{name} must be >= 1")
+    return value
+
+
+def non_negative_int_arg(args: dict[str, Any], name: str, *, default: int) -> int:
+    value = args.get(name, default)
+    if value < 0:
+        raise ValueError(f"{name} must be >= 0")
+    return value
+
+
 def classify_shell_command_policy(command: str) -> str:
     blocked_tokens = (">", "<", "|", "&&", ";", "`", "$(")
     if any(token in command for token in blocked_tokens):

tests/test_workspace_tools.py

@@ -89,6 +89,15 @@ def test_path_escape_is_blocked(self) -> None:
             with self.assertRaises(ToolExecutionError):
                 registry.execute("workspace_read_file", {"path": "../outside.txt"})
 
+    def test_read_file_rejects_negative_max_bytes(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            (Path(tmp) / "file.txt").write_text("content", encoding="utf-8")
+            registry = build_workspace_tool_registry(tmp)
+
+            with self.assertRaises(ToolExecutionError) as ctx:
+                registry.execute("workspace_read_file", {"path": "file.txt", "max_bytes": -1})
+            self.assertIn("max_bytes", str(ctx.exception))
+
     def test_shell_and_git_status_tools(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             registry = build_workspace_tool_registry(tmp)
@@ -138,6 +147,14 @@ def test_shell_inspect_rejects_unbalanced_quotes(self) -> None:
             with self.assertRaises(ToolExecutionError):
                 registry.execute("workspace_run_shell_inspect", {"command": "cat 'unterminated"})
 
+    def test_shell_inspect_rejects_non_positive_timeout(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            registry = build_workspace_tool_registry(tmp)
+
+            with self.assertRaises(ToolExecutionError) as ctx:
+                registry.execute("workspace_run_shell_inspect", {"command": "pwd", "timeout_seconds": 0})
+            self.assertIn("timeout_seconds", str(ctx.exception))
+
     def test_cli_workspace_tools_outputs_metadata(self) -> None:
         output = io.StringIO()
 
@@ -180,6 +197,14 @@ def test_list_files_truncates_when_limit_exceeded(self) -> None:
             self.assertEqual(len(result["files"]), 2)
             self.assertTrue(result["truncated"])
 
+    def test_list_files_rejects_non_positive_limit(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            registry = build_workspace_tool_registry(tmp)
+
+            with self.assertRaises(ToolExecutionError) as ctx:
+                registry.execute("workspace_list_files", {"pattern": "*.txt", "limit": 0})
+            self.assertIn("limit", str(ctx.exception))
+
     def test_search_text_truncates_when_limit_exceeded(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             for i in range(5):
@@ -190,6 +215,14 @@ def test_search_text_truncates_when_limit_exceeded(self) -> None:
             self.assertEqual(len(result["matches"]), 5)
             self.assertTrue(result["truncated"])
 
+    def test_search_text_rejects_non_positive_limit(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            registry = build_workspace_tool_registry(tmp)
+
+            with self.assertRaises(ToolExecutionError) as ctx:
+                registry.execute("workspace_search_text", {"pattern": "x", "limit": 0})
+            self.assertIn("limit", str(ctx.exception))
+
 
 if __name__ == "__main__":
     unittest.main()