Apply canonical AI outputs through scoped identity-checked changes
Foundry Go checks / check (push) Successful in 2m56s
Foundry Go checks / check (pull_request) Successful in 2m47s

This commit is contained in:
2026-11-08 00:23:00 +08:00
parent 0d7fc4dbe5
commit fa76773efc
12 changed files with 771 additions and 66 deletions
+6
View File
@@ -0,0 +1,6 @@
BEGIN;
-- Legacy requests retain NULL. Canonical operation identity cannot be replayed
-- under another transport key to admit a second paid execution.
ALTER TABLE canvas_ai_runs ADD COLUMN logical_operation_id uuid;
CREATE UNIQUE INDEX canvas_ai_logical_operation ON canvas_ai_runs(project,workspace_id,account_id,logical_operation_id) WHERE logical_operation_id IS NOT NULL;
COMMIT;
+23 -27
View File
@@ -4,6 +4,7 @@ import (
"context"
"database/sql"
"encoding/json"
"errors"
"gitea.entroia.com/Zayden-Jung/foundry-core/internal/codex"
)
@@ -37,7 +38,11 @@ func (s *service) startCanvasStep(ctx context.Context, run canvasQueueRun, step
return false, err
}
data := canvasNodeData(snapshot.Document)[step.NodeID]
if data == nil || canvasTargetFingerprint(step.BoardID, step.NodeID, data) != step.Fingerprint {
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)
return false, err
@@ -151,37 +156,28 @@ func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun
if err != nil {
return "", err
}
var doc map[string]any
if err = json.Unmarshal(snapshot.Document, &doc); err != nil {
target := canvasNodeData(snapshot.Document)[step.NodeID]
identityOK, err := canvasAIIdentityMatches(ctx, tx, run.Project, run.Workspace, step)
if err != nil {
return "", err
}
var target map[string]any
for _, v := range doc["nodes"].([]any) {
n := v.(map[string]any)
if n["id"] == step.NodeID {
target = n["data"].(map[string]any)
}
}
if target == nil || canvasTargetFingerprint(step.BoardID, step.NodeID, target) != step.Fingerprint {
status = "conflict"
code = "target_changed"
if !identityOK || target == nil || canvasTargetFingerprint(step.BoardID, step.NodeID, target) != step.Fingerprint {
status, code = "conflict", "target_changed"
} else {
for key, value := range patch {
target[key] = value
}
encoded, _ := json.Marshal(doc)
normalized, ok := normalizeCanvas(encoded)
if !ok {
status = "failed"
code = "generation_output_too_large"
} else {
snapshot.Document = normalized
snapshot.Version++
if err = s.saveCanvas(queueRequest(ctx), tx, run.Project, run.Workspace, snapshot); err != nil {
change, err := prepareCanvasAIPatch(queueRequest(ctx), tx, run.Project, run.Workspace, snapshot, step.NodeID, patch)
if err != nil {
if !errors.Is(err, errAIInvalidPatch) && !errors.Is(err, errCanvasModel) && !errors.Is(err, errGraphCapacity) {
return "", err
}
version = snapshot.Version
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.applied", response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "patch": patch, "canvas_version": snapshot.Version}); err != nil {
status, code = "failed", "generation_output_too_large"
} else {
if err = s.saveCanvasAIChange(queueRequest(ctx), tx, run.Project, run.Workspace, run.Account, change); err != nil {
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)
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.applied", applied); err != nil {
return "", err
}
}
+15 -10
View File
@@ -111,6 +111,14 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr
if err = json.Unmarshal(frozenRaw, &frozen); err != nil {
return err
}
identityOK, err := canvasAIIdentityMatches(r.Context(), tx, project, workspace, frozen)
if err != nil {
return err
}
if !identityOK {
reject(w, 409, "target_changed")
return nil
}
patch := response{}
switch output.Kind {
case "text":
@@ -158,29 +166,26 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr
reject(w, 409, "ownership_conflict")
return nil
}
for key, value := range patch {
target[key] = value
}
encoded, _ := json.Marshal(doc)
normalized, ok := normalizeCanvas(encoded)
if !ok {
change, err := prepareCanvasAIPatch(r, tx, project, workspace, snapshot, node, patch)
if err != nil {
reject(w, 413, "canvas_too_large")
return nil
}
snapshot.Document = normalized
snapshot.Version++
if err = s.saveCanvas(r, tx, project, workspace, snapshot); err != nil {
snapshot = change.Snapshot
if err = s.saveCanvasAIChange(r, tx, project, workspace, account, change); err != nil {
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)
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)
if err = appendAIEvent(r.Context(), tx, project, workspace, id, "step.applied", result); err != nil {
return err
}
encoded, _ = json.Marshal(result)
encoded, _ := json.Marshal(result)
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_mutations(project,workspace_id,run_id,account_id,action,idempotency_key,payload_hash,response_body) VALUES($1,$2,$3,$4,'apply',$5,$6,$7)`, project, workspace, id, account, key, hash, encoded)
if err != nil {
return err
+190
View File
@@ -0,0 +1,190 @@
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
}
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
}
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)
if err != nil {
return false, err
}
if root == "" {
return step.CanonicalRoot == "" && step.TargetIncarnation == 0, 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
// 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 && next["ownerCanvasId"] == element["ownerCanvasId"] && 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"]}})
}
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
}
@@ -0,0 +1,372 @@
package platform
import (
"bytes"
"context"
"crypto/rand"
"crypto/sha256"
"database/sql"
"encoding/base64"
"encoding/hex"
"encoding/json"
"image"
"image/png"
"net"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"strconv"
"strings"
"sync/atomic"
"testing"
"gitea.entroia.com/Zayden-Jung/foundry-core/internal/codex"
)
func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) {
dsn := os.Getenv("PLATFORM_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("requires an isolated integration database")
}
db, err := sql.Open("pgx", dsn)
if err != nil {
t.Fatal(err)
}
defer db.Close()
ctx := context.Background()
var entropy [32]byte
_, _ = rand.Read(entropy[:])
project := "canonical-ai-" + hex.EncodeToString(entropy[:8])
token := base64.RawURLEncoding.EncodeToString(entropy[:])
digest := sha256.Sum256([]byte(token))
var account, workspace, root string
if err = db.QueryRowContext(ctx, `INSERT INTO project_accounts(project,id) VALUES($1,gen_random_uuid()) RETURNING id::text`, project).Scan(&account); err != nil {
t.Fatal(err)
}
defer func() {
for _, table := range []string{"canvas_ai_runs", "canvas_model_operations", "canvas_model_identity_map", "canvas_model_requests", "canvas_model_projects", "canvas_asset_bindings", "canvas_assets", "canvas_subgraphs", "workspaces", "project_accounts"} {
_, _ = db.ExecContext(ctx, "DELETE FROM "+table+" WHERE project=$1", project)
}
}()
if _, err = db.ExecContext(ctx, `INSERT INTO project_sessions(token_hash,project,account_id,expires_at) VALUES($1,$2,$3,now()+interval '1 hour')`, base64.RawURLEncoding.EncodeToString(digest[:]), project, account); err != nil {
t.Fatal(err)
}
if err = db.QueryRowContext(ctx, `INSERT INTO workspaces(project,name,owner_id) VALUES($1,'Canonical AI',$2) RETURNING id::text`, project, account).Scan(&workspace); err != nil {
t.Fatal(err)
}
if _, err = db.ExecContext(ctx, `INSERT INTO workspace_members(project,workspace_id,account_id,role) VALUES($1,$2,$3,'admin')`, project, workspace, account); err != nil {
t.Fatal(err)
}
if err = db.QueryRowContext(ctx, `INSERT INTO materials(project,workspace_id,kind,title,created_by,canvas_document) VALUES($1,$2,'board','Canonical AI',$3,$4) RETURNING id::text`, project, workspace, account, json.RawMessage(canvasFixture)).Scan(&root); err != nil {
t.Fatal(err)
}
temp, err := os.MkdirTemp("", "fc-canonical-ai-")
if err != nil {
t.Fatal(err)
}
defer os.RemoveAll(temp)
socket := filepath.Join(temp, "worker.sock")
listener, err := net.Listen("unix", socket)
if err != nil {
t.Fatal(err)
}
var pngBytes bytes.Buffer
if err = png.Encode(&pngBytes, image.NewRGBA(image.Rect(0, 0, 3, 2))); err != nil {
t.Fatal(err)
}
hooks := make(chan func(), 1)
var executions atomic.Int32
worker := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/canvas/models" {
reply(w, 200, response{"models": []any{codex.CanvasModel{ID: "text", Available: true, InputModalities: []string{"text", "image"}, OutputModalities: []string{"text"}, MaxInputBytes: 1 << 20, MaxExecutionMS: 90000}, codex.CanvasModel{ID: "image", Available: true, InputModalities: []string{"text", "image"}, OutputModalities: []string{"image"}, MaxInputBytes: 1 << 20, MaxExecutionMS: 180000}}})
return
}
if r.URL.Path != "/canvas/exec" {
reject(w, 404, "not_found")
return
}
var task codex.CanvasTask
if json.NewDecoder(r.Body).Decode(&task) != nil {
reject(w, 400, "invalid_request")
return
}
executions.Add(1)
select {
case hook := <-hooks:
hook()
default:
}
if task.OutputModality == "image" {
reply(w, 200, response{"output": codex.CanvasOutput{Kind: "image", Image: &codex.CanvasImage{MIME: "image/png", Data: pngBytes.Bytes()}}})
} else {
reply(w, 200, response{"result": "generated result"})
}
})}
go worker.Serve(listener)
defer worker.Close()
s := &service{db: db, origin: "https://api.example.com", canvasAI: codex.NewCanvasClient(socket)}
callbacks := map[string]string{project: "https://app.example.com/auth/callback"}
mux := http.NewServeMux()
prefix := "/projects/{project}/workspaces/{workspace}/materials/{material}/canvas"
mux.HandleFunc(prefix+"/model", s.protected(callbacks, s.getCanvasModel))
mux.HandleFunc(prefix+"/model/migrate", s.protected(callbacks, s.migrateCanonicalCanvas))
mux.HandleFunc(prefix+"/ai/runs", s.protected(callbacks, s.createCanvasAIRun))
mux.HandleFunc(prefix+"/ai/runs/{run}", s.protected(callbacks, s.getCanvasAIRun))
mux.HandleFunc(prefix+"/ai/runs/{run}/apply", s.protected(callbacks, s.applyCanvasAIOutput))
mux.HandleFunc(prefix+"/ai/runs/{run}/cancel", s.protected(callbacks, s.cancelCanvasAIRun))
mux.HandleFunc(prefix+"/ai/operations/{request}", s.protected(callbacks, s.getCanvasAIAdmission))
mux.HandleFunc("/projects/{project}/workspaces/{workspace}/assets/{asset}", s.protected(callbacks, s.getAsset))
base := func(board string) string {
return "/projects/" + project + "/workspaces/" + workspace + "/materials/" + board + "/canvas"
}
call := func(method, path string, input any, status int, version int64, key string) response {
t.Helper()
raw, _ := json.Marshal(input)
r := httptest.NewRequest(method, path, bytes.NewReader(raw))
r.Header.Set("Authorization", "Bearer "+token)
r.Header.Set("Accept", "application/vnd.foundry.canvas+json;version=2")
if version > 0 {
r.Header.Set("If-Match", `"`+strconv.FormatInt(version, 10)+`"`)
}
if key != "" {
r.Header.Set("Idempotency-Key", key)
}
w := httptest.NewRecorder()
mux.ServeHTTP(w, r)
if w.Code != status {
t.Fatalf("%s %s got %d %s want %d", method, path, w.Code, w.Body.String(), status)
}
result := response{}
if err := json.Unmarshal(w.Body.Bytes(), &result); err != nil {
t.Fatal(err)
}
return result
}
read := func() canvasModel {
t.Helper()
tx, e := db.BeginTx(ctx, nil)
if e != nil {
t.Fatal(e)
}
defer tx.Rollback()
model, e := loadCanvasModel(ctx, tx, project, workspace, root)
if e != nil {
t.Fatal(e)
}
return model
}
mutate := func(edit func(*canvasModel)) {
t.Helper()
tx, e := db.BeginTx(ctx, nil)
if e != nil {
t.Fatal(e)
}
defer tx.Rollback()
if _, _, e = lockMaterial(ctx, tx, project, workspace, root, account); e != nil {
t.Fatal(e)
}
before, e := loadCanvasModel(ctx, tx, project, workspace, root)
if e != nil {
t.Fatal(e)
}
after := cloneModel(before)
edit(&after)
participants := map[string]bool{}
for id := range after.Canvases {
after.Versions[id] = before.Versions[id] + 1
participants[id] = true
}
if e = validateCanvasModel(after); e != nil {
t.Fatal(e)
}
if e = s.saveCanvasModel(queueRequest(ctx), tx, project, workspace, account, before, after, participants); e != nil {
t.Fatal(e)
}
if e = tx.Commit(); e != nil {
t.Fatal(e)
}
}
migrationKey := modelTestID(6000)
migration := call("POST", base(root)+"/model/migrate", response{"protocol_version": 2, "request_id": migrationKey, "expected_versions": map[string]int64{root: 1}}, 200, 0, migrationKey)
node := ""
for _, entry := range migration["identity_map"].([]any) {
mapping := entry.(map[string]any)
if mapping["sourceId"] == "a" {
node = mapping["elementId"].(string)
}
}
if node == "" {
t.Fatal("missing migrated target")
}
mutate(func(model *canvasModel) {
model.Elements[node]["geometry"].(map[string]any)["width"] = float64(9000)
model.Elements[node]["opaque"] = map[string]any{"kept": true}
model.Elements[node]["data"].(map[string]any)["extension"] = map[string]any{"constructor": map[string]any{"kept": true}}
model.Elements["future"] = modelTestNode("future", root)
model.Elements["future"]["componentId"] = "future-component"
model.Elements["future"]["dataVersion"] = float64(12)
})
next := 6100
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, "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)
return call("POST", base(board)+"/ai/runs", input, 202, read().Versions[board], key)
}
execute := func(id string) {
t.Helper()
run, e := s.claimCanvasRun(ctx)
if e != nil || run == nil || run.ID != id {
t.Fatalf("claim %v %v", run, e)
}
s.executeCanvasRun(ctx, *run)
}
input, key := command(root, node, "text", 1)
sourceVersion := read().Versions[root]
admitted := call("POST", base(root)+"/ai/runs", input, 202, sourceVersion, key)
if admitted["admission"] != "accepted" || admitted["target_incarnation"] != float64(1) || admitted["model"] == nil {
t.Fatal("canonical admission lost identity/base")
}
retry := call("POST", base(root)+"/ai/runs", input, 202, sourceVersion, key)
if !modelEqual(admitted, retry) {
t.Fatal("admission retry changed result")
}
queried := call("GET", base(root)+"/ai/operations/"+key, nil, 200, 0, "")
if !modelEqual(admitted, queried) {
t.Fatal("lost admission response not recoverable")
}
unknown := call("GET", base(root)+"/ai/operations/"+modelTestID(6099), nil, 404, 0, "")
if unknown["status"] != "unknown" || unknown["terminal"] == true {
t.Fatal("404 pretended zero execution")
}
duplicate := response{}
for k, v := range input {
duplicate[k] = v
}
duplicateKey := modelTestID(6098)
duplicate["request_id"] = duplicateKey
call("POST", base(root)+"/ai/runs", duplicate, 409, read().Versions[root], duplicateKey)
mutate(func(model *canvasModel) { model.Elements[node]["geometry"].(map[string]any)["x"] = float64(33) })
execute(admitted["run_id"].(string))
saved := read()
data := saved.Elements[node]["data"].(map[string]any)
if data["text"] != "generated result" || saved.Elements[node]["geometry"].(map[string]any)["x"] != float64(33) || saved.Elements[node]["geometry"].(map[string]any)["width"] != float64(9000) || saved.Elements["future"] == nil || saved.Elements[node]["opaque"] == nil || data["extension"] == nil {
t.Fatal("AI lost canonical fields, geometry or future records")
}
// A content conflict retains its output; explicit application is still a narrow canonical write.
conflictRun := start(root, node, "text", 1)
hooks <- func() {
mutate(func(model *canvasModel) { model.Elements[node]["data"].(map[string]any)["text"] = "user edit" })
}
execute(conflictRun["run_id"].(string))
conflictPath := base(root) + "/ai/runs/" + conflictRun["run_id"].(string)
conflict := call("GET", conflictPath, nil, 200, 0, "")
if conflict["status"] != "conflict" {
t.Fatal("changed content was overwritten")
}
outputID := conflict["outputs"].([]any)[0].(map[string]any)["output"].(map[string]any)["output_id"].(string)
applyKey := modelTestID(6200)
expected := read().Versions[root]
applyInput := response{"node_id": node, "output_id": outputID}
applied := call("POST", conflictPath+"/apply", applyInput, 200, expected, applyKey)
again := call("POST", conflictPath+"/apply", applyInput, 200, expected, applyKey)
if !modelEqual(applied, again) || applied["model"] == nil || read().Elements[node]["geometry"].(map[string]any)["width"] != float64(9000) {
t.Fatal("explicit canonical application changed geometry or receipt")
}
// Keep one unapplied old output, then replace the target incarnation atomically.
stale := start(root, node, "text", 1)
hooks <- func() {
mutate(func(model *canvasModel) { model.Elements[node]["data"].(map[string]any)["text"] = "keep this" })
}
execute(stale["run_id"].(string))
stalePath := base(root) + "/ai/runs/" + stale["run_id"].(string)
staleResult := call("GET", stalePath, nil, 200, 0, "")
staleOutput := staleResult["outputs"].([]any)[0].(map[string]any)["output"].(map[string]any)["output_id"].(string)
old := start(root, node, "text", 1)
run, e := s.claimCanvasRun(ctx)
if e != nil || run == nil || run.ID != old["run_id"] {
t.Fatalf("claim stale %v", e)
}
if ok, e := s.startCanvasStep(ctx, *run, run.Plan.Steps[0]); e != nil || !ok {
t.Fatalf("start stale %v", e)
}
mutate(func(model *canvasModel) {
model.Elements[node]["incarnation"] = float64(2)
model.Tombstones[node] = 1
model.TombstoneOwners[node] = root
})
state, e := s.finishCanvasStep(ctx, *run, run.Plan.Steps[0], "late output", "", true)
if e != nil || state != "cancelled" {
t.Fatalf("restored target accepted old callback %s %v", state, e)
}
if e = s.completeCanvasRun(ctx, *run, nil); e != nil {
t.Fatal(e)
}
if read().Elements[node]["data"].(map[string]any)["text"] != "keep this" {
t.Fatal("restore was overwritten")
}
call("POST", stalePath+"/apply", response{"node_id": node, "output_id": staleOutput}, 409, read().Versions[root], modelTestID(6201))
bad, badKey := command(root, node, "text", 1)
call("POST", base(root)+"/ai/runs", bad, 409, read().Versions[root], badKey)
restored := start(root, node, "text", 2)
cancelPath := base(root) + "/ai/runs/" + restored["run_id"].(string) + "/cancel"
call("POST", cancelPath, response{}, 200, 0, modelTestID(6202))
if executions.Load() != 3 {
t.Fatal("duplicate/cancelled/old execution called the model")
}
// Image output resolves a managed resource while retaining the element envelope.
imageID := modelTestID(6300)
mutate(func(model *canvasModel) {
record := modelTestNode(imageID, root)
record["componentId"] = "plain-image"
record["data"].(map[string]any)["src"] = "data:image/png;base64," + base64.StdEncoding.EncodeToString(pngBytes.Bytes())
record["geometry"].(map[string]any)["width"] = float64(10000)
model.Elements[imageID] = record
})
imageRun := start(root, imageID, "image", 1)
execute(imageRun["run_id"].(string))
imageRecord := read().Elements[imageID]
imageData := imageRecord["data"].(map[string]any)
if imageRecord["geometry"].(map[string]any)["width"] != float64(10000) || imageData["assetId"] == nil || !strings.HasPrefix(modelString(imageData["src"]), s.assetURL(project, workspace, "")) {
t.Fatal("image result lost geometry or resource identity")
}
request := httptest.NewRequest("GET", modelString(imageData["src"]), nil)
request.Header.Set("Authorization", "Bearer "+token)
download := httptest.NewRecorder()
mux.ServeHTTP(download, request)
if download.Code != 200 || !bytes.Equal(download.Body.Bytes(), pngBytes.Bytes()) {
t.Fatal("canonical generated asset is unreadable")
}
moved := start(root, imageID, "image", 1)
moving, e := s.claimCanvasRun(ctx)
if e != nil || moving == nil || moving.ID != moved["run_id"] {
t.Fatalf("claim move %v", e)
}
if ok, e := s.startCanvasStep(ctx, *moving, moving.Plan.Steps[0]); e != nil || !ok {
t.Fatalf("start move %v", e)
}
child, frame := modelTestID(6301), modelTestID(6302)
mutate(func(model *canvasModel) {
entry := modelTestNode(frame, root)
entry["componentId"] = "frame"
model.Elements[frame] = entry
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)
}
if e = s.completeCanvasRun(ctx, *moving, nil); e != nil {
t.Fatal(e)
}
childRun := start(child, imageID, "image", 1)
execute(childRun["run_id"].(string))
if read().Elements[imageID]["ownerCanvasId"] != child {
t.Fatal("child result moved its target back")
}
}
+58
View File
@@ -0,0 +1,58 @@
package platform
import (
"encoding/json"
"testing"
)
func TestCanonicalAIPatchPreservesOpaqueModelAndGeometry(t *testing.T) {
before := modelTestDocument()
before.Elements["a"]["geometry"].(map[string]any)["width"] = float64(9000)
before.Elements["a"]["extension"] = map[string]any{"kept": true}
before.Elements["a"]["data"].(map[string]any)["opaque"] = map[string]any{"constructor": map[string]any{"kept": true}}
before.Elements["future"] = modelTestNode("future", modelRootID)
before.Elements["future"]["dataVersion"] = float64(12)
original := cloneModel(before)
after, err := prepareModelAIPatch(before, modelRootID, "a", response{"text": "generated"})
if err != nil {
t.Fatal(err)
}
if !modelEqual(before, original) {
t.Fatal("AI staging changed its source")
}
if after.Elements["a"]["data"].(map[string]any)["text"] != "generated" || after.Versions[modelRootID] != 2 {
t.Fatal("AI text/version missing")
}
if !modelEqual(after.Elements["a"]["geometry"], before.Elements["a"]["geometry"]) || !modelEqual(after.Elements["a"]["extension"], before.Elements["a"]["extension"]) || !modelEqual(after.Elements["future"], before.Elements["future"]) {
t.Fatal("AI staging replaced unrelated fields")
}
if _, err = prepareModelAIPatch(before, modelRootID, "a", response{"geometry": map[string]any{}}); err != errAIInvalidPatch {
t.Fatalf("geometry patch was accepted: %v", err)
}
if _, err = prepareModelAIPatch(before, modelRootID, "future", response{"text": "lost"}); err != errAITargetChanged {
t.Fatalf("future schema was interpreted: %v", err)
}
if _, err = prepareModelAIPatch(before, modelChildID, "a", response{"text": "wrong owner"}); err != errAITargetChanged {
t.Fatalf("wrong owner was accepted: %v", err)
}
}
func TestCanonicalAIProjectionDoesNotExecuteFutureConnections(t *testing.T) {
model := modelTestDocument()
model.Elements["b"] = modelTestNode("b", modelRootID)
model.Elements["future-edge"] = map[string]any{"id": "future-edge", "incarnation": float64(1), "structure": "connection", "componentId": "connection", "dataVersion": float64(9), "ownerCanvasId": modelRootID, "source": map[string]any{"elementId": "a"}, "target": map[string]any{"elementId": "b"}, "path": nil, "data": map[string]any{}}
if err := validateCanvasModel(model); err != nil {
t.Fatal(err)
}
projection, err := projectCanvasModel(model)
if err != nil {
t.Fatal(err)
}
var document map[string]any
if err = json.Unmarshal(projection[modelRootID], &document); err != nil {
t.Fatal(err)
}
if len(document["edges"].([]any)) != 0 || model.Elements["future-edge"] == nil {
t.Fatal("future connection was executed or removed")
}
}
+14 -12
View File
@@ -15,18 +15,20 @@ type canvasAIStaticInput struct {
Content map[string]any `json:"content"`
}
type canvasAIStepPlan struct {
StaticInputs []canvasAIStaticInput `json:"static_inputs"`
MaxExecutionMS int `json:"max_execution_ms"`
ID string `json:"step_id"`
BoardID string `json:"board_id"`
NodeID string `json:"node_id"`
Kind string `json:"kind"`
ModelID string `json:"model_id"`
Config map[string]any `json:"config"`
FrozenContent map[string]any `json:"frozen_content"`
Fingerprint string `json:"target_fingerprint"`
Status string `json:"status"`
Dependencies []string `json:"dependencies"`
CanonicalRoot string `json:"canonical_root,omitempty"`
TargetIncarnation int64 `json:"target_incarnation,omitempty"`
StaticInputs []canvasAIStaticInput `json:"static_inputs"`
MaxExecutionMS int `json:"max_execution_ms"`
ID string `json:"step_id"`
BoardID string `json:"board_id"`
NodeID string `json:"node_id"`
Kind string `json:"kind"`
ModelID string `json:"model_id"`
Config map[string]any `json:"config"`
FrozenContent map[string]any `json:"frozen_content"`
Fingerprint string `json:"target_fingerprint"`
Status string `json:"status"`
Dependencies []string `json:"dependencies"`
}
type canvasAIEdgePlan struct {
ID string `json:"logical_edge_id"`
+35 -1
View File
@@ -14,7 +14,7 @@ func publicAIPlan(plan canvasAIPlan) response {
for _, source := range step.StaticInputs {
static = append(static, response{"logical_edge_id": source.EdgeID, "board_id": source.BoardID, "node_id": source.NodeID, "kind": source.Kind, "source": "frozen_snapshot"})
}
steps = append(steps, response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "kind": step.Kind, "model_id": step.ModelID, "max_execution_ms": step.MaxExecutionMS, "status": step.Status, "dependencies": step.Dependencies, "static_inputs": static})
steps = append(steps, response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "target_incarnation": step.TargetIncarnation, "canonical_root": step.CanonicalRoot, "kind": step.Kind, "model_id": step.ModelID, "max_execution_ms": step.MaxExecutionMS, "status": step.Status, "dependencies": step.Dependencies, "static_inputs": static})
}
return response{"steps": steps, "edges": plan.Edges, "frozen_versions": plan.Versions}
}
@@ -158,3 +158,37 @@ func (s *service) getCanvasAIRun(w http.ResponseWriter, r *http.Request, project
reply(w, 200, result)
return nil
}
// Resolve admission by the original client key, before the caller knows a run ID.
// Absence is unknown, not a terminal proof that an earlier invocation had no effect.
func (s *service) getCanvasAIAdmission(w http.ResponseWriter, r *http.Request, project, account string) error {
if r.Method != http.MethodGet {
reject(w, 405, "method_not_allowed")
return nil
}
key := r.PathValue("request")
if !uuidSlugPattern.MatchString(key) {
reject(w, 400, "invalid_request_id")
return nil
}
workspace, board := r.PathValue("workspace"), r.PathValue("material")
tx, err := s.db.BeginTx(r.Context(), nil)
if err != nil {
return err
}
defer tx.Rollback()
if _, _, err = lockMaterial(r.Context(), tx, project, workspace, board, account); err != nil {
return err
}
var body json.RawMessage
err = tx.QueryRowContext(r.Context(), `SELECT create_response FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND account_id=$4 AND idempotency_key=$5`, project, workspace, board, account, key).Scan(&body)
if err == sql.ErrNoRows {
reply(w, 404, response{"protocol_version": 2, "request_id": key, "status": "unknown", "error": "ai_admission_unknown"})
return nil
}
if err != nil {
return err
}
reply(w, 200, body)
return nil
}
+52 -14
View File
@@ -47,11 +47,15 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj
return nil
}
var input struct {
NodeID string `json:"node_id"`
ModelID string `json:"model_id"`
Prompt string `json:"prompt"`
Parameters map[string]any `json:"parameters"`
Propagate bool `json:"propagate"`
ProtocolVersion int `json:"protocol_version,omitempty"`
OperationID string `json:"operation_id,omitempty"`
RequestID string `json:"request_id,omitempty"`
TargetIncarnation int64 `json:"target_incarnation,omitempty"`
NodeID string `json:"node_id"`
ModelID string `json:"model_id"`
Prompt string `json:"prompt"`
Parameters map[string]any `json:"parameters"`
Propagate bool `json:"propagate"`
}
decoder := json.NewDecoder(bytes.NewReader(raw))
decoder.DisallowUnknownFields()
@@ -59,6 +63,10 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj
reject(w, 400, "invalid_ai_request")
return nil
}
if input.OperationID != "" && !uuidSlugPattern.MatchString(input.OperationID) {
reject(w, 400, "invalid_ai_request")
return nil
}
if input.Parameters == nil {
input.Parameters = map[string]any{}
}
@@ -105,6 +113,16 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj
if err != sql.ErrNoRows {
return err
}
if input.OperationID != "" {
var admitted bool
if err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND logical_operation_id=$4)`, project, workspace, account, input.OperationID).Scan(&admitted); err != nil {
return err
}
if admitted {
reject(w, 409, "operation_identity_conflict")
return nil
}
}
parent, err := loadCanvas(r, tx, project, workspace, board)
if err != nil {
return err
@@ -130,11 +148,25 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj
reject(w, 422, "model_incompatible")
return nil
}
target["ai"] = map[string]any{"enabled": true, "modelId": input.ModelID, "prompt": input.Prompt, "parameters": input.Parameters, "propagate": input.Propagate}
doc["schema_version"] = 2
encoded, _ := json.Marshal(doc)
normalized, ok := normalizeCanvas(encoded)
if !ok {
root, incarnation, _, err := canvasAIIdentity(r.Context(), tx, project, workspace, board, input.NodeID)
if err != nil {
return err
}
if root != "" {
if input.ProtocolVersion != 2 || input.RequestID != key || !uuidSlugPattern.MatchString(input.OperationID) || input.TargetIncarnation < 1 {
reject(w, 400, "invalid_ai_protocol")
return nil
}
if input.TargetIncarnation != incarnation {
reject(w, 409, "target_changed")
return nil
}
} else if input.ProtocolVersion != 0 || input.RequestID != "" || input.TargetIncarnation != 0 || input.OperationID != "" {
reject(w, 409, "canonical_migration_required")
return nil
}
change, err := prepareCanvasAIPatch(r, tx, project, workspace, parent, input.NodeID, response{"ai": map[string]any{"enabled": true, "modelId": input.ModelID, "prompt": input.Prompt, "parameters": input.Parameters, "propagate": input.Propagate}})
if err != nil {
reject(w, 400, "invalid_ai_request")
return nil
}
@@ -149,8 +181,7 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj
for _, model := range models {
catalogue[model.ID] = model
}
parent.Document = normalized
parent.Version++
parent = change.Snapshot
plan, err := loadCanvasAIPlan(r, tx, project, workspace, account, parent, input.NodeID, input.Propagate)
if err != nil {
if err.Error() == "execution_cycle" {
@@ -160,6 +191,9 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj
}
return nil
}
if err = bindCanvasAIIdentities(r.Context(), tx, project, workspace, &plan); err != nil {
return err
}
stepKinds := map[string]string{}
for _, step := range plan.Steps {
stepKinds[step.ID] = step.Kind
@@ -287,12 +321,16 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj
planJSON, _ := json.Marshal(plan)
poll := "/projects/" + url.PathEscape(project) + "/workspaces/" + workspace + "/materials/" + board + "/canvas/ai/runs/" + id
result := response{"run_id": id, "status": "queued", "poll_url": poll, "plan": publicAIPlan(plan), "canvas_version": parent.Version}
if root != "" {
result["protocol_version"], result["request_id"], result["admission"], result["operation_id"] = 2, key, "accepted", input.OperationID
change.addCanonical(result)
}
resultJSON, _ := json.Marshal(result)
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_runs(project,workspace_id,board_id,id,account_id,idempotency_key,payload_hash,status,plan,create_response) VALUES($1,$2,$3,$4,$5,$6,$7,'queued',$8,$9)`, project, workspace, board, id, account, key, hash, planJSON, resultJSON)
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_runs(project,workspace_id,board_id,id,account_id,idempotency_key,payload_hash,status,plan,create_response,logical_operation_id) VALUES($1,$2,$3,$4,$5,$6,$7,'queued',$8,$9,NULLIF($10,'')::uuid)`, project, workspace, board, id, account, key, hash, planJSON, resultJSON, input.OperationID)
if err != nil {
return err
}
if err = s.saveCanvas(r, tx, project, workspace, parent); err != nil {
if err = s.saveCanvasAIChange(r, tx, project, workspace, account, change); err != nil {
return err
}
if err = appendAIEvent(r.Context(), tx, project, workspace, id, "run.queued", response{"status": "queued"}); err != nil {
+1 -1
View File
@@ -39,7 +39,7 @@ func (s *service) canvasCapabilities(w http.ResponseWriter, r *http.Request, pro
}
}
}
ai := response{"available": textAvailable || imageAvailable, "text": textAvailable, "image": imageAvailable}
ai := response{"canonical_target_identity": true, "admission_protocol_version": 2, "admission_receipt_query": true, "available": textAvailable || imageAvailable, "text": textAvailable, "image": imageAvailable}
if modelErr != nil {
ai["unavailable_reason"] = "codex_unavailable"
}
+4 -1
View File
@@ -83,6 +83,9 @@ func loadCanvasModel(ctx context.Context, tx *sql.Tx, project, workspace, root s
// projection for internal AI readers, never the source of a canonical GET/replay.
func (s *service) saveCanvasModel(r *http.Request, tx *sql.Tx, project, workspace, account string, before, after canvasModel, participants map[string]bool) error {
ctx := r.Context()
if err := invalidateCanonicalAIIdentities(r, tx, project, workspace, before, after); err != nil {
return err
}
if _, err := tx.ExecContext(ctx, `SELECT set_config('foundry.canonical_writer','on',true)`); err != nil {
return err
}
@@ -249,7 +252,7 @@ func projectCanvasModel(model canvasModel) (map[string]json.RawMessage, error) {
}
for _, id := range elements {
element := model.Elements[id]
if element["structure"] != "connection" {
if element["structure"] != "connection" || element["componentId"] != "connection" || modelInt(element["dataVersion"]) != 1 {
continue
}
owner := modelString(element["ownerCanvasId"])
+1
View File
@@ -155,6 +155,7 @@ func Register(mux *http.ServeMux, db *sql.DB, callbacks map[string]string, origi
mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}", s.protected(callbacks, s.material))
mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas", s.protected(callbacks, s.canvas))
mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/models", s.protected(callbacks, s.canvasAIModels))
mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/operations/{request}", s.protected(callbacks, s.getCanvasAIAdmission))
mux.HandleFunc("POST /projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/runs", s.protected(callbacks, s.createCanvasAIRun))
mux.HandleFunc("GET /projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/runs", s.protected(callbacks, s.listCanvasAIRuns))
mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/runs/{run}", s.protected(callbacks, s.getCanvasAIRun))