diff --git a/db/011-canvas-ai-target-location.sql b/db/011-canvas-ai-target-location.sql new file mode 100644 index 0000000..f543637 --- /dev/null +++ b/db/011-canvas-ai-target-location.sql @@ -0,0 +1,9 @@ +BEGIN; +-- The frozen board remains the plan origin; result versions belong to the board +-- that actually received the output after any owner-only move. +ALTER TABLE canvas_ai_steps ADD COLUMN applied_board_id uuid; +ALTER TABLE canvas_ai_steps ADD FOREIGN KEY(project,workspace_id,applied_board_id) REFERENCES materials(project,workspace_id,id); +UPDATE canvas_ai_steps SET applied_board_id=board_id WHERE applied_version IS NOT NULL; +CREATE UNIQUE INDEX canvas_ai_canonical_target_busy ON canvas_ai_steps(project,workspace_id,(frozen_input->>'canonical_root'),node_id,(frozen_input->>'target_incarnation')) + WHERE status IN('queued','running') AND frozen_input->>'canonical_root' IS NOT NULL; +COMMIT; diff --git a/internal/platform/canvas_ai_apply.go b/internal/platform/canvas_ai_apply.go index 4a5567a..2fbf9f3 100644 --- a/internal/platform/canvas_ai_apply.go +++ b/internal/platform/canvas_ai_apply.go @@ -14,7 +14,19 @@ func (s *service) startCanvasStep(ctx context.Context, run canvasQueueRun, step return false, err } defer tx.Rollback() - _, role, permissionErr := lockMaterial(ctx, tx, run.Project, run.Workspace, step.BoardID, run.Account) + if err = lockCanvasAIWorkspace(ctx, tx, run.Project, run.Workspace); err != nil { + return false, err + } + board, identityOK, err := resolveCanvasAITarget(ctx, tx, run.Project, run.Workspace, step) + if err != nil { + return false, err + } + if !identityOK { + tx.Rollback() + _, err = s.finishCanvasStep(ctx, run, step, "", "target_removed", false) + return false, err + } + _, role, permissionErr := lockMaterial(ctx, tx, run.Project, run.Workspace, board, run.Account) if permissionErr != nil || role == "viewer" { tx.Rollback() if permissionErr != nil && permissionErr != sql.ErrNoRows { @@ -33,15 +45,11 @@ func (s *service) startCanvasStep(ctx context.Context, run canvasQueueRun, step _, err = s.finishCanvasStep(ctx, run, step, "", "generation_cancelled", false) return false, err } - snapshot, err := loadCanvas(queueRequest(ctx), tx, run.Project, run.Workspace, step.BoardID) + snapshot, err := loadCanvas(queueRequest(ctx), tx, run.Project, run.Workspace, board) if err != nil { return false, err } data := canvasNodeData(snapshot.Document)[step.NodeID] - identityOK, err := canvasAIIdentityMatches(ctx, tx, run.Project, run.Workspace, step) - if err != nil { - return false, err - } if !identityOK || data == nil || canvasTargetFingerprint(step.BoardID, step.NodeID, data) != step.Fingerprint { tx.Rollback() _, err = s.finishCanvasStep(ctx, run, step, "", "target_changed", false) @@ -108,7 +116,7 @@ func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun status = "skipped" case "target_changed": status = "conflict" - case "generation_cancelled": + case "generation_cancelled", "target_removed": status = "cancelled" default: if code != "" { @@ -119,6 +127,15 @@ func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun status = "cancelled" code = "generation_cancelled" } + board, identityOK, err := resolveCanvasAITarget(ctx, tx, run.Project, run.Workspace, step) + if err != nil { + return "", err + } + eventBoard := board + if !identityOK { + eventBoard = step.BoardID + } + var appliedBoard any var output any var version any if code == "" && started { @@ -141,10 +158,10 @@ func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun patch["text"] = generated.Text } - if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.output", response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "output": output}); err != nil { + if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.output", response{"step_id": step.ID, "board_id": eventBoard, "origin_board_id": step.BoardID, "canonical_root": step.CanonicalRoot, "target_incarnation": step.TargetIncarnation, "node_id": step.NodeID, "output": output}); err != nil { return "", err } - _, role, permissionErr := lockMaterial(ctx, tx, run.Project, run.Workspace, step.BoardID, run.Account) + _, role, permissionErr := lockCanvasAIRunRoute(ctx, tx, run.Project, run.Workspace, eventBoard, run.Account) if permissionErr != nil && permissionErr != sql.ErrNoRows { return "", permissionErr } @@ -152,14 +169,15 @@ func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun status = "failed" code = "permission_revoked" } else { - snapshot, err := loadCanvas(queueRequest(ctx), tx, run.Project, run.Workspace, step.BoardID) + snapshot, err := loadCanvas(queueRequest(ctx), tx, run.Project, run.Workspace, eventBoard) if err != nil { return "", err } target := canvasNodeData(snapshot.Document)[step.NodeID] - identityOK, err := canvasAIIdentityMatches(ctx, tx, run.Project, run.Workspace, step) - if err != nil { - return "", err + if image, ok := output.(response); ok && image["kind"] == "image" && identityOK { + if _, err = tx.ExecContext(ctx, `INSERT INTO canvas_asset_bindings(project,workspace_id,asset_id,board_id) VALUES($1,$2,$3,$4) ON CONFLICT DO NOTHING`, run.Project, run.Workspace, image["assetId"], eventBoard); err != nil { + return "", err + } } if !identityOK || target == nil || canvasTargetFingerprint(step.BoardID, step.NodeID, target) != step.Fingerprint { status, code = "conflict", "target_changed" @@ -175,8 +193,9 @@ func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun return "", err } version = change.Snapshot.Version - applied := response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "patch": patch, "canvas_version": change.Snapshot.Version} - change.addCanonical(applied) + appliedBoard = eventBoard + applied := response{"step_id": step.ID, "board_id": eventBoard, "node_id": step.NodeID, "patch": patch, "canvas_version": change.Snapshot.Version} + change.addTargetContext(applied, step) if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.applied", applied); err != nil { return "", err } @@ -189,7 +208,7 @@ func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun encoded, _ := json.Marshal(output) outputJSON = encoded } - _, err = tx.ExecContext(ctx, `UPDATE canvas_ai_steps SET status=$5,output=$6,error_code=NULLIF($7,''),applied_version=$8,completed_at=now() WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND id=$4`, run.Project, run.Workspace, run.ID, step.ID, status, outputJSON, code, version) + _, err = tx.ExecContext(ctx, `UPDATE canvas_ai_steps SET status=$5,output=$6,error_code=NULLIF($7,''),applied_version=$8,applied_board_id=$9,completed_at=now() WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND id=$4`, run.Project, run.Workspace, run.ID, step.ID, status, outputJSON, code, version, appliedBoard) if err != nil { return "", err } diff --git a/internal/platform/canvas_ai_cancel.go b/internal/platform/canvas_ai_cancel.go index 0aa408c..2a5c71a 100644 --- a/internal/platform/canvas_ai_cancel.go +++ b/internal/platform/canvas_ai_cancel.go @@ -63,7 +63,7 @@ func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } defer tx.Rollback() - _, role, err := lockMaterial(r.Context(), tx, project, workspace, board, account) + _, role, err := lockCanvasAIRunRoute(r.Context(), tx, project, workspace, board, account) if err != nil { return err } @@ -108,7 +108,14 @@ func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } for _, step := range plan.Steps { - _, role, err := lockMaterial(r.Context(), tx, project, workspace, step.BoardID, account) + target, ok, e := resolveCanvasAITarget(r.Context(), tx, project, workspace, step) + if e != nil { + return e + } + if !ok { + target = step.BoardID + } + _, role, err := lockCanvasAIRunRoute(r.Context(), tx, project, workspace, target, account) if err != nil { return err } @@ -159,7 +166,7 @@ func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } } - rows, err := tx.QueryContext(r.Context(), `SELECT id,board_id::text,node_id,status,applied_version FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 ORDER BY ordinal`, project, workspace, id) + rows, err := tx.QueryContext(r.Context(), `SELECT id,COALESCE(applied_board_id,board_id)::text,node_id,status,applied_version FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 ORDER BY ordinal`, project, workspace, id) if err != nil { return err } diff --git a/internal/platform/canvas_ai_explicit_apply.go b/internal/platform/canvas_ai_explicit_apply.go index 7374ccb..7792f5c 100644 --- a/internal/platform/canvas_ai_explicit_apply.go +++ b/internal/platform/canvas_ai_explicit_apply.go @@ -52,7 +52,7 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr return err } defer tx.Rollback() - _, role, err := lockMaterial(r.Context(), tx, project, workspace, board, account) + _, role, err := lockCanvasAIRunRoute(r.Context(), tx, project, workspace, board, account) if err != nil { return err } @@ -106,26 +106,9 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr if err != nil { return err } - if canonical && targetBoard != input.TargetCanvasID { - return rejectRequest(409, "target_changed") - } if node != input.NodeID || status != "conflict" || applied.Valid || runStatus == "cancelled" { return rejectRequest(409, "output_not_applicable") } - _, role, err = lockMaterial(r.Context(), tx, project, workspace, targetBoard, account) - if err != nil { - return err - } - if role == "viewer" { - return rejectRequest(403, "forbidden") - } - snapshot, err := loadCanvas(r, tx, project, workspace, targetBoard) - if err != nil { - return err - } - if snapshot.Version != expected { - return rejectRequest(412, "version_conflict") - } var output struct { Kind string `json:"kind"` Text string `json:"text"` @@ -142,13 +125,30 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr if canonical && (frozen.TargetIncarnation != input.TargetIncarnation || frozen.CanonicalRoot == "") { return rejectRequest(409, "target_changed") } - identityOK, err := canvasAIIdentityMatches(r.Context(), tx, project, workspace, frozen) + targetBoard, identityOK, err := resolveCanvasAITarget(r.Context(), tx, project, workspace, frozen) if err != nil { return err } if !identityOK { return rejectRequest(409, "target_changed") } + if canonical && targetBoard != input.TargetCanvasID { + return rejectRequest(409, "target_changed") + } + _, role, err = lockMaterial(r.Context(), tx, project, workspace, targetBoard, account) + if err != nil { + return err + } + if role == "viewer" { + return rejectRequest(403, "forbidden") + } + snapshot, err := loadCanvas(r, tx, project, workspace, targetBoard) + if err != nil { + return err + } + if snapshot.Version != expected { + return rejectRequest(412, "version_conflict") + } patch := response{} switch output.Kind { case "text": @@ -199,12 +199,12 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr return err } - _, err = tx.ExecContext(r.Context(), `UPDATE canvas_ai_steps SET applied_version=$5 WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND id=$4`, project, workspace, id, stepID, snapshot.Version) + _, err = tx.ExecContext(r.Context(), `UPDATE canvas_ai_steps SET applied_version=$5,applied_board_id=$6 WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND id=$4`, project, workspace, id, stepID, snapshot.Version, targetBoard) if err != nil { return err } result := response{"run_id": id, "step_id": stepID, "board_id": targetBoard, "node_id": node, "output_id": input.OutputID, "canvas_version": snapshot.Version, "patch": patch} - change.addCanonical(result) + change.addTargetContext(result, frozen) if receipt != nil { result["status"] = "applied" bindCanvasAIReceipt(result, receipt) diff --git a/internal/platform/canvas_ai_image_inputs.go b/internal/platform/canvas_ai_image_inputs.go index 1b0b606..1c5c79a 100644 --- a/internal/platform/canvas_ai_image_inputs.go +++ b/internal/platform/canvas_ai_image_inputs.go @@ -166,7 +166,17 @@ func (s *service) hydrateCanvasTask(ctx context.Context, run canvasQueueRun, ste return err } defer tx.Rollback() - _, role, err := lockMaterial(ctx, tx, run.Project, run.Workspace, step.BoardID, run.Account) + if err = lockCanvasAIWorkspace(ctx, tx, run.Project, run.Workspace); err != nil { + return err + } + board, ok, err := resolveCanvasAITarget(ctx, tx, run.Project, run.Workspace, step) + if err != nil { + return err + } + if !ok { + return errors.New("target_removed") + } + _, role, err := lockMaterial(ctx, tx, run.Project, run.Workspace, board, run.Account) if err != nil { return err } diff --git a/internal/platform/canvas_ai_model.go b/internal/platform/canvas_ai_model.go index 2a1f18a..a088ca1 100644 --- a/internal/platform/canvas_ai_model.go +++ b/internal/platform/canvas_ai_model.go @@ -170,15 +170,54 @@ func bindCanvasAIIdentities(ctx context.Context, tx *sql.Tx, project, workspace } return nil } -func canvasAIIdentityMatches(ctx context.Context, tx *sql.Tx, project, workspace string, step canvasAIStepPlan) (bool, error) { - root, incarnation, kind, err := canvasAIIdentity(ctx, tx, project, workspace, step.BoardID, step.NodeID) + +// 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 false, err + return "", "", err } - if root == "" { - return step.CanonicalRoot == "" && step.TargetIncarnation == 0, nil + 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) + } } - return root == step.CanonicalRoot && incarnation > 0 && incarnation == step.TargetIncarnation && kind == step.Kind, nil } // Whole identity transitions cancel old jobs even when delete/restore happens @@ -190,11 +229,30 @@ func invalidateCanonicalAIIdentities(r *http.Request, tx *sql.Tx, project, works continue } next := after.Elements[id] - if next != nil && next["ownerCanvasId"] == element["ownerCanvasId"] && modelInt(next["incarnation"]) == modelInt(element["incarnation"]) && next["componentId"] == element["componentId"] && modelInt(next["dataVersion"]) == modelInt(element["dataVersion"]) { + if next != nil && modelInt(next["incarnation"]) == modelInt(element["incarnation"]) && next["componentId"] == element["componentId"] && modelInt(next["dataVersion"]) == modelInt(element["dataVersion"]) { continue } - board := modelString(element["ownerCanvasId"]) - removed[board] = append(removed[board], map[string]any{"id": id, "data": map[string]any{"kind": element["componentId"]}}) + 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]}) diff --git a/internal/platform/canvas_ai_model_integration_test.go b/internal/platform/canvas_ai_model_integration_test.go index bacf8ce..a93e9b6 100644 --- a/internal/platform/canvas_ai_model_integration_test.go +++ b/internal/platform/canvas_ai_model_integration_test.go @@ -177,6 +177,12 @@ func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) { after.Versions[id] = before.Versions[id] + 1 participants[id] = true } + for id := range before.Canvases { + if _, active := after.Canvases[id]; !active { + after.Versions[id] = before.Versions[id] + 1 + participants[id] = true + } + } if e = validateCanvasModel(after); e != nil { t.Fatal(e) } @@ -211,7 +217,14 @@ func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) { command := func(board, target, model string, incarnation int64) (response, string) { next += 2 key := modelTestID(next) - return response{"protocol_version": 2, "request_id": key, "operation_id": modelTestID(next + 1), "target_incarnation": incarnation, "expected_versions": read().Versions, "node_id": target, "model_id": model, "prompt": "Generate", "parameters": response{}, "propagate": false}, key + return response{"protocol_version": 2, "request_id": key, "operation_id": modelTestID(next + 1), "target_incarnation": incarnation, "expected_versions": func() map[string]int64 { + model := read() + versions := map[string]int64{} + for id := range model.Canvases { + versions[id] = model.Versions[id] + } + return versions + }(), "node_id": target, "model_id": model, "prompt": "Generate", "parameters": response{}, "propagate": false}, key } start := func(board, target, model string, incarnation int64) response { input, key := command(board, target, model, incarnation) @@ -377,10 +390,32 @@ func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) { model.Canvases[child] = modelCanvas{ID: child, Parent: &modelParent{CanvasID: root, EntryElementID: frame}} model.Elements[imageID]["ownerCanvasId"] = child }) - state, e = s.finishCanvasStep(ctx, *moving, moving.Plan.Steps[0], "wrong owner", "", true) - if e != nil || state != "cancelled" { - t.Fatalf("moved target accepted old output %s %v", state, e) + busyInput, busyKey := command(child, imageID, "image", 1) + busy := call("POST", base(child)+"/ai/runs", busyInput, 409, read().Versions[child], busyKey) + if busy["error"] != "node_busy" { + t.Fatal("owner move admitted a duplicate paid task") } + beforeMovedOutput := read() + state, e = s.finishCanvasStepOutput(ctx, *moving, moving.Plan.Steps[0], codex.CanvasOutput{Kind: "image", Image: &codex.CanvasImage{MIME: "image/png", Data: pngBytes.Bytes()}}, "", true) + if e != nil || state != "succeeded" { + t.Fatalf("owner-only move rejected output %s %v", state, e) + } + afterMovedOutput := read() + if afterMovedOutput.Elements[imageID]["ownerCanvasId"] != child || afterMovedOutput.Versions[root] != beforeMovedOutput.Versions[root] || afterMovedOutput.Versions[child] != beforeMovedOutput.Versions[child]+1 || afterMovedOutput.Elements[imageID]["geometry"].(map[string]any)["width"] != float64(10000) { + t.Fatal("relocated result rewrote ownership, geometry or the origin version") + } + movedStatus := call("GET", base(root)+"/ai/runs/"+moving.ID, nil, 200, 0, "") + foundRelocation := false + for _, value := range movedStatus["events"].([]any) { + event := value.(map[string]any) + if event["type"] == "step.applied" { + foundRelocation = event["board_id"] == child && event["origin_board_id"] == root && event["canonical_root"] == root && event["model"].(map[string]any)["complete"] == true + } + } + if !foundRelocation || movedStatus["applied_versions"].(map[string]any)[child] != float64(afterMovedOutput.Versions[child]) { + t.Fatal("moved output lost current-owner version or canonical context") + } + if e = s.completeCanvasRun(ctx, *moving, nil); e != nil { t.Fatal(e) } @@ -430,4 +465,53 @@ func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) { if read().Elements[imageID]["ownerCanvasId"] != child { t.Fatal("child result moved its target back") } + queuedMove := start(child, imageID, "image", 1) + mutate(func(model *canvasModel) { + model.Elements[imageID]["ownerCanvasId"] = root + delete(model.Canvases, child) + }) + execute(queuedMove["run_id"].(string)) + archivedRoute := base(child) + "/ai/runs/" + queuedMove["run_id"].(string) + archivedResult := call("GET", archivedRoute, nil, 200, 0, "") + if archivedResult["status"] != "succeeded" || read().Elements[imageID]["ownerCanvasId"] != root { + t.Fatal("queued move or archived admission route stopped the run") + } + lateArchiveKey := modelTestID(6500) + if result := call("POST", archivedRoute+"/cancel", response{"protocol_version": 2, "request_id": lateArchiveKey, "operation_id": modelTestID(6501)}, 200, 0, lateArchiveKey); result["run_status"] != "succeeded" { + t.Fatal("archived route lost cancellation identity") + } + // Content changes still conflict after a move; explicit application targets the current owner. + movedConflict := start(root, node, "text", 2) + hooks <- func() { + mutate(func(model *canvasModel) { + model.Canvases[child] = modelCanvas{ID: child, Parent: &modelParent{CanvasID: root, EntryElementID: frame}} + model.Elements[node]["ownerCanvasId"] = child + model.Elements[node]["data"].(map[string]any)["text"] = "keep moved edit" + }) + } + execute(movedConflict["run_id"].(string)) + movedConflictPath := base(root) + "/ai/runs/" + movedConflict["run_id"].(string) + conflictMovedResult := call("GET", movedConflictPath, nil, 200, 0, "") + if conflictMovedResult["status"] != "conflict" || read().Elements[node]["data"].(map[string]any)["text"] != "keep moved edit" { + t.Fatal("move allowed a content conflict overwrite") + } + movedOutput := conflictMovedResult["outputs"].([]any)[0].(map[string]any)["output"].(map[string]any)["output_id"].(string) + movedApplyKey := modelTestID(6502) + movedApplied := call("POST", movedConflictPath+"/apply", response{"protocol_version": 2, "request_id": movedApplyKey, "operation_id": modelTestID(6503), "node_id": node, "output_id": movedOutput, "target_incarnation": 2, "target_canvas_id": child}, 200, read().Versions[child], movedApplyKey) + if movedApplied["board_id"] != child || movedApplied["origin_board_id"] != root || read().Elements[node]["ownerCanvasId"] != child { + t.Fatal("explicit application wrote to the frozen owner") + } + doomed := start(child, node, "text", 2) + mutate(func(model *canvasModel) { + record := model.Elements[node] + model.DeadRecords[node] = record + model.Tombstones[node] = 2 + model.TombstoneOwners[node] = child + delete(model.Elements, node) + }) + execute(doomed["run_id"].(string)) + if result := call("GET", base(child)+"/ai/runs/"+doomed["run_id"].(string), nil, 200, 0, ""); result["status"] != "cancelled" { + t.Fatal("deleting a moved target failed to revoke its old job") + } + } diff --git a/internal/platform/canvas_ai_query.go b/internal/platform/canvas_ai_query.go index 6ed9c78..b6d3cfb 100644 --- a/internal/platform/canvas_ai_query.go +++ b/internal/platform/canvas_ai_query.go @@ -51,7 +51,7 @@ func (s *service) getCanvasAIRun(w http.ResponseWriter, r *http.Request, project return err } defer tx.Rollback() - if _, _, err = lockMaterial(r.Context(), tx, project, workspace, board, account); err != nil { + if _, _, err = lockCanvasAIRunRoute(r.Context(), tx, project, workspace, board, account); err != nil { return err } var status string @@ -68,14 +68,21 @@ func (s *service) getCanvasAIRun(w http.ResponseWriter, r *http.Request, project } checked := map[string]bool{board: true} for _, step := range plan.Steps { - if !checked[step.BoardID] { - if _, _, err = lockMaterial(r.Context(), tx, project, workspace, step.BoardID, account); err != nil { - return err + target, ok, e := resolveCanvasAITarget(r.Context(), tx, project, workspace, step) + if e != nil { + return e + } + if !ok { + target = step.BoardID + } + if !checked[target] { + if _, _, e = lockCanvasAIRunRoute(r.Context(), tx, project, workspace, target, account); e != nil { + return e } - checked[step.BoardID] = true + checked[target] = true } } - rows, err := tx.QueryContext(r.Context(), `SELECT id,board_id::text,node_id,status,error_code,applied_version,CASE WHEN output IS NULL THEN NULL ELSE output-'text' END FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 ORDER BY ordinal`, project, workspace, id) + rows, err := tx.QueryContext(r.Context(), `SELECT id,COALESCE(applied_board_id,board_id)::text,node_id,status,error_code,applied_version,CASE WHEN output IS NULL THEN NULL ELSE output-'text' END FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 ORDER BY ordinal`, project, workspace, id) if err != nil { return err } @@ -177,7 +184,7 @@ func (s *service) getCanvasAIAdmission(w http.ResponseWriter, r *http.Request, p return err } defer tx.Rollback() - if _, _, err = lockMaterial(r.Context(), tx, project, workspace, board, account); err != nil { + if _, _, err = lockCanvasAIRunRoute(r.Context(), tx, project, workspace, board, account); err != nil { return err } var body json.RawMessage diff --git a/internal/platform/canvas_ai_runs.go b/internal/platform/canvas_ai_runs.go index 7b5ae34..1f3bdf1 100644 --- a/internal/platform/canvas_ai_runs.go +++ b/internal/platform/canvas_ai_runs.go @@ -293,7 +293,11 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return rejectRequest(413, "context_too_large") } var busy bool - err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND node_id=$4 AND status IN('queued','running'))`, project, workspace, step.BoardID, step.NodeID).Scan(&busy) + if step.CanonicalRoot != "" { + err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 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, step.CanonicalRoot, step.NodeID, step.TargetIncarnation).Scan(&busy) + } else { + err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND node_id=$4 AND status IN('queued','running'))`, project, workspace, step.BoardID, step.NodeID).Scan(&busy) + } if err != nil { return err } diff --git a/internal/platform/canvas_capabilities.go b/internal/platform/canvas_capabilities.go index 694e19b..305b091 100644 --- a/internal/platform/canvas_capabilities.go +++ b/internal/platform/canvas_capabilities.go @@ -39,7 +39,7 @@ func (s *service) canvasCapabilities(w http.ResponseWriter, r *http.Request, pro } } } - ai := response{"canonical_target_identity": true, "admission_protocol_version": 2, "admission_receipt_query": true, "mutation_protocol_version": 2, "terminal_rejection_receipts": true, "available": textAvailable || imageAvailable, "text": textAvailable, "image": imageAvailable} + ai := response{"canonical_target_identity": true, "owner_relocation": true, "admission_protocol_version": 2, "admission_receipt_query": true, "mutation_protocol_version": 2, "terminal_rejection_receipts": true, "available": textAvailable || imageAvailable, "text": textAvailable, "image": imageAvailable} if modelErr != nil { ai["unavailable_reason"] = "codex_unavailable" } diff --git a/internal/platform/canvas_model_store.go b/internal/platform/canvas_model_store.go index 878686a..c8994e1 100644 --- a/internal/platform/canvas_model_store.go +++ b/internal/platform/canvas_model_store.go @@ -193,7 +193,8 @@ func (s *service) saveCanvasModel(r *http.Request, tx *sql.Tx, project, workspac } continue } - // AI invalidation/asset bindings still run through the existing service checks. + // Asset bindings still run through service checks; canonical identity + // invalidation has already run on the normalized before/after models. if err = s.saveCanvas(r, tx, project, workspace, canvasSnapshot{BoardID: id, Version: after.Versions[id], Document: projection[id]}); err != nil { return err } diff --git a/internal/platform/subgraph_migration_http.go b/internal/platform/subgraph_migration_http.go index 555155b..639b085 100644 --- a/internal/platform/subgraph_migration_http.go +++ b/internal/platform/subgraph_migration_http.go @@ -44,14 +44,22 @@ func (s *service) saveCanvas(r *http.Request, tx *sql.Tx, project, workspace str if err := tx.QueryRowContext(r.Context(), `SELECT COALESCE(canvas_document,'{"schema_version":1,"nodes":[],"edges":[],"viewport":{"x":0,"y":0,"zoom":1}}'::jsonb) FROM materials WHERE project=$1 AND workspace_id=$2 AND id=$3 FOR UPDATE`, project, workspace, snapshot.BoardID).Scan(&before); err != nil { return err } - if err := invalidateCanvasAITargets(r, tx, project, workspace, snapshot.BoardID, before, snapshot.Document); err != nil { + root, err := canonicalRoot(r.Context(), tx, project, workspace, snapshot.BoardID) + if err != nil { return err } + // Canonical lifecycle invalidation runs once against normalized identities. + // Its compatibility projection must not mistake an owner move for deletion. + if root == "" { + if err := invalidateCanvasAITargets(r, tx, project, workspace, snapshot.BoardID, before, snapshot.Document); err != nil { + return err + } + } if err := s.attachCanvasAssets(r, tx, project, workspace, snapshot.BoardID, snapshot.Document); err != nil { return err } - _, err := tx.ExecContext(r.Context(), `UPDATE materials SET canvas_document=$4,canvas_version=$5 WHERE project=$1 AND workspace_id=$2 AND id=$3`, project, workspace, snapshot.BoardID, snapshot.Document, snapshot.Version) + _, err = tx.ExecContext(r.Context(), `UPDATE materials SET canvas_document=$4,canvas_version=$5 WHERE project=$1 AND workspace_id=$2 AND id=$3`, project, workspace, snapshot.BoardID, snapshot.Document, snapshot.Version) return err }