feat: add batch task approval settings and fix modal dropdown scrolling

This commit is contained in:
Ed1s0nZ
2026-09-13 18:02:43 +08:00
parent c563567502
commit ac101d8476
13 changed files with 387 additions and 32 deletions
+35
View File
@@ -0,0 +1,35 @@
package database
import (
"path/filepath"
"testing"
"go.uber.org/zap"
)
func TestBatchHITLLegacyMigration(t *testing.T) {
db, err := NewDB(filepath.Join(t.TempDir(), "legacy.db"), zap.NewNop())
if err != nil {
t.Fatal(err)
}
defer db.Close()
if err := db.CreateBatchQueue("legacy", "test", "", "eino_single", "manual", "", nil, "", 1, nil); err != nil {
t.Fatal(err)
}
if _, err := db.Exec("ALTER TABLE batch_task_queues DROP COLUMN hitl_policy"); err != nil {
t.Fatal(err)
}
if err := db.migrateBatchTaskQueuesTable(); err != nil {
t.Fatal(err)
}
if err := db.migrateBatchTaskQueuesTable(); err != nil {
t.Fatal(err)
}
row, err := db.GetBatchQueue("legacy")
if err != nil {
t.Fatal(err)
}
if row.HITLPolicy != "" {
t.Fatalf("legacy queue must inherit: %+v", row)
}
}
+17 -7
View File
@@ -15,6 +15,7 @@ type BatchTaskQueueRow struct {
Title sql.NullString
Role sql.NullString
AgentMode sql.NullString
HITLPolicy string
ScheduleMode sql.NullString
CronExpr sql.NullString
NextRunAt sql.NullTime
@@ -56,7 +57,12 @@ func (db *DB) CreateBatchQueue(
projectID string,
concurrency int,
tasks []map[string]interface{},
hitlPolicies ...string,
) error {
policy := ""
if len(hitlPolicies) > 0 {
policy = hitlPolicies[0]
}
tx, err := db.Begin()
if err != nil {
return fmt.Errorf("开始事务失败: %w", err)
@@ -74,8 +80,8 @@ func (db *DB) CreateBatchQueue(
projectIDVal = strings.TrimSpace(projectID)
}
_, err = tx.Exec(
"INSERT INTO batch_task_queues (id, title, role, agent_mode, schedule_mode, cron_expr, next_run_at, schedule_enabled, project_id, concurrency, status, created_at, current_index) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
queueID, title, role, agentMode, scheduleMode, cronExpr, nextRunAtValue, 1, projectIDVal, concurrency, "pending", now, 0,
"INSERT INTO batch_task_queues (id, title, role, agent_mode, hitl_policy, schedule_mode, cron_expr, next_run_at, schedule_enabled, project_id, concurrency, status, created_at, current_index) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
queueID, title, role, agentMode, policy, scheduleMode, cronExpr, nextRunAtValue, 1, projectIDVal, concurrency, "pending", now, 0,
)
if err != nil {
return fmt.Errorf("创建批量任务队列失败: %w", err)
@@ -104,7 +110,7 @@ func (db *DB) CreateBatchQueue(
return tx.Commit()
}
const batchQueueSelectColumns = `id, title, role, agent_mode, schedule_mode, cron_expr, next_run_at, schedule_enabled, last_schedule_trigger_at, last_schedule_error, last_run_error, project_id, concurrency, status, created_at, started_at, completed_at, current_index`
const batchQueueSelectColumns = `id, title, role, agent_mode, hitl_policy, schedule_mode, cron_expr, next_run_at, schedule_enabled, last_schedule_trigger_at, last_schedule_error, last_run_error, project_id, concurrency, status, created_at, started_at, completed_at, current_index`
// GetBatchQueue 获取批量任务队列
func (db *DB) GetBatchQueue(queueID string) (*BatchTaskQueueRow, error) {
@@ -113,7 +119,7 @@ func (db *DB) GetBatchQueue(queueID string) (*BatchTaskQueueRow, error) {
err := db.QueryRow(
"SELECT "+batchQueueSelectColumns+" FROM batch_task_queues WHERE id = ?",
queueID,
).Scan(&row.ID, &row.Title, &row.Role, &row.AgentMode, &row.ScheduleMode, &row.CronExpr, &row.NextRunAt, &row.ScheduleEnabled, &row.LastScheduleTriggerAt, &row.LastScheduleError, &row.LastRunError, &row.ProjectID, &row.Concurrency, &row.Status, &createdAt, &row.StartedAt, &row.CompletedAt, &row.CurrentIndex)
).Scan(&row.ID, &row.Title, &row.Role, &row.AgentMode, &row.HITLPolicy, &row.ScheduleMode, &row.CronExpr, &row.NextRunAt, &row.ScheduleEnabled, &row.LastScheduleTriggerAt, &row.LastScheduleError, &row.LastRunError, &row.ProjectID, &row.Concurrency, &row.Status, &createdAt, &row.StartedAt, &row.CompletedAt, &row.CurrentIndex)
if err == sql.ErrNoRows {
return nil, nil
}
@@ -148,7 +154,7 @@ func (db *DB) GetAllBatchQueues() ([]*BatchTaskQueueRow, error) {
for rows.Next() {
var row BatchTaskQueueRow
var createdAt string
if err := rows.Scan(&row.ID, &row.Title, &row.Role, &row.AgentMode, &row.ScheduleMode, &row.CronExpr, &row.NextRunAt, &row.ScheduleEnabled, &row.LastScheduleTriggerAt, &row.LastScheduleError, &row.LastRunError, &row.ProjectID, &row.Concurrency, &row.Status, &createdAt, &row.StartedAt, &row.CompletedAt, &row.CurrentIndex); err != nil {
if err := rows.Scan(&row.ID, &row.Title, &row.Role, &row.AgentMode, &row.HITLPolicy, &row.ScheduleMode, &row.CronExpr, &row.NextRunAt, &row.ScheduleEnabled, &row.LastScheduleTriggerAt, &row.LastScheduleError, &row.LastRunError, &row.ProjectID, &row.Concurrency, &row.Status, &createdAt, &row.StartedAt, &row.CompletedAt, &row.CurrentIndex); err != nil {
return nil, fmt.Errorf("扫描批量任务队列失败: %w", err)
}
parsedTime, parseErr := time.Parse("2006-01-02 15:04:05", createdAt)
@@ -220,7 +226,7 @@ func (db *DB) ListBatchQueuesForAccess(limit, offset int, status, keyword, userI
for rows.Next() {
var row BatchTaskQueueRow
var createdAt string
if err := rows.Scan(&row.ID, &row.Title, &row.Role, &row.AgentMode, &row.ScheduleMode, &row.CronExpr, &row.NextRunAt, &row.ScheduleEnabled, &row.LastScheduleTriggerAt, &row.LastScheduleError, &row.LastRunError, &row.ProjectID, &row.Concurrency, &row.Status, &createdAt, &row.StartedAt, &row.CompletedAt, &row.CurrentIndex); err != nil {
if err := rows.Scan(&row.ID, &row.Title, &row.Role, &row.AgentMode, &row.HITLPolicy, &row.ScheduleMode, &row.CronExpr, &row.NextRunAt, &row.ScheduleEnabled, &row.LastScheduleTriggerAt, &row.LastScheduleError, &row.LastRunError, &row.ProjectID, &row.Concurrency, &row.Status, &createdAt, &row.StartedAt, &row.CompletedAt, &row.CurrentIndex); err != nil {
return nil, fmt.Errorf("扫描批量任务队列失败: %w", err)
}
parsedTime, parseErr := time.Parse("2006-01-02 15:04:05", createdAt)
@@ -411,7 +417,11 @@ func (db *DB) UpdateBatchQueueCurrentIndex(queueID string, currentIndex int) err
}
// UpdateBatchQueueMetadata 更新批量任务队列标题、角色、代理模式和并发数
func (db *DB) UpdateBatchQueueMetadata(queueID, title, role, agentMode string, concurrency int) error {
func (db *DB) UpdateBatchQueueMetadata(queueID, title, role, agentMode string, concurrency int, hitlPolicies ...string) error {
if len(hitlPolicies) > 0 {
_, err := db.Exec("UPDATE batch_task_queues SET title = ?, role = ?, agent_mode = ?, concurrency = ?, hitl_policy = ? WHERE id = ?", title, role, agentMode, concurrency, hitlPolicies[0], queueID)
return err
}
_, err := db.Exec(
"UPDATE batch_task_queues SET title = ?, role = ?, agent_mode = ?, concurrency = ? WHERE id = ?",
title, role, agentMode, concurrency, queueID,
+11
View File
@@ -468,6 +468,7 @@ func (db *DB) initTables() error {
title TEXT,
role TEXT,
agent_mode TEXT NOT NULL DEFAULT 'eino_single',
hitl_policy TEXT NOT NULL DEFAULT '',
schedule_mode TEXT NOT NULL DEFAULT 'manual',
cron_expr TEXT,
next_run_at DATETIME,
@@ -1379,6 +1380,16 @@ func (db *DB) migrateBatchTaskQueuesTable() error {
}
}
var hitlPolicyCount int
if err := db.QueryRow("SELECT COUNT(*) FROM pragma_table_info('batch_task_queues') WHERE name='hitl_policy'").Scan(&hitlPolicyCount); err != nil {
return fmt.Errorf("检查队列审批字段失败: %w", err)
}
if hitlPolicyCount == 0 {
if _, err := db.Exec("ALTER TABLE batch_task_queues ADD COLUMN hitl_policy TEXT NOT NULL DEFAULT ''"); err != nil {
return fmt.Errorf("添加队列审批字段失败: %w", err)
}
}
var concurrencyCount int
err = db.QueryRow("SELECT COUNT(*) FROM pragma_table_info('batch_task_queues') WHERE name='concurrency'").Scan(&concurrencyCount)
if err != nil {
+12 -6
View File
@@ -1849,6 +1849,7 @@ func filterSlice[T any](items []T, keep func(T) bool) []T {
// BatchTaskRequest 批量任务请求
type BatchTaskRequest struct {
HITLPolicy string `json:"hitlPolicy"`
Title string `json:"title"` // 任务标题(可选)
Tasks []string `json:"tasks" binding:"required"` // 任务列表,每行一个任务
Role string `json:"role,omitempty"` // 角色名称(可选,空字符串表示默认角色)
@@ -1923,7 +1924,7 @@ func (h *AgentHandler) CreateBatchQueue(c *gin.Context) {
nextRunAt = &next
}
queue, createErr := h.batchTaskManager.CreateBatchQueue(req.Title, req.Role, agentMode, scheduleMode, cronExpr, req.ProjectID, nextRunAt, req.Concurrency, validTasks)
queue, createErr := h.batchTaskManager.CreateBatchQueue(req.Title, req.Role, agentMode, scheduleMode, cronExpr, req.ProjectID, nextRunAt, req.Concurrency, validTasks, req.HITLPolicy)
if createErr != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": createErr.Error()})
return
@@ -2118,16 +2119,21 @@ func (h *AgentHandler) PauseBatchQueue(c *gin.Context) {
func (h *AgentHandler) UpdateBatchQueueMetadata(c *gin.Context) {
queueID := c.Param("queueId")
var req struct {
Title string `json:"title"`
Role string `json:"role"`
AgentMode string `json:"agentMode"`
Concurrency *int `json:"concurrency"`
HITLPolicy *string `json:"hitlPolicy"`
Title string `json:"title"`
Role string `json:"role"`
AgentMode string `json:"agentMode"`
Concurrency *int `json:"concurrency"`
}
if err := c.ShouldBindJSON(&req); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
if err := h.batchTaskManager.UpdateQueueMetadata(queueID, req.Title, req.Role, req.AgentMode, req.Concurrency); err != nil {
var policies []string
if req.HITLPolicy != nil {
policies = append(policies, *req.HITLPolicy)
}
if err := h.batchTaskManager.UpdateQueueMetadata(queueID, req.Title, req.Role, req.AgentMode, req.Concurrency, policies...); err != nil {
c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
return
}
+28
View File
@@ -0,0 +1,28 @@
package handler
import "fmt"
// Empty policy preserves the global defaults for queues created before this setting existed.
func validateBatchHITLPolicy(policy string) error {
switch policy {
case "", "off", "human", "audit_agent", "review_edit":
return nil
default:
return fmt.Errorf("不支持的队列审批设置: %s", policy)
}
}
func (h *AgentHandler) batchHITLRequest(policy string) *HITLRequest {
req := h.hitlEffectiveDefaultRequest()
switch policy {
case "off":
req.Enabled, req.Mode = false, "off"
case "human":
req.Enabled, req.Mode, req.Reviewer = true, "approval", "human"
case "audit_agent":
req.Enabled, req.Mode, req.Reviewer = true, "approval", "audit_agent"
case "review_edit":
req.Enabled, req.Mode, req.Reviewer = true, "review_edit", "audit_agent"
}
return req
}
+121
View File
@@ -0,0 +1,121 @@
package handler
import (
"path/filepath"
"testing"
"cyberstrike-ai/internal/config"
"cyberstrike-ai/internal/database"
"go.uber.org/zap"
)
func TestBatchHITLPolicyPersistence(t *testing.T) {
db, err := database.NewDB(filepath.Join(t.TempDir(), "batch.db"), zap.NewNop())
if err != nil {
t.Fatal(err)
}
defer db.Close()
m := NewBatchTaskManager(zap.NewNop())
m.SetDB(db)
q, err := m.CreateBatchQueue("approval", "", "eino_single", "manual", "", "", nil, 1, []string{"test"}, "audit_agent")
if err != nil {
t.Fatal(err)
}
reloaded := NewBatchTaskManager(zap.NewNop())
reloaded.SetDB(db)
if err := reloaded.LoadFromDB(); err != nil {
t.Fatal(err)
}
got, ok := reloaded.GetBatchQueue(q.ID)
if !ok || got.HITLPolicy != "audit_agent" {
t.Fatalf("reload: %+v", got)
}
if err := reloaded.UpdateQueueMetadata(q.ID, "renamed", "", "", nil); err != nil {
t.Fatal(err)
}
row, err := db.GetBatchQueue(q.ID)
if err != nil || row.HITLPolicy != "audit_agent" {
t.Fatalf("unrelated edit lost policy: %+v, %v", row, err)
}
if err := reloaded.UpdateQueueMetadata(q.ID, "renamed", "", "", nil, ""); err != nil {
t.Fatal(err)
}
row, err = db.GetBatchQueue(q.ID)
if err != nil || row.HITLPolicy != "" {
t.Fatalf("reset failed: %+v, %v", row, err)
}
if err := reloaded.UpdateQueueMetadata(q.ID, "renamed", "", "", nil, "invalid"); err == nil {
t.Fatal("accepted invalid policy")
}
reloaded.UpdateTaskStatus(q.ID, got.Tasks[0].ID, BatchTaskStatusRunning, "", "")
if err := reloaded.UpdateQueueMetadata(q.ID, "renamed", "", "", nil, "off"); err == nil {
t.Fatal("changed policy during single-task execution")
}
}
func TestBatchHITLActivation(t *testing.T) {
db, err := database.NewDB(filepath.Join(t.TempDir(), "hitl.db"), zap.NewNop())
if err != nil {
t.Fatal(err)
}
defer db.Close()
timeout := 60
h := &AgentHandler{
config: &config.Config{Hitl: config.HitlConfig{
DefaultMode: "review_edit", DefaultReviewer: "audit_agent",
DefaultTimeoutSeconds: &timeout, ToolWhitelist: []string{"safe_tool"},
}},
hitlManager: NewHITLManager(db, zap.NewNop()),
}
if err := h.hitlManager.EnsureSchema(); err != nil {
t.Fatal(err)
}
for _, tc := range []struct {
policy, mode, reviewer string
enabled bool
}{
{"", "review_edit", "audit_agent", true},
{"off", "off", "audit_agent", false},
{"human", "approval", "human", true},
{"audit_agent", "approval", "audit_agent", true},
{"review_edit", "review_edit", "audit_agent", true},
} {
t.Run(tc.policy, func(t *testing.T) {
req := h.batchHITLRequest(tc.policy)
if req.Mode != tc.mode || req.Reviewer != tc.reviewer || req.Enabled != tc.enabled || req.TimeoutSeconds != timeout {
t.Fatalf("bad request: %+v", req)
}
h.activateHITLForConversation("batch-test", req)
defer h.hitlManager.DeactivateConversation("batch-test")
if h.HITLNeedsToolApproval("batch-test", "unsafe_tool") != tc.enabled {
t.Fatal("approval gate differs from policy")
}
if h.HITLNeedsToolApproval("batch-test", "safe_tool") {
t.Fatal("global whitelist lost")
}
})
}
}
func TestBatchHITLPersistenceFailure(t *testing.T) {
db, err := database.NewDB(filepath.Join(t.TempDir(), "closed.db"), zap.NewNop())
if err != nil {
t.Fatal(err)
}
m := NewBatchTaskManager(zap.NewNop())
m.SetDB(db)
q, err := m.CreateBatchQueue("test", "", "eino_single", "manual", "", "", nil, 1, []string{"test"}, "human")
if err != nil {
t.Fatal(err)
}
db.Close()
if err := m.UpdateQueueMetadata(q.ID, "changed", "", "", nil, "off"); err == nil {
t.Fatal("save failure hidden")
}
if q.HITLPolicy != "human" || q.Title != "test" {
t.Fatal("failed write changed in-memory policy")
}
if _, err := m.CreateBatchQueue("test", "", "eino_single", "manual", "", "", nil, 1, []string{"test"}, "audit_agent"); err == nil {
t.Fatal("create failure hidden")
}
}
+22
View File
@@ -220,6 +220,28 @@ func (h *AgentHandler) executeOneBatchSubTask(queueID string, queue *BatchTaskQu
registered = true
h.batchTaskManager.SetTaskCancel(queueID, task.ID, timeoutCancel)
if err := validateBatchHITLPolicy(queue.HITLPolicy); err != nil {
finishStatus = "failed"
h.batchTaskManager.UpdateTaskStatus(queueID, task.ID, BatchTaskStatusFailed, "", err.Error())
return
}
if h.hitlManager == nil {
finishStatus = "failed"
h.batchTaskManager.UpdateTaskStatus(queueID, task.ID, BatchTaskStatusFailed, "", "审批服务未初始化")
return
}
hitlReq := h.batchHITLRequest(queue.HITLPolicy)
if err := h.hitlManager.SaveConversationConfig(conversationID, hitlReq); err != nil {
finishStatus = "failed"
h.batchTaskManager.UpdateTaskStatus(queueID, task.ID, BatchTaskStatusFailed, "", "保存审批设置失败: "+err.Error())
return
}
h.activateHITLForConversation(conversationID, hitlReq)
defer h.hitlManager.DeactivateConversation(conversationID)
taskCtx = multiagent.WithHITLToolInterceptor(taskCtx, func(ctx context.Context, toolName, arguments string) (string, error) {
return h.interceptHITLForEinoTool(ctx, cancelWithCause, conversationID, assistantMessageID, sendEvent, toolName, arguments)
})
progressCallback := h.createProgressCallback(taskCtx, cancelWithCause, conversationID, assistantMessageID, sendEvent)
taskCtx = mcp.WithMCPConversationID(taskCtx, conversationID)
taskCtx = mcp.WithToolRunRegistry(taskCtx, h.tasks)
+39 -13
View File
@@ -74,8 +74,9 @@ type BatchTaskQueue struct {
ID string `json:"id"`
Title string `json:"title,omitempty"`
Role string `json:"role,omitempty"` // 角色名称(空字符串表示默认角色)
AgentMode string `json:"agentMode"` // single | eino_single | deep | plan_execute | supervisor
ScheduleMode string `json:"scheduleMode"` // manual | cron
HITLPolicy string `json:"hitlPolicy"`
AgentMode string `json:"agentMode"` // single | eino_single | deep | plan_execute | supervisor
ScheduleMode string `json:"scheduleMode"` // manual | cron
CronExpr string `json:"cronExpr,omitempty"`
NextRunAt *time.Time `json:"nextRunAt,omitempty"`
ScheduleEnabled bool `json:"scheduleEnabled"`
@@ -185,7 +186,15 @@ func (m *BatchTaskManager) CreateBatchQueue(
nextRunAt *time.Time,
concurrency int,
tasks []string,
hitlPolicies ...string,
) (*BatchTaskQueue, error) {
policy := ""
if len(hitlPolicies) > 0 {
policy = hitlPolicies[0]
}
if err := validateBatchHITLPolicy(policy); err != nil {
return nil, err
}
// 输入校验
if utf8.RuneCountInString(title) > MaxBatchQueueTitleLen {
return nil, fmt.Errorf("标题不能超过 %d 个字符", MaxBatchQueueTitleLen)
@@ -203,6 +212,7 @@ func (m *BatchTaskManager) CreateBatchQueue(
queueID := time.Now().Format("20060102150405") + "-" + generateShortID()
queue := &BatchTaskQueue{
ID: queueID,
HITLPolicy: policy,
Title: title,
Role: role,
ProjectID: strings.TrimSpace(projectID),
@@ -255,8 +265,9 @@ func (m *BatchTaskManager) CreateBatchQueue(
queue.ProjectID,
queue.Concurrency,
dbTasks,
policy,
); err != nil {
m.logger.Warn("batch queue DB create failed", zap.String("queueId", queueID), zap.Error(err))
return nil, fmt.Errorf("保存任务队列失败: %w", err)
}
}
@@ -305,6 +316,7 @@ func (m *BatchTaskManager) loadQueueFromDB(queueID string) *BatchTaskQueue {
queue := &BatchTaskQueue{
ID: queueRow.ID,
HITLPolicy: queueRow.HITLPolicy,
AgentMode: "eino_single",
ScheduleMode: "manual",
Status: queueRow.Status,
@@ -549,6 +561,7 @@ func (m *BatchTaskManager) LoadFromDB() error {
queue := &BatchTaskQueue{
ID: queueRow.ID,
HITLPolicy: queueRow.HITLPolicy,
AgentMode: "eino_single",
ScheduleMode: "manual",
Status: queueRow.Status,
@@ -743,7 +756,7 @@ func batchQueueConcurrencyFromRow(row *database.BatchTaskQueueRow) int {
}
// UpdateQueueMetadata 更新队列标题、角色、代理模式和并发数(非 running 时可用)
func (m *BatchTaskManager) UpdateQueueMetadata(queueID, title, role, agentMode string, concurrency *int) error {
func (m *BatchTaskManager) UpdateQueueMetadata(queueID, title, role, agentMode string, concurrency *int, hitlPolicies ...string) error {
if utf8.RuneCountInString(title) > MaxBatchQueueTitleLen {
return fmt.Errorf("标题不能超过 %d 个字符", MaxBatchQueueTitleLen)
}
@@ -761,6 +774,21 @@ func (m *BatchTaskManager) UpdateQueueMetadata(queueID, title, role, agentMode s
return fmt.Errorf("队列正在运行中,无法修改")
}
policy := queue.HITLPolicy
if len(hitlPolicies) > 0 {
if !queueAllowsTaskListMutationLocked(queue) {
return fmt.Errorf("队列有正在执行的任务,无法修改审批设置")
}
policy = hitlPolicies[0]
if err := validateBatchHITLPolicy(policy); err != nil {
return err
}
}
nextConcurrency := queue.Concurrency
if concurrency != nil {
nextConcurrency = normalizeBatchQueueConcurrency(*concurrency)
}
// 如果未传 agentMode,保留原值
if strings.TrimSpace(agentMode) != "" {
agentMode = config.NormalizeAgentMode(agentMode)
@@ -768,18 +796,16 @@ func (m *BatchTaskManager) UpdateQueueMetadata(queueID, title, role, agentMode s
agentMode = queue.AgentMode
}
if m.db != nil {
if err := m.db.UpdateBatchQueueMetadata(queueID, title, role, agentMode, nextConcurrency, policy); err != nil {
return fmt.Errorf("保存任务队列失败: %w", err)
}
}
queue.Title = title
queue.Role = role
queue.AgentMode = agentMode
if concurrency != nil {
queue.Concurrency = normalizeBatchQueueConcurrency(*concurrency)
}
if m.db != nil {
if err := m.db.UpdateBatchQueueMetadata(queueID, title, role, agentMode, queue.Concurrency); err != nil {
m.logger.Warn("batch queue DB metadata update failed", zap.String("queueId", queueID), zap.Error(err))
}
}
queue.Concurrency = nextConcurrency
queue.HITLPolicy = policy
return nil
}