298 lines
11 KiB
Go
298 lines
11 KiB
Go
package platform
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"gitea.entroia.com/Zayden-Jung/foundry-core/internal/codex"
|
|
)
|
|
|
|
func (s *service) startCanvasStep(ctx context.Context, run canvasQueueRun, step canvasAIStepPlan) (bool, error) {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
defer tx.Rollback()
|
|
if err = lockCanvasAIWorkspace(ctx, tx, run.Project, run.Workspace); err != nil {
|
|
return false, err
|
|
}
|
|
board, identityOK, err := resolveCanvasAITarget(ctx, tx, run.Project, run.Workspace, step)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if !identityOK {
|
|
tx.Rollback()
|
|
_, err = s.finishCanvasStep(ctx, run, step, "", "target_removed", false)
|
|
return false, err
|
|
}
|
|
_, role, permissionErr := lockMaterial(ctx, tx, run.Project, run.Workspace, board, run.Account)
|
|
if permissionErr != nil || role == "viewer" {
|
|
tx.Rollback()
|
|
if permissionErr != nil && permissionErr != sql.ErrNoRows {
|
|
return false, permissionErr
|
|
}
|
|
_, err = s.finishCanvasStep(ctx, run, step, "", "permission_revoked", false)
|
|
return false, err
|
|
}
|
|
var active bool
|
|
err = tx.QueryRowContext(ctx, `SELECT status='running' AND lease_id=$4 AND NOT cancel_requested FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND id=$3 FOR UPDATE`, run.Project, run.Workspace, run.ID, run.Lease).Scan(&active)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
if !active {
|
|
tx.Rollback()
|
|
_, err = s.finishCanvasStep(ctx, run, step, "", "generation_cancelled", false)
|
|
return false, err
|
|
}
|
|
snapshot, err := loadCanvas(queueRequest(ctx), tx, run.Project, run.Workspace, board)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
data := canvasNodeData(snapshot.Document)[step.NodeID]
|
|
if !identityOK || data == nil || canvasTargetFingerprint(step.BoardID, step.NodeID, data) != step.Fingerprint {
|
|
tx.Rollback()
|
|
_, err = s.finishCanvasStep(ctx, run, step, "", "target_changed", false)
|
|
return false, err
|
|
}
|
|
result, err := tx.ExecContext(ctx, `UPDATE canvas_ai_steps SET status='running',started_at=now() WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND id=$4 AND status='queued'`, run.Project, run.Workspace, run.ID, step.ID)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
count, _ := result.RowsAffected()
|
|
if count != 1 {
|
|
return false, nil
|
|
}
|
|
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.running", response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID}); err != nil {
|
|
return false, err
|
|
}
|
|
for _, edge := range run.Plan.Edges {
|
|
if edge.Target == step.ID {
|
|
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "edge.active", response{"logical_edge_id": edge.ID, "step_id": step.ID, "status": "active"}); err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
func (s *service) finishCanvasStep(ctx context.Context, run canvasQueueRun, step canvasAIStepPlan, text, code string, started bool) (string, error) {
|
|
return s.finishCanvasStepOutput(ctx, run, step, codex.CanvasOutput{Kind: "text", Text: text}, code, started)
|
|
}
|
|
func (s *service) finishCanvasStepOutput(ctx context.Context, run canvasQueueRun, step canvasAIStepPlan, generated codex.CanvasOutput, code string, started bool) (string, error) {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
defer tx.Rollback()
|
|
var locked string
|
|
err = tx.QueryRowContext(ctx, `SELECT id::text FROM workspaces WHERE project=$1 AND id=$2 FOR UPDATE`, run.Project, run.Workspace).Scan(&locked)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var runStatus string
|
|
var cancelled bool
|
|
var lease sql.NullString
|
|
err = tx.QueryRowContext(ctx, `SELECT status,cancel_requested,lease_id::text FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND id=$3 FOR UPDATE`, run.Project, run.Workspace, run.ID).Scan(&runStatus, &cancelled, &lease)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
var oldStatus string
|
|
err = tx.QueryRowContext(ctx, `SELECT status FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND id=$4 FOR UPDATE`, run.Project, run.Workspace, run.ID, step.ID).Scan(&oldStatus)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if oldStatus != "queued" && oldStatus != "running" {
|
|
return oldStatus, nil
|
|
}
|
|
if runStatus != "running" || !lease.Valid || lease.String != run.Lease {
|
|
return oldStatus, nil
|
|
}
|
|
status := "succeeded"
|
|
switch code {
|
|
case "dependency_failed":
|
|
status = "skipped"
|
|
case "target_changed":
|
|
status = "conflict"
|
|
case "generation_cancelled", "target_removed":
|
|
status = "cancelled"
|
|
default:
|
|
if code != "" {
|
|
status = "failed"
|
|
}
|
|
}
|
|
if cancelled {
|
|
status = "cancelled"
|
|
code = "generation_cancelled"
|
|
}
|
|
board, identityOK, err := resolveCanvasAITarget(ctx, tx, run.Project, run.Workspace, step)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
eventBoard := board
|
|
if !identityOK {
|
|
eventBoard = step.BoardID
|
|
}
|
|
var appliedBoard any
|
|
var appliedEvent response
|
|
var output any
|
|
var version any
|
|
if code == "" && started {
|
|
var outputID string
|
|
err = tx.QueryRowContext(ctx, `SELECT gen_random_uuid()::text`).Scan(&outputID)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
patch := response{}
|
|
if generated.Kind == "image" {
|
|
imageOutput, err := s.persistCanvasImage(ctx, tx, run, outputID, generated.Image)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
output = imageOutput
|
|
patch["src"] = imageOutput["src"]
|
|
patch["assetId"] = imageOutput["assetId"]
|
|
} else {
|
|
output = response{"output_id": outputID, "kind": "text", "text": generated.Text}
|
|
patch["text"] = generated.Text
|
|
}
|
|
|
|
_, role, permissionErr := lockCanvasAIRunRoute(ctx, tx, run.Project, run.Workspace, eventBoard, run.Account)
|
|
if permissionErr != nil && permissionErr != sql.ErrNoRows {
|
|
return "", permissionErr
|
|
}
|
|
if permissionErr != nil || role == "viewer" {
|
|
status = "failed"
|
|
code = "permission_revoked"
|
|
} else {
|
|
snapshot, err := loadCanvas(queueRequest(ctx), tx, run.Project, run.Workspace, eventBoard)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
target := canvasNodeData(snapshot.Document)[step.NodeID]
|
|
if image, ok := output.(response); ok && image["kind"] == "image" && identityOK {
|
|
if _, err = tx.ExecContext(ctx, `INSERT INTO canvas_asset_bindings(project,workspace_id,asset_id,board_id) VALUES($1,$2,$3,$4) ON CONFLICT DO NOTHING`, run.Project, run.Workspace, image["assetId"], eventBoard); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
if !identityOK || target == nil || canvasTargetFingerprint(step.BoardID, step.NodeID, target) != step.Fingerprint {
|
|
status, code = "conflict", "target_changed"
|
|
} else {
|
|
change, err := prepareCanvasAIPatch(queueRequest(ctx), tx, run.Project, run.Workspace, snapshot, step.NodeID, patch)
|
|
if err != nil {
|
|
if !errors.Is(err, errAIInvalidPatch) && !errors.Is(err, errCanvasModel) && !errors.Is(err, errGraphCapacity) {
|
|
return "", err
|
|
}
|
|
status, code = "failed", "generation_output_too_large"
|
|
} else {
|
|
if err = s.saveCanvasAIChange(queueRequest(ctx), tx, run.Project, run.Workspace, run.Account, change); err != nil {
|
|
return "", err
|
|
}
|
|
version = change.Snapshot.Version
|
|
appliedBoard = eventBoard
|
|
applied := response{"step_id": step.ID, "board_id": eventBoard, "node_id": step.NodeID, "patch": patch, "canvas_version": change.Snapshot.Version}
|
|
change.addTargetContext(applied, step)
|
|
appliedEvent = applied
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if output != nil {
|
|
event := response{"step_id": step.ID, "board_id": eventBoard, "origin_board_id": step.BoardID, "canonical_root": step.CanonicalRoot, "target_incarnation": step.TargetIncarnation, "node_id": step.NodeID, "output": output}
|
|
if status == "conflict" && identityOK && step.CanonicalRoot != "" && eventBoard != step.BoardID {
|
|
model, e := loadCanvasModel(ctx, tx, run.Project, run.Workspace, step.CanonicalRoot)
|
|
if e != nil {
|
|
return "", e
|
|
}
|
|
event["model"], event["document_format_version"], event["canvas_version"] = model.view(nil), canvasModelFormat, model.Versions[eventBoard]
|
|
}
|
|
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.output", event); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
if appliedEvent != nil {
|
|
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step.applied", appliedEvent); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
var outputJSON any
|
|
if output != nil {
|
|
encoded, _ := json.Marshal(output)
|
|
outputJSON = encoded
|
|
}
|
|
_, err = tx.ExecContext(ctx, `UPDATE canvas_ai_steps SET status=$5,output=$6,error_code=NULLIF($7,''),applied_version=$8,applied_board_id=$9,completed_at=now() WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND id=$4`, run.Project, run.Workspace, run.ID, step.ID, status, outputJSON, code, version, appliedBoard)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if status != "succeeded" {
|
|
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "step."+status, response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "status": status, "error": code}); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
edgeStatus := "completed"
|
|
if status == "failed" || status == "conflict" {
|
|
edgeStatus = "failed"
|
|
} else if status != "succeeded" {
|
|
edgeStatus = "skipped"
|
|
}
|
|
for _, edge := range run.Plan.Edges {
|
|
if edge.Target == step.ID {
|
|
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "edge."+edgeStatus, response{"logical_edge_id": edge.ID, "step_id": step.ID, "status": edgeStatus}); err != nil {
|
|
return "", err
|
|
}
|
|
}
|
|
}
|
|
if err = tx.Commit(); err != nil {
|
|
return "", err
|
|
}
|
|
return status, nil
|
|
}
|
|
func (s *service) completeCanvasRun(ctx context.Context, run canvasQueueRun, _ map[string]canvasStepResult) error {
|
|
tx, err := s.db.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer tx.Rollback()
|
|
var status string
|
|
var cancelled bool
|
|
var lease sql.NullString
|
|
err = tx.QueryRowContext(ctx, `SELECT status,cancel_requested,lease_id::text FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND id=$3 FOR UPDATE`, run.Project, run.Workspace, run.ID).Scan(&status, &cancelled, &lease)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if status != "running" || !lease.Valid || lease.String != run.Lease {
|
|
return nil
|
|
}
|
|
var succeeded, failed, conflicts, cancelledSteps, active int
|
|
err = tx.QueryRowContext(ctx, `SELECT count(*) FILTER(WHERE status='succeeded'),count(*) FILTER(WHERE status='failed'),count(*) FILTER(WHERE status='conflict'),count(*) FILTER(WHERE status='cancelled'),count(*) FILTER(WHERE status IN('queued','running')) FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3`, run.Project, run.Workspace, run.ID).Scan(&succeeded, &failed, &conflicts, &cancelledSteps, &active)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if active > 0 {
|
|
return nil
|
|
}
|
|
status = "succeeded"
|
|
if cancelled {
|
|
status = "cancelled"
|
|
} else if failed+conflicts+cancelledSteps > 0 {
|
|
status = "failed"
|
|
if succeeded > 0 {
|
|
status = "partially_succeeded"
|
|
} else if cancelledSteps > 0 && failed+conflicts == 0 {
|
|
status = "cancelled"
|
|
} else if conflicts > 0 && failed == 0 {
|
|
status = "conflict"
|
|
}
|
|
}
|
|
_, err = tx.ExecContext(ctx, `UPDATE canvas_ai_runs SET status=$4,completed_at=now(),lease_id=NULL,lease_until=NULL WHERE project=$1 AND workspace_id=$2 AND id=$3`, run.Project, run.Workspace, run.ID, status)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err = appendAIEvent(ctx, tx, run.Project, run.Workspace, run.ID, "run.completed", response{"status": status}); err != nil {
|
|
return err
|
|
}
|
|
return tx.Commit()
|
|
}
|