359 lines
14 KiB
Go
359 lines
14 KiB
Go
package platform
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"strings"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"gitea.entroia.com/Zayden-Jung/foundry-core/internal/codex"
|
|
)
|
|
|
|
func appendAIEvent(rctx context.Context, tx *sql.Tx, project, workspace, run, kind string, data any) error {
|
|
raw, err := json.Marshal(data)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
_, err = tx.ExecContext(rctx, `WITH next AS (UPDATE canvas_ai_runs SET last_seq=last_seq+1 WHERE project=$1 AND workspace_id=$2 AND id=$3 RETURNING last_seq) INSERT INTO canvas_ai_events(project,workspace_id,run_id,seq,type,data) SELECT $1,$2,$3,last_seq,$4,$5 FROM next`, project, workspace, run, kind, raw)
|
|
return err
|
|
}
|
|
func (s *service) createCanvasAIRun(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
|
|
}
|
|
expected, ok := expectedVersion(w, r)
|
|
if !ok {
|
|
return nil
|
|
}
|
|
raw, err := io.ReadAll(http.MaxBytesReader(w, r.Body, 256<<10))
|
|
if err != nil {
|
|
reject(w, 413, "context_too_large")
|
|
return nil
|
|
}
|
|
var input struct {
|
|
ProtocolVersion int `json:"protocol_version,omitempty"`
|
|
OperationID string `json:"operation_id,omitempty"`
|
|
RequestID string `json:"request_id,omitempty"`
|
|
TargetIncarnation int64 `json:"target_incarnation,omitempty"`
|
|
ExpectedVersions map[string]int64 `json:"expected_versions,omitempty"`
|
|
NodeID string `json:"node_id"`
|
|
ModelID string `json:"model_id"`
|
|
Prompt string `json:"prompt"`
|
|
Parameters map[string]any `json:"parameters"`
|
|
Propagate bool `json:"propagate"`
|
|
}
|
|
decoder := json.NewDecoder(bytes.NewReader(raw))
|
|
decoder.DisallowUnknownFields()
|
|
if !utf8.Valid(raw) || decoder.Decode(&input) != nil || decoder.Decode(new(any)) != io.EOF || !canvasID.MatchString(input.NodeID) || strings.TrimSpace(input.Prompt) == "" || !canvasString(input.Prompt, 16000) {
|
|
reject(w, 400, "invalid_ai_request")
|
|
return nil
|
|
}
|
|
if input.OperationID != "" && !uuidSlugPattern.MatchString(input.OperationID) {
|
|
reject(w, 400, "invalid_ai_request")
|
|
return nil
|
|
}
|
|
if input.Parameters == nil {
|
|
input.Parameters = map[string]any{}
|
|
}
|
|
params, _ := json.Marshal(input.Parameters)
|
|
if len(params) > 16<<10 {
|
|
reject(w, 413, "context_too_large")
|
|
return nil
|
|
}
|
|
hash := payloadDigest(input)
|
|
if input.ProtocolVersion != 0 {
|
|
// The version header is part of the immutable canonical admission request.
|
|
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 {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
kind, role, err := lockMaterial(r.Context(), tx, project, workspace, board, account)
|
|
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" {
|
|
return rejectRequest(400, "invalid_material")
|
|
}
|
|
if role == "viewer" {
|
|
return rejectRequest(403, "forbidden")
|
|
}
|
|
var previousHash string
|
|
var previous json.RawMessage
|
|
err = tx.QueryRowContext(r.Context(), `SELECT payload_hash,create_response FROM canvas_ai_runs WHERE project=$1 AND account_id=$2 AND board_id=$3 AND idempotency_key=$4 AND workspace_id=$5`, project, account, board, key, workspace).Scan(&previousHash, &previous)
|
|
if err == nil {
|
|
if previousHash != hash {
|
|
reject(w, 409, "idempotency_conflict")
|
|
return nil
|
|
}
|
|
var result struct {
|
|
Version int64 `json:"canvas_version"`
|
|
}
|
|
_ = json.Unmarshal(previous, &result)
|
|
setVersion(w, result.Version)
|
|
reply(w, 202, previous)
|
|
return nil
|
|
}
|
|
if err != sql.ErrNoRows {
|
|
return err
|
|
}
|
|
if input.OperationID != "" {
|
|
var admitted bool
|
|
if err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND account_id=$3 AND logical_operation_id=$4)`, project, workspace, account, input.OperationID).Scan(&admitted); err != nil {
|
|
return err
|
|
}
|
|
if admitted {
|
|
reject(w, 409, "operation_identity_conflict")
|
|
return nil
|
|
}
|
|
}
|
|
parent, err := loadCanvas(r, tx, project, workspace, board)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if parent.Version != expected {
|
|
return rejectRequest(412, "version_conflict")
|
|
}
|
|
var doc map[string]any
|
|
_ = json.Unmarshal(parent.Document, &doc)
|
|
var target map[string]any
|
|
for _, v := range doc["nodes"].([]any) {
|
|
n := v.(map[string]any)
|
|
if n["id"] == input.NodeID {
|
|
target = n["data"].(map[string]any)
|
|
}
|
|
}
|
|
if target == nil {
|
|
return rejectRequest(404, "not_found")
|
|
}
|
|
if !canvasChoice(target["kind"], "text", "script", "plain-text", "image", "plain-image") {
|
|
return rejectRequest(422, "model_incompatible")
|
|
}
|
|
root, incarnation, _, err := canvasAIIdentity(r.Context(), tx, project, workspace, board, input.NodeID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if root != "" {
|
|
if input.ProtocolVersion != 2 || input.RequestID != key || !uuidSlugPattern.MatchString(input.OperationID) || input.TargetIncarnation < 1 {
|
|
return rejectRequest(400, "invalid_ai_protocol")
|
|
}
|
|
if input.TargetIncarnation != incarnation {
|
|
return rejectRequest(409, "target_changed")
|
|
}
|
|
} else if input.ProtocolVersion != 0 || input.RequestID != "" || input.TargetIncarnation != 0 || input.OperationID != "" || input.ExpectedVersions != 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 {
|
|
return rejectRequest(400, "invalid_ai_request")
|
|
}
|
|
if change.Before != nil {
|
|
if code := validateCanvasAIVersions(*change.Before, board, expected, input.ExpectedVersions); code != "" {
|
|
status := http.StatusBadRequest
|
|
if code == "version_conflict" {
|
|
status = http.StatusPreconditionFailed
|
|
}
|
|
return rejectRequest(status, code)
|
|
}
|
|
}
|
|
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
|
|
defer cancel()
|
|
models, err := s.canvasAI.Models(ctx)
|
|
if err != nil {
|
|
return rejectRequest(503, "codex_unavailable")
|
|
}
|
|
catalogue := map[string]codex.CanvasModel{}
|
|
for _, model := range models {
|
|
catalogue[model.ID] = model
|
|
}
|
|
parent = change.Snapshot
|
|
plan, err := loadCanvasAIPlan(r, tx, project, workspace, account, parent, input.NodeID, input.Propagate)
|
|
if err != nil {
|
|
if err.Error() == "execution_cycle" {
|
|
return rejectRequest(409, "execution_cycle")
|
|
} else {
|
|
return rejectRequest(413, "graph_limit_exceeded")
|
|
}
|
|
}
|
|
if err = bindCanvasAIIdentities(r.Context(), tx, project, workspace, &plan); err != nil {
|
|
return err
|
|
}
|
|
stepKinds := map[string]string{}
|
|
for _, step := range plan.Steps {
|
|
stepKinds[step.ID] = step.Kind
|
|
}
|
|
freezeCtx, stopFreeze := context.WithTimeout(r.Context(), 10*time.Second)
|
|
defer stopFreeze()
|
|
frozenImages := map[string]map[string]any{}
|
|
for index, step := range plan.Steps {
|
|
if step.Status == "skipped" {
|
|
continue
|
|
}
|
|
model, exists := catalogue[step.ModelID]
|
|
if !exists || !model.Available {
|
|
return rejectRequest(422, "model_unavailable")
|
|
}
|
|
modality := "text"
|
|
if canvasChoice(step.Kind, "image", "plain-image") {
|
|
modality = "image"
|
|
}
|
|
if !canvasModelSupports(model, modality) {
|
|
return rejectRequest(422, "model_incompatible")
|
|
}
|
|
imageSources := map[string]bool{}
|
|
if canvasChoice(step.Kind, "image", "plain-image") {
|
|
imageSources[step.ID] = true
|
|
}
|
|
for _, edge := range plan.Edges {
|
|
if edge.Target == step.ID && canvasChoice(stepKinds[edge.Source], "image", "plain-image") {
|
|
imageSources[edge.Source] = true
|
|
}
|
|
}
|
|
for _, source := range step.StaticInputs {
|
|
if canvasChoice(source.Kind, "image", "plain-image") {
|
|
imageSources[canvasNodeKey(source.BoardID, source.NodeID)] = true
|
|
}
|
|
}
|
|
if len(imageSources) > 0 && !canvasModelAccepts(model, "image") {
|
|
return rejectRequest(422, "model_incompatible")
|
|
}
|
|
if len(imageSources) > 4 {
|
|
return rejectRequest(413, "context_too_large")
|
|
}
|
|
plan.Steps[index].MaxExecutionMS = model.MaxExecutionMS
|
|
if canvasChoice(step.Kind, "image", "plain-image") {
|
|
content := frozenImages[step.ID]
|
|
var e error
|
|
if content == nil {
|
|
content, e = s.freezeCanvasImage(freezeCtx, tx, project, workspace, step.BoardID, account, step.FrozenContent)
|
|
frozenImages[step.ID] = content
|
|
}
|
|
if e != nil {
|
|
return rejectRequest(422, "model_incompatible")
|
|
}
|
|
plan.Steps[index].FrozenContent = content
|
|
}
|
|
|
|
for staticIndex, source := range step.StaticInputs {
|
|
if !canvasChoice(source.Kind, "image", "plain-image") {
|
|
continue
|
|
}
|
|
key := canvasNodeKey(source.BoardID, source.NodeID)
|
|
content := frozenImages[key]
|
|
var e error
|
|
if content == nil {
|
|
content, e = s.freezeCanvasImage(freezeCtx, tx, project, workspace, source.BoardID, account, source.Content)
|
|
frozenImages[key] = content
|
|
}
|
|
if e != nil {
|
|
return rejectRequest(422, "model_incompatible")
|
|
}
|
|
plan.Steps[index].StaticInputs[staticIndex].Content = content
|
|
}
|
|
if _, e := assembleCanvasTask(plan, plan.Steps[index], nil); e != nil {
|
|
return rejectRequest(413, "context_too_large")
|
|
}
|
|
|
|
parameters, _ := step.Config["parameters"].(map[string]any)
|
|
if len(parameters) > 0 {
|
|
return rejectRequest(400, "invalid_ai_request")
|
|
}
|
|
prompt, _ := step.Config["prompt"].(string)
|
|
text, _ := step.FrozenContent["text"].(string)
|
|
if len(prompt)+len(text) > model.MaxInputBytes {
|
|
return rejectRequest(413, "context_too_large")
|
|
}
|
|
var busy bool
|
|
if step.CanonicalRoot != "" {
|
|
err = tx.QueryRowContext(r.Context(), `SELECT EXISTS(SELECT 1 FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND frozen_input->>'canonical_root'=$3 AND node_id=$4 AND frozen_input->>'target_incarnation'=$5::bigint::text AND status IN('queued','running'))`, project, workspace, step.CanonicalRoot, step.NodeID, step.TargetIncarnation).Scan(&busy)
|
|
} else {
|
|
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)
|
|
}
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if busy {
|
|
return rejectRequest(409, "node_busy")
|
|
}
|
|
}
|
|
// A project-scoped advisory lock makes the queue limit atomic across workspaces.
|
|
_, err = tx.ExecContext(r.Context(), `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, project+"/canvas-ai-queue")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
var queued int
|
|
err = tx.QueryRowContext(r.Context(), `SELECT count(*) FROM canvas_ai_runs WHERE project=$1 AND status IN('queued','running')`, project).Scan(&queued)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if queued >= 128 {
|
|
w.Header().Set("Retry-After", "5")
|
|
return rejectRequest(429, "queue_full")
|
|
}
|
|
var id string
|
|
err = tx.QueryRowContext(r.Context(), `SELECT gen_random_uuid()::text`).Scan(&id)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
planJSON, _ := json.Marshal(plan)
|
|
poll := "/projects/" + url.PathEscape(project) + "/workspaces/" + workspace + "/materials/" + board + "/canvas/ai/runs/" + id
|
|
result := response{"run_id": id, "status": "queued", "poll_url": poll, "plan": publicAIPlan(plan), "canvas_version": parent.Version}
|
|
if root != "" {
|
|
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 {
|
|
return err
|
|
}
|
|
if err = s.saveCanvasAIChange(r, tx, project, workspace, account, change); err != nil {
|
|
return err
|
|
}
|
|
if err = appendAIEvent(r.Context(), tx, project, workspace, id, "run.queued", response{"status": "queued"}); err != nil {
|
|
return err
|
|
}
|
|
for ordinal, step := range plan.Steps {
|
|
frozen, _ := json.Marshal(step)
|
|
_, err = tx.ExecContext(r.Context(), `INSERT INTO canvas_ai_steps(project,workspace_id,run_id,id,board_id,node_id,ordinal,status,model_id,frozen_input,target_fingerprint) VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)`, project, workspace, id, step.ID, step.BoardID, step.NodeID, ordinal, step.Status, step.ModelID, frozen, step.Fingerprint)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = appendAIEvent(r.Context(), tx, project, workspace, id, "step."+step.Status, response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "status": step.Status}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
setVersion(w, parent.Version)
|
|
return commitCanvasAIReceipt(w, r, tx, project, workspace, account, receipt, 202, "applied", result)
|
|
}
|