Add atomic connections across canvas subgraph boundaries

internal/platform/canvas_capabilities.go

@@ -44,7 +44,7 @@ func (s *service) canvasCapabilities(w http.ResponseWriter, r *http.Request, pro
 		"contract_version":          "canvas-v2-2026-11-06",
 		"supported_schema_versions": []int{1, 2},
 		"edge_features":             response{"appearance": true, "markers": true, "bezier_chain": true, "animation_configuration": true, "execution_events": true},
-		"subgraphs":                 response{"create": true, "migrate": true, "undo": true, "members": true, "dissolve": true},
+		"subgraphs":                 response{"create": true, "migrate": true, "undo": true, "members": true, "dissolve": true, "connections": true},
 		"ai":                        ai,
 		"limits":                    response{"max_depth": 8, "max_nodes": 500, "max_edges": 2000, "max_frame_ports": 4000, "max_curve_knots": 64, "max_snapshot_bytes": canvasLimit, "max_active_tasks": 128, "max_run_steps": 64, "max_execution_depth": 32, "max_input_images": 4, "max_input_image_bytes": 20 << 20, "max_coordinate": 1000000},
 		"retention":                 response{"idempotency_min_seconds": 86400, "undo_receipt_seconds": 604800},

internal/platform/canvas_connections.go

@@ -0,0 +1,351 @@
+package platform
+
+import (
+	"crypto/sha256"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"net/http"
+	"sort"
+	"time"
+)
+
+// Endpoints name real nodes, never transient projected copies or client-created ports.
+type canvasConnectionEndpoint struct {
+	Board  string `json:"board_id"`
+	Node   string `json:"node_id"`
+	Handle string `json:"handle"`
+}
+type canvasConnectionInput struct {
+	Source canvasConnectionEndpoint `json:"source"`
+	Target canvasConnectionEndpoint `json:"target"`
+	Edge   struct {
+		ID   string         `json:"id"`
+		Data map[string]any `json:"data,omitempty"`
+	} `json:"edge"`
+	Expected map[string]int64 `json:"expected_versions"`
+}
+
+// Paths run from a leaf board upwards to (but excluding) the coordinator.
+func connectionPath(r *http.Request, tx *sql.Tx, project, workspace, coordinator, board string) ([]subgraphRelation, error) {
+	path := []subgraphRelation{}
+	seen := map[string]bool{}
+	for board != coordinator {
+		if seen[board] || len(path) >= 8 {
+			return nil, errGraphOwnership
+		}
+		seen[board] = true
+		relation, err := readSubgraphRelation(r, tx, project, workspace, board)
+		if err == sql.ErrNoRows {
+			return nil, errGraphOwnership
+		}
+		if err != nil {
+			return nil, err
+		}
+		if relation.Archived {
+			return nil, errGraphOwnership
+		}
+		path = append(path, relation)
+		board = relation.Parent
+	}
+	return path, nil
+}
+
+func connectionNode(doc map[string]any, id string) map[string]any {
+	for _, value := range doc["nodes"].([]any) {
+		node := value.(map[string]any)
+		if node["id"] == id {
+			return node
+		}
+	}
+	return nil
+}
+
+// Reuse ports by logical endpoint, including legacy ports with non-derived IDs.
+// A derived ID collision must never retarget an existing port.
+func connectionPort(data map[string]any, node, handle, direction string) (string, bool, error) {
+	ports, _ := data["ports"].([]any)
+	sum := sha256.Sum256([]byte(direction + "/" + node + "/" + handle))
+	id := "p-" + hex.EncodeToString(sum[:16])
+	collision := false
+	for _, value := range ports {
+		port := value.(map[string]any)
+		if port["direction"] == direction && port["nodeId"] == node && port["handle"] == handle {
+			return port["id"].(string), false, nil
+		}
+		collision = collision || port["id"] == id
+	}
+	if collision {
+		return "", false, errGraphOwnership
+	}
+	if len(ports) >= 4000 {
+		return "", false, errGraphCapacity
+	}
+	data["ports"] = append(ports, map[string]any{"id": id, "direction": direction, "nodeId": node, "handle": handle})
+	return id, true, nil
+}
+
+func exposeConnectionEndpoint(endpoint canvasConnectionEndpoint, direction string, path []subgraphRelation, documents map[string]map[string]any) (canvasConnectionEndpoint, []any, error) {
+	node := connectionNode(documents[endpoint.Board], endpoint.Node)
+	if node == nil {
+		return endpoint, nil, errGraphOwnership
+	}
+	data := node["data"].(map[string]any)
+	if !canvasChoice(data["kind"], "text", "script", "plain-text", "image", "plain-image") {
+		return endpoint, nil, errGraphOwnership
+	}
+	created := []any{}
+	for _, relation := range path {
+		frame := connectionNode(documents[relation.Parent], relation.Frame)
+		if frame == nil {
+			return endpoint, nil, errGraphOwnership
+		}
+		data := frame["data"].(map[string]any)
+		sub, _ := data["subgraph"].(map[string]any)
+		if relation.Child != endpoint.Board || data["kind"] != "frame" || sub["boardId"] != relation.Child {
+			return endpoint, nil, errGraphOwnership
+		}
+		id, added, err := connectionPort(data, endpoint.Node, endpoint.Handle, direction)
+		if err != nil {
+			return endpoint, nil, err
+		}
+		if added {
+			created = append(created, response{"board_id": relation.Parent, "frame_id": relation.Frame, "port_id": id})
+		}
+		endpoint = canvasConnectionEndpoint{Board: relation.Parent, Node: relation.Frame, Handle: "port:" + id}
+	}
+	return endpoint, created, nil
+}
+
+func (s *service) createCanvasConnection(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 canvasConnectionInput
+	if !decode(w, r, &input) {
+		return nil
+	}
+	for _, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
+		if !uuidSlugPattern.MatchString(endpoint.Board) || !canvasID.MatchString(endpoint.Node) || !canvasChoice(endpoint.Handle, "top", "right", "bottom", "left") {
+			reject(w, 400, "invalid_connection_request")
+			return nil
+		}
+	}
+	if !canvasID.MatchString(input.Edge.ID) {
+		reject(w, 400, "invalid_edge")
+		return nil
+	}
+	limits, _ := json.Marshal(response{"edges": []any{input.Edge}})
+	if canvasExceededGraphLimits(limits) {
+		reject(w, 413, "graph_limit_exceeded")
+		return nil
+	}
+	if input.Edge.Data != nil && !validCanvasEdgeData(input.Edge.Data) {
+		reject(w, 400, "invalid_edge")
+		return nil
+	}
+	if len(input.Expected) == 0 {
+		reject(w, 428, "version_required")
+		return nil
+	}
+	workspace, coordinator := r.PathValue("workspace"), r.PathValue("material")
+	tx, err := s.db.BeginTx(r.Context(), nil)
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+	kind, role, err := lockMaterial(r.Context(), tx, project, workspace, coordinator, account)
+	if err != nil {
+		return err
+	}
+	if kind != "board" {
+		reject(w, 400, "invalid_material")
+		return nil
+	}
+	if role == "viewer" {
+		reject(w, 403, "forbidden")
+		return nil
+	}
+	action, hash := "create_connection", payloadDigest(input)
+	done, err := transactionRetry(w, r, tx, project, account, coordinator, action, key, hash)
+	if err != nil || done {
+		return err
+	}
+	participants := map[string]bool{coordinator: true}
+	paths := [][]subgraphRelation{}
+	for _, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
+		kind, role, err := lockMaterial(r.Context(), tx, project, workspace, endpoint.Board, account)
+		if err != nil {
+			return err
+		}
+		if role == "viewer" {
+			reject(w, 403, "forbidden")
+			return nil
+		}
+		if kind != "board" {
+			reject(w, 400, "invalid_material")
+			return nil
+		}
+		path, err := connectionPath(r, tx, project, workspace, coordinator, endpoint.Board)
+		if err == errGraphOwnership {
+			reject(w, 409, "ownership_conflict")
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		paths = append(paths, path)
+		participants[endpoint.Board] = true
+		for _, relation := range path {
+			participants[relation.Parent] = true
+		}
+	}
+	// Require the lowest common ancestor, so the logical edge has one owner.
+	left := map[string]bool{}
+	for _, relation := range paths[0] {
+		left[relation.Child] = true
+	}
+	for _, relation := range paths[1] {
+		if left[relation.Child] {
+			reject(w, 409, "connection_owner_mismatch")
+			return nil
+		}
+	}
+	ids := []string{}
+	for id := range participants {
+		ids = append(ids, id)
+	}
+	sort.Strings(ids)
+	before := []canvasSnapshot{}
+	snapshots := map[string]canvasSnapshot{}
+	documents := map[string]map[string]any{}
+	for _, id := range ids {
+		_, role, err := lockMaterial(r.Context(), tx, project, workspace, id, account)
+		if err != nil {
+			return err
+		}
+		if role == "viewer" {
+			reject(w, 403, "forbidden")
+			return nil
+		}
+		snapshot, err := loadCanvas(r, tx, project, workspace, id)
+		if err != nil {
+			return err
+		}
+		snapshots[id] = snapshot
+		before = append(before, snapshot)
+		var doc map[string]any
+		if err = json.Unmarshal(snapshot.Document, &doc); err != nil {
+			return err
+		}
+		documents[id] = doc
+	}
+	if len(input.Expected) != len(ids) {
+		reply(w, 428, response{"error": "version_required", "details": response{"required_board_ids": ids}})
+		return nil
+	}
+	for _, id := range ids {
+		expected, exists := input.Expected[id]
+		if !exists {
+			reply(w, 428, response{"error": "version_required", "details": response{"required_board_ids": ids}})
+			return nil
+		}
+		if expected != snapshots[id].Version {
+			reject(w, 412, "version_conflict")
+			return nil
+		}
+	}
+	for _, value := range documents[coordinator]["edges"].([]any) {
+		if value.(map[string]any)["id"] == input.Edge.ID {
+			reject(w, 409, "edge_id_conflict")
+			return nil
+		}
+	}
+	mapped := []canvasConnectionEndpoint{}
+	created := []any{}
+	for i, endpoint := range []canvasConnectionEndpoint{input.Source, input.Target} {
+		direction := "output"
+		if i == 1 {
+			direction = "input"
+		}
+		exposed, ports, err := exposeConnectionEndpoint(endpoint, direction, paths[i], documents)
+		if err == errGraphOwnership {
+			reject(w, 409, "ownership_conflict")
+			return nil
+		}
+		if err == errGraphCapacity {
+			reject(w, 413, "graph_limit_exceeded")
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+		mapped = append(mapped, exposed)
+		created = append(created, ports...)
+	}
+	edge := map[string]any{"id": input.Edge.ID, "source": mapped[0].Node, "sourceHandle": mapped[0].Handle, "target": mapped[1].Node, "targetHandle": mapped[1].Handle}
+	if input.Edge.Data != nil {
+		edge["data"] = input.Edge.Data
+	}
+	documents[coordinator]["edges"] = append(documents[coordinator]["edges"].([]any), edge)
+	updated := []canvasSnapshot{}
+	versions := map[string]int64{}
+	for _, id := range ids {
+		documents[id]["schema_version"] = 2
+		raw, _ := json.Marshal(documents[id])
+		if len(raw) > canvasLimit {
+			reject(w, 413, "canvas_too_large")
+			return nil
+		}
+		if canvasExceededGraphLimits(raw) {
+			reject(w, 413, "graph_limit_exceeded")
+			return nil
+		}
+		normalized, ok := normalizeCanvas(raw)
+		if !ok {
+			reject(w, 400, "invalid_canvas")
+			return nil
+		}
+		snapshot := snapshots[id]
+		snapshot.Document = normalized
+		snapshot.Version++
+		updated = append(updated, snapshot)
+		snapshots[id] = snapshot
+		versions[id] = snapshot.Version
+	}
+	// Nothing is persisted until both complete port chains and all snapshots validate.
+	for _, snapshot := range updated {
+		if err = s.saveCanvas(r, tx, project, workspace, snapshot); err != nil {
+			return err
+		}
+	}
+	var id string
+	var expiry time.Time
+	if err = tx.QueryRowContext(r.Context(), `SELECT gen_random_uuid()::text,now()+interval '7 days'`).Scan(&id, &expiry); err != nil {
+		return err
+	}
+	result := response{"transaction_id": id, "parent": snapshots[coordinator], "snapshots": updated, "connection": response{"board_id": coordinator, "edge": edge, "source": input.Source, "target": input.Target}, "created_ports": created, "undo_receipt": response{"transaction_id": id, "valid_for_seconds": 604800, "expires_at": expiry}}
+	encoded, _ := json.Marshal(result)
+	prior, _ := json.Marshal(before)
+	after, _ := json.Marshal(versions)
+	_, 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) VALUES($1,$2,$3,$4,$5,$6,$7,$8,201,$9,$10,$11)`, project, workspace, coordinator, id, account, action, key, hash, encoded, prior, after)
+	if err != nil {
+		return err
+	}
+	if err = tx.Commit(); err != nil {
+		return err
+	}
+	setVersion(w, versions[coordinator])
+	reply(w, 201, result)
+	return nil
+}

internal/platform/canvas_connections_integration_test.go

@@ -0,0 +1,309 @@
+package platform
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strconv"
+	"testing"
+)
+
+func TestCanvasConnectionTransactions(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 random [12]byte
+	_, _ = rand.Read(random[:])
+	project := "connections-" + hex.EncodeToString(random[:])
+	defer func() {
+		for _, table := range []string{"canvas_transactions", "canvas_subgraphs", "workspaces", "project_accounts"} {
+			_, _ = db.ExecContext(ctx, "DELETE FROM "+table+" WHERE project=$1", project)
+		}
+	}()
+	tokens := []string{}
+	accounts := []string{}
+	for range 3 {
+		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)
+		}
+		tokens = append(tokens, token)
+		accounts = append(accounts, account)
+	}
+	var workspace, root string
+	if err = db.QueryRowContext(ctx, `INSERT INTO workspaces(project,name,owner_id) VALUES($1,'Connections',$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 VALUES($1,$2,$3,$4,now())`, project, workspace, accounts[i], role); 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','Root',$3,$4) RETURNING id::text`, project, workspace, accounts[0], json.RawMessage(canvasFixture)).Scan(&root); err != nil {
+		t.Fatal(err)
+	}
+	mux := http.NewServeMux()
+	Register(mux, db, map[string]string{project: "https://app.example.com/auth/callback"}, "https://api.example.com", nil)
+	base := "/projects/" + project + "/workspaces/" + workspace + "/materials/"
+	path := func(board string) string { return base + board + "/canvas" }
+	do := func(method, url, token string, body any, version int64, key string) *httptest.ResponseRecorder {
+		raw, _ := json.Marshal(body)
+		r := httptest.NewRequest(method, url, 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.Quote(strconv.FormatInt(version, 10)))
+		}
+		if key != "" {
+			r.Header.Set("Idempotency-Key", key)
+		}
+		w := httptest.NewRecorder()
+		mux.ServeHTTP(w, r)
+		return w
+	}
+	call := func(method, url, token string, body any, status int, version int64, key string) response {
+		t.Helper()
+		w := do(method, url, token, body, version, key)
+		if w.Code != status {
+			t.Fatalf("%s %s got %d %s want %d", method, url, w.Code, w.Body.String(), status)
+		}
+		out := response{}
+		if err := json.Unmarshal(w.Body.Bytes(), &out); err != nil {
+			t.Fatal(err)
+		}
+		return out
+	}
+	key := func(n int) string { return fmt.Sprintf("a0000000-0000-4000-8000-%012d", n) }
+	versions := func(ids ...string) map[string]int64 {
+		t.Helper()
+		out := map[string]int64{}
+		for _, id := range ids {
+			out[id] = int64(call("GET", path(id), tokens[0], nil, 200, 0, "")["version"].(float64))
+		}
+		return out
+	}
+	create := func(board, frame string, n int) string {
+		t.Helper()
+		v := versions(board)[board]
+		result := call("POST", path(board)+"/subgraphs", tokens[0], response{"frame": response{"id": frame, "title": frame, "position": response{"x": 10, "y": 20}, "width": 600, "height": 400, "labelMode": "fixed"}}, 201, v, key(n))
+		return result["child_board"].(map[string]any)["id"].(string)
+	}
+	seed := func(board string, nodes ...string) {
+		t.Helper()
+		values := []any{}
+		for _, id := range nodes {
+			values = append(values, response{"id": id, "type": "material", "position": response{"x": 10, "y": 20}, "data": response{"kind": "text", "title": id, "text": "preserved"}})
+		}
+		call("PATCH", path(board), tokens[0], response{"document": response{"schema_version": 2, "nodes": values, "edges": []any{}, "viewport": response{"x": 0, "y": 0, "zoom": 1}}}, 200, 1, "")
+	}
+	left := create(root, "left", 1)
+	right := create(root, "right", 2)
+	seed(left, "l1", "l2")
+	seed(right, "r1")
+	leaf := create(left, "nested", 3)
+	seed(leaf, "deep")
+	command := func(source, node, target, targetNode, id string, ids ...string) response {
+		return response{"source": response{"board_id": source, "node_id": node, "handle": "right"}, "target": response{"board_id": target, "node_id": targetNode, "handle": "left"}, "edge": response{"id": id, "data": response{"appearance": response{"color": "#123456", "width": 2}, "path": response{"type": "bezier-chain", "knots": []any{response{"id": "k", "position": response{"x": 200, "y": 300}, "in": response{"x": -20, "y": 0}, "out": response{"x": 20, "y": 0}, "mode": "symmetric"}}}}}, "expected_versions": versions(ids...)}
+	}
+	firstBody := command(leaf, "deep", root, "a", "external", root, left, leaf)
+	beforeLeaf := call("GET", path(leaf), tokens[0], nil, 200, 0, "")["document"]
+	first := call("POST", path(root)+"/connections", tokens[0], firstBody, 201, 0, key(4))
+	if len(first["created_ports"].([]any)) != 2 {
+		t.Fatal("nested connection missing ports")
+	}
+	retry := call("POST", path(root)+"/connections", tokens[0], firstBody, 201, 0, key(4))
+	if retry["transaction_id"] != first["transaction_id"] {
+		t.Fatal("retry duplicated operation")
+	}
+	changedBody := command(leaf, "deep", root, "a", "different", root, left, leaf)
+	call("POST", path(root)+"/connections", tokens[0], changedBody, 409, 0, key(4))
+	afterLeaf := call("GET", path(leaf), tokens[0], nil, 200, 0, "")["document"]
+	if payloadDigest(beforeLeaf) != payloadDigest(afterLeaf) {
+		t.Fatal("connection changed leaf content")
+	}
+	parent := first["parent"].(map[string]any)
+	edge := first["connection"].(map[string]any)["edge"].(map[string]any)
+	if edge["source"] != "left" || edge["target"] != "a" || edge["data"].(map[string]any)["appearance"].(map[string]any)["color"] != "#123456" {
+		t.Fatal("connection edge lost mapping/style")
+	}
+	if len(parent["document"].(map[string]any)["edges"].([]any)) != 2 {
+		t.Fatal("existing edge lost")
+	}
+	undo := call("POST", path(root)+"/transactions/"+first["transaction_id"].(string)+"/undo", tokens[0], response{"expected_versions": versions(root, left, leaf)}, 200, 0, key(5))
+	if len(undo["snapshots"].([]any)) != 3 {
+		t.Fatal("undo did not restore complete chain")
+	}
+	for board, frame := range map[string]string{root: "left", left: "nested"} {
+		document := call("GET", path(board), tokens[0], nil, 200, 0, "")["document"]
+		raw, _ := json.Marshal(document)
+		if len(canvasNodeData(raw)[frame]["ports"].([]any)) != 0 {
+			t.Fatal("undo left ports behind")
+		}
+	}
+	redoBody := command(leaf, "deep", root, "a", "external", root, left, leaf)
+	redone := call("POST", path(root)+"/connections", tokens[0], redoBody, 201, 0, key(6))
+	if payloadDigest(redone["connection"]) != payloadDigest(first["connection"]) {
+		t.Fatal("redo changed stable ports")
+	}
+	reused := call("POST", path(root)+"/connections", tokens[0], command(leaf, "deep", root, "b", "second", root, left, leaf), 201, 0, key(7))
+	if len(reused["created_ports"].([]any)) != 0 {
+		t.Fatal("same source endpoint duplicated ports")
+	}
+	sibling := call("POST", path(root)+"/connections", tokens[0], command(left, "l1", right, "r1", "sibling", root, left, right), 201, 0, key(8))
+	if len(sibling["created_ports"].([]any)) != 2 {
+		t.Fatal("sibling connection missing boundary ports")
+	}
+	reverse := call("POST", path(root)+"/connections", tokens[0], command(root, "b", leaf, "deep", "reverse", root, left, leaf), 201, 0, key(9))
+	if len(reverse["created_ports"].([]any)) != 2 {
+		t.Fatal("input chain reused output ports")
+	}
+	// Missing/stale participant versions, bad endpoints and permission failures are atomic.
+	stable := versions(root, left, leaf)
+	missing := command(leaf, "deep", root, "a", "missing", root, left)
+	required := call("POST", path(root)+"/connections", tokens[0], missing, 428, 0, key(10))
+	if len(required["details"].(map[string]any)["required_board_ids"].([]any)) != 3 {
+		t.Fatal("version response omitted participants")
+	}
+	stale := command(leaf, "deep", root, "a", "stale", root, left, leaf)
+	stale["expected_versions"].(map[string]int64)[leaf]--
+	call("POST", path(root)+"/connections", tokens[0], stale, 412, 0, key(11))
+	call("POST", path(root)+"/connections", tokens[0], command(leaf, "absent", root, "a", "bad-node", root, left, leaf), 409, 0, key(12))
+	call("POST", path(root)+"/connections", tokens[0], command(root, "icon", leaf, "deep", "icon-edge", root, left, leaf), 409, 0, key(13))
+	call("POST", path(root)+"/connections", tokens[0], command(leaf, "deep", left, "l1", "wrong-owner", root, left, leaf), 409, 0, key(14))
+	call("POST", path(root)+"/connections", tokens[0], command(leaf, "deep", root, "a", "external", root, left, leaf), 409, 0, key(15))
+	call("POST", path(root)+"/connections", tokens[1], command(leaf, "deep", root, "a", "viewer", root, left, leaf), 403, 0, key(16))
+	call("POST", path(root)+"/connections", tokens[2], command(leaf, "deep", root, "a", "outsider", root, left, leaf), 404, 0, key(17))
+	invalid := command(leaf, "deep", root, "a", "handle", root, left, leaf)
+	invalid["source"].(response)["handle"] = "port:fake"
+	call("POST", path(root)+"/connections", tokens[0], invalid, 400, 0, key(18))
+	if payloadDigest(stable) != payloadDigest(versions(root, left, leaf)) {
+		t.Fatal("rejected transaction partially changed versions")
+	}
+	// The actual nested mapping is resolvable by the same reader used by AI propagation.
+	graph := map[string]canvasSnapshot{}
+	for _, board := range []string{root, left, leaf} {
+		read := call("GET", path(board), tokens[0], nil, 200, 0, "")
+		raw, _ := json.Marshal(read["document"])
+		graph[board] = canvasSnapshot{BoardID: board, Version: int64(read["version"].(float64)), Document: raw}
+	}
+	reader := readerFromSnapshots(graph, map[string]int{})
+	reader.snapshots = map[string]canvasSnapshot{}
+	reader.documents = map[string]map[string]any{}
+	reader.parents = map[string]subgraphRelation{}
+	reader.parentKnown = map[string]bool{}
+	resolved, err := reader.resolve(root, edge["source"].(string), edge["sourceHandle"].(string), "output", 0)
+	if err != nil || resolved != canvasNodeKey(leaf, "deep") {
+		t.Fatal("AI reader cannot resolve newly created port chain")
+	}
+	// Ordinary saves must still reject port retargeting.
+	readParent := call("GET", path(root), tokens[0], nil, 200, 0, "")
+	parentDoc := readParent["document"].(map[string]any)
+	for _, value := range parentDoc["nodes"].([]any) {
+		node := value.(map[string]any)
+		if node["id"] == "left" {
+			node["data"].(map[string]any)["ports"].([]any)[0].(map[string]any)["nodeId"] = "l2"
+		}
+	}
+	call("PATCH", path(root), tokens[0], response{"document": parentDoc}, 409, int64(readParent["version"].(float64)), "")
+	var otherWorkspace, otherBoard, unrelatedBoard string
+	if err = db.QueryRowContext(ctx, `INSERT INTO workspaces(project,name,owner_id) VALUES($1,'Other',$2) RETURNING id::text`, project, accounts[0]).Scan(&otherWorkspace); err != nil {
+		t.Fatal(err)
+	}
+	if _, err = db.ExecContext(ctx, `INSERT INTO workspace_members VALUES($1,$2,$3,'admin',now())`, project, otherWorkspace, accounts[0]); err != nil {
+		t.Fatal(err)
+	}
+	for _, target := range []struct {
+		workspace string
+		id        *string
+	}{{otherWorkspace, &otherBoard}, {workspace, &unrelatedBoard}} {
+		if err = db.QueryRowContext(ctx, `INSERT INTO materials(project,workspace_id,kind,title,created_by,canvas_document) VALUES($1,$2,'board','Other',$3,$4) RETURNING id::text`, project, target.workspace, accounts[0], json.RawMessage(canvasFixture)).Scan(target.id); err != nil {
+			t.Fatal(err)
+		}
+	}
+	call("POST", path(root)+"/connections", tokens[0], command(left, "l1", otherBoard, "a", "cross-workspace", root, left), 404, 0, key(21))
+	call("POST", path(root)+"/connections", tokens[0], command(left, "l1", unrelatedBoard, "a", "unrelated", root, left), 409, 0, key(22))
+	legacyRequest := httptest.NewRequest("POST", path(root)+"/connections", bytes.NewReader([]byte("{}")))
+	legacyRequest.Header.Set("Authorization", "Bearer "+tokens[0])
+	legacyResponse := httptest.NewRecorder()
+	mux.ServeHTTP(legacyResponse, legacyRequest)
+	if legacyResponse.Code != 409 {
+		t.Fatal("connection mutation accepted a legacy client")
+	}
+	caps := call("GET", path(root)+"/capabilities", tokens[0], nil, 200, 0, "")
+	if caps["subgraphs"].(map[string]any)["connections"] != true {
+		t.Fatal("capabilities omitted connection transaction")
+	}
+	// Four simultaneous retries create exactly one edge/receipt and advance versions once.
+	concurrent := command(left, "l2", root, "a", "concurrent", root, left)
+	prior := versions(root, left)
+	start := make(chan struct{})
+	results := make(chan *httptest.ResponseRecorder, 4)
+	for range 4 {
+		go func() { <-start; results <- do("POST", path(root)+"/connections", tokens[0], concurrent, 0, key(19)) }()
+	}
+	close(start)
+	transaction := ""
+	for range 4 {
+		w := <-results
+		if w.Code != 201 {
+			t.Fatalf("concurrent request: %d %s", w.Code, w.Body.String())
+		}
+		var result response
+		_ = json.Unmarshal(w.Body.Bytes(), &result)
+		id := result["transaction_id"].(string)
+		if transaction != "" && transaction != id {
+			t.Fatal("concurrent retry duplicated receipt")
+		}
+		transaction = id
+	}
+	for id, v := range versions(root, left) {
+		if v != prior[id]+1 {
+			t.Fatal("concurrent retry advanced version twice")
+		}
+	}
+	// A full coordinator rejects the entire operation, including new leaf ports.
+	snapshot := call("GET", path(root), tokens[0], nil, 200, 0, "")
+	doc := snapshot["document"].(map[string]any)
+	edges := []any{}
+	for i := 0; i < 2000; i++ {
+		edges = append(edges, response{"id": fmt.Sprint("full-", i), "source": "a", "target": "b", "sourceHandle": "right", "targetHandle": "left"})
+	}
+	doc["edges"] = edges
+	call("PATCH", path(root), tokens[0], response{"document": doc}, 200, int64(snapshot["version"].(float64)), "")
+	capacity := command(left, "l2", root, "a", "capacity", root, left)
+	capacity["source"].(response)["handle"] = "top"
+	beforeFailure := versions(root, left)
+	beforeLeft := call("GET", path(left), tokens[0], nil, 200, 0, "")["document"]
+	call("POST", path(root)+"/connections", tokens[0], capacity, 413, 0, key(20))
+	afterLeft := call("GET", path(left), tokens[0], nil, 200, 0, "")["document"]
+	if payloadDigest(beforeLeft) != payloadDigest(afterLeft) || payloadDigest(beforeFailure) != payloadDigest(versions(root, left)) {
+		t.Fatal("capacity failure left ports or versions behind")
+	}
+	var receipts int
+	if err = db.QueryRowContext(ctx, `SELECT count(*) FROM canvas_transactions WHERE project=$1 AND action='create_connection' AND idempotency_key=$2`, project, key(20)).Scan(&receipts); err != nil || receipts != 0 {
+		t.Fatal("rejected connection left a receipt")
+	}
+}

internal/platform/canvas_connections_test.go

@@ -0,0 +1,65 @@
+package platform
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+)
+
+func TestConnectionPortReuseCollisionAndCapacity(t *testing.T) {
+	data := map[string]any{"ports": []any{map[string]any{"id": "custom-stable", "nodeId": "n", "handle": "left", "direction": "input"}}}
+	id, created, err := connectionPort(data, "n", "left", "input")
+	if err != nil || created || id != "custom-stable" {
+		t.Fatal("existing logical port was not reused")
+	}
+	id, created, err = connectionPort(data, "n", "left", "output")
+	if err != nil || !created || id == "custom-stable" {
+		t.Fatal("input and output ports were conflated")
+	}
+	data["ports"].([]any)[1].(map[string]any)["nodeId"] = "other"
+	if _, _, err = connectionPort(data, "n", "left", "output"); err != errGraphOwnership {
+		t.Fatal("derived ID collision retargeted a port")
+	}
+	ports := []any{}
+	for i := 0; i < 4000; i++ {
+		ports = append(ports, map[string]any{"id": fmt.Sprint(i), "nodeId": "other", "handle": "left", "direction": "input"})
+	}
+	data["ports"] = ports
+	if _, _, err = connectionPort(data, "n", "right", "output"); err != errGraphCapacity || len(data["ports"].([]any)) != 4000 {
+		t.Fatal("port capacity partially mutated frame")
+	}
+}
+
+func TestConnectionExposureBuildsNestedDirectionChain(t *testing.T) {
+	frame := func(child string) map[string]any {
+		return map[string]any{"id": "frame", "data": map[string]any{"kind": "frame", "subgraph": map[string]any{"boardId": child}, "ports": []any{}}}
+	}
+	docs := map[string]map[string]any{
+		"root":   {"nodes": []any{frame("middle")}},
+		"middle": {"nodes": []any{frame("leaf")}},
+		"leaf":   {"nodes": []any{map[string]any{"id": "n", "data": map[string]any{"kind": "plain-image", "src": "https://example.com/image.png"}}}},
+	}
+	before, _ := json.Marshal(docs["leaf"])
+	path := []subgraphRelation{{Parent: "middle", Frame: "frame", Child: "leaf"}, {Parent: "root", Frame: "frame", Child: "middle"}}
+	endpoint := canvasConnectionEndpoint{Board: "leaf", Node: "n", Handle: "bottom"}
+	exposed, created, err := exposeConnectionEndpoint(endpoint, "output", path, docs)
+	if err != nil || exposed.Board != "root" || exposed.Node != "frame" || len(created) != 2 {
+		t.Fatalf("incomplete chain: %+v %v", exposed, err)
+	}
+	inner := connectionNode(docs["middle"], "frame")["data"].(map[string]any)["ports"].([]any)[0].(map[string]any)
+	outer := connectionNode(docs["root"], "frame")["data"].(map[string]any)["ports"].([]any)[0].(map[string]any)
+	if inner["nodeId"] != "n" || inner["handle"] != "bottom" || outer["nodeId"] != "frame" || outer["handle"] != "port:"+inner["id"].(string) {
+		t.Fatal("outer port skipped a child boundary")
+	}
+	again, created, err := exposeConnectionEndpoint(endpoint, "output", path, docs)
+	if err != nil || again != exposed || len(created) != 0 {
+		t.Fatal("repeated exposure duplicated ports")
+	}
+	if _, created, err = exposeConnectionEndpoint(endpoint, "input", path, docs); err != nil || len(created) != 2 {
+		t.Fatal("reverse direction did not create independent input chain")
+	}
+	after, _ := json.Marshal(docs["leaf"])
+	if string(before) != string(after) {
+		t.Fatal("exposure modified the actual child node")
+	}
+}

internal/platform/canvas_undo.go

@@ -71,7 +71,7 @@ func (s *service) undoCanvasTransaction(w http.ResponseWriter, r *http.Request,
 		reject(w, 409, "transaction_already_undone")
 		return nil
 	}
-	if originalAction != "create_subgraph" && originalAction != "migrate_subgraphs" && !strings.HasPrefix(originalAction, "dissolve_subgraph/") && !strings.HasPrefix(originalAction, "move_members/") {
+	if originalAction != "create_subgraph" && originalAction != "migrate_subgraphs" && originalAction != "create_connection" && !strings.HasPrefix(originalAction, "dissolve_subgraph/") && !strings.HasPrefix(originalAction, "move_members/") {
 		reject(w, 409, "undo_not_supported")
 		return nil
 	}

internal/platform/http.go

@@ -147,6 +147,7 @@ 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("/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))
 	mux.HandleFunc("/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/subgraphs/{frame}/dissolve", s.protected(callbacks, s.dissolveSubgraph))