package app import ( "crypto/sha256" "encoding/hex" "encoding/json" "errors" "math" "strings" "time" "unicode" "unicode/utf8" "gorm.io/gorm" "infinite-canvas/backend/internal/kernel" "infinite-canvas/backend/internal/model" ) const cloudAgentOperation = "cloud_agent" // A run is one immutable, durable model turn backed by a normal task. Subsequent // turns reference the previous run, not a mutable in-memory conversation. This // reuses transactional billing, worker leases, cancellation and text replay. type CloudAgentRequest struct { ReasoningMode string `json:"reasoningMode,omitempty"` ProfileRevision string `json:"profileRevision,omitempty"` CanvasID string `json:"canvasId"` Prompt string `json:"prompt"` Model string `json:"model,omitempty"` LogicalModelID string `json:"logicalModelId,omitempty"` ChannelID string `json:"channelId,omitempty"` ChannelModelKey string `json:"channelModelKey,omitempty"` PermissionMode string `json:"permissionMode"` SkillIDs []string `json:"skillIds,omitempty"` ContextScope []string `json:"contextScope"` Budget struct { MaxCredits float64 `json:"maxCredits"` MaxGenerationTasks int `json:"maxGenerationTasks,omitempty"` MaxVideoSeconds int `json:"maxVideoSeconds,omitempty"` } `json:"budget"` IdempotencyKey string `json:"idempotencyKey"` } type cloudAgentState struct { Version int `json:"version"` Request CloudAgentRequest `json:"request"` ParentID string `json:"parentId"` Fingerprint string `json:"fingerprint"` Skills []cloudAgentSkill `json:"skills,omitempty"` Profile cloudAgentProfileSnapshot `json:"profile"` Policy cloudAgentPolicySnapshot `json:"policy"` } type CloudAgentRun struct { ID string `json:"id"` CanvasID string `json:"canvasId"` ParentID string `json:"parentId,omitempty"` Status string `json:"status"` Revision int64 `json:"revision"` CleanupPending bool `json:"cleanupPending,omitempty"` FailureMessage string `json:"failureMessage,omitempty"` PermissionMode string `json:"permissionMode"` Model string `json:"model"` CreatedAt time.Time `json:"createdAt"` UpdatedAt time.Time `json:"updatedAt"` Events []CloudAgentEvent `json:"events,omitempty"` Skills []cloudAgentSkill `json:"skills,omitempty"` Approval *cloudAgentApproval `json:"approval,omitempty"` SpentCredits float64 `json:"spentCredits"` Step int `json:"step"` ActiveMessage map[string]string `json:"activeMessage,omitempty"` } func validateCloudAgentRequest(req *CloudAgentRequest) error { if req == nil { return BadAuthRequest("请求不能为空") } // IDs and protocol selectors are identifiers, not free-form text. Keep their // validation in one place so byte/rune and Unicode handling cannot drift. if err := validateCloudAgentID(req.CanvasID, "画布 ID", 80); err != nil { return err } req.CanvasID = strings.TrimSpace(req.CanvasID) if err := validateCloudAgentPrompt(req.Prompt, 16000); err != nil { return err } req.Prompt = strings.TrimSpace(req.Prompt) if err := validateCloudAgentID(req.IdempotencyKey, "幂等键", 128); err != nil || utf8.RuneCountInString(req.IdempotencyKey) < 8 { return BadAuthRequest("需要 8–128 个字符的幂等键") } if req.PermissionMode != "read_only" && req.PermissionMode != "request_approval" && req.PermissionMode != "auto" { return BadAuthRequest("无效的 Agent 执行权限") } for value, spec := range map[string]struct { label string limit int }{ "model": {"模型标识", 160}, "logicalModelId": {"逻辑模型 ID", 80}, "channelId": {"渠道 ID", 80}, "channelModelKey": {"渠道模型标识", 160}, } { if value == "model" && req.Model == "" || value == "logicalModelId" && req.LogicalModelID == "" || value == "channelId" && req.ChannelID == "" || value == "channelModelKey" && req.ChannelModelKey == "" { continue } var candidate string switch value { case "model": candidate = req.Model case "logicalModelId": candidate = req.LogicalModelID case "channelId": candidate = req.ChannelID case "channelModelKey": candidate = req.ChannelModelKey } if err := validateCloudAgentID(candidate, spec.label, spec.limit); err != nil { return err } } if req.ReasoningMode != "" && req.ReasoningMode != "off" && req.ReasoningMode != "auto" && req.ReasoningMode != "deep" { return BadAuthRequest("无效的 Agent 推理模式") } if req.ProfileRevision != "" { if err := validateCloudAgentID(req.ProfileRevision, "偏好版本", 120); err != nil { return err } } if req.LogicalModelID != "" { if req.ChannelID != "" || req.ChannelModelKey != "" { return BadAuthRequest("逻辑模型和系统渠道不能混用") } } else if req.ChannelID == "" || req.ChannelModelKey == "" { return BadAuthRequest("请选择后端受管文本模型;Agent 不接受浏览器自定义密钥或上游地址") } else if req.Model != "" && req.Model != req.ChannelModelKey { return BadAuthRequest("渠道模型标识与 model 不一致") } if req.Budget.MaxGenerationTasks < 0 || req.Budget.MaxVideoSeconds < 0 { return BadAuthRequest("生成任务和视频秒数预算不能为负数") } seen := map[string]bool{} for i, id := range req.SkillIDs { id = strings.TrimSpace(id) if err := validateCloudAgentID(id, "技能 ID", 80); err != nil || seen[id] { return BadAuthRequest("技能 ID 无效或重复") } req.SkillIDs[i] = id seen[id] = true } if req.PermissionMode == "read_only" && (req.Budget.MaxGenerationTasks != 0 || req.Budget.MaxVideoSeconds != 0) { return BadAuthRequest("只读模式不能设置生成预算") } if len(req.ContextScope) > 1 || (len(req.ContextScope) == 1 && req.ContextScope[0] != "canvas") { return BadAuthRequest("当前仅支持已保存画布摘要,其他上下文尚未开放") } if len(req.ContextScope) == 1 && req.ContextScope[0] == "canvas" { req.ContextScope[0] = "canvas" } if math.IsNaN(req.Budget.MaxCredits) || math.IsInf(req.Budget.MaxCredits, 0) || req.Budget.MaxCredits <= 0 || req.Budget.MaxCredits > 1000000 { return BadAuthRequest("本轮积分上限必须大于 0 且不超过 1000000") } return nil } func validateCloudAgentPrompt(value string, maxRunes int) error { if !utf8.ValidString(value) || strings.TrimSpace(value) == "" || utf8.RuneCountInString(strings.TrimSpace(value)) > maxRunes { return BadAuthRequest("需要有效画布和 1–16000 个字符的提示词") } for _, r := range value { if unicode.IsControl(r) && r != '\n' && r != '\r' && r != '\t' { return BadAuthRequest("提示词不能包含控制字符") } } return nil } func cloudAgentID(userID, key string) string { sum := sha256.Sum256([]byte(userID + "\x00" + key)) return "ag" + hex.EncodeToString(sum[:16]) } func cloudAgentFingerprint(req CloudAgentRequest, parent string) string { data, _ := json.Marshal(struct { Request CloudAgentRequest Parent string }{req, parent}) sum := sha256.Sum256(data) return hex.EncodeToString(sum[:]) } func agentRunOutput(task *model.Task, state cloudAgentState) *CloudAgentRun { status := string(task.Status) if task.Status == model.TaskStatusSucceeded { status = "completed" } return &CloudAgentRun{ID: task.ID, CanvasID: task.ProjectID, ParentID: state.ParentID, Status: status, PermissionMode: state.Request.PermissionMode, Model: task.Model, CreatedAt: task.CreatedAt, UpdatedAt: task.UpdatedAt, Skills: state.Skills} } func (s *Service) cloudAgentTask(userID, id string) (*model.Task, cloudAgentState, error) { var input struct { Agent cloudAgentState `json:"cloudAgent"` } task, err := s.repo.TaskForUser(userID, id) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { err = kernel.NotFound("Agent 运行不存在") } return nil, input.Agent, err } if task.Operation != cloudAgentOperation { return nil, input.Agent, kernel.NotFound("Agent 运行不存在") } if json.Unmarshal([]byte(task.InputJSON), &input) == nil && input.Agent.Version == 1 && task.ID == cloudAgentID(userID, input.Agent.Request.IdempotencyKey) { return task, input.Agent, nil } // 成功、失败和取消任务的 InputJSON 都可能按存储配额策略压缩;执行 // 记录才是完成后轮次的持久来源。运行中如果无法解析执行记录,不能 // 返回一个可操作的审批/活动任务状态。 run, err := s.repo.CloudAgent(userID, id) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { err = kernel.NotFound("Agent 运行不存在") } return nil, input.Agent, err } state, err := cloudAgentDecode(run) if err != nil { if cloudAgentTaskTerminal(task.Status) || cloudAgentRunTerminal(run.Status) { // 终态运行仍应可查询,但绝不从损坏的 runtime 伪造审批、权限、 // 活动任务或技能内容。TaskForUser 已完成归属校验,画布 ID 仅 // 使用任务自身的归属字段作为安全展示降级。 return task, cloudAgentState{Version: 1, Request: CloudAgentRequest{CanvasID: task.ProjectID}}, nil } return nil, input.Agent, kernel.NotFound("Agent 运行不存在") } if task.ID != cloudAgentID(userID, state.Request.IdempotencyKey) || task.ProjectID != state.Request.CanvasID { return nil, input.Agent, kernel.NotFound("Agent 运行不存在") } input.Agent = cloudAgentState{Version: 1, Request: state.Request, ParentID: state.ParentID, Fingerprint: state.Fingerprint, Skills: state.Skills, Profile: state.Profile, Policy: state.Policy} return task, input.Agent, nil } func cloudAgentTaskTerminal(status model.TaskStatus) bool { return status == model.TaskStatusSucceeded || status == model.TaskStatusFailed || status == model.TaskStatusCancelled } func cloudAgentRunTerminal(status string) bool { return status == "completed" || status == "failed" || status == "cancelled" } func (s *Service) CloudAgentRun(userID, id string) (*CloudAgentRun, error) { task, state, err := s.cloudAgentTask(userID, id) if err != nil { return nil, err } // A durable execution is authoritative once it exists. In particular, do // not try to reconstruct/overwrite a terminal run whose task input or // runtime blob was compacted or damaged; the output path has a deliberately // read-only, minimal fallback for that case. The ensure path is only for a // legacy root task whose execution row has not been created yet. if _, lookupErr := s.repo.CloudAgent(userID, id); errors.Is(lookupErr, gorm.ErrRecordNotFound) { if err := s.ensureCloudAgentExecution(task, state); err != nil { return nil, err } } else if lookupErr != nil { return nil, lookupErr } return s.cloudAgentExecutionOutput(task, state) } // CloudAgentRunIfChanged keeps idle event streams on a small indexed read. // The persisted revision, not a process-local notification, is authoritative // across instances and after missed/disconnected notifications. func (s *Service) CloudAgentRunIfChanged(userID, id string, revision int64) (*CloudAgentRun, error) { current, err := s.repo.CloudAgentRevision(userID, id) if errors.Is(err, gorm.ErrRecordNotFound) { return nil, kernel.NotFound("Agent 运行不存在") } if err != nil { return nil, err } if current == revision { return nil, nil } return s.CloudAgentRun(userID, id) } // CreateCloudAgentRun validates every capability before admission. The task PK // is deterministic per user/key. Competing requests may race, but task creation // and credit reservation share a transaction: only one can commit. func (s *Service) CreateCloudAgentRun(userID string, req CloudAgentRequest, parentID string) (*CloudAgentRun, error) { if err := validateCloudAgentRequest(&req); err != nil { return nil, err } if userID == "" { return nil, kernel.Unauthorized("请先登录") } canvas, err := s.repo.CanvasProjectForUser(userID, req.CanvasID) if err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { return nil, kernel.NotFound("画布不存在或尚未保存到服务端,请先完成画布同步") } return nil, err } // Resolve and freeze the effective preference document before idempotency // lookup. A retry without an explicit revision must still refer to the same // immutable input; a changed profile therefore cannot silently create a // different run under the same key. profile, err := s.cloudAgentProfileSnapshot(userID, req.CanvasID) if err != nil { return nil, err } if req.ProfileRevision != "" && req.ProfileRevision != profile.Revision { return nil, creationConflict("Agent 偏好已变化,请重新读取后提交") } req.ProfileRevision = profile.Revision id := cloudAgentID(userID, req.IdempotencyKey) fingerprint := cloudAgentFingerprint(req, parentID) if existing, state, lookupErr := s.cloudAgentTask(userID, id); lookupErr == nil { if state.Fingerprint == "" || state.Fingerprint != fingerprint { return nil, kernel.NewAppError(409, "幂等键已用于不同请求,请使用新的幂等键") } return s.CloudAgentRun(userID, existing.ID) } else { var appErr *AppError if !errors.As(lookupErr, &appErr) || appErr.Status != 404 { return nil, lookupErr } } var history []providerTextMessage if parentID != "" { parent, _, parentErr := s.cloudAgentTask(userID, parentID) if parentErr != nil { return nil, parentErr } if parent.ProjectID != req.CanvasID { return nil, kernel.Forbidden("不能跨画布追加 Agent 消息") } if parent.Status == model.TaskStatusQueued || parent.Status == model.TaskStatusRunning { return nil, kernel.NewAppError(409, "上一轮仍在执行,请等待结束") } if err := s.advanceCloudAgentByID(userID, parentID); err != nil { return nil, err } parentRun, err := s.CloudAgentRun(userID, parentID) if err != nil { return nil, err } if !cloudAgentRunTerminal(parentRun.Status) || parentRun.CleanupPending { return nil, kernel.NewAppError(409, "上一轮 Agent 尚未结束") } parentExecution, err := s.repo.CloudAgent(userID, parentID) if err != nil { return nil, err } parentState, err := cloudAgentDecode(parentExecution) if err != nil { return nil, WrapAppError(409, "上一轮 Agent 历史记录不完整,无法继续对话;请新建对话", err) } history = parentState.TextHistory if history == nil { history = cloudAgentLegacyHistory(parentState.Canonical.Messages, parent.Prompt) } text, err := cloudAgentContinuationReply(parent, parentRun) if err != nil { return nil, err } // The user's goal survives a failed first model call too. Tool facts are // context, not authorization to replay a write or charge a second time. history = append(history, providerTextMessage{Role: "user", Content: parent.Prompt}, providerTextMessage{Role: "assistant", Content: text}) } // Bound prompt growth without silently injecting a huge canvas or transcript. if len(history) > 16 { return nil, BadAuthRequest("对话已达到 8 轮上下文上限,请新建对话") } encodedHistory, err := json.Marshal(history) if err != nil { return nil, err } if len(encodedHistory) > 64000 { return nil, BadAuthRequest("对话上下文超过 64KB,请新建对话") } skillSnapshots, err := s.cloudAgentSkills(userID, req.SkillIDs) if err != nil { return nil, err } canvasSummary := "" if len(req.ContextScope) != 0 { canvasSummary, err = cloudAgentCanvasSummary(canvas) if err != nil { return nil, err } } system, policy, err := compileCloudAgentPolicies(req, skillSnapshots, canvasSummary, profile) if err != nil { return nil, err } state := cloudAgentState{Version: 1, Request: req, ParentID: parentID, Fingerprint: fingerprint, Skills: skillSnapshots, Profile: profile, Policy: policy} canonical := cloudAgentCanonical(system, history, req.Prompt, req) input := map[string]any{"mode": "text", "prompt": req.Prompt, "textHistory": history, "textOptions": map[string]any{"stream": true, "thinking": cloudAgentReasoningEnabled(policy.ReasoningMode)}, "cloudAgent": state, "agentRequests": map[string]any{"canonical": canonical}, "config": map[string]any{"channelId": req.ChannelID, "channelModelKey": req.ChannelModelKey, "model": firstNonEmpty(req.ChannelModelKey, req.Model), "systemPrompt": system}} task, err := s.CreateTask(userID, CreateTaskRequest{ProjectID: req.CanvasID, Type: "canvas_text", Operation: cloudAgentOperation, Prompt: req.Prompt, Model: req.Model, LogicalModelID: req.LogicalModelID, Input: input, admission: &taskAdmission{ID: id, MaxCharge: int64(math.Floor(req.Budget.MaxCredits * float64(CreditScale)))}}) if err != nil { // A concurrent identical request may have won the transaction. Never // replace its result or reserve credits a second time. if existing, stored, readErr := s.cloudAgentTask(userID, id); readErr == nil { if stored.Fingerprint == "" || stored.Fingerprint != fingerprint { return nil, kernel.NewAppError(409, "幂等键已用于不同请求") } return s.CloudAgentRun(userID, existing.ID) } return nil, err } return s.CloudAgentRun(userID, task.ID) } // 旧运行记录没有单独保存历史;只从模型请求中当前用户消息之前的严格交替前缀恢复。 func cloudAgentLegacyHistory(messages []map[string]interface{}, currentPrompt string) []providerTextMessage { currentIndex := -1 for index, message := range messages { role, _ := message["role"].(string) content, _ := message["content"].(string) if role != "user" && role != "assistant" { break } if role == "user" && content == currentPrompt { currentIndex = index } } if currentIndex < 0 || currentIndex%2 != 0 { return nil } history := make([]providerTextMessage, 0, currentIndex) for index := 0; index < currentIndex; index++ { role, _ := messages[index]["role"].(string) content, ok := messages[index]["content"].(string) if !ok || role != []string{"user", "assistant"}[index%2] { return nil } history = append(history, providerTextMessage{Role: role, Content: content}) } return history } func cloudAgentCanvasSummary(canvas *model.CanvasProject) (string, error) { var payload struct { Nodes []struct { ID string `json:"id"` Type string `json:"type"` Title string `json:"title"` Metadata map[string]any `json:"metadata"` } `json:"nodes"` } if err := json.Unmarshal([]byte(canvas.PayloadJSON), &payload); err != nil { return "", BadAuthRequest("服务端画布内容无法解析,请先重新同步") } nodes := make([]map[string]any, 0) for index, node := range payload.Nodes { if index == 80 { break } descriptor, known := cloudAgentNodeCapabilityForType(node.Type) if !known { return "", BadAuthRequest("画布包含当前 Agent 不支持的节点类型") } item := map[string]any{"id": truncateRunes(node.ID, 100), "type": truncateRunes(node.Type, 40), "title": truncateRunes(node.Title, 300)} projected, err := cloudAgentProjectNodeFields(map[string]any{"title": node.Title}, node.Metadata, descriptor, descriptor.SummaryFields, 600, false, 0) if err != nil { return "", err } for key, value := range projected { item[key] = value } nodes = append(nodes, item) } data, err := json.Marshal(map[string]any{"title": truncateRunes(canvas.Title, 240), "savedAt": canvas.UpdatedAt, "totalNodes": len(payload.Nodes), "includedNodes": len(nodes), "nodes": nodes}) if err != nil { return "", err } if len(data) > 64000 { return "", BadAuthRequest("画布摘要超过 64KB,请缩小画布后重试") } return string(data), nil }