feat(task): auto-inject scheduling params into script template via scheduling_map

Add scheduling_map field to ParameterSchema so Application creators can declare that a parameter (e.g. NP) maps to a scheduling field (e.g. cpus). The backend auto-injects the scheduling value into script template variables before rendering, eliminating duplicate user input. The frontend hides mapped parameters from the form and injects their values on submit.
fix(task): prevent RecoverStuckTasks from re-enqueueing in-flight tasks
2026-04-22 10:26:52 +08:00 · 2026-04-21 17:19:10 +08:00 · 2026-04-21 13:35:21 +08:00 · 2026-04-21 13:23:40 +08:00 · 2026-04-21 11:31:07 +08:00 · 2026-04-21 10:57:38 +08:00
29 changed files with 2274 additions and 1317 deletions
@@ -3,7 +3,6 @@ package main
 import (
 	"bytes"
 	"encoding/json"
-	"fmt"
 	"net/http"
 	"testing"

@@ -66,6 +65,8 @@ func jobSubmitViaAPI(t *testing.T, env *testenv.TestEnv, script string) int32 {
 	return job.JobID
 }

+// [已弃用] 以下测试依赖 POST /api/v1/jobs/submit，该接口已被 POST /tasks 取代。
+/*
 // TestIntegration_Jobs_Submit verifies POST /api/v1/jobs/submit creates a new job.
 func TestIntegration_Jobs_Submit(t *testing.T) {
 	env := testenv.NewTestEnv(t)
@@ -220,3 +221,4 @@ func TestIntegration_Jobs_History(t *testing.T) {
 		t.Fatalf("cancelled job %d not found in history", jobID)
 	}
 }
+*/
@@ -53,7 +53,7 @@ func TestRouterRegistration(t *testing.T) {
 		method string
 		path   string
 	}{
-		{"POST", "/api/v1/jobs/submit"},
+		// {"POST", "/api/v1/jobs/submit"}, // [已弃用] 已被 POST /tasks 取代
 		{"GET", "/api/v1/jobs"},
 		{"GET", "/api/v1/jobs/history"},
 		{"GET", "/api/v1/jobs/:id"},
@@ -42,80 +42,6 @@
    }
  ],
  "paths": {
-    "/jobs/submit": {
-      "post": {
-        "tags": ["Jobs"],
-        "summary": "Submit a new job",
-        "description": "Submits a Slurm job with the specified script and optional parameters.",
-        "operationId": "submitJob",
-        "requestBody": {
-          "required": true,
-          "content": {
-            "application/json": {
-              "schema": {
-                "$ref": "#/components/schemas/SubmitJobRequest"
-              }
-            }
-          }
-        },
-        "responses": {
-          "201": {
-            "description": "Job submitted successfully",
-            "content": {
-              "application/json": {
-                "schema": {
-                  "allOf": [
-                    { "$ref": "#/components/schemas/ApiResponseSuccess" },
-                    {
-                      "type": "object",
-                      "properties": {
-                        "data": {
-                          "$ref": "#/components/schemas/JobResponse"
-                        }
-                      }
-                    }
-                  ]
-                }
-              }
-            }
-          },
-          "400": {
-            "description": "Invalid request body or missing required fields",
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ApiResponseError"
-                },
-                "examples": {
-                  "invalid_body": {
-                    "value": {
-                      "success": false,
-                      "error": "invalid request body"
-                    }
-                  },
-                  "missing_script": {
-                    "value": {
-                      "success": false,
-                      "error": "script is required"
-                    }
-                  }
-                }
-              }
-            }
-          },
-          "502": {
-            "description": "Slurm backend error",
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ApiResponseError"
-                }
-              }
-            }
-          }
-        }
-      }
-    },
    "/jobs": {
      "get": {
        "tags": ["Jobs"],
@@ -2034,52 +1960,7 @@
          }
        }
      },
-      "SubmitJobRequest": {
-        "type": "object",
-        "required": ["script"],
-        "properties": {
-          "script": {
-            "type": "string",
-            "description": "Job script content"
-          },
-          "partition": {
-            "type": "string",
-            "description": "Target partition"
-          },
-          "qos": {
-            "type": "string",
-            "description": "Quality of Service"
-          },
-          "cpus": {
-            "type": "integer",
-            "format": "int32",
-            "description": "Number of CPUs required"
-          },
-          "memory": {
-            "type": "string",
-            "description": "Memory requirement (e.g. \"4G\")"
-          },
-          "time_limit": {
-            "type": "string",
-            "description": "Time limit (e.g. \"1:00:00\")"
-          },
-          "job_name": {
-            "type": "string",
-            "description": "Job name"
-          },
-          "environment": {
-            "type": "object",
-            "additionalProperties": {
-              "type": "string"
-            },
-            "description": "Environment variables"
-          },
-          "work_dir": {
-            "type": "string",
-            "description": "Job working directory"
-          }
-        }
-      },
+
      "JobResponse": {
        "type": "object",
        "properties": {
@@ -8,12 +8,15 @@ import (
 	"go.uber.org/zap"
 )

-// TaskPollable defines the interface for refreshing stale task statuses.
+// TaskPollable defines the interface for refreshing stale task statuses
+// and recovering stuck tasks.
 type TaskPollable interface {
 	RefreshStaleTasks(ctx context.Context) error
+	RecoverStuckTasks(ctx context.Context)
 }

-// TaskPoller periodically polls Slurm for task status updates via TaskPollable.
+// TaskPoller periodically polls Slurm for task status updates and recovers
+// stuck tasks via TaskPollable.
 type TaskPoller struct {
 	taskSvc  TaskPollable
 	interval time.Duration
@@ -31,9 +34,11 @@ func NewTaskPoller(taskSvc TaskPollable, interval time.Duration, logger *zap.Log
 	}
 }

-// Start launches the background goroutine that periodically refreshes stale tasks.
+// Start launches background goroutines that periodically refresh stale tasks
+// and recover stuck tasks.
 func (p *TaskPoller) Start(ctx context.Context) {
 	ctx, p.cancel = context.WithCancel(ctx)
+
 	p.wg.Add(1)
 	go func() {
 		defer p.wg.Done()
@@ -50,9 +55,24 @@ func (p *TaskPoller) Start(ctx context.Context) {
 			}
 		}
 	}()
+
+	p.wg.Add(1)
+	go func() {
+		defer p.wg.Done()
+		ticker := time.NewTicker(5 * time.Minute)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				p.taskSvc.RecoverStuckTasks(ctx)
+			}
+		}
+	}()
 }

-// Stop cancels the background goroutine and waits for it to finish.
+// Stop cancels the background goroutines and waits for them to finish.
 func (p *TaskPoller) Stop() {
 	if p.cancel != nil {
 		p.cancel()
@@ -25,6 +25,8 @@ func (m *mockTaskPollable) RefreshStaleTasks(ctx context.Context) error {
 	return nil
 }

+func (m *mockTaskPollable) RecoverStuckTasks(ctx context.Context) {}
+
 func (m *mockTaskPollable) getCallCount() int {
 	m.mu.Lock()
 	defer m.mu.Unlock()
@@ -22,29 +22,31 @@ func NewJobHandler(jobSvc *service.JobService, logger *zap.Logger) *JobHandler {
 	return &JobHandler{jobSvc: jobSvc, logger: logger}
 }

+// [已弃用] SubmitJob 已被 POST /tasks 取代。
+// 保留方法体以防需要回滚。
 // SubmitJob handles POST /api/v1/jobs/submit.
-func (h *JobHandler) SubmitJob(c *gin.Context) {
-	var req model.SubmitJobRequest
-	if err := c.ShouldBindJSON(&req); err != nil {
-		h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "invalid request body"))
-		server.BadRequest(c, "invalid request body")
-		return
-	}
-	if req.Script == "" {
-		h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "script is required"))
-		server.BadRequest(c, "script is required")
-		return
-	}
-
-	resp, err := h.jobSvc.SubmitJob(c.Request.Context(), &req)
-	if err != nil {
-		h.logger.Error("handler error", zap.String("method", "SubmitJob"), zap.Int("status", http.StatusBadGateway), zap.Error(err))
-		server.ErrorWithStatus(c, http.StatusBadGateway, "slurm error: "+err.Error())
-		return
-	}
-
-	server.Created(c, resp)
-}
+// func (h *JobHandler) SubmitJob(c *gin.Context) {
+// 	var req model.SubmitJobRequest
+// 	if err := c.ShouldBindJSON(&req); err != nil {
+// 		h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "invalid request body"))
+// 		server.BadRequest(c, "invalid request body")
+// 		return
+// 	}
+// 	if req.Script == "" {
+// 		h.logger.Warn("bad request", zap.String("method", "SubmitJob"), zap.String("error", "script is required"))
+// 		server.BadRequest(c, "script is required")
+// 		return
+// 	}
+//
+// 	resp, err := h.jobSvc.SubmitJob(c.Request.Context(), &req)
+// 	if err != nil {
+// 		h.logger.Error("handler error", zap.String("method", "SubmitJob"), zap.Int("status", http.StatusBadGateway), zap.Error(err))
+// 		server.ErrorWithStatus(c, http.StatusBadGateway, "slurm error: "+err.Error())
+// 		return
+// 	}
+//
+// 	server.Created(c, resp)
+// }

 // GetJobs handles GET /api/v1/jobs with pagination.
 func (h *JobHandler) GetJobs(c *gin.Context) {
@@ -1,9 +1,7 @@
 package handler

 import (
-	"bytes"
 	"encoding/json"
-	"fmt"
 	"net/http"
 	"net/http/httptest"
 	"testing"
@@ -23,7 +21,7 @@ func setupJobRouter(h *JobHandler) *gin.Engine {
 	v1 := r.Group("/api/v1")
 	jobs := v1.Group("/jobs")
 	{
-		jobs.POST("/submit", h.SubmitJob)
+		// jobs.POST("/submit", h.SubmitJob) // [已弃用] 已被 POST /tasks 取代
 		jobs.GET("", h.GetJobs)
 		jobs.GET("/history", h.GetJobHistory)
 		jobs.GET("/:id", h.GetJob)
@@ -61,6 +59,8 @@ func handlerLogs(logs *observer.ObservedLogs) []observer.LoggedEntry {
 	return handler
 }

+// [已弃用] SubmitJob 相关测试已被禁用，该接口已被 POST /tasks 取代。
+/*
 func TestSubmitJob_Success(t *testing.T) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/slurm/v0.0.40/job/submit", func(w http.ResponseWriter, r *http.Request) {
@@ -171,6 +171,9 @@ func TestSubmitJob_SlurmError(t *testing.T) {
 		t.Fatalf("expected 502, got %d: %s", w.Code, w.Body.String())
 	}
 }
+*/
+
+// --- Logging verification tests ---

 func TestGetJobs_Success(t *testing.T) {
 	mux := http.NewServeMux()
@@ -462,6 +465,7 @@ func TestGetJobHistory_DefaultPagination(t *testing.T) {
 	}
 }

+/*
 func TestSubmitJob_InvalidBody(t *testing.T) {
 	mux := http.NewServeMux()
 	srv, handler := setupJobHandler(mux)
@@ -479,9 +483,11 @@ func TestSubmitJob_InvalidBody(t *testing.T) {
 		t.Fatalf("expected 400, got %d: %s", w.Code, w.Body.String())
 	}
 }
+*/

 // --- Logging verification tests ---

+/*
 func TestSubmitJob_InvalidBody_LogsWarn(t *testing.T) {
 	mux := http.NewServeMux()
 	srv, handler, logs := setupJobHandlerWithObserver(mux)
@@ -614,6 +620,7 @@ func TestSubmitJob_Success_NoHandlerLogs(t *testing.T) {
 		t.Errorf("expected no handler log entries on success, got %d", len(hLogs))
 	}
 }
+*/

 func TestGetJobs_Error_LogsError(t *testing.T) {
 	mux := http.NewServeMux()
@@ -24,6 +24,7 @@ type Application struct {
 	Category       string          `gorm:"size:255" json:"category,omitempty"`              // 分类
 	ScriptTemplate string          `gorm:"type:text;not null" json:"script_template"`       // 脚本模板
 	Parameters     json.RawMessage `gorm:"type:json" json:"parameters,omitempty"`           // 参数表单JSON
+	Environment    json.RawMessage `gorm:"type:text" json:"environment,omitempty"`          // 环境变量JSON
 	Scope          string          `gorm:"size:50;default:'system'" json:"scope,omitempty"` // 作用域(system/user)
 	CreatedBy      int64           `json:"created_by,omitempty"`                            // 创建者ID
 	CreatedAt      time.Time       `json:"created_at"`                                      // 创建时间
@@ -43,6 +44,7 @@ type ParameterSchema struct {
 	Default     string   `json:"default,omitempty"`     // 默认值
 	Options     []string `json:"options,omitempty"`     // 枚举选项列表
 	Description   string   `json:"description,omitempty"` // 参数说明
+	SchedulingMap string   `json:"scheduling_map,omitempty"` // maps to a scheduling param
 }

 // CreateApplicationRequest 是创建应用的 API 请求。
@@ -53,6 +55,7 @@ type CreateApplicationRequest struct {
 	Category       string          `json:"category,omitempty"`                 // 分类
 	ScriptTemplate string          `json:"script_template" binding:"required"` // 脚本模板（必填）
 	Parameters     json.RawMessage `json:"parameters,omitempty"`               // 参数表单JSON
+	Environment    map[string]string `json:"environment,omitempty"`            // 环境变量
 	Scope          string          `json:"scope,omitempty"`                    // 作用域
 }

@@ -64,6 +67,7 @@ type UpdateApplicationRequest struct {
 	Category       *string          `json:"category,omitempty"`        // 分类
 	ScriptTemplate *string          `json:"script_template,omitempty"` // 脚本模板
 	Parameters     *json.RawMessage `json:"parameters,omitempty"`      // 参数表单JSON
+	Environment    *json.RawMessage `json:"environment,omitempty"`     // 环境变量
 	Scope          *string          `json:"scope,omitempty"`           // 作用域
 }

@@ -57,15 +57,27 @@ type JobResponse struct {

 	// Resources
 	Cpus          *int32 `json:"cpus,omitempty"`            // 分配/请求的 CPU 核数
+	CpusPerTask   *int32 `json:"cpus_per_task,omitempty"`   // 每任务 CPU 核数
 	Tasks         *int32 `json:"tasks,omitempty"`           // 任务数
 	NodeCount     *int32 `json:"node_count,omitempty"`      // 节点数
 	Nodes         string `json:"nodes,omitempty"`           // 分配的节点列表
 	BatchHost     string `json:"batch_host,omitempty"`      // 批处理主节点
+	MemoryPerCpu  *int64 `json:"memory_per_cpu,omitempty"`  // 每 CPU 内存 (MB)
+	MemoryPerNode *int64 `json:"memory_per_node,omitempty"` // 每节点内存 (MB)
+	MemoryUsed    *int64 `json:"memory_used,omitempty"`     // 实际峰值内存消耗 (MB，仅历史作业)
+
+	// TRES (Trackable Resources)
+	TresReqStr    string   `json:"tres_req_str,omitempty"`   // 请求的 TRES 字符串 (活跃作业)
+	TresAllocStr  string   `json:"tres_alloc_str,omitempty"` // 分配的 TRES 字符串 (活跃作业)
+	TresRequested string   `json:"tres_requested,omitempty"` // 请求的 TRES (历史作业)
+	TresAllocated string   `json:"tres_allocated,omitempty"` // 分配的 TRES (历史作业，最接近实际消耗)
+	GresDetail    []string `json:"gres_detail,omitempty"`    // GPU/GRES 详情

 	// Timing (Unix timestamp)
 	SubmitTime *int64 `json:"submit_time,omitempty"` // 提交时间
 	StartTime  *int64 `json:"start_time,omitempty"`  // 开始运行时间
 	EndTime    *int64 `json:"end_time,omitempty"`    // 结束/预计结束时间
+	Elapsed    *int32 `json:"elapsed,omitempty"`     // 运行时长 (秒，仅历史作业)

 	// Result
 	ExitCode *int32 `json:"exit_code,omitempty"` // 退出码 (nil 表示未结束)
@@ -10,7 +10,7 @@ import (
 )

 type JobHandler interface {
-	SubmitJob(c *gin.Context)
+	// SubmitJob(c *gin.Context) // [已弃用] 已被 POST /tasks 取代
 	GetJobs(c *gin.Context)
 	GetJobHistory(c *gin.Context)
 	GetJob(c *gin.Context)
@@ -73,7 +73,7 @@ func NewRouter(jobH JobHandler, clusterH ClusterHandler, appH ApplicationHandler
 	v1 := r.Group("/api/v1")

 	jobs := v1.Group("/jobs")
-	jobs.POST("/submit", jobH.SubmitJob)
+	// jobs.POST("/submit", jobH.SubmitJob) // [已弃用] 已被 POST /tasks 取代
 	jobs.GET("", jobH.GetJobs)
 	jobs.GET("/history", jobH.GetJobHistory)
 	jobs.GET("/:id", jobH.GetJob)
@@ -144,7 +144,7 @@ func NewTestRouter() *gin.Engine {

 func registerPlaceholderRoutes(v1 *gin.RouterGroup) {
 	jobs := v1.Group("/jobs")
-	jobs.POST("/submit", notImplemented)
+	// jobs.POST("/submit", notImplemented) // [已弃用] 已被 POST /tasks 取代
 	jobs.GET("", notImplemented)
 	jobs.GET("/history", notImplemented)
 	jobs.GET("/:id", notImplemented)
@@ -17,7 +17,7 @@ func TestAllRoutesRegistered(t *testing.T) {
 		method string
 		path   string
 	}{
-		{"POST", "/api/v1/jobs/submit"},
+		// {"POST", "/api/v1/jobs/submit"}, // [已弃用] 已被 POST /tasks 取代
 		{"GET", "/api/v1/jobs"},
 		{"GET", "/api/v1/jobs/history"},
 		{"GET", "/api/v1/jobs/:id"},
@@ -42,6 +42,13 @@ func derefInt32ToStr(i *int32) string {
 	return strconv.FormatInt(int64(*i), 10)
 }

+func derefInt64ToStr(i *int64) string {
+	if i == nil {
+		return ""
+	}
+	return strconv.FormatInt(*i, 10)
+}
+
 func uint32NoValString(v *slurm.Uint32NoVal) string {
 	if v == nil {
 		return ""
@@ -444,6 +444,26 @@ func TestClusterService_GetPartition_ErrorLogging(t *testing.T) {
 	}
 }

+func TestDerefInt64ToStr(t *testing.T) {
+	t.Run("nil returns empty", func(t *testing.T) {
+		if got := derefInt64ToStr(nil); got != "" {
+			t.Errorf("derefInt64ToStr(nil) = %q, want empty", got)
+		}
+	})
+	t.Run("non-nil returns string", func(t *testing.T) {
+		v := int64(4096)
+		if got := derefInt64ToStr(&v); got != "4096" {
+			t.Errorf("derefInt64ToStr(4096) = %q, want %q", got, "4096")
+		}
+	})
+	t.Run("zero value", func(t *testing.T) {
+		v := int64(0)
+		if got := derefInt64ToStr(&v); got != "0" {
+			t.Errorf("derefInt64ToStr(0) = %q, want %q", got, "0")
+		}
+	})
+}
+
 func TestClusterService_GetDiag_ErrorLogging(t *testing.T) {
 	srv := errorServer()
 	defer srv.Close()
@@ -37,7 +37,7 @@ func (s *JobService) SubmitJob(ctx context.Context, req *model.SubmitJobRequest)
 		jobDesc.CurrentWorkingDirectory = &req.WorkDir
 	}
 	if req.CPUs > 0 {
-		jobDesc.MinimumCpus = slurm.Ptr(req.CPUs)
+		jobDesc.CpusPerTask = slurm.Ptr(req.CPUs)
 	}
 	if req.TimeLimit != "" {
 		if mins, err := strconv.ParseInt(req.TimeLimit, 10, 64); err == nil {
@@ -49,12 +49,15 @@ func (s *JobService) SubmitJob(ctx context.Context, req *model.SubmitJobRequest)
 		"PATH=/usr/local/bin:/usr/bin:/bin",
 		"HOME=/root",
 	}
+	for k, v := range req.Environment {
+		jobDesc.Environment = append(jobDesc.Environment, k+"="+v)
+	}

 	if req.MemoryPerNode != nil {
-		jobDesc.MemoryPerNode = &slurm.Uint64NoVal{Number: req.MemoryPerNode}
+		jobDesc.MemoryPerNode = &slurm.Uint64NoVal{Set: slurm.Ptr(true), Number: req.MemoryPerNode}
 	}
 	if req.MemoryPerCpu != nil {
-		jobDesc.MemoryPerCpu = &slurm.Uint64NoVal{Number: req.MemoryPerCpu}
+		jobDesc.MemoryPerCpu = &slurm.Uint64NoVal{Set: slurm.Ptr(true), Number: req.MemoryPerCpu}
 	}
 	if req.Nodes != nil {
 		jobDesc.Nodes = req.Nodes
@@ -461,6 +464,88 @@ func mapUint32NoValToInt32(v *slurm.Uint32NoVal) *int32 {
 	return nil
 }

+func mapUint16NoValToInt32(v *slurm.Uint16NoVal) *int32 {
+	if v != nil && v.Number != nil {
+		n := int32(*v.Number)
+		return &n
+	}
+	return nil
+}
+
+func mapUint64NoValToInt64(v *slurm.Uint64NoVal) *int64 {
+	if v != nil && v.Number != nil && *v.Number != 0 {
+		return v.Number
+	}
+	return nil
+}
+
+// formatTresList serializes a TresList to Slurm-style TRES string.
+// Format: "type=count,type:name=count" (e.g. "billing=4,cpu=4,mem=16384M").
+func formatTresList(tl slurm.TresList) string {
+	if len(tl) == 0 {
+		return ""
+	}
+	parts := make([]string, 0, len(tl))
+	for _, t := range tl {
+		if t.Type == nil || t.Count == nil {
+			continue
+		}
+		key := *t.Type
+		if t.Name != nil && *t.Name != "" {
+			key += ":" + *t.Name
+		}
+		parts = append(parts, fmt.Sprintf("%s=%d", key, *t.Count))
+	}
+	return strings.Join(parts, ",")
+}
+
+// parseGresDetail parses a comma-separated GRES string into a slice.
+func parseGresDetail(s string) []string {
+	if s == "" {
+		return nil
+	}
+	parts := strings.Split(s, ",")
+	result := make([]string, 0, len(parts))
+	for _, p := range parts {
+		p = strings.TrimSpace(p)
+		if p != "" {
+			result = append(result, p)
+		}
+	}
+	if len(result) == 0 {
+		return nil
+	}
+	return result
+}
+
+// extractMemoryFromSteps extracts peak memory usage (in MB) from a SlurmDBD
+// job's step TRES data. Despite the confusing naming, Slurm stores actual
+// resource consumption (CPU time, memory RSS) in Tres.Requested.Max, not in
+// Tres.Consumed (which holds I/O output data like disk/network).
+// It scans ALL steps and returns the maximum mem value found.
+// Returns nil if no memory data is available.
+func extractMemoryFromSteps(steps slurm.StepList) *int64 {
+	var peakMB int64
+	for _, step := range steps {
+		if step.Tres == nil || step.Tres.Requested == nil {
+			continue
+		}
+		for _, t := range step.Tres.Requested.Max {
+			if t.Type != nil && *t.Type == "mem" && t.Count != nil {
+				// Slurm reports memory in bytes, convert to MB
+				mb := *t.Count / (1024 * 1024)
+				if mb > peakMB {
+					peakMB = mb
+				}
+			}
+		}
+	}
+	if peakMB > 0 {
+		return &peakMB
+	}
+	return nil
+}
+
 // mapJobInfo maps SDK JobInfo to API JobResponse.
 func mapJobInfo(ji *slurm.JobInfo) model.JobResponse {
 	resp := model.JobResponse{}
@@ -485,6 +570,14 @@ func mapJobInfo(ji *slurm.JobInfo) model.JobResponse {
 	resp.Tasks = mapUint32NoValToInt32(ji.Tasks)
 	resp.NodeCount = mapUint32NoValToInt32(ji.NodeCount)
 	resp.BatchHost = derefStr(ji.BatchHost)
+	resp.CpusPerTask = mapUint16NoValToInt32(ji.CpusPerTask)
+	resp.MemoryPerCpu = mapUint64NoValToInt64(ji.MemoryPerCpu)
+	resp.MemoryPerNode = mapUint64NoValToInt64(ji.MemoryPerNode)
+	resp.TresReqStr = derefStr(ji.TresReqStr)
+	resp.TresAllocStr = derefStr(ji.TresAllocStr)
+	if ji.GresDetail != nil {
+		resp.GresDetail = []string(ji.GresDetail)
+	}
 	if ji.SubmitTime != nil && ji.SubmitTime.Number != nil {
 		resp.SubmitTime = ji.SubmitTime.Number
 	}
@@ -555,10 +648,22 @@ func mapSlurmdbJob(j *slurm.Job) model.JobResponse {
 	}
 	if j.Required != nil {
 		resp.Cpus = j.Required.CPUs
+		resp.CpusPerTask = nil // SlurmDBD Job doesn't expose CpusPerTask directly
+		resp.MemoryPerCpu = mapUint64NoValToInt64(j.Required.MemoryPerCpu)
+		resp.MemoryPerNode = mapUint64NoValToInt64(j.Required.MemoryPerNode)
 	}
 	if j.AllocationNodes != nil {
 		resp.NodeCount = j.AllocationNodes
 	}
+	if j.Tres != nil {
+		resp.TresAllocated = formatTresList(j.Tres.Allocated)
+		resp.TresRequested = formatTresList(j.Tres.Requested)
+	}
+	if j.Time != nil {
+		resp.Elapsed = j.Time.Elapsed
+	}
+	resp.MemoryUsed = extractMemoryFromSteps(j.Steps)
+	resp.GresDetail = parseGresDetail(derefStr(j.UsedGres))
 	resp.WorkDir = derefStr(j.WorkingDirectory)
 	return resp
 }
@@ -78,8 +78,8 @@ func TestSubmitJob_WithOptionalFields(t *testing.T) {
 		if body.Job.Partition != nil {
 			t.Error("expected partition nil for empty string")
 		}
-		if body.Job.MinimumCpus != nil {
-			t.Error("expected minimum_cpus nil when CPUs=0")
+		if body.Job.CpusPerTask != nil {
+			t.Error("expected cpus_per_task nil when CPUs=0")
 		}

 		jobID := int32(456)
@@ -885,18 +885,22 @@ func TestSubmitJob_AllSchedulingFields(t *testing.T) {
 		if j.Name == nil || *j.Name != "full-test" {
 			t.Errorf("Name mismatch: %v", j.Name)
 		}
-		if j.MinimumCpus == nil || *j.MinimumCpus != int32(8) {
-			t.Errorf("MinimumCpus mismatch: %v", j.MinimumCpus)
+		// CPUs=8 maps to CpusPerTask, then overridden by explicit CpusPerTask=2
+		if j.CpusPerTask == nil || *j.CpusPerTask != cpusPerTask {
+			t.Errorf("CpusPerTask mismatch: got %v, want %d (explicit CpusPerTask overrides CPUs)", j.CpusPerTask, cpusPerTask)
+		}
+		if j.MinimumCpus != nil {
+			t.Errorf("MinimumCpus should be nil, got %v", j.MinimumCpus)
 		}

 		// --- 22 new scheduling fields ---

 		// MemoryPerNode → *Uint64NoVal
-		if j.MemoryPerNode == nil || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
+		if j.MemoryPerNode == nil || j.MemoryPerNode.Set == nil || !*j.MemoryPerNode.Set || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
 			t.Errorf("MemoryPerNode mismatch: %v", j.MemoryPerNode)
 		}
 		// MemoryPerCpu → *Uint64NoVal
-		if j.MemoryPerCpu == nil || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
+		if j.MemoryPerCpu == nil || j.MemoryPerCpu.Set == nil || !*j.MemoryPerCpu.Set || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
 			t.Errorf("MemoryPerCpu mismatch: %v", j.MemoryPerCpu)
 		}
 		// Nodes → *string
@@ -1151,10 +1155,10 @@ func TestSubmitJob_MemoryBothSet(t *testing.T) {
 		j := body.Job

 		// Both memory fields should be mapped independently
-		if j.MemoryPerNode == nil || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
+		if j.MemoryPerNode == nil || j.MemoryPerNode.Set == nil || !*j.MemoryPerNode.Set || j.MemoryPerNode.Number == nil || *j.MemoryPerNode.Number != memoryPerNode {
 			t.Errorf("MemoryPerNode mismatch: %v", j.MemoryPerNode)
 		}
-		if j.MemoryPerCpu == nil || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
+		if j.MemoryPerCpu == nil || j.MemoryPerCpu.Set == nil || !*j.MemoryPerCpu.Set || j.MemoryPerCpu.Number == nil || *j.MemoryPerCpu.Number != memoryPerCpu {
 			t.Errorf("MemoryPerCpu mismatch: %v", j.MemoryPerCpu)
 		}

@@ -123,3 +123,28 @@ func RandomSuffix(n int) string {
 	}
 	return string(b)
 }
+
+func ResolveSchedulingMap(field string, task *model.Task) string {
+	switch field {
+	case "cpus":
+		return derefInt32ToStr(task.Cpus)
+	case "memory_per_node":
+		return derefInt64ToStr(task.MemoryPerNode)
+	case "memory_per_cpu":
+		return derefInt64ToStr(task.MemoryPerCpu)
+	case "nodes":
+		return derefStr(task.Nodes)
+	case "tasks":
+		return derefInt32ToStr(task.Tasks)
+	case "cpus_per_task":
+		return derefInt32ToStr(task.CpusPerTask)
+	case "partition":
+		return task.Partition
+	case "time_limit":
+		return derefInt32ToStr(task.TimeLimit)
+	case "qos":
+		return derefStr(task.QOS)
+	default:
+		return ""
+	}
+}
@@ -0,0 +1,72 @@
+package service
+
+import (
+	"testing"
+
+	"gcy_hpc_server/internal/model"
+)
+
+func strPtr(v string) *string { return &v }
+
+func TestResolveSchedulingMap(t *testing.T) {
+	cpus := int32Ptr(8)
+	memPerNode := int64Ptr(4096)
+	memPerCpu := int64Ptr(512)
+	nodes := strPtr("2-4")
+	tasks := int32Ptr(4)
+	cpusPerTask := int32Ptr(2)
+	timeLimit := int32Ptr(60)
+	qos := strPtr("high")
+
+	task := &model.Task{
+		Partition:     "gpu",
+		Cpus:          cpus,
+		MemoryPerNode: memPerNode,
+		MemoryPerCpu:  memPerCpu,
+		Nodes:         nodes,
+		Tasks:         tasks,
+		CpusPerTask:   cpusPerTask,
+		TimeLimit:     timeLimit,
+		QOS:           qos,
+	}
+
+	tests := []struct {
+		field string
+		want  string
+	}{
+		{"cpus", "8"},
+		{"memory_per_node", "4096"},
+		{"memory_per_cpu", "512"},
+		{"nodes", "2-4"},
+		{"tasks", "4"},
+		{"cpus_per_task", "2"},
+		{"partition", "gpu"},
+		{"time_limit", "60"},
+		{"qos", "high"},
+		{"unknown_field", ""},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.field, func(t *testing.T) {
+			got := ResolveSchedulingMap(tt.field, task)
+			if got != tt.want {
+				t.Errorf("ResolveSchedulingMap(%q) = %q, want %q", tt.field, got, tt.want)
+			}
+		})
+	}
+}
+
+func TestResolveSchedulingMap_NilFields(t *testing.T) {
+	// All scheduling fields are nil/empty — should return empty strings
+	task := &model.Task{}
+	for _, field := range []string{"cpus", "memory_per_node", "memory_per_cpu", "nodes", "tasks", "cpus_per_task", "time_limit", "qos"} {
+		got := ResolveSchedulingMap(field, task)
+		if got != "" {
+			t.Errorf("ResolveSchedulingMap(%q) with nil fields = %q, want empty", field, got)
+		}
+	}
+	// partition is a plain string, not a pointer — empty string is the zero value
+	if got := ResolveSchedulingMap("partition", task); got != "" {
+		t.Errorf("ResolveSchedulingMap(partition) = %q, want empty", got)
+	}
+}
@@ -29,12 +29,29 @@ type TaskService struct {
 	logger      *zap.Logger

 	// async processing
-	taskCh   chan int64 // buffered channel, cap=16
+	taskCh   chan int64 // buffered channel for task IDs awaiting processing
 	cancelFn context.CancelFunc
 	wg       sync.WaitGroup
 	mu       sync.Mutex // protects taskCh from send-on-closed
 	started  bool       // prevent double-start
 	stopped  bool
+	// inflight tracks task IDs currently being processed by the worker goroutine.
+	//
+	// Why it exists: taskCh is an in-memory Go channel — all pending taskIDs are
+	// lost when the server restarts. RecoverStuckTasks is responsible for
+	// recovering those lost tasks from the DB. However, GetStuckTasks uses a
+	// broad query (status NOT IN completed/failed AND updated_at < 5min ago) that
+	// also matches tasks being actively processed by the worker (e.g. a slow
+	// download). Without inflight, RecoverStuckTasks would reset those tasks to
+	// "submitted" and re-enqueue them, causing double-processing.
+	//
+	// How it works:
+	//   - ProcessTask stores the taskID on entry, deletes on exit (via defer).
+	//   - RecoverStuckTasks checks inflight before re-enqueueing; in-flight tasks
+	//     are skipped.
+	//   - On server restart inflight is empty (in-memory), so all genuinely stuck
+	//     tasks are correctly recovered without false negatives.
+	inflight sync.Map // map[int64]struct{}
 }

 func NewTaskService(
@@ -56,7 +73,7 @@ func NewTaskService(
 		jobSvc:      jobSvc,
 		workDirBase: workDirBase,
 		logger:      logger,
-		taskCh:      make(chan int64, 16),
+		taskCh:      make(chan int64, 10000),
 	}
 }

@@ -169,6 +186,9 @@ func (s *TaskService) CreateTask(ctx context.Context, req *model.CreateTaskReque

 // ProcessTask runs the full synchronous processing pipeline for a task.
 func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
+	s.inflight.Store(taskID, struct{}{})
+	defer s.inflight.Delete(taskID)
+
 	// 1. Fetch task
 	task, err := s.taskStore.GetByID(ctx, taskID)
 	if err != nil {
@@ -178,6 +198,24 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
 		return fmt.Errorf("task %d not found", taskID)
 	}

+	// Defense-in-depth against duplicate processing. When the same taskID enters
+	// taskCh multiple times (e.g. submitted normally + RecoverStuckTasks also
+	// enqueues it before the worker picks up the first copy), the worker processes
+	// them sequentially. The first invocation changes status from "submitted" to
+	// "preparing"; the second invocation reads the latest DB status, sees
+	// non-submitted, and safely skips.
+	//
+	// This does NOT block retries: processWithRetry sets status back to "submitted"
+	// before re-enqueueing, so the retried invocation passes this check and
+	// continues from the saved currentStep.
+	if task.Status != model.TaskStatusSubmitted {
+		s.logger.Debug("skipping task with non-submitted status",
+			zap.Int64("task_id", taskID),
+			zap.String("status", string(task.Status)),
+		)
+		return nil
+	}
+
 	fail := func(step, msg string) error {
 		_ = s.taskStore.UpdateStatus(ctx, taskID, model.TaskStatusFailed, msg)
 		_ = s.taskStore.UpdateRetryState(ctx, taskID, model.TaskStatusFailed, step, task.RetryCount)
@@ -263,7 +301,15 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
 		}
 	}

-	// 13-14. Set ready + submitting
+	// 13-14. Set ready + submitting (guard: skip if already submitted to Slurm)
+	if task.SlurmJobID != nil {
+		s.logger.Info("task already has slurm job, skipping submission",
+			zap.Int64("task_id", taskID),
+			zap.Int32("slurm_job_id", *task.SlurmJobID),
+		)
+		return nil
+	}
+
 	if err := s.taskStore.UpdateRetryState(ctx, taskID, model.TaskStatusReady, model.TaskStepSubmitting, 0); err != nil {
 		return fail(model.TaskStepSubmitting, fmt.Sprintf("update status to ready: %v", err))
 	}
@@ -276,6 +322,14 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
 		}
 	}

+	// 15a. Parse app environment
+	var appEnv map[string]string
+	if len(app.Environment) > 0 {
+		if err := json.Unmarshal(app.Environment, &appEnv); err != nil {
+			return fail(model.TaskStepSubmitting, fmt.Sprintf("parse application environment: %v", err))
+		}
+	}
+
 	// 16. Parse task values
 	values := make(map[string]string)
 	if len(task.Values) > 0 {
@@ -284,16 +338,55 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
 		}
 	}

+	// 16a. Auto-inject WORK_DIR if the app defines it as a parameter.
+	// The work directory is created by the server, not provided by the user.
+	for _, p := range params {
+		if p.Name == "WORK_DIR" {
+			values["WORK_DIR"] = workDir
+			break
+		}
+	}
+
+	// 16b. Map input_file_ids to file-type parameters by order.
+	// User selects files via FilePicker; we assign their IDs to file/directory
+	// params sequentially so the backend can resolve them to filenames later.
+	var inputFileIDs []int64
+	if len(task.InputFileIDs) > 0 {
+		if err := json.Unmarshal(task.InputFileIDs, &inputFileIDs); err != nil {
+			return fail(model.TaskStepSubmitting, fmt.Sprintf("parse input file ids: %v", err))
+		}
+	}
+	if len(inputFileIDs) > 0 {
+		fileParamIdx := 0
+		for _, p := range params {
+			if p.Type != model.ParamTypeFile && p.Type != model.ParamTypeDirectory {
+				continue
+			}
+			if fileParamIdx < len(inputFileIDs) {
+				values[p.Name] = strconv.FormatInt(inputFileIDs[fileParamIdx], 10)
+				fileParamIdx++
+			}
+		}
+	}
+
+	// 16b-3. Auto-inject scheduling params based on scheduling_map.
+	// If an Application parameter declares scheduling_map, the corresponding
+	// scheduling field value overrides any user-provided value.
+	for _, p := range params {
+		if p.SchedulingMap == "" {
+			continue
+		}
+		if val := ResolveSchedulingMap(p.SchedulingMap, task); val != "" {
+			values[p.Name] = val
+		}
+	}
+
+	// 16c. Validate all params (WORK_DIR and file params now have values).
 	if err := ValidateParams(params, values); err != nil {
 		return fail(model.TaskStepSubmitting, err.Error())
 	}

-	if strings.Contains(app.ScriptTemplate, "$WORK_DIR") {
-		values["WORK_DIR"] = workDir
-	}
-
-	// Resolve file-type parameters: user sends file_id, we replace with filename.
-	// Only query the database if there are file/directory-type parameters with values.
+	// 16d. Resolve file-type parameter values: file_id → filename.
 	var fileLookupIDs []int64
 	for _, p := range params {
 		if p.Type != model.ParamTypeFile && p.Type != model.ParamTypeDirectory {
@@ -386,6 +479,7 @@ func (s *TaskService) ProcessTask(ctx context.Context, taskID int64) error {
 		Dependency:     task.Dependency,
 		Requeue:        task.Requeue,
 		KillOnNodeFail: task.KillOnNodeFail,
+		Environment:    appEnv,
 	})
 	if err != nil {
 		return fail(model.TaskStepSubmitting, fmt.Sprintf("submit job: %v", err))
@@ -453,7 +547,7 @@ func uniqueInt64s(ids []int64) []int64 {

 func (s *TaskService) mapSlurmStateToTaskStatus(slurmState []string) string {
 	if len(slurmState) == 0 {
-		return model.TaskStatusRunning
+		return ""
 	}

 	state := strings.ToUpper(slurmState[0])
@@ -467,7 +561,8 @@ func (s *TaskService) mapSlurmStateToTaskStatus(slurmState []string) string {
 	case "FAILED", "CANCELLED", "TIMEOUT", "NODE_FAIL", "OUT_OF_MEMORY", "PREEMPTED":
 		return model.TaskStatusFailed
 	default:
-		return model.TaskStatusRunning
+		s.logger.Warn("unrecognized slurm state, skipping update", zap.String("state", state))
+		return ""
 	}
 }

@@ -498,15 +593,16 @@ func (s *TaskService) refreshTaskStatus(ctx context.Context, taskID int64) error
 	}

 	newStatus := s.mapSlurmStateToTaskStatus(jobResp.State)
-	if newStatus != task.Status {
+	if newStatus == "" || newStatus == task.Status {
+		return nil
+	}
+
 	s.logger.Info("updating task status from slurm",
 		zap.Int64("task_id", taskID),
 		zap.String("old_status", task.Status),
 		zap.String("new_status", newStatus),
 	)
 	return s.taskStore.UpdateStatus(ctx, taskID, newStatus, "")
-	}
-	return nil
 }

 func (s *TaskService) RefreshStaleTasks(ctx context.Context) error {
@@ -618,7 +714,7 @@ func (s *TaskService) StopProcessor() {

 	s.mu.Lock()
 	drainCh := s.taskCh
-	s.taskCh = make(chan int64, 16)
+	s.taskCh = make(chan int64, 10000)
 	s.mu.Unlock()

 	for taskID := range drainCh {
@@ -652,12 +748,55 @@ func (s *TaskService) processWithRetry(ctx context.Context, taskID int64) {
 }

 func (s *TaskService) RecoverStuckTasks(ctx context.Context) {
+	// RecoverStuckTasks recovers tasks that are "stuck" — they exist in the DB
+	// with a non-terminal status but are not being processed.
+	//
+	// Scenarios that create stuck tasks:
+	//
+	//   1. Server restart: taskCh is an in-memory Go channel, all pending IDs are
+	//      lost on process exit. Tasks that were queued but never picked up by the
+	//      worker remain in "submitted" status in DB with no one to process them.
+	//
+	//   2. Server crash mid-processing: the worker had advanced a task to
+	//      "preparing"/"downloading" and then died. The task sits in that
+	//      intermediate state with no SlurmJobID and no worker to continue.
+	//
+	//   3. Channel full: SubmitAsync dropped a task because taskCh was at
+	//      capacity. The task stays "submitted" but was never enqueued.
+	//
+	// The bug this fix addresses:
+	//
+	//   GetStuckTasks queries: status NOT IN (completed, failed) AND updated_at <
+	//   5min ago. This also matches tasks currently being processed by the worker
+	//   whose step is slow (>5 min, e.g. downloading large files) and hasn't
+	//   refreshed updated_at. Without the inflight check below, this function
+	//   would reset such a task to "submitted" and re-enqueue it, causing the
+	//   same task to be processed by two concurrent invocations of ProcessTask.
+	//
+	// Fix: the inflight sync.Map tracks taskIDs currently inside ProcessTask.
+	// Tasks found in inflight are skipped here. On server restart inflight is
+	// empty (it's in-memory), so all genuinely stuck tasks from scenarios 1-3
+	// above are correctly recovered.
+
 	tasks, err := s.taskStore.GetStuckTasks(ctx, 5*time.Minute)
 	if err != nil {
 		s.logger.Error("failed to get stuck tasks", zap.Error(err))
 		return
 	}
 	for i := range tasks {
+		if tasks[i].SlurmJobID != nil {
+			s.logger.Info("skipping stuck task recovery, already in slurm",
+				zap.Int64("taskID", tasks[i].ID),
+				zap.Int32("slurm_job_id", *tasks[i].SlurmJobID),
+			)
+			continue
+		}
+		if _, ok := s.inflight.Load(tasks[i].ID); ok {
+			s.logger.Debug("skipping in-flight task",
+				zap.Int64("taskID", tasks[i].ID),
+			)
+			continue
+		}
 		_ = s.taskStore.UpdateStatus(ctx, tasks[i].ID, model.TaskStatusSubmitted, "")
 		s.mu.Lock()
 		if !s.stopped {
@@ -97,7 +97,7 @@ func TestTaskService_MapSlurmState_AllStates(t *testing.T) {
 		{[]string{"OUT_OF_MEMORY"}, model.TaskStatusFailed},
 		{[]string{"PREEMPTED"}, model.TaskStatusFailed},
 		{[]string{"SPECIAL_EXIT"}, model.TaskStatusRunning},
-		{[]string{"unknown_state"}, model.TaskStatusRunning},
+		{[]string{"unknown_state"}, ""},
 		{[]string{"pending"}, model.TaskStatusQueued},
 		{[]string{"Running"}, model.TaskStatusRunning},
 	}
@@ -115,13 +115,13 @@ func TestTaskService_MapSlurmState_Empty(t *testing.T) {
 	defer env.close()

 	got := env.svc.mapSlurmStateToTaskStatus([]string{})
-	if got != model.TaskStatusRunning {
-		t.Errorf("mapSlurmStateToTaskStatus([]) = %q, want %q", got, model.TaskStatusRunning)
+	if got != "" {
+		t.Errorf("mapSlurmStateToTaskStatus([]) = %q, want empty string", got)
 	}

 	got = env.svc.mapSlurmStateToTaskStatus(nil)
-	if got != model.TaskStatusRunning {
-		t.Errorf("mapSlurmStateToTaskStatus(nil) = %q, want %q", got, model.TaskStatusRunning)
+	if got != "" {
+		t.Errorf("mapSlurmStateToTaskStatus(nil) = %q, want empty string", got)
 	}
 }

@@ -1048,8 +1048,12 @@ func TestProcessTask_SchedulingParams(t *testing.T) {
 	if j.Partition == nil || *j.Partition != "gpu" {
 		t.Errorf("Partition = %v, want %q", j.Partition, "gpu")
 	}
-	if j.MinimumCpus == nil || *j.MinimumCpus != int32(8) {
-		t.Errorf("MinimumCpus = %v, want 8", j.MinimumCpus)
+	// CPUs=8 maps to CpusPerTask, then overridden by explicit CpusPerTask=2
+	if j.CpusPerTask == nil || *j.CpusPerTask != int32(2) {
+		t.Errorf("CpusPerTask = %v, want 2 (explicit CpusPerTask overrides CPUs)", j.CpusPerTask)
+	}
+	if j.MinimumCpus != nil {
+		t.Errorf("MinimumCpus should be nil, got %v", j.MinimumCpus)
 	}
 	if j.TimeLimit == nil || j.TimeLimit.Number == nil || *j.TimeLimit.Number != int64(60) {
 		t.Errorf("TimeLimit = %v, want 60", j.TimeLimit)
@@ -1175,8 +1179,8 @@ func TestProcessTask_PartialSchedulingParams(t *testing.T) {
 		t.Errorf("Partition = %v, want %q", j.Partition, "debug")
 	}

-	if j.MinimumCpus != nil {
-		t.Errorf("MinimumCpus = %v, want nil (no cpus set)", j.MinimumCpus)
+	if j.CpusPerTask != nil {
+		t.Errorf("CpusPerTask = %v, want nil (no cpus set)", j.CpusPerTask)
 	}
 	if j.TimeLimit == nil {
 		t.Errorf("TimeLimit = nil, want non-nil (default should be injected)")
@@ -1262,3 +1266,49 @@ func TestProcessTask_PartialSchedulingParams(t *testing.T) {
 		t.Errorf("KillOnNodeFail = %v, want nil", j.KillOnNodeFail)
 	}
 }
+
+func TestTaskService_ProcessTask_SchedulingMapInjection(t *testing.T) {
+	jobID := int32(42)
+
+	var capturedReq slurm.JobSubmitReq
+
+	env := newTaskTestEnv(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if err := json.NewDecoder(r.Body).Decode(&capturedReq); err != nil {
+			t.Fatalf("decode request body: %v", err)
+		}
+		json.NewEncoder(w).Encode(slurm.OpenapiJobSubmitResponse{
+			Result: &slurm.JobSubmitResponseMsg{JobID: &jobID},
+		})
+	}))
+	defer env.close()
+
+	params := json.RawMessage(`[
+		{"name": "NP", "type": "integer", "scheduling_map": "cpus", "required": true}
+	]`)
+	appID := env.createApp(t, "sched-map-app", "#!/bin/bash\nmpirun -np $NP my_app", params)
+
+	cpus := int32(8)
+	task, err := env.svc.CreateTask(context.Background(), &model.CreateTaskRequest{
+		AppID:    appID,
+		TaskName: "sched-map-test",
+		Cpus:     &cpus,
+	})
+	if err != nil {
+		t.Fatalf("CreateTask: %v", err)
+	}
+
+	if err := env.svc.ProcessTask(context.Background(), task.ID); err != nil {
+		t.Fatalf("ProcessTask: %v", err)
+	}
+
+	if capturedReq.Script == nil {
+		t.Fatal("submitted script is nil")
+	}
+
+	if !strings.Contains(*capturedReq.Script, "'8'") {
+		t.Errorf("rendered script does not contain shell-escaped scheduling value:\n%s", *capturedReq.Script)
+	}
+	if !strings.Contains(*capturedReq.Script, "mpirun -np '8'") {
+		t.Errorf("rendered script does not contain expected mpirun command:\n%s", *capturedReq.Script)
+	}
+}
@@ -357,8 +357,9 @@ func (s *UploadService) CompleteUpload(ctx context.Context, sessionID int64) (*m
 			keys[i] = fmt.Sprintf("%schunk_%05d", minioPrefix, i)
 		}
 		go func() {
-			bgCtx := context.Background()
-			if delErr := s.storage.RemoveObjects(bgCtx, s.cfg.Bucket, keys, storage.RemoveObjectsOptions{}); delErr != nil {
+			ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+			defer cancel()
+			if delErr := s.storage.RemoveObjects(ctx, s.cfg.Bucket, keys, storage.RemoveObjectsOptions{}); delErr != nil {
 				s.logger.Warn("delete temp chunks", zap.Error(delErr))
 			}
 		}()
@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"

 	"gcy_hpc_server/internal/model"

@@ -52,6 +53,15 @@ func (s *ApplicationStore) Create(ctx context.Context, req *model.CreateApplicat
 		params = json.RawMessage(`[]`)
 	}

+	var envJSON json.RawMessage
+	if len(req.Environment) > 0 {
+		b, err := json.Marshal(req.Environment)
+		if err != nil {
+			return 0, fmt.Errorf("marshal environment: %w", err)
+		}
+		envJSON = b
+	}
+
 	app := &model.Application{
 		Name:           req.Name,
 		Description:    req.Description,
@@ -59,6 +69,7 @@ func (s *ApplicationStore) Create(ctx context.Context, req *model.CreateApplicat
 		Category:       req.Category,
 		ScriptTemplate: req.ScriptTemplate,
 		Parameters:     params,
+		Environment:    envJSON,
 		Scope:          req.Scope,
 	}
 	if err := s.db.WithContext(ctx).Create(app).Error; err != nil {
@@ -87,6 +98,9 @@ func (s *ApplicationStore) Update(ctx context.Context, id int64, req *model.Upda
 	if req.Parameters != nil {
 		updates["parameters"] = *req.Parameters
 	}
+	if req.Environment != nil {
+		updates["environment"] = *req.Environment
+	}
 	if req.Scope != nil {
 		updates["scope"] = *req.Scope
 	}
@@ -49,7 +49,7 @@ func TestAllRoutesRegistered(t *testing.T) {
 		method string
 		path   string
 	}{
-		{"POST", "/api/v1/jobs/submit"},
+		// {"POST", "/api/v1/jobs/submit"}, // [已弃用] 已被 POST /tasks 取代
 		{"GET", "/api/v1/jobs"},
 		{"GET", "/api/v1/jobs/history"},
 		{"GET", "/api/v1/jobs/1"},
@@ -82,8 +82,8 @@ func TestAllRoutesRegistered(t *testing.T) {
 		{"GET", "/api/v1/tasks"},
 	}

-	if len(routes) != 30 {
-		t.Fatalf("expected 31 routes, got %d", len(routes))
+	if len(routes) != 29 {
+		t.Fatalf("expected 30 routes, got %d", len(routes))
 	}

 	for _, r := range routes {
@@ -9,20 +9,21 @@
    "preview": "vite preview"
  },
  "dependencies": {
-    "@element-plus/icons-vue": "^2.3.2",
-    "axios": "^1.15.0",
-    "element-plus": "^2.13.7",
-    "vue": "^3.5.32",
-    "vue-router": "^5.0.4"
+    "@element-plus/icons-vue": "^2.3.1",
+    "axios": "^1.6.5",
+    "element-plus": "^2.5.2",
+    "vue": "^3.4.15",
+    "vue-router": "^4.2.5"
  },
  "devDependencies": {
-    "@types/node": "^24.12.2",
-    "@vitejs/plugin-vue": "^6.0.5",
-    "@vue/tsconfig": "^0.9.1",
-    "typescript": "~6.0.2",
-    "unplugin-auto-import": "^21.0.0",
-    "unplugin-vue-components": "^32.0.0",
-    "vite": "^8.0.4",
-    "vue-tsc": "^3.2.6"
+    "@playwright/test": "^1.59.1",
+    "@types/node": "^20.11.5",
+    "@vitejs/plugin-vue": "^5.0.3",
+    "@vue/tsconfig": "^0.5.1",
+    "typescript": "~5.3.3",
+    "unplugin-auto-import": "^0.17.3",
+    "unplugin-vue-components": "^0.26.0",
+    "vite": "^5.0.12",
+    "vue-tsc": "^2.0.24"
  }
 }
@@ -6,6 +6,7 @@ export interface ParameterSchema {
  default?: string
  options?: string[]
  description?: string
+  scheduling_map?: string
 }

 export interface Application {
@@ -34,7 +34,7 @@

        <template v-if="selectedApp">
          <el-form-item
-            v-for="param in (selectedApp.parameters || [])"
+            v-for="param in visibleParams"
            :key="param.name"
            :label="param.label || param.name"
          >
@@ -57,7 +57,13 @@
              @update:model-value="(val: string | number | boolean) => { values[param.name] = String(val) }"
            />

-            <el-input v-else-if="param.type === 'file' || param.type === 'directory'" disabled placeholder="文件选择功能开发中" />
+            <template v-else-if="param.type === 'file' || param.type === 'directory'">
+              <div v-if="getFileForParam(param)" style="display: flex; align-items: center; gap: 8px">
+                <el-tag>{{ getFileForParam(param).name }}</el-tag>
+                <span style="color: #909399; font-size: 12px">ID: {{ getFileForParam(param).id }}</span>
+              </div>
+              <span v-else style="color: #909399; font-size: 13px">请通过上方"关联文件"选择</span>
+            </template>
          </el-form-item>
        </template>

@@ -115,6 +121,33 @@ const showFilePicker = ref(false)

 const selectedApp = computed(() => appList.value.find(a => a.id === selectedAppId.value))

+const autoParams = new Set(['WORK_DIR'])
+
+const visibleParams = computed(() =>
+  (selectedApp.value?.parameters || []).filter((p: any) => !autoParams.has(p.name) && !p.scheduling_map)
+)
+
+const fileParams = computed(() =>
+  (selectedApp.value?.parameters || []).filter(
+    (p: any) => p.type === 'file' || p.type === 'directory'
+  )
+)
+
+const fileParamMapping = computed(() => {
+  const mapping: Record<string, string> = {}
+  fileParams.value.forEach((p: any, i: number) => {
+    if (selectedFiles.value[i]) {
+      mapping[p.name] = String(selectedFiles.value[i].id)
+    }
+  })
+  return mapping
+})
+
+const getFileForParam = (param: any) => {
+  const index = fileParams.value.findIndex((p: any) => p.name === param.name)
+  return selectedFiles.value[index]
+}
+
 watch(selectedAppId, () => { values.value = {} })

 onMounted(async () => {
@@ -126,12 +159,33 @@ onMounted(async () => {
  }
 })

+const resolveSchedMapValue = (mapField: string): string | undefined => {
+  switch (mapField) {
+    case 'cpus': return form.cpus != null ? String(form.cpus) : undefined
+    case 'memory_per_node': return form.memory_per_node != null ? String(form.memory_per_node) : undefined
+    case 'nodes': return form.nodes || undefined
+    case 'tasks': return form.tasks != null ? String(form.tasks) : undefined
+    case 'cpus_per_task': return form.cpus_per_task != null ? String(form.cpus_per_task) : undefined
+    case 'partition': return form.partition || undefined
+    default: return undefined
+  }
+}
+
 const handleSubmit = async () => {
  if (!selectedAppId.value) { ElMessage.warning('请选择应用'); return }
  submitting.value = true
  try {
    const taskName = form.task_name.trim() || `task_${selectedAppId.value}_${Date.now()}`
-    const resp = await createTask({ ...form, task_name: taskName, job_name: taskName, app_id: selectedAppId.value, values: values.value, file_ids: selectedFiles.value.map(f => f.id) })
+    const mergedValues = { ...values.value, ...fileParamMapping.value }
+    // Auto-inject scheduling_map values
+    const schedParams = (selectedApp.value?.parameters || []).filter((p: any) => p.scheduling_map)
+    for (const p of schedParams) {
+      const val = resolveSchedMapValue(p.scheduling_map)
+      if (val !== undefined && val !== '') {
+        mergedValues[p.name] = val
+      }
+    }
+    const resp = await createTask({ ...form, task_name: taskName, job_name: taskName, app_id: selectedAppId.value, values: mergedValues, file_ids: selectedFiles.value.map(f => f.id) })
    if (resp.success) {
      ElMessage.success('任务提交成功')
      router.push('/tasks')
@@ -1,18 +1,17 @@
 {
  "extends": "@vue/tsconfig/tsconfig.dom.json",
  "compilerOptions": {
+    "incremental": true,
    "tsBuildInfoFile": "./node_modules/.tmp/tsconfig.app.tsbuildinfo",
    "types": ["vite/client"],
    "baseUrl": ".",
    "paths": {
      "@/*": ["src/*"]
    },
-    "ignoreDeprecations": "6.0",

    /* Linting */
    "noUnusedLocals": true,
    "noUnusedParameters": true,
-    "erasableSyntaxOnly": true,
    "noFallthroughCasesInSwitch": true
  },
  "include": ["src/**/*.ts", "src/**/*.tsx", "src/**/*.vue"]
@@ -1,8 +1,9 @@
 {
  "compilerOptions": {
+    "composite": true,
    "tsBuildInfoFile": "./node_modules/.tmp/tsconfig.node.tsbuildinfo",
-    "target": "es2023",
-    "lib": ["ES2023"],
+    "target": "es2022",
+    "lib": ["ES2022"],
    "module": "esnext",
    "types": ["node"],
    "skipLibCheck": true,
@@ -10,14 +11,12 @@
    /* Bundler mode */
    "moduleResolution": "bundler",
    "allowImportingTsExtensions": true,
-    "verbatimModuleSyntax": true,
    "moduleDetection": "force",
    "noEmit": true,

    /* Linting */
    "noUnusedLocals": true,
    "noUnusedParameters": true,
-    "erasableSyntaxOnly": true,
    "noFallthroughCasesInSwitch": true
  },
  "include": ["vite.config.ts"]
Author	SHA1	Message	Date
dailz	5591b67f75	feat(task): auto-inject scheduling params into script template via scheduling_map Add scheduling_map field to ParameterSchema so Application creators can declare that a parameter (e.g. NP) maps to a scheduling field (e.g. cpus). The backend auto-injects the scheduling value into script template variables before rendering, eliminating duplicate user input. The frontend hides mapped parameters from the form and injects their values on submit.	2026-04-22 10:26:52 +08:00
dailz	435ab285c1	fix(task): prevent RecoverStuckTasks from re-enqueueing in-flight tasks RecoverStuckTasks scans for tasks with updated_at > 5min ago and re-enqueues them. This incorrectly matched tasks actively being processed by the worker (e.g. slow downloads), causing double-processing. Add inflight sync.Map to track taskIDs currently inside ProcessTask. RecoverStuckTasks skips tasks found in inflight. On server restart inflight is empty (in-memory), so genuinely stuck tasks are still correctly recovered. Also: increase taskCh buffer 16→10000, add periodic RecoverStuckTasks goroutine in TaskPoller (every 5min), and add status guard in ProcessTask as defense-in-depth against duplicate enqueues.	2026-04-21 17:19:10 +08:00
dailz	8955e513aa	fix(upload): add 30s timeout to background chunk cleanup goroutine The cleanup goroutine used context.Background() with no timeout, so if MinIO accepted TCP connections but never responded, the goroutine would block indefinitely. Now uses context.WithTimeout to prevent leaks.	2026-04-21 13:35:21 +08:00
dailz	f13377ca7d	fix(task): return empty string for unknown/empty Slurm states instead of defaulting to running mapSlurmStateToTaskStatus previously defaulted to 'running' for empty state arrays and unrecognized states. This was too aggressive — treating unknown as actively running could cause incorrect status updates when Slurm returns unexpected or empty state data. Now empty/unknown states return an empty string, and refreshTaskStatus skips the update in that case.	2026-04-21 13:23:40 +08:00
dailz	43329d2333	docs(openapi): remove deprecated POST /jobs/submit endpoint and SubmitJobRequest schema	2026-04-21 11:31:07 +08:00
dailz	b90942de77	fix(task): prevent duplicate Slurm job submission on backend restart RecoverStuckTasks now skips tasks that already have a slurm_job_id, and ProcessTask adds a guard before the submitting step to prevent re-submission even if a task is incorrectly re-enqueued. Also deprecates POST /api/v1/jobs/submit endpoint (replaced by POST /tasks) and comments out related handlers and tests.	2026-04-21 10:57:38 +08:00
dailz	4fd331ebd8	feat(application): add Environment field and inject into Slurm job submission	2026-04-21 10:23:31 +08:00
dailzandSisyphus	d9ca9233b3	fix(service): correct CPU/memory mapping and add TRES/memory_used extraction - Map CPUs to CpusPerTask (not MinimumCpus) for consistent SlurmDBD history - Add Set:true to memory Uint64NoVal on submission - Filter number=0 in mapUint64NoValToInt64 to avoid false zeros - Extract peak memory from Steps.Tres.Requested.Max across all steps - Add formatTresList, parseGresDetail, extractMemoryFromSteps helpers - Update mapJobInfo and mapSlurmdbJob with new field mappings Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>	2026-04-20 17:10:19 +08:00
dailzandSisyphus	d79656c728	feat(model): add resource fields to JobResponse (CPU, memory, TRES, elapsed) Add CpusPerTask, MemoryPerCpu, MemoryPerNode, MemoryUsed, TRES strings, GresDetail, and Elapsed fields to capture full Slurm job resource info. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>	2026-04-20 17:09:52 +08:00
dailzandSisyphus	0a0e5fd7d4	fix(web): downgrade deps for TS 5.3 compatibility vue-tsc 1.8.27 -> 2.0.24 (monkey-patching broke with TS 5.3). Remove erasableSyntaxOnly, ignoreDeprecations (TS 6.0 only). Remove verbatimModuleSyntax, target es2023 -> es2022. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>	2026-04-20 13:41:46 +08:00
dailzandSisyphus	22963d0e6f	fix(web): map selected files to file-type params and hide auto-injected params File/directory type app parameters now show the associated file name instead of a disabled input. WORK_DIR is hidden from the form as it is auto-injected by the backend. Selected files are mapped to file params via fileParamMapping on submit. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>	2026-04-20 13:41:37 +08:00
dailzandSisyphus	08ca4da691	fix(service): inject WORK_DIR and map file_ids before param validation Previously ValidateParams ran before WORK_DIR injection and file_ids mapping, causing required parameter missing errors for auto-handled params. Now the execution order is: inject WORK_DIR, map file_ids to file params, validate, resolve. Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>	2026-04-20 13:41:28 +08:00