tools: 强制 spark_submit 的 script_path 必须来自 upload_file

在 spark_submit handler 中增加 server-minted 校验：script_path 必须通过
本服务器 upload_file 的 UploadStore.Validate，否则立即返回清晰错误，避免
远程 agent 把自身本地路径转发给服务器侧 spark-submit 时出现难以定位的
FileNotFoundException。

校验对 nil UploadStore 是安全的：当 d.UploadStore 为 nil 时直接跳过，保留
不强制依赖 store 的测试或降级场景。当前 testDepsWithDataDir 始终会注入
UploadStore，所以测试走的是真实校验路径。

同步更新了 script_path 的工具描述，明确声明非 upload_file 铸造的路径会被
拒绝。这是对 7920dc9 中“path 永远在最后”规则的收紧——现在不仅位置固定，
而且必须是本机 uploads 目录下的有效上传文件。

测试调整：
- 新增 TestSparkSubmit_RejectsNonMintedPath：使用 t.TempDir() 下未通过
  upload_file 写入的文件作为 script_path，断言返回错误并包含
  "not from upload_file"。
- TestSparkSubmit_MissingRequiredField / TestSparkSubmit_BadSparkConfValue
  原来使用 "/tmp/script.py" 作为占位路径，现在会先通过 UploadStore.Save
  生成一个铸造路径，确保它们继续分别验证 master 缺失和 spark_conf 值类型
  错误，而不是被新的路径守卫拦截。

Co-Authored-By: Claude <noreply@anthropic.com>

internal/mcp/tools/spark_submit.go

@@ -32,7 +32,7 @@ func NewSparkSubmitTool() mcp.Tool {
 		),
 		mcp.WithString("script_path",
 			mcp.Required(),
-			mcp.Description("Absolute path to the script. Pass the path field returned by upload_file — do not construct your own path. The script is always the last argv element."),
+			mcp.Description("Absolute path to the script. Pass the `path` field returned by upload_file — do not construct your own path. The script is always the last argv element. Paths not minted by upload_file on this server are rejected."),
 		),
 		mcp.WithString("queue",
 			mcp.Required(),
@@ -79,6 +79,16 @@ func (d *Deps) SparkSubmitHandler(ctx context.Context, req mcp.CallToolRequest)
 	if err != nil {
 		return errResult("spark_submit: " + err.Error()), nil
 	}
+
+	// Reject paths that didn't come through upload_file on this server.
+	// Cluster-local paths and arbitrary host paths are unreachable from the MCP
+	// server's spark-submit, so we fail fast with a clear message rather than
+	// letting spark-submit produce a confusing FileNotFoundException later.
+	if d.UploadStore != nil {
+		if _, err := d.UploadStore.Validate(scriptPath); err != nil {
+			return errResult(fmt.Errorf("spark_submit: script_path %q is not from upload_file on this server; upload the file first via upload_file and pass back the path field: %w", scriptPath, err).Error()), nil
+		}
+	}
 	queue, err := req.RequireString("queue")
 	if err != nil {
 		return errResult("spark_submit: " + err.Error()), nil

internal/mcp/tools/spark_submit_path_test.go

@@ -97,11 +97,16 @@ func TestSparkSubmit_StructuredCommand(t *testing.T) {
 
 func TestSparkSubmit_MissingRequiredField(t *testing.T) {
 	deps, _ := testDepsWithDataDir(t)
+	store := deps.UploadStore
+	_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
+	if err != nil {
+		t.Fatalf("save upload: %v", err)
+	}
 
 	req := newToolRequest(SparkSubmitName, map[string]any{
 		"cluster_id":      "cluster-echo",
 		"deploy_mode":     "cluster",
-		"script_path":     "/tmp/script.py",
+		"script_path":     mintedPath,
 		"queue":           "default",
 		"executor_memory": "2G",
 		"executor_cores":  1,
@@ -125,12 +130,17 @@ func TestSparkSubmit_MissingRequiredField(t *testing.T) {
 
 func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
 	deps, _ := testDepsWithDataDir(t)
+	store := deps.UploadStore
+	_, _, _, _, mintedPath, err := store.Save([]byte("# dummy\n"), "dummy.py")
+	if err != nil {
+		t.Fatalf("save upload: %v", err)
+	}
 
 	req := newToolRequest(SparkSubmitName, map[string]any{
 		"cluster_id":      "cluster-echo",
 		"master":          "yarn",
 		"deploy_mode":     "cluster",
-		"script_path":     "/tmp/script.py",
+		"script_path":     mintedPath,
 		"queue":           "default",
 		"executor_memory": "2G",
 		"executor_cores":  1,
@@ -152,3 +162,37 @@ func TestSparkSubmit_BadSparkConfValue(t *testing.T) {
 		t.Errorf("error text=%q, want mention of spark_conf", text.Text)
 	}
 }
+
+func TestSparkSubmit_RejectsNonMintedPath(t *testing.T) {
+	deps, _ := testDepsWithDataDir(t)
+
+	unminted := filepath.Join(t.TempDir(), "unminted.py")
+	if err := os.WriteFile(unminted, []byte("print('not from upload_file')\n"), 0o644); err != nil {
+		t.Fatalf("write unminted file: %v", err)
+	}
+
+	req := newToolRequest(SparkSubmitName, map[string]any{
+		"cluster_id":      "cluster-echo",
+		"master":          "yarn",
+		"deploy_mode":     "cluster",
+		"script_path":     unminted,
+		"queue":           "default",
+		"executor_memory": "2G",
+		"executor_cores":  1,
+		"num_executors":   4,
+	})
+	res, err := deps.SparkSubmitHandler(context.Background(), req)
+	if err != nil {
+		t.Fatalf("handler error: %v", err)
+	}
+	if !res.IsError {
+		t.Fatalf("expected error result, got: %v", res.Content)
+	}
+	text, ok := mcp.AsTextContent(res.Content[0])
+	if !ok {
+		t.Fatalf("content is not text: %T", res.Content[0])
+	}
+	if !strings.Contains(text.Text, "not from upload_file") {
+		t.Errorf("error text=%q, want mention of not from upload_file", text.Text)
+	}
+}