foundry-core/internal/platform/canvas_ai_query.go

package platform

import (
	"database/sql"
	"encoding/json"
	"net/http"
	"strconv"
)

func publicAIPlan(plan canvasAIPlan) response {
	steps := []any{}
	for _, step := range plan.Steps {
		static := []any{}
		for _, source := range step.StaticInputs {
			static = append(static, response{"logical_edge_id": source.EdgeID, "board_id": source.BoardID, "node_id": source.NodeID, "kind": source.Kind, "source": "frozen_snapshot"})
		}
		steps = append(steps, response{"step_id": step.ID, "board_id": step.BoardID, "node_id": step.NodeID, "kind": step.Kind, "model_id": step.ModelID, "max_execution_ms": step.MaxExecutionMS, "status": step.Status, "dependencies": step.Dependencies, "static_inputs": static})
	}
	return response{"steps": steps, "edges": plan.Edges, "frozen_versions": plan.Versions}
}
func (s *service) getCanvasAIRun(w http.ResponseWriter, r *http.Request, project, account string) error {
	if r.Method != http.MethodGet {
		reject(w, 405, "method_not_allowed")
		return nil
	}
	id := r.PathValue("run")
	if !uuidSlugPattern.MatchString(id) {
		reject(w, 404, "not_found")
		return nil
	}
	after := int64(0)
	limit := 100
	var err error
	if raw := r.URL.Query().Get("after_seq"); raw != "" {
		after, err = strconv.ParseInt(raw, 10, 64)
	}
	if err != nil || after < 0 {
		reject(w, 400, "invalid_pagination")
		return nil
	}
	if raw := r.URL.Query().Get("limit"); raw != "" {
		limit, err = strconv.Atoi(raw)
	}
	if err != nil || limit < 1 || limit > 100 {
		reject(w, 400, "invalid_pagination")
		return nil
	}
	workspace, board := r.PathValue("workspace"), r.PathValue("material")
	tx, err := s.db.BeginTx(r.Context(), nil)
	if err != nil {
		return err
	}
	defer tx.Rollback()
	if _, _, err = lockMaterial(r.Context(), tx, project, workspace, board, account); err != nil {
		return err
	}
	var status string
	var rawPlan json.RawMessage
	var last int64
	var errorCode sql.NullString
	err = tx.QueryRowContext(r.Context(), `SELECT status,plan,last_seq,error_code FROM canvas_ai_runs WHERE project=$1 AND workspace_id=$2 AND board_id=$3 AND id=$4`, project, workspace, board, id).Scan(&status, &rawPlan, &last, &errorCode)
	if err != nil {
		return err
	}
	var plan canvasAIPlan
	if err = json.Unmarshal(rawPlan, &plan); err != nil {
		return err
	}
	checked := map[string]bool{board: true}
	for _, step := range plan.Steps {
		if !checked[step.BoardID] {
			if _, _, err = lockMaterial(r.Context(), tx, project, workspace, step.BoardID, account); err != nil {
				return err
			}
			checked[step.BoardID] = true
		}
	}
	rows, err := tx.QueryContext(r.Context(), `SELECT id,board_id::text,node_id,status,error_code,applied_version,CASE WHEN output IS NULL THEN NULL ELSE output-'text' END FROM canvas_ai_steps WHERE project=$1 AND workspace_id=$2 AND run_id=$3 ORDER BY ordinal`, project, workspace, id)
	if err != nil {
		return err
	}
	steps := []any{}
	outputs := []any{}
	versions := map[string]int64{}
	for rows.Next() {
		var step, node, stepBoard, stepStatus string
		var code sql.NullString
		var version sql.NullInt64
		var output []byte
		if err = rows.Scan(&step, &stepBoard, &node, &stepStatus, &code, &version, &output); err != nil {
			rows.Close()
			return err
		}
		item := response{"step_id": step, "board_id": stepBoard, "node_id": node, "status": stepStatus}
		if code.Valid {
			item["error"] = code.String
		}
		if version.Valid {
			item["applied_version"] = version.Int64
			if version.Int64 > versions[stepBoard] {
				versions[stepBoard] = version.Int64
			}
		}
		if len(output) > 0 {
			outputs = append(outputs, response{"step_id": step, "output": json.RawMessage(output)})
		}
		steps = append(steps, item)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	rows, err = tx.QueryContext(r.Context(), `SELECT seq,type,data,created_at FROM canvas_ai_events WHERE project=$1 AND workspace_id=$2 AND run_id=$3 AND seq>$4 ORDER BY seq LIMIT $5`, project, workspace, id, after, limit)
	if err != nil {
		return err
	}
	events := []any{}
	next := after
	size := 0
	for rows.Next() {
		var seq int64
		var kind string
		var data json.RawMessage
		var timestamp any
		if err = rows.Scan(&seq, &kind, &data, &timestamp); err != nil {
			rows.Close()
			return err
		}
		if size+len(data) > 800<<10 && len(events) > 0 {
			break
		}
		var event response
		if err = json.Unmarshal(data, &event); err != nil {
			rows.Close()
			return err
		}
		if event == nil {
			event = response{}
		}
		event["seq"] = seq
		event["run_id"] = id
		event["type"] = kind
		event["timestamp"] = timestamp
		events = append(events, event)
		next = seq
		size += len(data)
	}
	err = rows.Err()
	rows.Close()
	if err != nil {
		return err
	}
	result := response{"run_id": id, "status": status, "plan": publicAIPlan(plan), "steps": steps, "events": events, "next_seq": next, "has_more": next < last, "outputs": outputs, "applied_versions": versions, "retention_days": 30}
	if errorCode.Valid {
		result["error"] = errorCode.String
	}
	reply(w, 200, result)
	return nil
}