Implement Canvas V2 subgraphs, assets and asynchronous AI execution

internal/codex/canvas.go

@@ -0,0 +1,155 @@
+package codex
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"sort"
+	"strings"
+	"time"
+)
+
+const CanvasInputLimit = 1 << 20
+
+type CanvasModel struct {
+	ID               string         `json:"id"`
+	Label            string         `json:"label"`
+	Provider         string         `json:"provider"`
+	InputModalities  []string       `json:"input_modalities"`
+	OutputModalities []string       `json:"output_modalities"`
+	Available        bool           `json:"available"`
+	MaxExecutionMS   int            `json:"max_execution_ms"`
+	MaxInputBytes    int            `json:"max_input_bytes"`
+	ParametersSchema map[string]any `json:"parameters_schema"`
+	Backend          string         `json:"-"`
+}
+
+// Only curated catalog fields leave the worker. Raw CLI cache metadata,
+// local paths, authentication files and backend configuration stay private.
+func workerCanvasModels() ([]CanvasModel, error) {
+	raw, err := os.ReadFile("/home/node/.codex/models_cache.json")
+	if err != nil {
+		return nil, err
+	}
+	var cache struct {
+		Models []struct {
+			Slug       string   `json:"slug"`
+			Label      string   `json:"display_name"`
+			Visibility string   `json:"visibility"`
+			Inputs     []string `json:"input_modalities"`
+		} `json:"models"`
+	}
+	if err = json.Unmarshal(raw, &cache); err != nil {
+		return nil, err
+	}
+	result := []CanvasModel{}
+	for _, entry := range cache.Models {
+		if entry.Visibility != "list" || entry.Slug == "" || strings.HasPrefix(entry.Slug, "-") || len(entry.Slug) > 128 {
+			continue
+		}
+		sum := sha256.Sum256([]byte("codex-cli/" + entry.Slug))
+		inputs := []string{}
+		for _, value := range entry.Inputs {
+			if value == "text" || value == "image" {
+				inputs = append(inputs, value)
+			}
+		}
+		if len(inputs) == 0 {
+			continue
+		}
+		label := entry.Label
+		if label == "" {
+			label = entry.Slug
+		}
+		if len(label) > 128 {
+			label = entry.Slug
+		}
+		result = append(result, CanvasModel{ID: "codex-" + hex.EncodeToString(sum[:16]), Label: label, Provider: "codex-cli", InputModalities: inputs, OutputModalities: []string{"text"}, Available: true, MaxExecutionMS: 90000, MaxInputBytes: CanvasInputLimit, ParametersSchema: map[string]any{"type": "object", "properties": map[string]any{}, "additionalProperties": false}, Backend: entry.Slug})
+	}
+	verified := map[string]bool{}
+	for _, id := range strings.Split(os.Getenv("FOUNDRY_CANVAS_IMAGE_MODELS"), ",") {
+		id = strings.TrimSpace(id)
+		if id != "" {
+			verified[id] = true
+		}
+	}
+	textModels := append([]CanvasModel{}, result...)
+	for _, model := range textModels {
+		if !verified[model.Backend] {
+			continue
+		}
+		sum := sha256.Sum256([]byte("codex-cli/" + model.Backend + "/image"))
+		model.ID = "codex-" + hex.EncodeToString(sum[:16])
+		model.Label += " · Image"
+		model.OutputModalities = []string{"image"}
+		model.MaxExecutionMS = 180000
+		result = append(result, model)
+	}
+	sort.Slice(result, func(i, j int) bool { return result[i].ID < result[j].ID })
+	return result, nil
+}
+
+type CanvasClient struct{ client *http.Client }
+
+func NewCanvasClient(socket string) *CanvasClient {
+	if socket == "" {
+		return nil
+	}
+	transport := &http.Transport{DisableKeepAlives: true, DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+		return (&net.Dialer{}).DialContext(ctx, "unix", socket)
+	}}
+	return &CanvasClient{client: &http.Client{Transport: transport, Timeout: 182 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}}
+}
+func (c *CanvasClient) Models(ctx context.Context) ([]CanvasModel, error) {
+	if c == nil {
+		return nil, errors.New("codex_unavailable")
+	}
+	request, err := http.NewRequestWithContext(ctx, "GET", "http://codex/canvas/models", nil)
+	if err != nil {
+		return nil, err
+	}
+	response, err := c.client.Do(request)
+	if err != nil {
+		return nil, err
+	}
+	defer response.Body.Close()
+	if response.StatusCode != 200 {
+		return nil, errors.New("codex_unavailable")
+	}
+	raw, err := io.ReadAll(io.LimitReader(response.Body, 256<<10))
+	if err != nil {
+		return nil, err
+	}
+	var result struct {
+		Models []CanvasModel `json:"models"`
+	}
+	if err = json.Unmarshal(raw, &result); err != nil {
+		return nil, err
+	}
+	return result.Models, nil
+}
+
+type CanvasTask struct {
+	OutputModality string         `json:"output_modality,omitempty"`
+	Images         []CanvasImage  `json:"images,omitempty"`
+	ModelID        string         `json:"model_id"`
+	Prompt         string         `json:"prompt"`
+	Parameters     map[string]any `json:"parameters"`
+}
+
+func (c *CanvasClient) Generate(ctx context.Context, input CanvasTask) (string, error) {
+	output, err := c.GenerateOutput(ctx, input)
+	if err != nil {
+		return "", err
+	}
+	if output.Kind != "text" {
+		return "", errors.New("model_incompatible")
+	}
+	return output.Text, nil
+}

internal/codex/canvas_artifacts.go

@@ -0,0 +1,103 @@
+package codex
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+)
+
+const CanvasImageBytesLimit = 20 << 20
+const canvasArtifactResponseLimit = 32 << 20
+
+type CanvasImage struct {
+	MIME string `json:"mime_type"`
+	Data []byte `json:"data"`
+}
+type CanvasOutput struct {
+	Kind  string       `json:"kind"`
+	Text  string       `json:"text,omitempty"`
+	Image *CanvasImage `json:"image,omitempty"`
+}
+
+// Encoded image bytes exist only on the private Unix IPC channel. Public Canvas
+// jobs expose stable asset IDs/URLs and never serialize these bytes to browsers.
+func (c *CanvasClient) GenerateOutput(ctx context.Context, input CanvasTask) (CanvasOutput, error) {
+	if c == nil {
+		return CanvasOutput{}, errors.New("codex_unavailable")
+	}
+	if len(input.Prompt) > CanvasInputLimit || len(input.Images) > 4 {
+		return CanvasOutput{}, errors.New("context_too_large")
+	}
+	total := 0
+	for _, image := range input.Images {
+		total += len(image.Data)
+	}
+	if total > CanvasImageBytesLimit {
+		return CanvasOutput{}, errors.New("context_too_large")
+	}
+	raw, err := json.Marshal(input)
+	if err != nil {
+		return CanvasOutput{}, err
+	}
+	request, err := http.NewRequestWithContext(ctx, "POST", "http://codex/canvas/exec", bytes.NewReader(raw))
+	if err != nil {
+		return CanvasOutput{}, err
+	}
+	response, err := c.client.Do(request)
+	if err != nil {
+		return CanvasOutput{}, err
+	}
+	defer response.Body.Close()
+	raw, err = io.ReadAll(io.LimitReader(response.Body, canvasArtifactResponseLimit+1))
+	if err != nil || len(raw) > canvasArtifactResponseLimit {
+		return CanvasOutput{}, errors.New("generation_failed")
+	}
+	var result struct {
+		Result string        `json:"result"`
+		Output *CanvasOutput `json:"output"`
+		Error  string        `json:"error"`
+	}
+	if json.Unmarshal(raw, &result) != nil {
+		return CanvasOutput{}, errors.New("generation_failed")
+	}
+	if response.StatusCode != 200 {
+		switch result.Error {
+		case "codex_timeout":
+			return CanvasOutput{}, errors.New("generation_timeout")
+		case "codex_busy":
+			if response.StatusCode == 429 {
+				return CanvasOutput{}, errors.New("codex_busy")
+			}
+			return CanvasOutput{}, errors.New("generation_failed")
+		case "context_too_large":
+			return CanvasOutput{}, errors.New("context_too_large")
+		case "model_unavailable":
+			return CanvasOutput{}, errors.New("model_unavailable")
+		default:
+			return CanvasOutput{}, errors.New("generation_failed")
+		}
+	}
+	if result.Output == nil {
+		if input.OutputModality == "image" {
+			return CanvasOutput{}, errors.New("generation_failed")
+		}
+		return CanvasOutput{Kind: "text", Text: result.Result}, nil
+	}
+	output := *result.Output
+	switch output.Kind {
+	case "text":
+		if len(output.Text) > maxOutput || output.Image != nil || input.OutputModality == "image" {
+			return CanvasOutput{}, errors.New("generation_failed")
+		}
+	case "image":
+		if output.Image == nil || len(output.Image.Data) == 0 || len(output.Image.Data) > CanvasImageBytesLimit || input.OutputModality != "image" {
+			return CanvasOutput{}, errors.New("generation_failed")
+		}
+	default:
+		return CanvasOutput{}, errors.New("generation_failed")
+	}
+	return output, nil
+}

internal/codex/canvas_artifacts_test.go

@@ -0,0 +1,62 @@
+package codex
+
+import (
+	"encoding/base64"
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestCanvasArtifactDirectoryIsolation(t *testing.T) {
+	root := t.TempDir()
+	generated := filepath.Join(root, "generated_images")
+	if err := os.Mkdir(generated, 0700); err != nil {
+		t.Fatal(err)
+	}
+	// A minimal PNG header is sufficient for the worker's transport sniff. Core
+	// performs full decode/dimension validation before accepting it as an asset.
+	encoded, _ := base64.StdEncoding.DecodeString("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+j1WQAAAAASUVORK5CYII=")
+	if err := os.WriteFile(filepath.Join(root, "unrelated.png"), encoded, 0600); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := readCanvasGeneratedImage(generated); err == nil {
+		t.Fatal("unrelated image used as generated output")
+	}
+	if err := os.WriteFile(filepath.Join(generated, "result.png"), encoded, 0600); err != nil {
+		t.Fatal(err)
+	}
+	result, err := readCanvasGeneratedImage(generated)
+	if err != nil || result.MIME != "image/png" || len(result.Data) != len(encoded) {
+		t.Fatalf("generated output: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(generated, "second.png"), encoded, 0600); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := readCanvasGeneratedImage(generated); err == nil {
+		t.Fatal("multiple images silently dropped")
+	}
+	if err := os.Remove(filepath.Join(generated, "second.png")); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Remove(filepath.Join(generated, "result.png")); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.Symlink(filepath.Join(root, "unrelated.png"), filepath.Join(generated, "escape.png")); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := readCanvasGeneratedImage(generated); err == nil {
+		t.Fatal("symlink escaped job output directory")
+	}
+}
+func TestCanvasAgentCompletionRequiresSuccessfulTurn(t *testing.T) {
+	if _, ok := canvasAgentText([]byte("{\"type\":\"item.completed\",\"item\":{\"type\":\"agent_message\",\"text\":\"partial\"}}\n")); ok {
+		t.Fatal("partial generation accepted")
+	}
+	if _, ok := canvasAgentText([]byte("{\"type\":\"turn.completed\"}\n{\"type\":\"error\"}\n")); ok {
+		t.Fatal("failed generation accepted")
+	}
+	text, ok := canvasAgentText([]byte("{\"type\":\"item.completed\",\"item\":{\"type\":\"agent_message\",\"text\":\"done\"}}\n{\"type\":\"turn.completed\"}\n"))
+	if !ok || text != "done" {
+		t.Fatal("successful generation rejected")
+	}
+}

internal/codex/canvas_worker.go

@@ -0,0 +1,197 @@
+package codex
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/fs"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"syscall"
+	"time"
+)
+
+func executeCanvasTask(w http.ResponseWriter, r *http.Request, input CanvasTask, model CanvasModel) {
+	imageOutput := input.OutputModality == "image"
+	if input.OutputModality != "" && input.OutputModality != "text" && !imageOutput {
+		fail(w, 400, "invalid_ai_request")
+		return
+	}
+	if imageOutput != (len(model.OutputModalities) == 1 && model.OutputModalities[0] == "image") {
+		fail(w, 422, "model_incompatible")
+		return
+	}
+	duration := 90 * time.Second
+	if imageOutput {
+		duration = 180 * time.Second
+	}
+	ctx, cancel := context.WithTimeout(r.Context(), duration)
+	defer cancel()
+	job, err := os.MkdirTemp("", "foundry-canvas-")
+	if err != nil {
+		fail(w, 503, "codex_unavailable")
+		return
+	}
+	defer os.RemoveAll(job)
+	home := filepath.Join(job, "home")
+	if os.Mkdir(home, 0700) != nil || os.Symlink("/home/node/.codex/auth.json", filepath.Join(home, "auth.json")) != nil {
+		fail(w, 503, "codex_unavailable")
+		return
+	}
+	args := []string{"--ask-for-approval", "never", "--disable", "shell_tool", "--disable", "apps", "--disable", "plugins", "--disable", "multi_agent", "--disable", "hooks", "--disable", "memories", "-c", `web_search="disabled"`}
+	if !imageOutput {
+		args = append(args, "--disable", "image_generation")
+	}
+	args = append(args, "exec", "--json", "--sandbox", "read-only", "--skip-git-repo-check", "--ephemeral", "--ignore-user-config", "--ignore-rules", "--model", model.Backend)
+	total := 0
+	for i, image := range input.Images {
+		total += len(image.Data)
+		if total > CanvasImageBytesLimit || len(image.Data) == 0 || http.DetectContentType(image.Data) != image.MIME {
+			fail(w, 400, "invalid_ai_request")
+			return
+		}
+		extension := map[string]string{"image/png": "png", "image/jpeg": "jpeg", "image/webp": "webp", "image/gif": "gif"}[image.MIME]
+		if extension == "" {
+			fail(w, 400, "invalid_ai_request")
+			return
+		}
+		path := filepath.Join(job, fmt.Sprintf("input-%d.%s", i, extension))
+		if os.WriteFile(path, image.Data, 0600) != nil {
+			fail(w, 503, "codex_unavailable")
+			return
+		}
+		args = append(args, "--image", path)
+	}
+	args = append(args, "-")
+	prompt := input.Prompt
+	if imageOutput {
+		prompt = "Use the built-in image_generation tool to generate exactly one raster image following the supplied canvas instruction. Do not use shell, SVG, external URLs or substitute a textual description. Return a brief confirmation after generating the image.\n" + prompt
+	}
+	if len(prompt) > CanvasInputLimit {
+		fail(w, 413, "context_too_large")
+		return
+	}
+	cmd := exec.CommandContext(ctx, executable, args...)
+	cmd.Dir = "/work"
+	cmd.Env = []string{"PATH=/usr/local/bin:/usr/bin:/bin", "HOME=/home/node", "CODEX_HOME=" + home, "SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt"}
+	cmd.Stdin = strings.NewReader(prompt)
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error {
+		err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+		if errors.Is(err, syscall.ESRCH) {
+			return os.ErrProcessDone
+		}
+		return err
+	}
+	cmd.WaitDelay = 2 * time.Second
+	output := outputBuffer{limit: 2 << 20}
+	cmd.Stdout = &output
+	if err = cmd.Run(); err != nil {
+		if errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			fail(w, 504, "codex_timeout")
+		} else {
+			fail(w, 502, "generation_failed")
+		}
+		return
+	}
+	if output.exceeded {
+		fail(w, 502, "generation_failed")
+		return
+	}
+	text, completed := canvasAgentText(output.buffer.Bytes())
+	if !completed {
+		fail(w, 502, "generation_failed")
+		return
+	}
+	result := CanvasOutput{Kind: "text", Text: text}
+	if imageOutput {
+		image, err := readCanvasGeneratedImage(filepath.Join(home, "generated_images"))
+		if err != nil {
+			fail(w, 502, "generation_failed")
+			return
+		}
+		result = CanvasOutput{Kind: "image", Image: image}
+	} else if len(text) > maxOutput {
+		fail(w, 502, "generation_failed")
+		return
+	}
+	writeJSON(w, 200, map[string]any{"output": result})
+}
+func canvasAgentText(raw []byte) (string, bool) {
+	result := ""
+	completed := false
+	for _, line := range bytes.Split(bytes.TrimSpace(raw), []byte("\n")) {
+		var event struct {
+			Type string `json:"type"`
+			Item struct {
+				Type string `json:"type"`
+				Text string `json:"text"`
+			} `json:"item"`
+		}
+		if json.Unmarshal(line, &event) != nil {
+			return "", false
+		}
+		if event.Type == "turn.failed" || event.Type == "error" {
+			return "", false
+		}
+		if event.Type == "item.completed" && event.Item.Type == "agent_message" {
+			result = event.Item.Text
+		}
+		if event.Type == "turn.completed" {
+			completed = true
+		}
+	}
+	return result, completed
+}
+func readCanvasGeneratedImage(root string) (*CanvasImage, error) {
+	var result *CanvasImage
+	err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error {
+		if walkErr != nil {
+			return walkErr
+		}
+		if entry.Type()&os.ModeSymlink != 0 {
+			return errors.New("invalid_artifact")
+		}
+		if entry.IsDir() {
+			return nil
+		}
+		info, err := entry.Info()
+		if err != nil {
+			return err
+		}
+		if !info.Mode().IsRegular() {
+			return errors.New("invalid_artifact")
+		}
+		ext := strings.ToLower(filepath.Ext(path))
+		if ext != ".png" && ext != ".jpeg" && ext != ".jpg" && ext != ".webp" && ext != ".gif" {
+			return nil
+		}
+		if result != nil || info.Size() <= 0 || info.Size() > CanvasImageBytesLimit {
+			return errors.New("invalid_artifact")
+		}
+		data, err := os.ReadFile(path)
+		if err != nil {
+			return err
+		}
+		mime := http.DetectContentType(data)
+		switch mime {
+		case "image/png", "image/jpeg", "image/webp", "image/gif":
+		default:
+			return errors.New("invalid_artifact")
+		}
+		result = &CanvasImage{MIME: mime, Data: data}
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	if result == nil {
+		return nil, errors.New("image_missing")
+	}
+	return result, nil
+}

internal/codex/worker.go

@@ -22,10 +22,15 @@ var errOutputLimit = errors.New("Codex output limit exceeded")
 type outputBuffer struct {
 	buffer   bytes.Buffer
 	exceeded bool
+	limit    int
 }
 
 func (b *outputBuffer) Write(p []byte) (int, error) {
-	if len(p) > maxOutput-b.buffer.Len() {
+	limit := b.limit
+	if limit == 0 {
+		limit = maxOutput
+	}
+	if len(p) > limit-b.buffer.Len() {
 		b.exceeded = true
 		return 0, errOutputLimit
 	}
@@ -61,17 +66,63 @@ func ServeWorker(socket string) error {
 			execute(w, r, input.Prompt)
 		}
 	})
-	server := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: timeout + 15*time.Second}
+	mux.HandleFunc("GET /canvas/models", func(w http.ResponseWriter, r *http.Request) {
+		models, err := workerCanvasModels()
+		if err != nil {
+			fail(w, 503, "codex_unavailable")
+			return
+		}
+		writeJSON(w, 200, map[string]any{"models": models})
+	})
+	mux.HandleFunc("POST /canvas/exec", func(w http.ResponseWriter, r *http.Request) {
+		var input CanvasTask
+		decoder := json.NewDecoder(http.MaxBytesReader(w, r.Body, canvasArtifactResponseLimit+CanvasInputLimit))
+		decoder.DisallowUnknownFields()
+		if decoder.Decode(&input) != nil || decoder.Decode(new(any)) != io.EOF || len(input.Prompt) > CanvasInputLimit || strings.TrimSpace(input.Prompt) == "" || len(input.Parameters) > 0 || len(input.Images) > 4 {
+			fail(w, 400, "invalid_ai_request")
+			return
+		}
+		models, err := workerCanvasModels()
+		if err != nil {
+			fail(w, 503, "codex_unavailable")
+			return
+		}
+		var selected *CanvasModel
+		for _, model := range models {
+			if model.ID == input.ModelID {
+				copy := model
+				selected = &copy
+			}
+		}
+		if selected == nil {
+			fail(w, 422, "model_unavailable")
+			return
+		}
+		select {
+		case slot <- struct{}{}:
+			defer func() { <-slot }()
+		default:
+			w.Header().Set("Retry-After", "5")
+			fail(w, 429, "codex_busy")
+			return
+		}
+		executeCanvasTask(w, r, input, *selected)
+	})
+	server := &http.Server{Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 10 * time.Second, WriteTimeout: 195 * time.Second}
 	return server.Serve(listener)
 }
 
-func execute(w http.ResponseWriter, r *http.Request, prompt string) {
+func execute(w http.ResponseWriter, r *http.Request, prompt string) { executeModel(w, r, prompt, "") }
+func executeModel(w http.ResponseWriter, r *http.Request, prompt, model string) {
 	ctx, cancel := context.WithTimeout(r.Context(), timeout)
 	defer cancel()
 	cmd := exec.CommandContext(ctx, executable, "--ask-for-approval", "never",
 		"--disable", "shell_tool", "--disable", "apps", "--disable", "plugins",
 		"--disable", "multi_agent", "--disable", "hooks", "--disable", "memories",
 		"-c", `web_search="disabled"`, "exec", "--json", "--sandbox", "read-only", "--skip-git-repo-check", "--ephemeral", "-")
+	if model != "" {
+		cmd.Args = append(cmd.Args[:len(cmd.Args)-1], "--model", model, "-")
+	}
 	cmd.Dir = "/work"
 	cmd.Env = []string{"PATH=/usr/local/bin:/usr/bin:/bin", "HOME=/home/node", "CODEX_HOME=/home/node/.codex", "SSL_CERT_FILE=/etc/ssl/certs/ca-certificates.crt"}
 	cmd.Stdin = strings.NewReader(prompt)