Resolve canonical AI targets at their current canvas owner
This commit is contained in:
@@ -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;
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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]})
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user