feat(canvas): Agent 审批 - 节点直接生成时自动关闭同节点的待审批任务
用户在画布节点上直接提交生成后，同画布 Agent 对该节点等待中的 generate_media 审批关闭为 superseded_by_node，Agent 带着「用户已手动提交」的结果继续，不重复提交、不扣费。审批卡片识别目标节点正在生成时禁用「同意执行」并提示原因。
This commit is contained in:
ddcat666 committed
1 parent
8586565272
commit
354d86a70b
9 files changed
+411
-10
No files matched your search
@@ -0,0 +1,159 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
)
|
||||
|
||||
// cloudAgentApprovalSuperseded 是审批的第三种决定：用户没有点 Agent 卡片里的
|
||||
// “同意/拒绝”，而是直接在画布节点上提交了同一节点的生成。Agent 不再提交自己的
|
||||
// 任务、不扣费，也不接管该任务的结果回写（画布自己的生成链路负责回写）。
|
||||
const cloudAgentApprovalSuperseded = "superseded_by_node"
|
||||
|
||||
// cloudAgentDecisionConflict 给重复/冲突的审批决定一个用户能看懂的原因。
|
||||
func cloudAgentDecisionConflict(previous string) error {
|
||||
if previous == cloudAgentApprovalSuperseded {
|
||||
return creationConflict("你已在画布节点上直接生成，本次审批已自动关闭")
|
||||
}
|
||||
return creationConflict("该审批已有不同决定")
|
||||
}
|
||||
|
||||
// supersedeCloudAgentApprovalsForNodeTask 在普通画布生成任务创建成功后调用：
|
||||
// 如果同一画布上有 Agent 正在等待对同一节点的 generate_media 审批，就把该审批
|
||||
// 关闭为 superseded_by_node，并让 Agent 带着“用户已手动提交”的工具结果继续。
|
||||
//
|
||||
// 这是尽力而为的联动：任何失败只记日志，不影响用户已经创建成功的任务。
|
||||
func (s *Service) supersedeCloudAgentApprovalsForNodeTask(userID string, task *model.Task, input map[string]any) {
|
||||
if s == nil || s.repo == nil || task == nil || task.AgentRunID != "" || strings.TrimSpace(task.ProjectID) == "" {
|
||||
return
|
||||
}
|
||||
metadata, _ := input["metadata"].(map[string]any)
|
||||
nodeID := strings.TrimSpace(stringValue(metadata["nodeId"]))
|
||||
if nodeID == "" || stringValue(metadata["source"]) == "cloud_agent" {
|
||||
return
|
||||
}
|
||||
// 批量图片等场景结果落在新节点上，sourceNodeId 才是用户点击生成的节点。
|
||||
nodeIDs := []string{nodeID}
|
||||
if source := strings.TrimSpace(stringValue(metadata["sourceNodeId"])); source != "" && source != nodeID {
|
||||
nodeIDs = append(nodeIDs, source)
|
||||
}
|
||||
runIDs, err := s.repo.CloudAgentIDsWaitingApprovalForCanvas(userID, task.ProjectID)
|
||||
if err != nil {
|
||||
slog.Warn("agent approval supersede lookup failed", "canvas", task.ProjectID, "task", task.ID, "error", err)
|
||||
return
|
||||
}
|
||||
for _, runID := range runIDs {
|
||||
target, err := s.supersedeCloudAgentApproval(userID, runID, nodeIDs, task)
|
||||
if err != nil {
|
||||
slog.Warn("agent approval supersede failed", "run", runID, "task", task.ID, "error", err)
|
||||
continue
|
||||
}
|
||||
if target != "" {
|
||||
s.resumeCloudAgentAfterSupersede(userID, runID, target, task.ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// cloudAgentApprovalTargetNode 返回待审批 generate_media 调用的目标节点。
|
||||
// 其他工具（批量、删除、计划等）不参与联动。
|
||||
func cloudAgentApprovalTargetNode(approval *cloudAgentApproval) (string, bool) {
|
||||
if approval == nil || approval.Decision != "" || approval.Call.Function.Name != "generate_media" {
|
||||
return "", false
|
||||
}
|
||||
var args struct {
|
||||
NodeID string `json:"nodeId"`
|
||||
}
|
||||
if err := json.Unmarshal([]byte(approval.Call.Function.Arguments), &args); err != nil {
|
||||
return "", false
|
||||
}
|
||||
nodeID := strings.TrimSpace(args.NodeID)
|
||||
return nodeID, nodeID != ""
|
||||
}
|
||||
|
||||
// supersedeCloudAgentApproval 返回被关闭审批的目标节点；不匹配时返回空串。
|
||||
// 与 DecideCloudAgentApproval 一样按 revision CAS 重试：两边同时点击时先落库者生效，
|
||||
// 后到的一方看到状态已变化直接退出，不会重复提交任务。
|
||||
func (s *Service) supersedeCloudAgentApproval(userID, runID string, nodeIDs []string, task *model.Task) (string, error) {
|
||||
for attempt := 0; attempt < 8; attempt++ {
|
||||
if attempt > 0 {
|
||||
time.Sleep(time.Duration(attempt) * 20 * time.Millisecond)
|
||||
}
|
||||
run, err := s.repo.CloudAgent(userID, runID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if run.Status != "waiting_approval" {
|
||||
return "", nil
|
||||
}
|
||||
state, err := cloudAgentDecodeForExecution(run)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
target, ok := cloudAgentApprovalTargetNode(state.Approval)
|
||||
if !ok || !cloudAgentContainsString(nodeIDs, target) || state.CallIndex >= len(state.Calls) {
|
||||
return "", nil
|
||||
}
|
||||
approvalID := state.Approval.ID
|
||||
if _, decided := state.Decisions[approvalID]; decided {
|
||||
return "", nil
|
||||
}
|
||||
call := state.Calls[state.CallIndex]
|
||||
s.storageMu.Lock()
|
||||
err = s.repo.MutateCloudAgent(userID, runID, run.Revision, func(current *model.CloudAgentExecution, repo *repository.Repository) error {
|
||||
applyCloudAgentApprovalSupersede(current, &state, call, approvalID, target, task.ID)
|
||||
if err := repo.ReleaseCloudAgentResourceLeases(userID, approvalID); err != nil {
|
||||
return err
|
||||
}
|
||||
return cloudAgentSave(current, &state)
|
||||
})
|
||||
s.storageMu.Unlock()
|
||||
if errors.Is(err, repository.ErrCreationConflict) {
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return target, nil
|
||||
}
|
||||
return "", repository.ErrCreationConflict
|
||||
}
|
||||
|
||||
// applyCloudAgentApprovalSupersede 在同一个检查点里记录决定、关闭审批，并把
|
||||
// “用户已手动提交”作为本次工具调用的结果写入历史（同时推进到下一个调用）。
|
||||
func applyCloudAgentApprovalSupersede(current *model.CloudAgentExecution, state *cloudAgentRuntime, call cloudAgentCall, approvalID, nodeID, taskID string) {
|
||||
state.Decisions[approvalID] = cloudAgentApprovalSuperseded
|
||||
state.Approval = nil
|
||||
state.event(current.ID, "approval_decided", map[string]any{
|
||||
"approvalId": approvalID,
|
||||
"decision": cloudAgentApprovalSuperseded,
|
||||
"nodeId": nodeID,
|
||||
"taskId": taskID,
|
||||
"text": "你已在画布节点上直接提交了生成，本次审批已自动关闭；Agent 不会重复提交或扣费，结果由画布回写。",
|
||||
})
|
||||
cloudAgentRecordToolResult(current, state, call, map[string]any{
|
||||
"status": "superseded_by_user",
|
||||
"taskSubmitted": false,
|
||||
"nodeId": nodeID,
|
||||
"userTaskId": taskID,
|
||||
"summary": fmt.Sprintf("用户已在画布节点 %s 上直接提交该生成（taskId=%s）。Agent 未提交任务、未扣费；不要再为该节点调用 generate_media，结果由画布自动回写。", nodeID, taskID),
|
||||
}, nil)
|
||||
if current.Status == "waiting_approval" {
|
||||
current.Status = "running"
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Service) resumeCloudAgentAfterSupersede(userID, runID, nodeID, taskID string) {
|
||||
prompt := fmt.Sprintf("用户没有使用审批卡片，而是直接在画布节点 %s 上提交了刚才等待审批的生成（taskId=%s）。该调用已记录为用户手动提交：Agent 没有提交任务、没有扣费，生成结果由画布回写。请继续剩余步骤，不要再次为该节点调用 generate_media。", nodeID, taskID)
|
||||
if err := s.saveCloudAgentPiResumePrompt(userID, runID, prompt); err != nil {
|
||||
slog.Warn("agent approval supersede resume prompt failed", "run", runID, "error", err)
|
||||
return
|
||||
}
|
||||
s.resumeCloudAgentPi(userID, runID)
|
||||
}
|
||||
@@ -0,0 +1,135 @@
|
||||
package app
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"infinite-canvas/backend/internal/model"
|
||||
)
|
||||
|
||||
// waitingAgentMediaApproval 把 Agent 推进到“等待生成审批”，返回运行和审批 ID。
|
||||
func waitingAgentMediaApproval(t *testing.T, s *Service, a cloudAgentMediaArgs) (*model.CloudAgentExecution, string) {
|
||||
t.Helper()
|
||||
run, _ := agentMediaRun(t, s, a, "request_approval")
|
||||
if err := s.advanceCloudAgentByID("user", run.ID); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
run, _ = s.repo.CloudAgent("user", run.ID)
|
||||
state, err := cloudAgentDecode(run)
|
||||
if err != nil || run.Status != "waiting_approval" || state.Approval == nil {
|
||||
t.Fatalf("expected waiting approval: status=%s err=%v", run.Status, err)
|
||||
}
|
||||
return run, state.Approval.ID
|
||||
}
|
||||
|
||||
// userNodeTaskRequest 复用 Agent 同一份生成请求，但以普通画布节点生成的身份提交。
|
||||
func userNodeTaskRequest(t *testing.T, s *Service, run *model.CloudAgentExecution, metadata map[string]any) CreateTaskRequest {
|
||||
t.Helper()
|
||||
state, _ := cloudAgentDecode(run)
|
||||
req, _, err := s.prepareCloudAgentMedia(run, &state, state.Calls[state.CallIndex])
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
req.Input["metadata"] = metadata
|
||||
return req
|
||||
}
|
||||
|
||||
func TestNodeGenerationSupersedesMatchingAgentApproval(t *testing.T) {
|
||||
s, db, a := agentMediaFixture(t)
|
||||
run, approvalID := waitingAgentMediaApproval(t, s, a)
|
||||
req := userNodeTaskRequest(t, s, run, map[string]any{"nodeId": a.NodeID, "source": "canvas"})
|
||||
task, err := s.CreateTask("user", req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
run, _ = s.repo.CloudAgent("user", run.ID)
|
||||
state, err := cloudAgentDecode(run)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if run.Status != "running" || state.Approval != nil || state.Decisions[approvalID] != cloudAgentApprovalSuperseded || state.CallIndex != 1 {
|
||||
t.Fatalf("approval not superseded: status=%s approval=%+v decision=%q index=%d", run.Status, state.Approval, state.Decisions[approvalID], state.CallIndex)
|
||||
}
|
||||
decided := false
|
||||
for _, event := range state.Events {
|
||||
if event.Type == "approval_decided" && event.Payload["decision"] == cloudAgentApprovalSuperseded {
|
||||
decided = event.Payload["approvalId"] == approvalID && event.Payload["taskId"] == task.ID && event.Payload["nodeId"] == a.NodeID
|
||||
}
|
||||
}
|
||||
if !decided {
|
||||
t.Fatal("missing superseded approval_decided event")
|
||||
}
|
||||
content, ok := cloudAgentToolMessage(&state, "media-call")
|
||||
if !ok || !strings.Contains(content, "superseded_by_user") || !strings.Contains(content, task.ID) || cloudAgentToolContentIsError(content) {
|
||||
t.Fatalf("tool result does not tell the Agent the user submitted: %q", content)
|
||||
}
|
||||
if state.MediaTaskID != "" || state.Generations != 0 {
|
||||
t.Fatalf("Agent took over the user task: media=%s generations=%d", state.MediaTaskID, state.Generations)
|
||||
}
|
||||
var count, leases int64
|
||||
db.Model(&model.Task{}).Where("type = ?", "canvas_video").Count(&count)
|
||||
db.Model(&model.CloudAgentResourceLease{}).Where("owner_id = ?", approvalID).Count(&leases)
|
||||
if count != 1 || leases != 0 {
|
||||
t.Fatalf("expected only the user task and released leases: tasks=%d leases=%d", count, leases)
|
||||
}
|
||||
|
||||
// 卡片上的“同意/拒绝”晚到时不能再提交任务，并给出可读原因。
|
||||
for _, decision := range []string{"approve", "reject"} {
|
||||
err = s.DecideCloudAgentApproval("user", run.ID, approvalID, decision, "")
|
||||
if err == nil || !strings.Contains(err.Error(), "直接生成") {
|
||||
t.Fatalf("late %s must be rejected with a superseded reason: %v", decision, err)
|
||||
}
|
||||
}
|
||||
db.Model(&model.Task{}).Where("type = ?", "canvas_video").Count(&count)
|
||||
if count != 1 {
|
||||
t.Fatalf("late approval submitted a duplicate task: %d", count)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNodeGenerationLeavesUnrelatedAgentApprovals(t *testing.T) {
|
||||
for name, metadata := range map[string]map[string]any{
|
||||
"other node": {"nodeId": "hero", "source": "canvas"},
|
||||
"no node": {"source": "canvas"},
|
||||
"agent origin": {"nodeId": "video-shot-1", "source": "cloud_agent"},
|
||||
} {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
s, _, a := agentMediaFixture(t)
|
||||
run, approvalID := waitingAgentMediaApproval(t, s, a)
|
||||
state, _ := cloudAgentDecode(run)
|
||||
s.supersedeCloudAgentApprovalsForNodeTask("user", &model.Task{ID: "user-task", ProjectID: "agent-canvas"}, map[string]any{"metadata": metadata})
|
||||
if name == "agent origin" {
|
||||
// 显式带 AgentRunID 的任务同样跳过。
|
||||
s.supersedeCloudAgentApprovalsForNodeTask("user", &model.Task{ID: "agent-task", ProjectID: "agent-canvas", AgentRunID: run.ID}, map[string]any{"metadata": map[string]any{"nodeId": a.NodeID}})
|
||||
}
|
||||
// 其他画布上的同名节点也不能关闭本画布的审批。
|
||||
s.supersedeCloudAgentApprovalsForNodeTask("user", &model.Task{ID: "other-canvas-task", ProjectID: "other-canvas"}, map[string]any{"metadata": map[string]any{"nodeId": a.NodeID}})
|
||||
after, _ := s.repo.CloudAgent("user", run.ID)
|
||||
afterState, _ := cloudAgentDecode(after)
|
||||
if after.Status != "waiting_approval" || afterState.Approval == nil || afterState.Approval.ID != approvalID || afterState.CallIndex != state.CallIndex {
|
||||
t.Fatalf("unrelated task changed the approval: status=%s approval=%+v", after.Status, afterState.Approval)
|
||||
}
|
||||
if err := s.DecideCloudAgentApproval("user", run.ID, approvalID, "reject", ""); err != nil {
|
||||
t.Fatalf("approval should still be decidable: %v", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNodeGenerationSupersedeViaSourceNode(t *testing.T) {
|
||||
s, _, a := agentMediaFixture(t)
|
||||
run, approvalID := waitingAgentMediaApproval(t, s, a)
|
||||
// 批量图片的结果落在新子节点，sourceNodeId 才是被审批的那个节点。
|
||||
s.supersedeCloudAgentApprovalsForNodeTask("user", &model.Task{ID: "batch-child-task", ProjectID: "agent-canvas"}, map[string]any{"metadata": map[string]any{"nodeId": "batch-child", "sourceNodeId": a.NodeID}})
|
||||
after, _ := s.repo.CloudAgent("user", run.ID)
|
||||
state, _ := cloudAgentDecode(after)
|
||||
if state.Decisions[approvalID] != cloudAgentApprovalSuperseded || state.Approval != nil {
|
||||
t.Fatalf("sourceNodeId did not supersede the approval: %+v", state.Approval)
|
||||
}
|
||||
// 重复通知是幂等的：不会再写第二条工具结果。
|
||||
s.supersedeCloudAgentApprovalsForNodeTask("user", &model.Task{ID: "batch-child-task-2", ProjectID: "agent-canvas"}, map[string]any{"metadata": map[string]any{"nodeId": a.NodeID}})
|
||||
again, _ := s.repo.CloudAgent("user", run.ID)
|
||||
if again.Revision != after.Revision {
|
||||
t.Fatalf("repeated supersede mutated the run: %d -> %d", after.Revision, again.Revision)
|
||||
}
|
||||
}
|
||||
@@ -2449,7 +2449,7 @@ func (s *Service) DecideCloudAgentApproval(userID, id, approvalID, decision, rea
|
||||
if previous == decision && (settings == nil || state.DecisionSettings[approvalID] == creationHash(settings)) {
|
||||
return nil
|
||||
}
|
||||
return creationConflict("该审批已有不同决定")
|
||||
return cloudAgentDecisionConflict(previous)
|
||||
}
|
||||
if run.Status != "waiting_approval" || state.Approval == nil || state.Approval.ID != approvalID {
|
||||
return creationConflict("审批不存在或已过期")
|
||||
@@ -2472,7 +2472,7 @@ func (s *Service) DecideCloudAgentApproval(userID, id, approvalID, decision, rea
|
||||
if previous == decision && (settings == nil || state.DecisionSettings[approvalID] == creationHash(settings)) {
|
||||
return nil
|
||||
}
|
||||
return creationConflict("该审批已有不同决定")
|
||||
return cloudAgentDecisionConflict(previous)
|
||||
}
|
||||
if run.Status != "waiting_approval" || state.Approval == nil || state.Approval.ID != approvalID {
|
||||
return creationConflict("审批不存在或已过期")
|
||||
|
||||
@@ -209,6 +209,10 @@ func (s *Service) CreateTask(userID string, req CreateTaskRequest) (*model.Task,
|
||||
}
|
||||
s.recordActivity(userID, "task", 1)
|
||||
_ = s.log(userID, task.ID, "info", "任务已进入队列", "")
|
||||
if req.admission == nil {
|
||||
// 用户直接在节点上生成时，关闭 Agent 对同一节点仍在等待的生成审批。
|
||||
s.supersedeCloudAgentApprovalsForNodeTask(userID, &task, normalizedInput)
|
||||
}
|
||||
return taskForOutput(task), nil
|
||||
}
|
||||
|
||||
|
||||
@@ -149,6 +149,17 @@ func (r *Repository) CloudAgentForActiveTask(userID, taskID string) (*model.Clou
|
||||
})
|
||||
return &run, err
|
||||
}
|
||||
|
||||
// CloudAgentIDsWaitingApprovalForCanvas lists the caller's runs paused on an
|
||||
// approval for one canvas. Callers reload each run before mutating it.
|
||||
func (r *Repository) CloudAgentIDsWaitingApprovalForCanvas(userID, canvasID string) ([]string, error) {
|
||||
var ids []string
|
||||
err := r.db.Model(&model.CloudAgentExecution{}).
|
||||
Where("user_id = ? AND canvas_id = ? AND status = ?", userID, canvasID, "waiting_approval").
|
||||
Order("id").Limit(20).Pluck("id", &ids).Error
|
||||
return ids, err
|
||||
}
|
||||
|
||||
func (r *Repository) CloudAgentRoots() ([]model.Task, error) {
|
||||
var tasks []model.Task
|
||||
err := r.db.Where("operation = ? AND id NOT IN (SELECT id FROM cloud_agent_executions)", "cloud_agent").Order("created_at").Limit(50).Find(&tasks).Error
|
||||
|
||||
Reference in new issue
Block a user