Files
ttstd-infinite-canvas/backend/internal/app/cloud_agent_canvas_events.go
AetherNo2332 6a216dddc2 perf(cloud-agent): 画布事件 - 整理类变更只下发几何增量
一次整理几十个节点时,canvas_updated 仍按整节点下发完整 before/after,单条事件会超过
validateCloudAgentRuntime 的 128KiB 单事件上限,整轮直接失败。

- 新增 cloudAgentShrinkChange:仅当 before/after 除 position/width/height/zIndex 之外
  完全一致(且无键被删)时,只下发 id 与变化的几何字段;带内容或 metadata 变化的节点仍发
  完整 before/after,不影响前端的三方合并与生成任务状态保护。
- 新增用例覆盖"几何变化只发增量"与"内容变化 / 键被删仍发完整对象"。
2026-09-22 01:32:32 +08:00

221 lines
7.7 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
package app
import (
"reflect"
"infinite-canvas/backend/internal/repository"
)
// Persist the delta in the same transaction as the canvas and run checkpoint.
// Clients can project it without fetching the entire canvas for every task.
func cloudAgentCanvasEventRecorder(runID string, state *cloudAgentRuntime) cloudAgentMutationRecorder {
return func(repo *repository.Repository, input cloudAgentMutationInput) error {
if err := cloudAgentMutationRecorderForRun(runID)(repo, input); err != nil {
return err
}
return emitCloudAgentCanvasChange(repo, runID, state, input)
}
}
func cloudAgentObjectChanges(before, after any) []map[string]any {
previous := map[string]map[string]any{}
for _, item := range creationMaps(before) {
previous[stringValue(item["id"])] = item
}
changes := []map[string]any{}
for _, item := range creationMaps(after) {
old := previous[stringValue(item["id"])]
if !reflect.DeepEqual(old, item) {
oldFields, newFields := cloudAgentShrinkChange(old, item)
changes = append(changes, map[string]any{"before": oldFields, "after": newFields})
}
}
return changes
}
// cloudAgentGeometryFields 是"只改位置/尺寸"的字段集合。
var cloudAgentGeometryFields = []string{"position", "width", "height", "zIndex"}
// cloudAgentShrinkChange 把"只动了几何"的节点变更缩成几何增量。
//
// 画布增量在前端是**三方合并**(按字段合并,未出现的字段保持本地值),所以只发发生变化的
// 几何字段既安全又必要:一次整理 50 个节点时,若每个节点都带完整正文(分镜行、Markdown
// 正文可达十几 KB),单条 canvas_updated 事件就会超过 validateCloudAgentRuntime 的
// 单事件 128KiB 上限("Agent runtime event payload is too large"),整轮直接终止。
// 只有"除几何外完全一致"时才缩:带内容/metadata 变化的节点仍发完整 before/after,
// 以免影响前端对生成任务状态的保护逻辑。
func cloudAgentShrinkChange(before, after map[string]any) (map[string]any, map[string]any) {
if before == nil || after == nil {
return before, after
}
for key, value := range after {
if cloudAgentContainsString(cloudAgentGeometryFields, key) {
continue
}
if !reflect.DeepEqual(before[key], value) {
return before, after
}
}
for key, value := range before {
if cloudAgentContainsString(cloudAgentGeometryFields, key) {
continue
}
if _, exists := after[key]; !exists && value != nil {
return before, after
}
}
shrunkBefore := map[string]any{"id": after["id"]}
shrunkAfter := map[string]any{"id": after["id"]}
for _, key := range cloudAgentGeometryFields {
final, hasFinal := after[key]
initial, hasInitial := before[key]
if !hasFinal && !hasInitial {
continue
}
if reflect.DeepEqual(initial, final) {
continue
}
if hasInitial {
shrunkBefore[key] = initial
}
if hasFinal {
shrunkAfter[key] = final
}
}
return shrunkBefore, shrunkAfter
}
func emitCloudAgentCanvasChange(repo *repository.Repository, runID string, state *cloudAgentRuntime, input cloudAgentMutationInput) error {
before, err := creationDocument(input.BeforeJSON)
if err != nil {
return err
}
canvas, err := repo.CanvasProjectForUser(input.UserID, input.CanvasID)
if err != nil {
return err
}
after, err := creationDocument(canvas.PayloadJSON)
if err != nil {
return err
}
nodes := cloudAgentObjectChanges(before["nodes"], after["nodes"])
edges := cloudAgentObjectChanges(before["connections"], after["connections"])
if !cloudAgentPatchCoversDocument(before, after) {
// A partial delta must never acknowledge a revision containing omitted edits.
state.event(runID, "canvas_updated", map[string]any{"canvasId": input.CanvasID, "operation": input.Operation, "requiresRefresh": true})
return nil
}
actions := []map[string]any{}
previewByOperationAndNode := map[string]cloudAgentApprovalPreviewItem{}
previewByUniqueNode := map[string]cloudAgentApprovalPreviewItem{}
ambiguousPreviewNode := map[string]bool{}
if inputPreview := input.Preview; inputPreview != nil {
for _, item := range inputPreview.Items {
if item.NodeID == "" {
continue
}
previewByOperationAndNode[item.Operation+"\x00"+item.NodeID] = item
if _, exists := previewByUniqueNode[item.NodeID]; exists {
ambiguousPreviewNode[item.NodeID] = true
} else {
previewByUniqueNode[item.NodeID] = item
}
}
}
findPreview := func(operation, nodeID string) (cloudAgentApprovalPreviewItem, bool) {
if item, ok := previewByOperationAndNode[operation+"\x00"+nodeID]; ok {
return item, true
}
if !ambiguousPreviewNode[nodeID] {
item, ok := previewByUniqueNode[nodeID]
return item, ok
}
return cloudAgentApprovalPreviewItem{}, false
}
for _, change := range nodes {
node := change["after"].(map[string]any)
action := "updated"
if old, _ := change["before"].(map[string]any); old == nil {
action = "created"
}
entry := map[string]any{"action": action, "nodeId": node["id"], "title": node["title"], "nodeType": node["type"]}
previewOperation := "update_node"
if action == "created" {
previewOperation = "add_node"
}
if preview, ok := findPreview(previewOperation, stringValue(node["id"])); ok {
if preview.NodeTitle != "" {
entry["title"] = preview.NodeTitle
}
if len(preview.Fields) > 0 {
entry["fields"] = preview.Fields
}
if preview.ResultTitle != "" {
entry["resultTitle"] = preview.ResultTitle
}
if preview.Summary != "" {
entry["summary"] = preview.Summary
}
}
actions = append(actions, entry)
}
byID, err := creationObjects(after["nodes"])
if err != nil {
return err
}
for _, change := range edges {
edge := change["after"].(map[string]any)
if node := byID[stringValue(edge["fromNodeId"])]; node != nil {
entry := map[string]any{"action": "referenced", "nodeId": node["id"], "title": node["title"], "nodeType": node["type"], "targetNodeId": edge["toNodeId"]}
if target := byID[stringValue(edge["toNodeId"])]; target != nil {
entry["targetTitle"], entry["targetNodeType"] = target["title"], target["type"]
}
if preview, ok := findPreview("connect_nodes", stringValue(edge["fromNodeId"])); ok && preview.TargetNodeTitle != "" {
entry["targetTitle"], entry["targetNodeType"] = preview.TargetNodeTitle, preview.TargetNodeType
if preview.Summary != "" {
entry["summary"] = preview.Summary
}
}
actions = append(actions, entry)
}
}
payload := map[string]any{
"canvasId": input.CanvasID, "operation": input.Operation, "actions": actions,
"canvasPatch": map[string]any{"canvasId": input.CanvasID, "baseRevision": canvas.Revision - 1, "revision": canvas.Revision, "updatedAt": after["updatedAt"], "nodes": nodes, "connections": edges},
}
if input.Preview != nil {
payload["preview"] = input.Preview
payload["text"] = input.Preview.Description
}
if state.CallIndex >= 0 && state.CallIndex < len(state.Calls) {
payload["callId"] = state.Calls[state.CallIndex].ID
}
state.event(runID, "canvas_updated", payload)
return nil
}
func cloudAgentPatchCoversDocument(before, after map[string]any) bool {
for _, key := range []string{"nodes", "connections"} {
old, next := creationMaps(before[key]), creationMaps(after[key])
if len(next) < len(old) {
return false
}
// The delta merger updates existing items in place and appends new items.
for i, item := range old {
if item["id"] != next[i]["id"] {
return false
}
}
}
content := func(doc map[string]any) map[string]any {
result := map[string]any{}
for key, value := range doc {
if key != "nodes" && key != "connections" && key != "updatedAt" && key != "viewport" {
result[key] = value
}
}
return result
}
return reflect.DeepEqual(content(before), content(after))
}