From 07fd678cf3c44f8d91c36f821cd360d193790030 Mon Sep 17 00:00:00 2001 From: Codex Bot Date: Fri, 6 Nov 2026 19:38:11 +0800 Subject: [PATCH] Add atomic connections across canvas subgraph boundaries --- README.md | 104 ++++++ internal/platform/canvas_capabilities.go | 2 +- internal/platform/canvas_connections.go | 351 ++++++++++++++++++ .../canvas_connections_integration_test.go | 309 +++++++++++++++ internal/platform/canvas_connections_test.go | 65 ++++ internal/platform/canvas_undo.go | 2 +- internal/platform/http.go | 1 + 7 files changed, 832 insertions(+), 2 deletions(-) create mode 100644 internal/platform/canvas_connections.go create mode 100644 internal/platform/canvas_connections_integration_test.go create mode 100644 internal/platform/canvas_connections_test.go diff --git a/README.md b/README.md index eecadda..c23d762 100644 --- a/README.md +++ b/README.md @@ -212,6 +212,7 @@ Collapsing/scaling presentation does not rewrite child content. Projection is | Route | Required input / behavior | | --- | --- | +| POST B/connections | `{source,target,edge,expected_versions}` + Idempotency-Key; atomically creates/reuses boundary port chains and stores one edge in the lowest common ancestor | | POST B/subgraphs | Parent If-Match; `{frame:{id,title,position,width,height,labelMode}}`; returns 201, child_board, parent/child snapshots and receipt | | POST B/subgraphs/{frame}/members | `{source_board_id,target_board_id,node_ids,expected_versions}`; parent/child or siblings coordinated by their common parent | | POST B/subgraphs/{frame}/dissolve | `{expected_versions}`; preserves content in the parent and archives the emptied child | @@ -405,6 +406,109 @@ summary are retained with mode 0600 under on the ECS host, in addition to the database receipt. The frontend V2 integration remains a separate task; use a V2-aware client for this migrated test canvas. +#### Creating new cross-boundary connections + +Ordinary Canvas PATCH preserves existing frame ports and cannot add or retarget +ports. Migration/movement-generated ports are reusable, but creating a new +external connection to an unexposed child node requires this dedicated command: + +`POST B/connections` with user session authorization, +`Accept: application/vnd.foundry.canvas+json;version=2`, JSON content type and a +UUID `Idempotency-Key`. Availability is advertised as +`capabilities.subgraphs.connections: true`; clients must check the deployed +capability before calling it. + +`B` identifies the **lowest common ancestor board** of the two actual endpoints. +The coordinator may itself be a nested board. Both endpoints must be in this +same project/workspace and inside its active subgraph tree. A direct endpoint on +B is valid, as are siblings and descendants up to the existing depth limit. +When both endpoints are in the same board, use that board as B. + +```json +{ + "source": {"board_id":"","node_id":"child-text","handle":"right"}, + "target": {"board_id":"","node_id":"parent-image","handle":"left"}, + "edge": { + "id":"new-edge", + "data": { + "appearance":{"color":"#AA968B","width":2}, + "animation":{"mode":"executing","effect":"flow","direction":"forward","durationMs":1600}, + "execution":{"enabled":true} + } + }, + "expected_versions": {"":210,"":1} +} +``` + +Use canonical board UUIDs for endpoints and version keys. Eligible real node +kinds are text/script/plain-text/image/plain-image; handles must be one of +`top/right/bottom/left`. Do not submit projected node IDs, frame IDs as generated +content, or `port:*` handles. The server exposes a real endpoint through each +containing frame, creating or reusing a port with the matching +`direction/nodeId/handle` tuple. Existing matching ports retain their IDs, +including migration-created or custom stable IDs. A source chain uses output +ports, and a target chain uses input ports; these are distinct. No unbound port +or writable arbitrary child reference is accepted. + +The expected_versions map must contain **exactly every board on both paths**, +including the endpoint boards and B. For a grandchild-to-parent connection it +contains parent, intermediate child and grandchild; for siblings it contains +both children and their parent. No separate If-Match header is required for this +multi-board operation. Missing participants return 428 `version_required`; +when a nonempty map has missing/extra participant keys, the response includes +`details.required_board_ids`. Any stale version returns 412 `version_conflict`. +All participants require current editor/admin access. Calling a higher ancestor +instead of the lowest common ancestor returns 409 `connection_owner_mismatch`. +An unrelated/invalid ownership path returns 409 `ownership_conflict`; resources +outside the caller's scope retain the normal 403/404 behavior. + +Only `edge.id` and optional typed V2 `edge.data` are accepted. Its appearance, +animation, execution flag and bezier-chain path follow the normal V2 validation. +All knots are already in B's local coordinates; port creation does not move node +geometry or re-layout the edge. An existing edge ID in B returns 409 +`edge_id_conflict`. Graph, port and snapshot capacity failures return 413 without +saving any ports, edges, versions or receipt. + +Success is 201 with the coordinator's ETag and: + +```json +{ + "transaction_id":"", + "parent":{"board_id":"","version":211,"document":{}}, + "snapshots":[ + {"board_id":"","version":211,"document":{}}, + {"board_id":"","version":2,"document":{}} + ], + "connection":{ + "board_id":"", + "edge":{"id":"new-edge","source":"","sourceHandle":"port:","target":"parent-image","targetHandle":"left","data":{}}, + "source":{"board_id":"","node_id":"child-text","handle":"right"}, + "target":{"board_id":"","node_id":"parent-image","handle":"left"} + }, + "created_ports":[{"board_id":"","frame_id":"","port_id":""}], + "undo_receipt":{"transaction_id":"","valid_for_seconds":604800,"expires_at":""} +} +``` + +The JSON above abbreviates document values and shows a one-level example; nested +connections include every intermediate board in snapshots. The response edge is +the stored owner-board edge, while source/target preserve the logical endpoint +identities. created_ports contains only newly allocated ports; reused ports are +available in the snapshots. Each participating board advances exactly one +version (including endpoint boards whose node content remains unchanged), +consistent with other cross-board commands. Flush pending saves before creating +the connection, then absorb **all** returned versions/snapshots before resuming +autosave. Retrying an identical body with the same key returns the original 201 +response and does not advance versions again. Changing the body, including its +version map, with that key returns 409 `idempotency_conflict`. + +Undo uses the existing `POST B/transactions/{transaction_id}/undo` endpoint with +a new key and all current participant versions. It restores the previous edge +and port arrays atomically; subsequent incompatible edits cause 412. Existing +ports reused by the command remain after undo. Redo submits the command with +fresh versions and a new key. Ordinary PATCH can continue to edit/remove edges +using already exposed ports, but cannot retarget or directly delete the ports. + #### Reproducible session API examples These examples require Bash, curl and jq, an editor/admin login session, and an diff --git a/internal/platform/canvas_capabilities.go b/internal/platform/canvas_capabilities.go index 828a227..30391a2 100644 --- a/internal/platform/canvas_capabilities.go +++ b/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}, diff --git a/internal/platform/canvas_connections.go b/internal/platform/canvas_connections.go new file mode 100644 index 0000000..3659779 --- /dev/null +++ b/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 +} diff --git a/internal/platform/canvas_connections_integration_test.go b/internal/platform/canvas_connections_integration_test.go new file mode 100644 index 0000000..058458a --- /dev/null +++ b/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") + } +} diff --git a/internal/platform/canvas_connections_test.go b/internal/platform/canvas_connections_test.go new file mode 100644 index 0000000..363854d --- /dev/null +++ b/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") + } +} diff --git a/internal/platform/canvas_undo.go b/internal/platform/canvas_undo.go index 8de06e0..3d6b2f7 100644 --- a/internal/platform/canvas_undo.go +++ b/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 } diff --git a/internal/platform/http.go b/internal/platform/http.go index c25ed1c..9792af4 100644 --- a/internal/platform/http.go +++ b/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))