Persist queryable AI action outcomes and zero-effect rejection receipts
This commit is contained in:
@@ -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;
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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"})
|
||||
|
||||
@@ -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})
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user