diff --git a/db/010-canvas-ai-receipts.sql b/db/010-canvas-ai-receipts.sql new file mode 100644 index 0000000..f996970 --- /dev/null +++ b/db/010-canvas-ai-receipts.sql @@ -0,0 +1,29 @@ +BEGIN; +-- Keep transport receipts independently from execution retention. Their absence +-- must never be used as proof that a paid invocation did not happen. +CREATE TABLE canvas_ai_requests ( + project text NOT NULL, + workspace_id uuid NOT NULL, + account_id uuid NOT NULL, + request_id uuid NOT NULL, + board_id uuid NOT NULL, + run_id uuid, + action text NOT NULL CHECK(action IN('run.start','run.cancel','run.apply')), + operation_id uuid NOT NULL, + payload_hash text NOT NULL, + outcome text NOT NULL CHECK(outcome IN('applied','not-applied')), + response_status integer NOT NULL, + response_body jsonb NOT NULL, + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY(project,workspace_id,account_id,request_id), + FOREIGN KEY(project,workspace_id) REFERENCES workspaces(project,id) ON DELETE CASCADE +); +-- A known zero-effect rejection may be explicitly replanned under another key; +-- a completed logical action must never execute twice under different keys. +CREATE UNIQUE INDEX canvas_ai_request_operation ON canvas_ai_requests(project,workspace_id,account_id,operation_id) WHERE outcome='applied'; +-- Preserve accepted canonical requests that predate this receipt table. +INSERT INTO canvas_ai_requests(project,workspace_id,account_id,request_id,board_id,action,operation_id,payload_hash,outcome,response_status,response_body) +SELECT project,workspace_id,account_id,idempotency_key,board_id,'run.start',logical_operation_id,payload_hash,'applied',202,create_response +FROM canvas_ai_runs WHERE logical_operation_id IS NOT NULL AND create_response->>'protocol_version'='2' +ORDER BY created_at,id ON CONFLICT DO NOTHING; +COMMIT; diff --git a/internal/platform/canvas_ai_cancel.go b/internal/platform/canvas_ai_cancel.go index f2b85ac..0aa408c 100644 --- a/internal/platform/canvas_ai_cancel.go +++ b/internal/platform/canvas_ai_cancel.go @@ -22,12 +22,41 @@ func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, proj reject(w, 413, "context_too_large") return nil } - input := map[string]any{} - if len(raw) > 0 && (json.Unmarshal(raw, &input) != nil || input == nil || len(input) > 0) { + var input struct { + ProtocolVersion int `json:"protocol_version,omitempty"` + RequestID string `json:"request_id,omitempty"` + OperationID string `json:"operation_id,omitempty"` + } + if len(raw) > 0 && (json.Unmarshal(raw, &input) != nil || string(raw) == "null") { reject(w, 400, "invalid_ai_request") return nil } - hash := payloadDigest(input) + var fields map[string]any + if len(raw) > 0 { + if json.Unmarshal(raw, &fields) != nil || fields == nil { + reject(w, 400, "invalid_ai_request") + return nil + } + for key := range fields { + if key != "protocol_version" && key != "request_id" && key != "operation_id" { + reject(w, 400, "invalid_ai_request") + return nil + } + } + } + canonical := input.ProtocolVersion != 0 || input.RequestID != "" || input.OperationID != "" + if canonical && (input.ProtocolVersion != 2 || input.RequestID != key || !uuidSlugPattern.MatchString(input.OperationID) || !acceptsCanvasV2(r)) { + reject(w, 400, "invalid_ai_protocol") + return nil + } + if !canonical && len(fields) > 0 { + reject(w, 400, "invalid_ai_request") + return nil + } + hash := payloadDigest(map[string]any{}) + if canonical { + hash = payloadDigest(input) + } workspace, board := r.PathValue("workspace"), r.PathValue("material") tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { @@ -38,13 +67,25 @@ func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, proj if err != nil { return err } + var receipt *canvasAIRequest + if canonical { + receipt = &canvasAIRequest{Key: key, Hash: hash, Action: "run.cancel", Board: board, Run: id, Operation: input.OperationID} + if done, e := beginCanvasAIRequest(w, r, tx, project, workspace, account, receipt); done || e != nil { + return e + } + } + rejectRequest := func(status int, code string) error { + return rejectCanvasAIRequest(w, r, tx, project, workspace, account, receipt, status, code) + } if role == "viewer" { - reject(w, 403, "forbidden") - return nil + return rejectRequest(403, "forbidden") } var status string var rawPlan json.RawMessage err = tx.QueryRowContext(r.Context(), `SELECT status,plan FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND id=$4 FOR UPDATE`, project, workspace, board, id).Scan(&status, &rawPlan) + if err == sql.ErrNoRows { + return rejectRequest(404, "not_found") + } if err != nil { return err } @@ -72,8 +113,7 @@ func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } if role == "viewer" { - reject(w, 403, "forbidden") - return nil + return rejectRequest(403, "forbidden") } } if status == "queued" || status == "running" { @@ -145,14 +185,14 @@ func (s *service) cancelCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } result := response{"run_id": id, "status": status, "applied": applied, "cancelled_step_ids": cancelledSteps, "worker_stop_grace_ms": 1000} + if receipt != nil { + result["run_status"], result["status"] = status, "applied" + bindCanvasAIReceipt(result, receipt) + } encoded, _ := json.Marshal(result) _, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_mutations(project,workspace_id,run_id,account_id,action,idempotency_key,payload_hash,response_body) VALUES($1,$2,$3,$4,'cancel',$5,$6,$7)`, project, workspace, id, account, key, hash, encoded) if err != nil { return err } - if err = tx.Commit(); err != nil { - return err - } - reply(w, 200, result) - return nil + return commitCanvasAIReceipt(w, r, tx, project, workspace, account, receipt, 200, "applied", result) } diff --git a/internal/platform/canvas_ai_explicit_apply.go b/internal/platform/canvas_ai_explicit_apply.go index 91347cd..7374ccb 100644 --- a/internal/platform/canvas_ai_explicit_apply.go +++ b/internal/platform/canvas_ai_explicit_apply.go @@ -22,8 +22,13 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr return nil } var input struct { - NodeID string `json:"node_id"` - OutputID string `json:"output_id"` + ProtocolVersion int `json:"protocol_version,omitempty"` + RequestID string `json:"request_id,omitempty"` + OperationID string `json:"operation_id,omitempty"` + TargetIncarnation int64 `json:"target_incarnation,omitempty"` + TargetCanvasID string `json:"target_canvas_id,omitempty"` + NodeID string `json:"node_id"` + OutputID string `json:"output_id"` } if !decode(w, r, &input) { return nil @@ -32,7 +37,15 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr reject(w, 400, "invalid_ai_request") return nil } + canonical := input.ProtocolVersion != 0 || input.RequestID != "" || input.OperationID != "" || input.TargetIncarnation != 0 || input.TargetCanvasID != "" + if canonical && (input.ProtocolVersion != 2 || input.RequestID != key || !uuidSlugPattern.MatchString(input.OperationID) || input.TargetIncarnation < 1 || !uuidSlugPattern.MatchString(input.TargetCanvasID) || !acceptsCanvasV2(r)) { + reject(w, 400, "invalid_ai_protocol") + return nil + } hash := payloadDigest(input) + if canonical { + hash = payloadDigest(response{"input": input, "if_match": expected}) + } workspace, board := r.PathValue("workspace"), r.PathValue("material") tx, err := s.db.BeginTx(r.Context(), nil) if err != nil { @@ -43,12 +56,24 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr if err != nil { return err } + var receipt *canvasAIRequest + if canonical { + receipt = &canvasAIRequest{Key: key, Hash: hash, Action: "run.apply", Board: board, Run: id, Operation: input.OperationID} + if done, e := beginCanvasAIRequest(w, r, tx, project, workspace, account, receipt); done || e != nil { + return e + } + } + rejectRequest := func(status int, code string) error { + return rejectCanvasAIRequest(w, r, tx, project, workspace, account, receipt, status, code) + } if role == "viewer" { - reject(w, 403, "forbidden") - return nil + return rejectRequest(403, "forbidden") } var runStatus string err = tx.QueryRowContext(r.Context(), `SELECT status FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND id=$4 FOR UPDATE`, project, workspace, board, id).Scan(&runStatus) + if err == sql.ErrNoRows { + return rejectRequest(404, "not_found") + } if err != nil { return err } @@ -75,28 +100,31 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr var outputRaw, frozenRaw json.RawMessage var applied sql.NullInt64 err = tx.QueryRowContext(r.Context(), `SELECT id,board_id::text,node_id,status,output,frozen_input,applied_version FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND output->>'output_id'=$4 FOR UPDATE`, project, workspace, id, input.OutputID).Scan(&stepID, &targetBoard, &node, &status, &outputRaw, &frozenRaw, &applied) + if err == sql.ErrNoRows { + return rejectRequest(404, "not_found") + } if err != nil { return err } + if canonical && targetBoard != input.TargetCanvasID { + return rejectRequest(409, "target_changed") + } if node != input.NodeID || status != "conflict" || applied.Valid || runStatus == "cancelled" { - reject(w, 409, "output_not_applicable") - return nil + return rejectRequest(409, "output_not_applicable") } _, role, err = lockMaterial(r.Context(), tx, project, workspace, targetBoard, account) if err != nil { return err } if role == "viewer" { - reject(w, 403, "forbidden") - return nil + return rejectRequest(403, "forbidden") } snapshot, err := loadCanvas(r, tx, project, workspace, targetBoard) if err != nil { return err } if snapshot.Version != expected { - reject(w, 412, "version_conflict") - return nil + return rejectRequest(412, "version_conflict") } var output struct { Kind string `json:"kind"` @@ -111,40 +139,38 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr if err = json.Unmarshal(frozenRaw, &frozen); err != nil { return err } + if canonical && (frozen.TargetIncarnation != input.TargetIncarnation || frozen.CanonicalRoot == "") { + return rejectRequest(409, "target_changed") + } identityOK, err := canvasAIIdentityMatches(r.Context(), tx, project, workspace, frozen) if err != nil { return err } if !identityOK { - reject(w, 409, "target_changed") - return nil + return rejectRequest(409, "target_changed") } patch := response{} switch output.Kind { case "text": if !canvasChoice(frozen.Kind, "text", "script", "plain-text") || !utf8.ValidString(output.Text) || utf8.RuneCountInString(output.Text) > 100000 { - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } patch["text"] = output.Text case "image": if !canvasChoice(frozen.Kind, "image", "plain-image") || !uuidSlugPattern.MatchString(output.AssetID) || output.Src != s.assetURL(project, workspace, output.AssetID) { - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } readable, err := assetReadable(r, tx, project, workspace, output.AssetID, account) if err != nil { return err } if !readable { - reject(w, 404, "not_found") - return nil + return rejectRequest(404, "not_found") } patch["src"] = output.Src patch["assetId"] = output.AssetID default: - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } var doc map[string]any @@ -159,17 +185,14 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr } } if target == nil { - reject(w, 404, "not_found") - return nil + return rejectRequest(404, "not_found") } if target["kind"] != frozen.Kind { - reject(w, 409, "ownership_conflict") - return nil + return rejectRequest(409, "ownership_conflict") } change, err := prepareCanvasAIPatch(r, tx, project, workspace, snapshot, node, patch) if err != nil { - reject(w, 413, "canvas_too_large") - return nil + return rejectRequest(413, "canvas_too_large") } snapshot = change.Snapshot if err = s.saveCanvasAIChange(r, tx, project, workspace, account, change); err != nil { @@ -182,6 +205,10 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr } result := response{"run_id": id, "step_id": stepID, "board_id": targetBoard, "node_id": node, "output_id": input.OutputID, "canvas_version": snapshot.Version, "patch": patch} change.addCanonical(result) + if receipt != nil { + result["status"] = "applied" + bindCanvasAIReceipt(result, receipt) + } if err = appendAIEvent(r.Context(), tx, project, workspace, id, "step.applied", result); err != nil { return err } @@ -190,10 +217,6 @@ func (s *service) applyCanvasAIOutput(w http.ResponseWriter, r *http.Request, pr if err != nil { return err } - if err = tx.Commit(); err != nil { - return err - } setVersion(w, snapshot.Version) - reply(w, 200, result) - return nil + return commitCanvasAIReceipt(w, r, tx, project, workspace, account, receipt, 200, "applied", result) } diff --git a/internal/platform/canvas_ai_model_integration_test.go b/internal/platform/canvas_ai_model_integration_test.go index f39d1d0..71fdc0b 100644 --- a/internal/platform/canvas_ai_model_integration_test.go +++ b/internal/platform/canvas_ai_model_integration_test.go @@ -272,12 +272,17 @@ func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) { outputID := conflict["outputs"].([]any)[0].(map[string]any)["output"].(map[string]any)["output_id"].(string) applyKey := modelTestID(6200) expected := read().Versions[root] - applyInput := response{"node_id": node, "output_id": outputID} + applyInput := response{"protocol_version": 2, "request_id": applyKey, "operation_id": modelTestID(6199), "target_incarnation": 1, "target_canvas_id": root, "node_id": node, "output_id": outputID} applied := call("POST", conflictPath+"/apply", applyInput, 200, expected, applyKey) again := call("POST", conflictPath+"/apply", applyInput, 200, expected, applyKey) if !modelEqual(applied, again) || applied["model"] == nil || read().Elements[node]["geometry"].(map[string]any)["width"] != float64(9000) { t.Fatal("explicit canonical application changed geometry or receipt") } + applyReceipt := call("GET", base(root)+"/ai/operations/"+applyKey, nil, 200, 0, "") + if !modelEqual(applied, applyReceipt) || applied["action"] != "run.apply" || applied["status"] != "applied" { + t.Fatal("explicit apply receipt is not recoverable") + } + call("POST", conflictPath+"/apply", applyInput, 409, expected+1, applyKey) // Keep one unapplied old output, then replace the target incarnation atomically. stale := start(root, node, "text", 1) hooks <- func() { @@ -315,7 +320,16 @@ func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) { call("POST", base(root)+"/ai/runs", bad, 409, read().Versions[root], badKey) restored := start(root, node, "text", 2) cancelPath := base(root) + "/ai/runs/" + restored["run_id"].(string) + "/cancel" - call("POST", cancelPath, response{}, 200, 0, modelTestID(6202)) + cancelKey := modelTestID(6202) + cancelInput := response{"protocol_version": 2, "request_id": cancelKey, "operation_id": modelTestID(6203)} + cancelled := call("POST", cancelPath, cancelInput, 200, 0, cancelKey) + cancelAgain := call("POST", cancelPath, cancelInput, 200, 0, cancelKey) + cancelReceipt := call("GET", base(root)+"/ai/operations/"+cancelKey, nil, 200, 0, "") + if cancelled["status"] != "applied" || cancelled["run_status"] != "cancelled" || !modelEqual(cancelled, cancelAgain) || !modelEqual(cancelled, cancelReceipt) { + t.Fatal("cancel receipt changed on retry/query") + } + duplicateCancelKey := modelTestID(6204) + call("POST", cancelPath, response{"protocol_version": 2, "request_id": duplicateCancelKey, "operation_id": modelTestID(6203)}, 409, 0, duplicateCancelKey) if executions.Load() != 3 { t.Fatal("duplicate/cancelled/old execution called the model") } @@ -368,7 +382,27 @@ func TestCanonicalAIAdmissionAndLifecycle(t *testing.T) { staleInput, staleKey := command(child, imageID, "image", 1) staleVersions := staleInput["expected_versions"].(map[string]int64) staleVersions[root]-- // The target version is current; its ancestor context is not. - call("POST", base(child)+"/ai/runs", staleInput, 412, read().Versions[child], staleKey) + beforeRejected := read() + denied := call("POST", base(child)+"/ai/runs", staleInput, 412, read().Versions[child], staleKey) + if denied["status"] != "not-applied" || denied["terminal"] != true || denied["action"] != "run.start" || !modelEqual(beforeRejected, read()) { + t.Fatal("rejection was not a zero-effect receipt") + } + if queried := call("GET", base(child)+"/ai/operations/"+staleKey, nil, 200, 0, ""); !modelEqual(denied, queried) { + t.Fatal("negative admission receipt lost") + } + // A later context change cannot turn an old rejected key into a paid execution. + mutate(func(model *canvasModel) { model.Elements[imageID]["geometry"].(map[string]any)["x"] = float64(123) }) + if retried := call("POST", base(child)+"/ai/runs", staleInput, 412, beforeRejected.Versions[child], staleKey); !modelEqual(denied, retried) { + t.Fatal("rejected request was reevaluated") + } + missingRunKey := modelTestID(6400) + missingCancel := call("POST", base(child)+"/ai/runs/"+modelTestID(6401)+"/cancel", response{"protocol_version": 2, "request_id": missingRunKey, "operation_id": modelTestID(6402)}, 404, 0, missingRunKey) + if missingCancel["status"] != "not-applied" || missingCancel["terminal"] != true { + t.Fatal("missing run cancel has no stable rejection") + } + if queried := call("GET", base(child)+"/ai/operations/"+missingRunKey, nil, 200, 0, ""); !modelEqual(missingCancel, queried) { + t.Fatal("negative cancel receipt lost") + } missingInput, missingKey := command(child, imageID, "image", 1) delete(missingInput, "expected_versions") call("POST", base(child)+"/ai/runs", missingInput, 400, read().Versions[child], missingKey) diff --git a/internal/platform/canvas_ai_query.go b/internal/platform/canvas_ai_query.go index db6ecd7..6ed9c78 100644 --- a/internal/platform/canvas_ai_query.go +++ b/internal/platform/canvas_ai_query.go @@ -181,6 +181,14 @@ func (s *service) getCanvasAIAdmission(w http.ResponseWriter, r *http.Request, p return err } var body json.RawMessage + err = tx.QueryRowContext(r.Context(), `SELECT response_body FROM canvas_ai_requests WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND account_id=$4 AND request_id=$5`, project, workspace, board, account, key).Scan(&body) + if err == nil { + reply(w, 200, body) + return nil + } + if err != sql.ErrNoRows { + return err + } err = tx.QueryRowContext(r.Context(), `SELECT create_response FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND account_id=$4 AND idempotency_key=$5`, project, workspace, board, account, key).Scan(&body) if err == sql.ErrNoRows { reply(w, 404, response{"protocol_version": 2, "request_id": key, "status": "unknown", "error": "ai_admission_unknown"}) diff --git a/internal/platform/canvas_ai_receipts.go b/internal/platform/canvas_ai_receipts.go new file mode 100644 index 0000000..5d1c908 --- /dev/null +++ b/internal/platform/canvas_ai_receipts.go @@ -0,0 +1,83 @@ +package platform + +import ( + "database/sql" + "encoding/json" + "net/http" +) + +type canvasAIRequest struct { + Key, Hash, Action, Board, Run, Operation string +} + +// The workspace lock is acquired by the caller before this savepoint. Rejected +// requests roll back every staged effect and retain that lock while recording +// their zero-effect receipt, so a response can never race a second invocation. +func beginCanvasAIRequest(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account string, request *canvasAIRequest) (bool, error) { + if request == nil { + return false, nil + } + var board, run, action, hash string + var status int + var body json.RawMessage + err := tx.QueryRowContext(r.Context(), `SELECT board_id::text,COALESCE(run_id::text,''),action,payload_hash,response_status,response_body FROM canvas_ai_requests WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND request_id=$4`, project, workspace, account, request.Key).Scan(&board, &run, &action, &hash, &status, &body) + if err == nil { + if board != request.Board || run != request.Run || action != request.Action || hash != request.Hash { + reject(w, 409, "idempotency_conflict") + return true, nil + } + reply(w, status, body) + return true, nil + } + if err != sql.ErrNoRows { + return false, err + } + var completed bool + err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_requests WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND operation_id=$4 AND outcome='applied')`, project, workspace, account, request.Operation).Scan(&completed) + if err != nil { + return false, err + } + if completed { + reject(w, 409, "operation_identity_conflict") + return true, nil + } + _, err = tx.ExecContext(r.Context(), `SAVEPOINT canvas_ai_effects`) + return false, err +} +func bindCanvasAIReceipt(body response, request *canvasAIRequest) { + if request == nil { + return + } + body["protocol_version"], body["request_id"], body["operation_id"], body["action"] = 2, request.Key, request.Operation, request.Action + if request.Run != "" { + body["run_id"] = request.Run + } +} +func commitCanvasAIReceipt(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account string, request *canvasAIRequest, status int, outcome string, body response) error { + if request != nil { + bindCanvasAIReceipt(body, request) + raw, err := json.Marshal(body) + if err != nil { + return err + } + _, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_requests(project,workspace_id,account_id,request_id,board_id,run_id,action,operation_id,payload_hash,outcome,response_status,response_body) VALUES($1,$2,$3,$4,$5,NULLIF($6,'')::uuid,$7,$8,$9,$10,$11,$12)`, project, workspace, account, request.Key, request.Board, request.Run, request.Action, request.Operation, request.Hash, outcome, status, raw) + if err != nil { + return err + } + } + if err := tx.Commit(); err != nil { + return err + } + reply(w, status, body) + return nil +} +func rejectCanvasAIRequest(w http.ResponseWriter, r *http.Request, tx *sql.Tx, project, workspace, account string, request *canvasAIRequest, status int, code string) error { + if request == nil { + reject(w, status, code) + return nil + } + if _, err := tx.ExecContext(r.Context(), `ROLLBACK TO SAVEPOINT canvas_ai_effects`); err != nil { + return err + } + return commitCanvasAIReceipt(w, r, tx, project, workspace, account, request, status, "not-applied", response{"status": "not-applied", "terminal": true, "error": code}) +} diff --git a/internal/platform/canvas_ai_runs.go b/internal/platform/canvas_ai_runs.go index ee4634f..7b5ae34 100644 --- a/internal/platform/canvas_ai_runs.go +++ b/internal/platform/canvas_ai_runs.go @@ -91,13 +91,21 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj if err != nil { return err } + var receipt *canvasAIRequest + if input.ProtocolVersion == 2 && input.RequestID == key && uuidSlugPattern.MatchString(input.OperationID) && input.TargetIncarnation > 0 { + receipt = &canvasAIRequest{Key: key, Hash: hash, Action: "run.start", Board: board, Operation: input.OperationID} + if done, e := beginCanvasAIRequest(w, r, tx, project, workspace, account, receipt); done || e != nil { + return e + } + } + rejectRequest := func(status int, code string) error { + return rejectCanvasAIRequest(w, r, tx, project, workspace, account, receipt, status, code) + } if kind != "board" { - reject(w, 400, "invalid_material") - return nil + return rejectRequest(400, "invalid_material") } if role == "viewer" { - reject(w, 403, "forbidden") - return nil + return rejectRequest(403, "forbidden") } var previousHash string var previous json.RawMessage @@ -133,8 +141,7 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } if parent.Version != expected { - reject(w, 412, "version_conflict") - return nil + return rejectRequest(412, "version_conflict") } var doc map[string]any _ = json.Unmarshal(parent.Document, &doc) @@ -146,12 +153,10 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj } } if target == nil { - reject(w, 404, "not_found") - return nil + return rejectRequest(404, "not_found") } if !canvasChoice(target["kind"], "text", "script", "plain-text", "image", "plain-image") { - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } root, incarnation, _, err := canvasAIIdentity(r.Context(), tx, project, workspace, board, input.NodeID) if err != nil { @@ -159,21 +164,17 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj } if root != "" { if input.ProtocolVersion != 2 || input.RequestID != key || !uuidSlugPattern.MatchString(input.OperationID) || input.TargetIncarnation < 1 { - reject(w, 400, "invalid_ai_protocol") - return nil + return rejectRequest(400, "invalid_ai_protocol") } if input.TargetIncarnation != incarnation { - reject(w, 409, "target_changed") - return nil + return rejectRequest(409, "target_changed") } } else if input.ProtocolVersion != 0 || input.RequestID != "" || input.TargetIncarnation != 0 || input.OperationID != "" || input.ExpectedVersions != nil { - reject(w, 409, "canonical_migration_required") - return nil + return rejectRequest(409, "canonical_migration_required") } change, err := prepareCanvasAIPatch(r, tx, project, workspace, parent, input.NodeID, response{"ai": map[string]any{"enabled": true, "modelId": input.ModelID, "prompt": input.Prompt, "parameters": input.Parameters, "propagate": input.Propagate}}) if err != nil { - reject(w, 400, "invalid_ai_request") - return nil + return rejectRequest(400, "invalid_ai_request") } if change.Before != nil { if code := validateCanvasAIVersions(*change.Before, board, expected, input.ExpectedVersions); code != "" { @@ -181,16 +182,14 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj if code == "version_conflict" { status = http.StatusPreconditionFailed } - reject(w, status, code) - return nil + return rejectRequest(status, code) } } ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) defer cancel() models, err := s.canvasAI.Models(ctx) if err != nil { - reject(w, 503, "codex_unavailable") - return nil + return rejectRequest(503, "codex_unavailable") } catalogue := map[string]codex.CanvasModel{} for _, model := range models { @@ -200,11 +199,10 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj plan, err := loadCanvasAIPlan(r, tx, project, workspace, account, parent, input.NodeID, input.Propagate) if err != nil { if err.Error() == "execution_cycle" { - reject(w, 409, "execution_cycle") + return rejectRequest(409, "execution_cycle") } else { - reject(w, 413, "graph_limit_exceeded") + return rejectRequest(413, "graph_limit_exceeded") } - return nil } if err = bindCanvasAIIdentities(r.Context(), tx, project, workspace, &plan); err != nil { return err @@ -222,16 +220,14 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj } model, exists := catalogue[step.ModelID] if !exists || !model.Available { - reject(w, 422, "model_unavailable") - return nil + return rejectRequest(422, "model_unavailable") } modality := "text" if canvasChoice(step.Kind, "image", "plain-image") { modality = "image" } if !canvasModelSupports(model, modality) { - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } imageSources := map[string]bool{} if canvasChoice(step.Kind, "image", "plain-image") { @@ -248,12 +244,10 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj } } if len(imageSources) > 0 && !canvasModelAccepts(model, "image") { - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } if len(imageSources) > 4 { - reject(w, 413, "context_too_large") - return nil + return rejectRequest(413, "context_too_large") } plan.Steps[index].MaxExecutionMS = model.MaxExecutionMS if canvasChoice(step.Kind, "image", "plain-image") { @@ -264,8 +258,7 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj frozenImages[step.ID] = content } if e != nil { - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } plan.Steps[index].FrozenContent = content } @@ -282,26 +275,22 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj frozenImages[key] = content } if e != nil { - reject(w, 422, "model_incompatible") - return nil + return rejectRequest(422, "model_incompatible") } plan.Steps[index].StaticInputs[staticIndex].Content = content } if _, e := assembleCanvasTask(plan, plan.Steps[index], nil); e != nil { - reject(w, 413, "context_too_large") - return nil + return rejectRequest(413, "context_too_large") } parameters, _ := step.Config["parameters"].(map[string]any) if len(parameters) > 0 { - reject(w, 400, "invalid_ai_request") - return nil + return rejectRequest(400, "invalid_ai_request") } prompt, _ := step.Config["prompt"].(string) text, _ := step.FrozenContent["text"].(string) if len(prompt)+len(text) > model.MaxInputBytes { - reject(w, 413, "context_too_large") - return nil + return rejectRequest(413, "context_too_large") } var busy bool err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND node_id=$4 AND status IN('queued','running'))`, project, workspace, step.BoardID, step.NodeID).Scan(&busy) @@ -309,8 +298,7 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } if busy { - reject(w, 409, "node_busy") - return nil + return rejectRequest(409, "node_busy") } } // A project-scoped advisory lock makes the queue limit atomic across workspaces. @@ -325,8 +313,7 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj } if queued >= 128 { w.Header().Set("Retry-After", "5") - reject(w, 429, "queue_full") - return nil + return rejectRequest(429, "queue_full") } var id string err = tx.QueryRowContext(r.Context(), `SELECT gen_random_uuid()::text`).Scan(&id) @@ -340,6 +327,7 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj result["protocol_version"], result["request_id"], result["admission"], result["operation_id"] = 2, key, "accepted", input.OperationID change.addCanonical(result) } + bindCanvasAIReceipt(result, receipt) resultJSON, _ := json.Marshal(result) _, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_runs(project,workspace_id,board_id,id,account_id,idempotency_key,payload_hash,status,plan,create_response,logical_operation_id) VALUES($1,$2,$3,$4,$5,$6,$7,'queued',$8,$9,NULLIF($10,'')::uuid)`, project, workspace, board, id, account, key, hash, planJSON, resultJSON, input.OperationID) if err != nil { @@ -361,10 +349,6 @@ func (s *service) createCanvasAIRun(w http.ResponseWriter, r *http.Request, proj return err } } - if err = tx.Commit(); err != nil { - return err - } setVersion(w, parent.Version) - reply(w, 202, result) - return nil + return commitCanvasAIReceipt(w, r, tx, project, workspace, account, receipt, 202, "applied", result) } diff --git a/internal/platform/canvas_capabilities.go b/internal/platform/canvas_capabilities.go index ad8970a..694e19b 100644 --- a/internal/platform/canvas_capabilities.go +++ b/internal/platform/canvas_capabilities.go @@ -39,7 +39,7 @@ func (s *service) canvasCapabilities(w http.ResponseWriter, r *http.Request, pro } } } - ai := response{"canonical_target_identity": true, "admission_protocol_version": 2, "admission_receipt_query": true, "available": textAvailable || imageAvailable, "text": textAvailable, "image": imageAvailable} + ai := response{"canonical_target_identity": true, "admission_protocol_version": 2, "admission_receipt_query": true, "mutation_protocol_version": 2, "terminal_rejection_receipts": true, "available": textAvailable || imageAvailable, "text": textAvailable, "image": imageAvailable} if modelErr != nil { ai["unavailable_reason"] = "codex_unavailable" }