package platform import ( "context" "database/sql" "encoding/json" "errors" "net/http" ) var errAITargetChanged = errors.New("target_changed") var errAIInvalidPatch = errors.New("invalid_ai_patch") // The snapshot is an AI reader projection only. Canonical changes retain the // original rows, including unknown envelopes, geometry and reference metadata. type canvasAIChange struct { Snapshot canvasSnapshot Before *canvasModel After *canvasModel Incarnation int64 } // Until a narrower read-set protocol exists, admission binds every active // canvas in this canonical project. Archived versions are not graph inputs. // Workspace locking keeps validation, plan freezing and admission atomic. func validateCanvasAIVersions(model canvasModel, board string, header int64, expected map[string]int64) string { if len(expected) != len(model.Canvases) || expected[board] != header { return "invalid_ai_versions" } for id := range model.Canvases { version, ok := expected[id] if !ok || version < 1 || version > 9007199254740991 { return "invalid_ai_versions" } if version != model.Versions[id] { return "version_conflict" } } return "" } func prepareModelAIPatch(before canvasModel, board, node string, patch response) (canvasModel, error) { element := before.Elements[node] if element == nil || element["structure"] != "node" || element["ownerCanvasId"] != board || modelInt(element["dataVersion"]) != 1 || !canvasChoice(element["componentId"], "text", "script", "plain-text", "image", "plain-image") { return canvasModel{}, errAITargetChanged } after := cloneModel(before) data := after.Elements[node]["data"].(map[string]any) for key, value := range patch { if !canvasChoice(key, "ai", "text", "src", "assetId") { return canvasModel{}, errAIInvalidPatch } data[key] = value } if before.Versions[board] < 1 || before.Versions[board] >= 9007199254740991 { return canvasModel{}, errModelConflict } after.Versions[board]++ if err := validateCanvasModel(after); err != nil { return canvasModel{}, err } return after, nil } func prepareCanvasAIPatch(r *http.Request, tx *sql.Tx, project, workspace string, snapshot canvasSnapshot, node string, patch response) (canvasAIChange, error) { change := canvasAIChange{Snapshot: snapshot} root, err := canonicalRoot(r.Context(), tx, project, workspace, snapshot.BoardID) if err != nil { return change, err } if root != "" { before, err := loadCanvasModel(r.Context(), tx, project, workspace, root) if err != nil { return change, err } if before.Versions[snapshot.BoardID] != snapshot.Version { return change, errModelConflict } after, err := prepareModelAIPatch(before, snapshot.BoardID, node, patch) if err != nil { return change, err } projection, err := projectCanvasModel(after) if err != nil { return change, err } change.Before, change.After = &before, &after change.Incarnation = modelInt(before.Elements[node]["incarnation"]) change.Snapshot.Document, change.Snapshot.Version = projection[snapshot.BoardID], after.Versions[snapshot.BoardID] return change, nil } var document map[string]any if err = json.Unmarshal(snapshot.Document, &document); err != nil { return change, err } target := canvasNodeData(snapshot.Document)[node] if target == nil { return change, errAITargetChanged } // The data lookup owns a separate decode. Update the staged document itself. for _, raw := range document["nodes"].([]any) { element := raw.(map[string]any) if element["id"] != node { continue } data := element["data"].(map[string]any) for key, value := range patch { if !canvasChoice(key, "ai", "text", "src", "assetId") { return change, errAIInvalidPatch } data[key] = value } } document["schema_version"] = float64(2) encoded, err := json.Marshal(document) if err != nil { return change, err } normalized, ok := normalizeCanvas(encoded) if !ok { return change, errAIInvalidPatch } change.Snapshot.Document = normalized change.Snapshot.Version++ return change, nil } func (s *service) saveCanvasAIChange(r *http.Request, tx *sql.Tx, project, workspace, account string, change canvasAIChange) error { if change.After == nil { return s.saveCanvas(r, tx, project, workspace, change.Snapshot) } return s.saveCanvasModel(r, tx, project, workspace, account, *change.Before, *change.After, map[string]bool{change.Snapshot.BoardID: true}) } func (change canvasAIChange) addCanonical(result response) { if change.After == nil { return } board := change.Snapshot.BoardID result["document_format_version"], result["target_incarnation"] = canvasModelFormat, change.Incarnation result["model"] = change.After.view(map[string]int64{board: change.Snapshot.Version}) } func canvasAIIdentity(ctx context.Context, tx *sql.Tx, project, workspace, board, node string) (string, int64, string, error) { root, err := canonicalRoot(ctx, tx, project, workspace, board) if err != nil || root == "" { return root, 0, "", err } var incarnation, version int64 var kind string err = tx.QueryRowContext(ctx, `SELECT incarnation,record->>'componentId',(record->>'dataVersion')::bigint FROM canvas_model_elements WHERE project=$1 AND workspace_id=$2 AND root_id=$3 AND element_id=$4 AND owner_canvas_id=$5 AND NOT deleted AND record->>'structure'='node'`, project, workspace, root, node, board).Scan(&incarnation, &kind, &version) if err == sql.ErrNoRows { return root, 0, "", nil } if err != nil { return root, 0, "", err } if version != 1 || !canvasChoice(kind, "text", "script", "plain-text", "image", "plain-image") { return root, 0, "", nil } return root, incarnation, kind, nil } func bindCanvasAIIdentities(ctx context.Context, tx *sql.Tx, project, workspace string, plan *canvasAIPlan) error { for i, step := range plan.Steps { root, incarnation, kind, err := canvasAIIdentity(ctx, tx, project, workspace, step.BoardID, step.NodeID) if err != nil { return err } if root != "" && (incarnation < 1 || kind != step.Kind) { return errAITargetChanged } plan.Steps[i].CanonicalRoot, plan.Steps[i].TargetIncarnation = root, incarnation } return nil } // Run URLs remain anchored at their admission board, even after that partition // is archived. Canonical ownership grants are inherited from the same root. func lockCanvasAIRunRoute(ctx context.Context, tx *sql.Tx, project, workspace, board, account string) (string, string, error) { root, err := canonicalRoot(ctx, tx, project, workspace, board) if err != nil { return "", "", err } if root != "" { board = root } return lockMaterial(ctx, tx, project, workspace, board, account) } func lockCanvasAIWorkspace(ctx context.Context, tx *sql.Tx, project, workspace string) error { var id string return tx.QueryRowContext(ctx, `SELECT id::text FROM workspaces WHERE project=$1 AND id=$2 FOR UPDATE`, project, workspace).Scan(&id) } // Caller holds the workspace lock. Frozen identity is stable; owner is resolved // afresh at execution, hydration and write time, never written into the plan. func resolveCanvasAITarget(ctx context.Context, tx *sql.Tx, project, workspace string, step canvasAIStepPlan) (string, bool, error) { if step.CanonicalRoot == "" { root, err := canonicalRoot(ctx, tx, project, workspace, step.BoardID) return step.BoardID, err == nil && root == "" && step.TargetIncarnation == 0, err } var board, kind string var incarnation, version int64 err := tx.QueryRowContext(ctx, `SELECT e.owner_canvas_id::text,e.incarnation,e.record->>'componentId',(e.record->>'dataVersion')::bigint FROM canvas_model_elements e JOIN canvas_model_canvases c ON c.project=e.project AND c.workspace_id=e.workspace_id AND c.canvas_id=e.owner_canvas_id AND c.root_id=e.root_id WHERE e.project=$1 AND e.workspace_id=$2 AND e.root_id=$3 AND e.element_id=$4 AND NOT e.deleted AND NOT c.archived AND e.record->>'structure'='node'`, project, workspace, step.CanonicalRoot, step.NodeID).Scan(&board, &incarnation, &kind, &version) if err == sql.ErrNoRows { return "", false, nil } if err != nil { return "", false, err } return board, incarnation == step.TargetIncarnation && incarnation > 0 && kind == step.Kind && version == 1, nil } func canvasAIIdentityMatches(ctx context.Context, tx *sql.Tx, project, workspace string, step canvasAIStepPlan) (bool, error) { _, ok, err := resolveCanvasAITarget(ctx, tx, project, workspace, step) return ok, err } func (change canvasAIChange) addTargetContext(result response, step canvasAIStepPlan) { change.addCanonical(result) if step.CanonicalRoot != "" { result["origin_board_id"], result["canonical_root"] = step.BoardID, step.CanonicalRoot if change.After != nil && change.Snapshot.BoardID != step.BoardID { result["model"] = change.After.view(nil) } } } // Whole identity transitions cancel old jobs even when delete/restore happens // in a single atomic batch and the compatibility projection looks identical. func invalidateCanonicalAIIdentities(r *http.Request, tx *sql.Tx, project, workspace string, before, after canvasModel) error { removed := map[string][]any{} for id, element := range before.Elements { if element["structure"] != "node" { continue } next := after.Elements[id] if next != nil && modelInt(next["incarnation"]) == modelInt(element["incarnation"]) && next["componentId"] == element["componentId"] && modelInt(next["dataVersion"]) == modelInt(element["dataVersion"]) { continue } rows, err := tx.QueryContext(r.Context(), `SELECT DISTINCT board_id::text FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND frozen_input->>'canonical_root'=$3 AND node_id=$4 AND frozen_input->>'target_incarnation'=$5::bigint::text AND status IN('queued','running')`, project, workspace, before.Root, id, modelInt(element["incarnation"])) if err != nil { return err } boards := []string{} for rows.Next() { var board string if err = rows.Scan(&board); err != nil { rows.Close() return err } boards = append(boards, board) } err = rows.Err() rows.Close() if err != nil { return err } for _, board := range boards { removed[board] = append(removed[board], map[string]any{"id": id, "data": map[string]any{"kind": element["componentId"]}}) } } for _, board := range sortedKeys(removed) { raw, err := json.Marshal(map[string]any{"nodes": removed[board]}) if err != nil { return err } if err = invalidateCanvasAITargets(r, tx, project, workspace, board, raw, json.RawMessage(`{"nodes":[]}`)); err != nil { return err } } return nil }