Stage atomic canvas operation batches and durable receipts

internal/platform/canvas_operations.go

@@ -0,0 +1,147 @@
+package platform
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+)
+
+const canvasOperationsProtocol = 1
+
+type canvasOperation struct {
+	ID        string                     `json:"id"`
+	Type      string                     `json:"type"`
+	CanvasID  string                     `json:"canvas_id"`
+	ElementID string                     `json:"element_id"`
+	Record    json.RawMessage            `json:"record,omitempty"`
+	Patch     map[string]json.RawMessage `json:"patch,omitempty"`
+}
+type canvasOperationBatch struct {
+	ProtocolVersion  int               `json:"protocol_version"`
+	RequestID        string            `json:"request_id"`
+	ExpectedVersions map[string]int64  `json:"expected_versions"`
+	Operations       []canvasOperation `json:"operations"`
+}
+
+var errInvalidCanvasOperation = errors.New("invalid_canvas_operation")
+
+// applyCanvasOperations stages a complete batch. Input snapshots are never mutated.
+// HTTP/auth/version/idempotency are checked by the transaction coordinator.
+func applyCanvasOperations(initial map[string]canvasSnapshot, operations []canvasOperation) (map[string]canvasSnapshot, error) {
+	if len(operations) == 0 || len(operations) > 2500 {
+		return nil, errInvalidCanvasOperation
+	}
+	documents := make(map[string]map[string]json.RawMessage, len(initial))
+	for id, snapshot := range initial {
+		var document map[string]json.RawMessage
+		if json.Unmarshal(snapshot.Document, &document) != nil {
+			return nil, errInvalidCanvasOperation
+		}
+		documents[id] = document
+	}
+	seen := map[string]bool{}
+	for _, operation := range operations {
+		if !uuidSlugPattern.MatchString(operation.ID) || seen[operation.ID] || !canvasID.MatchString(operation.ElementID) {
+			return nil, errInvalidCanvasOperation
+		}
+		seen[operation.ID] = true
+		document := documents[operation.CanvasID]
+		if document == nil {
+			return nil, errInvalidCanvasOperation
+		}
+		collection := ""
+		switch operation.Type {
+		case "node.create", "node.patch", "node.delete":
+			collection = "nodes"
+		case "edge.create", "edge.patch", "edge.delete":
+			collection = "edges"
+		default:
+			return nil, errInvalidCanvasOperation
+		}
+		var records []map[string]json.RawMessage
+		if json.Unmarshal(document[collection], &records) != nil {
+			return nil, errInvalidCanvasOperation
+		}
+		index := -1
+		for i, record := range records {
+			var id string
+			_ = json.Unmarshal(record["id"], &id)
+			if id == operation.ElementID {
+				index = i
+				break
+			}
+		}
+		switch operation.Type {
+		case "node.create", "edge.create":
+			if index >= 0 || len(operation.Record) == 0 || len(operation.Patch) != 0 {
+				return nil, fmt.Errorf("%w: create conflict", errInvalidCanvasOperation)
+			}
+			var record map[string]json.RawMessage
+			var id string
+			if json.Unmarshal(operation.Record, &record) != nil || json.Unmarshal(record["id"], &id) != nil || id != operation.ElementID {
+				return nil, errInvalidCanvasOperation
+			}
+			records = append(records, record)
+		case "node.delete", "edge.delete":
+			if index < 0 || len(operation.Record) != 0 || len(operation.Patch) != 0 {
+				return nil, fmt.Errorf("%w: missing delete target", errInvalidCanvasOperation)
+			}
+			records = append(records[:index], records[index+1:]...)
+		case "node.patch", "edge.patch":
+			if index < 0 || len(operation.Record) != 0 || len(operation.Patch) == 0 {
+				return nil, errInvalidCanvasOperation
+			}
+			for field, value := range operation.Patch {
+				allowed := field == "data" || collection == "nodes" && (field == "position" || field == "width" || field == "height") || collection == "edges" && (field == "source" || field == "target" || field == "sourceHandle" || field == "targetHandle")
+				if !allowed {
+					return nil, errInvalidCanvasOperation
+				}
+				if field == "data" {
+					// Merge individual data fields; a geometry-only patch never rolls back titles/assets.
+					var before, patch map[string]json.RawMessage
+					if json.Unmarshal(records[index][field], &before) != nil || json.Unmarshal(value, &patch) != nil || patch == nil {
+						return nil, errInvalidCanvasOperation
+					}
+					if before == nil {
+						before = map[string]json.RawMessage{}
+					}
+					for key, item := range patch {
+						if string(item) == "null" {
+							delete(before, key)
+						} else {
+							before[key] = item
+						}
+					}
+					encoded, err := json.Marshal(before)
+					if err != nil {
+						return nil, err
+					}
+					records[index][field] = encoded
+				} else {
+					records[index][field] = append(json.RawMessage(nil), value...)
+				}
+			}
+		}
+		encoded, err := json.Marshal(records)
+		if err != nil {
+			return nil, err
+		}
+		document[collection] = encoded
+	}
+	result := make(map[string]canvasSnapshot, len(initial))
+	for id, document := range documents {
+		raw, err := json.Marshal(document)
+		if err != nil {
+			return nil, err
+		}
+		normalized, valid := normalizeCanvas(raw)
+		if !valid {
+			return nil, errInvalidCanvasOperation
+		}
+		snapshot := initial[id]
+		snapshot.Document = normalized
+		snapshot.Version++
+		result[id] = snapshot
+	}
+	return result, nil
+}

internal/platform/canvas_operations_http.go

@@ -0,0 +1,218 @@
+package platform
+
+import (
+	"database/sql"
+	"encoding/json"
+	"io"
+	"net/http"
+	"sort"
+	"time"
+)
+
+const canvasOperationsAction = "headless.operations.v1"
+
+func (s *service) submitCanvasOperations(w http.ResponseWriter, r *http.Request, project, account string) error {
+	if r.Method != http.MethodPost {
+		reject(w, 405, "method_not_allowed")
+		return nil
+	}
+	if !acceptsCanvasV2(r) {
+		reject(w, 409, "canvas_schema_upgrade_required")
+		return nil
+	}
+	key := canvasIdempotencyKey(r)
+	if !uuidSlugPattern.MatchString(key) {
+		reject(w, 400, "invalid_idempotency_key")
+		return nil
+	}
+	var input canvasOperationBatch
+	decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, canvasLimit))
+	decoder.DisallowUnknownFields()
+	if decoder.Decode(&input) != nil || decoder.Decode(new(any)) != io.EOF || input.ProtocolVersion != canvasOperationsProtocol || input.RequestID != key || len(input.Operations) == 0 || len(input.Operations) > 2500 || len(input.ExpectedVersions) == 0 || len(input.ExpectedVersions) > 64 {
+		reject(w, 400, "invalid_canvas_operation_batch")
+		return nil
+	}
+	workspace, board := r.PathValue("workspace"), r.PathValue("material")
+	if _, ok := input.ExpectedVersions[board]; !ok {
+		reject(w, 428, "version_required")
+		return nil
+	}
+	participants := make([]string, 0, len(input.ExpectedVersions))
+	for id, version := range input.ExpectedVersions {
+		if !uuidSlugPattern.MatchString(id) || version < 1 {
+			reject(w, 400, "invalid_canvas_operation_batch")
+			return nil
+		}
+		participants = append(participants, id)
+	}
+	for _, operation := range input.Operations {
+		if _, ok := input.ExpectedVersions[operation.CanvasID]; !ok {
+			reject(w, 428, "version_required")
+			return nil
+		}
+	}
+	sort.Strings(participants)
+	tx, err := s.db.BeginTx(r.Context(), nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	// lockMaterial serializes the workspace; sorted participants preserve a stable lock order.
+	for _, id := range participants {
+		kind, role, e := lockMaterial(r.Context(), tx, project, workspace, id, account)
+		if e != nil {
+			return e
+		}
+		if kind != "board" {
+			reject(w, 400, "invalid_material")
+			return nil
+		}
+		if role == "viewer" {
+			reject(w, 403, "forbidden")
+			return nil
+		}
+	}
+	// Resolve the same request before comparing expected versions: exact retry after lost ACK succeeds.
+	done, err := transactionRetry(w, r, tx, project, account, board, canvasOperationsAction, key, payloadDigest(input))
+	if err != nil {
+		return err
+	}
+	if done {
+		return nil
+	}
+	before := make([]canvasSnapshot, 0, len(participants))
+	initial := map[string]canvasSnapshot{}
+	for _, id := range participants {
+		snapshot, e := loadCanvas(r, tx, project, workspace, id)
+		if e != nil {
+			return e
+		}
+		if snapshot.Version != input.ExpectedVersions[id] {
+			return commitCanvasOperationReceipt(w, r, tx, project, account, board, input, 412, response{"error": "version_conflict", "status": "not-applied", "terminal": true, "request_id": key}, nil, nil)
+		}
+		initial[id] = snapshot
+		before = append(before, snapshot)
+	}
+	staged, err := applyCanvasOperations(initial, input.Operations)
+	if err != nil {
+		return commitCanvasOperationReceipt(w, r, tx, project, account, board, input, 400, response{"error": "invalid_canvas_operation", "status": "not-applied", "terminal": true, "request_id": key}, nil, nil)
+	}
+	snapshots := make([]canvasSnapshot, 0, len(participants))
+	versions := map[string]int64{}
+	// Validate all participant documents before writing any of them.
+	for _, id := range participants {
+		snapshot := staged[id]
+		valid, e := validateCanvasRelations(r, tx, project, workspace, id, initial[id].Document, snapshot.Document)
+		if e != nil {
+			return e
+		}
+		if !valid {
+			return commitCanvasOperationReceipt(w, r, tx, project, account, board, input, 409, response{"error": "ownership_conflict", "status": "not-applied", "terminal": true, "request_id": key}, nil, nil)
+		}
+		if !s.validateCanvasAIModels(w, r, snapshot.Document) {
+			return nil
+		}
+		valid, e = s.validateCanvasAssets(r, tx, project, workspace, account, snapshot.Document)
+		if e != nil {
+			return e
+		}
+		if !valid {
+			return commitCanvasOperationReceipt(w, r, tx, project, account, board, input, 400, response{"error": "invalid_asset_reference", "status": "not-applied", "terminal": true, "request_id": key}, nil, nil)
+		}
+		snapshots = append(snapshots, snapshot)
+		versions[id] = snapshot.Version
+	}
+	for _, snapshot := range snapshots {
+		if err = s.saveCanvas(r, tx, project, workspace, snapshot); err != nil {
+			return err
+		}
+	}
+	operationIDs := make([]string, 0, len(input.Operations))
+	for _, operation := range input.Operations {
+		operationIDs = append(operationIDs, operation.ID)
+	}
+	body := response{"protocol_version": canvasOperationsProtocol, "status": "applied", "request_id": key, "operation_ids": operationIDs, "snapshots": snapshots, "applied_versions": versions, "parent": staged[board]}
+	return commitCanvasOperationReceipt(w, r, tx, project, account, board, input, 200, body, before, versions)
+}
+
+// A terminal rejection also reserves the frozen key/body. Losing that response must
+// not turn a definite zero-write result into a future execution of the same request.
+func commitCanvasOperationReceipt(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, account, board string, input canvasOperationBatch, status int, body response, before []canvasSnapshot, versions map[string]int64) error {
+	var transactionID string
+	if err := tx.QueryRowContext(r.Context(), `SELECT gen_random_uuid()::text`).Scan(&transactionID); err != nil {
+		return err
+	}
+	body["transaction_id"] = transactionID
+	body["protocol_version"] = canvasOperationsProtocol
+	expires := time.Now().Add(7 * 24 * time.Hour)
+	bodyRaw, err := json.Marshal(body)
+	if err != nil {
+		return err
+	}
+	beforeRaw, err := json.Marshal(before)
+	if err != nil {
+		return err
+	}
+	versionsRaw, err := json.Marshal(versions)
+	if err != nil {
+		return err
+	}
+	_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_transactions(project,workspace_id,board_id,id,account_id,action,idempotency_key,payload_hash,response_status,response_body,before_state,after_versions,expires_at) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13)`, project, r.PathValue("workspace"), board, transactionID, account, canvasOperationsAction, input.RequestID, payloadDigest(input), status, bodyRaw, beforeRaw, versionsRaw, expires)
+	if err != nil {
+		return err
+	}
+	if err = tx.Commit(); err != nil {
+		return err
+	}
+	if version := versions[board]; version > 0 {
+		setVersion(w, version)
+	}
+	reply(w, status, body)
+	return nil
+}
+
+func (s *service) getCanvasOperation(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
+	}
+	tx, err := s.db.BeginTx(r.Context(), nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	workspace, board := r.PathValue("workspace"), r.PathValue("material")
+	if _, _, err = lockMaterial(r.Context(), tx, project, workspace, board, account); err != nil {
+		return err
+	}
+	var raw json.RawMessage
+	err = tx.QueryRowContext(r.Context(), `SELECT response_body FROM canvas_transactions WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND account_id=$4 AND action=$5 AND idempotency_key=$6`, project, workspace, board, account, canvasOperationsAction, key).Scan(&raw)
+	if err == sql.ErrNoRows {
+		reply(w, 404, response{"status": "unknown", "error": "operation_receipt_unknown", "request_id": key})
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	var result struct {
+		Snapshots []canvasSnapshot `json:"snapshots"`
+	}
+	if err = json.Unmarshal(raw, &result); err != nil {
+		return err
+	}
+	for _, snapshot := range result.Snapshots {
+		if _, _, err = lockMaterial(r.Context(), tx, project, workspace, snapshot.BoardID, account); err != nil {
+			return err
+		}
+	}
+	if err = tx.Rollback(); err != nil {
+		return err
+	}
+	reply(w, 200, raw)
+	return nil
+}

internal/platform/canvas_operations_integration_test.go

@@ -0,0 +1,166 @@
+package platform
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"sync"
+	"testing"
+)
+
+func TestCanvasOperationsTransactions(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 [12]byte
+	_, _ = rand.Read(entropy[:])
+	project := "operations-" + hex.EncodeToString(entropy[:])
+	defer func() {
+		for _, table := range []string{"canvas_transactions", "workspaces", "project_accounts"} {
+			_, _ = db.ExecContext(ctx, "DELETE FROM "+table+" WHERE project=$1", project)
+		}
+	}()
+	accounts, tokens := []string{}, []string{}
+	for range 2 {
+		var account 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)
+		}
+		var value [32]byte
+		_, _ = rand.Read(value[:])
+		token := base64.RawURLEncoding.EncodeToString(value[:])
+		hash := sha256.Sum256([]byte(token))
+		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(hash[:]), project, account); err != nil {
+			t.Fatal(err)
+		}
+		accounts, tokens = append(accounts, account), append(tokens, token)
+	}
+	var workspace string
+	if err = db.QueryRowContext(ctx, `INSERT INTO workspaces(project,name,owner_id) VALUES($1,'Operations',$2) RETURNING id::text`, project, accounts[0]).Scan(&workspace); err != nil {
+		t.Fatal(err)
+	}
+	for i, role := range []string{"admin", "viewer"} {
+		if _, err = db.ExecContext(ctx, `INSERT INTO workspace_members(project,workspace_id,account_id,role) VALUES($1,$2,$3,$4)`, project, workspace, accounts[i], role); err != nil {
+			t.Fatal(err)
+		}
+	}
+	boards := []string{}
+	for range 2 {
+		var board string
+		if err = db.QueryRowContext(ctx, `INSERT INTO materials(project,workspace_id,kind,title,created_by,canvas_document) VALUES($1,$2,'board','Board',$3,$4) RETURNING id::text`, project, workspace, accounts[0], json.RawMessage(operationsFixture)).Scan(&board); err != nil {
+			t.Fatal(err)
+		}
+		boards = append(boards, board)
+	}
+	mux := http.NewServeMux()
+	Register(mux, db, map[string]string{project: "https://app.example.com/auth/callback"}, "https://api.example.com", nil)
+	path := "/projects/" + project + "/workspaces/" + workspace + "/materials/" + boards[0] + "/canvas/operations"
+	key := func(n int) string { return fmt.Sprintf("a0000000-0000-4000-8000-%012d", n) }
+	do := func(method, path, token string, input canvasOperationBatch) *httptest.ResponseRecorder {
+		encoded, _ := json.Marshal(input)
+		request := httptest.NewRequest(method, path, bytes.NewReader(encoded))
+		request.Header.Set("Authorization", "Bearer "+token)
+		request.Header.Set("Accept", "application/vnd.foundry.canvas+json;version=2")
+		request.Header.Set("Idempotency-Key", input.RequestID)
+		writer := httptest.NewRecorder()
+		mux.ServeHTTP(writer, request)
+		return writer
+	}
+	call := func(method, path, token string, input canvasOperationBatch, status int) response {
+		t.Helper()
+		writer := do(method, path, token, input)
+		if writer.Code != status {
+			t.Fatalf("%s %s got %d %s want %d", method, path, writer.Code, writer.Body.String(), status)
+		}
+		out := response{}
+		if err := json.Unmarshal(writer.Body.Bytes(), &out); err != nil {
+			t.Fatal(err)
+		}
+		return out
+	}
+	version := func(board string) int64 {
+		t.Helper()
+		var value int64
+		if err := db.QueryRowContext(ctx, `SELECT canvas_version FROM materials WHERE project=$1 AND workspace_id=$2 AND id=$3`, project, workspace, board).Scan(&value); err != nil {
+			t.Fatal(err)
+		}
+		return value
+	}
+	batch := func(n int, expected int64) canvasOperationBatch {
+		return canvasOperationBatch{ProtocolVersion: 1, RequestID: key(n), ExpectedVersions: map[string]int64{boards[0]: expected}, Operations: []canvasOperation{{ID: key(n + 100), Type: "node.patch", CanvasID: boards[0], ElementID: "a", Patch: map[string]json.RawMessage{"data": json.RawMessage(`{"title":"renamed"}`)}}}}
+	}
+	firstBody := batch(1, 1)
+	first := call("POST", path, tokens[0], firstBody, 200)
+	retried := call("POST", path, tokens[0], firstBody, 200)
+	if first["transaction_id"] != retried["transaction_id"] || version(boards[0]) != 2 {
+		t.Fatal("retry duplicated writes or changed receipt identity")
+	}
+	queried := call("GET", path+"/"+key(1), tokens[0], canvasOperationBatch{}, 200)
+	if queried["status"] != "applied" || queried["transaction_id"] != first["transaction_id"] {
+		t.Fatal("lost acknowledgment cannot be resolved by request identity")
+	}
+	altered := batch(1, 2)
+	call("POST", path, tokens[0], altered, 409)
+	call("GET", path+"/"+key(1), tokens[1], canvasOperationBatch{}, 404)
+	call("POST", path, tokens[1], batch(2, 2), 403)
+	unknown := call("GET", path+"/"+key(999), tokens[0], canvasOperationBatch{}, 404)
+	if unknown["status"] != "unknown" || unknown["terminal"] == true {
+		t.Fatal("missing receipt must not claim terminal zero writes")
+	}
+	// Terminal zero-write responses must remain queryable and reserve their frozen body.
+	stale := batch(3, 1)
+	rejected := call("POST", path, tokens[0], stale, 412)
+	proof := call("GET", path+"/"+key(3), tokens[0], canvasOperationBatch{}, 200)
+	if proof["status"] != "not-applied" || proof["terminal"] != true || proof["transaction_id"] != rejected["transaction_id"] {
+		t.Fatal("terminal rejection was not persisted")
+	}
+	call("POST", path, tokens[0], batch(3, 2), 409)
+	if version(boards[0]) != 2 {
+		t.Fatal("rejected request performed a write")
+	}
+	// Concurrent exact retries serialize to one transaction and one version increment.
+	concurrent := batch(4, 2)
+	var wait sync.WaitGroup
+	results := make(chan *httptest.ResponseRecorder, 4)
+	for range 4 {
+		wait.Add(1)
+		go func() { defer wait.Done(); results <- do("POST", path, tokens[0], concurrent) }()
+	}
+	wait.Wait()
+	close(results)
+	for result := range results {
+		if result.Code != 200 {
+			t.Fatalf("concurrent retry got %d %s", result.Code, result.Body.String())
+		}
+	}
+	if version(boards[0]) != 3 {
+		t.Fatal("concurrent request executed more than once")
+	}
+	// A valid first participant and an invalid second participant produce no partial write.
+	atomic := batch(5, 3)
+	atomic.ExpectedVersions[boards[1]] = 1
+	atomic.Operations = append(atomic.Operations, canvasOperation{ID: key(500), Type: "node.delete", CanvasID: boards[1], ElementID: "missing"})
+	call("POST", path, tokens[0], atomic, 400)
+	if version(boards[0]) != 3 || version(boards[1]) != 1 {
+		t.Fatal("invalid multi-canvas batch partially committed")
+	}
+	if call("GET", path+"/"+key(5), tokens[0], canvasOperationBatch{}, 200)["status"] != "not-applied" {
+		t.Fatal("invalid batch lost its terminal receipt")
+	}
+}

internal/platform/canvas_operations_test.go

@@ -0,0 +1,63 @@
+package platform
+
+import (
+	"encoding/json"
+	"testing"
+)
+
+const operationsFixture = `{"schema_version":2,"nodes":[{"id":"a","type":"plainText","position":{"x":0,"y":0},"width":240,"height":72,"data":{"kind":"plain-text","title":"original","text":"hello"}}],"edges":[],"viewport":{"x":0,"y":0,"zoom":1}}`
+
+func TestCanvasOperationBatchStagesAndMerges(t *testing.T) {
+	initial := map[string]canvasSnapshot{"board": {BoardID: "board", Version: 7, Document: json.RawMessage(operationsFixture)}}
+	operations := []canvasOperation{
+		{ID: "11111111-1111-4111-8111-111111111111", Type: "node.patch", CanvasID: "board", ElementID: "a", Patch: map[string]json.RawMessage{"data": json.RawMessage(`{"title":"renamed"}`)}},
+		{ID: "22222222-2222-4222-8222-222222222222", Type: "node.patch", CanvasID: "board", ElementID: "a", Patch: map[string]json.RawMessage{"position": json.RawMessage(`{"x":90,"y":20}`)}},
+	}
+	result, err := applyCanvasOperations(initial, operations)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result["board"].Version != 8 {
+		t.Fatal("version did not advance once")
+	}
+	var document struct {
+		Nodes []struct {
+			Position struct{ X float64 }
+			Data     struct {
+				Title string
+				Text  string
+			}
+		}
+	}
+	if err = json.Unmarshal(result["board"].Document, &document); err != nil {
+		t.Fatal(err)
+	}
+	if document.Nodes[0].Data.Title != "renamed" || document.Nodes[0].Data.Text != "hello" || document.Nodes[0].Position.X != 90 {
+		t.Fatal("field merge lost unrelated content")
+	}
+	if string(initial["board"].Document) != operationsFixture {
+		t.Fatal("input snapshot mutated")
+	}
+}
+func TestCanvasOperationBatchRejectsWithoutPartialOutput(t *testing.T) {
+	initial := map[string]canvasSnapshot{"board": {BoardID: "board", Version: 1, Document: json.RawMessage(operationsFixture)}}
+	for _, operation := range []canvasOperation{
+		{ID: "11111111-1111-4111-8111-111111111111", Type: "node.patch", CanvasID: "board", ElementID: "a", Patch: map[string]json.RawMessage{"id": json.RawMessage(`"stolen"`)}},
+		{ID: "11111111-1111-4111-8111-111111111111", Type: "node.patch", CanvasID: "board", ElementID: "missing", Patch: map[string]json.RawMessage{"data": json.RawMessage(`{"title":"no"}`)}},
+		{ID: "11111111-1111-4111-8111-111111111111", Type: "edge.create", CanvasID: "board", ElementID: "edge", Record: json.RawMessage(`{"id":"edge","source":"a","target":"missing"}`)},
+	} {
+		if result, err := applyCanvasOperations(initial, []canvasOperation{operation}); err == nil || result != nil {
+			t.Fatal("invalid batch partially accepted")
+		}
+	}
+}
+func TestCanvasOperationBatchValidatesFinalGraphInsteadOfIntermediateState(t *testing.T) {
+	initial := map[string]canvasSnapshot{"board": {BoardID: "board", Version: 1, Document: json.RawMessage(operationsFixture)}}
+	operations := []canvasOperation{
+		{ID: "11111111-1111-4111-8111-111111111111", Type: "edge.create", CanvasID: "board", ElementID: "edge", Record: json.RawMessage(`{"id":"edge","source":"a","target":"b"}`)},
+		{ID: "22222222-2222-4222-8222-222222222222", Type: "node.create", CanvasID: "board", ElementID: "b", Record: json.RawMessage(`{"id":"b","type":"plainText","position":{"x":20,"y":20},"width":240,"height":72,"data":{"kind":"plain-text","title":"b","text":""}}`)},
+	}
+	if _, err := applyCanvasOperations(initial, operations); err != nil {
+		t.Fatal(err)
+	}
+}

internal/platform/http.go

@@ -147,6 +147,8 @@ func Register(mux *http.ServeMux, db *sql.DB, callbacks map[string]string, origi
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/runs/{run}/cancel", s.protected(callbacks, s.cancelCanvasAIRun))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai/runs/{run}/apply", s.protected(callbacks, s.applyCanvasAIOutput))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/capabilities", s.protected(callbacks, s.canvasCapabilities))
+	mux.HandleFunc("POST /projects/{project}/workspaces/{workspace}/materials/{material}/canvas/operations", s.protected(callbacks, s.submitCanvasOperations))
+	mux.HandleFunc("GET /projects/{project}/workspaces/{workspace}/materials/{material}/canvas/operations/{request}", s.protected(callbacks, s.getCanvasOperation))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/connections", s.protected(callbacks, s.createCanvasConnection))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/subgraphs", s.protected(callbacks, s.createSubgraph))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/subgraphs/migrate", s.protected(callbacks, s.migrateSubgraphs))