Verify runtime AI coordination against real HTTP and database state

2026-11-11 15:13:14 +08:00
parent 28ca98a506
commit 6131fb5d7b
3 changed files with 160 additions and 7 deletions
@@ -71,7 +71,7 @@ jobs:
          PYSLACK
      - name: Fetch pinned headless HTTP integration client
        env:
-          HEADLESS_CLIENT_REVISION: e3a23351609557f0ac862b271b3f5d62f269a806
+          HEADLESS_CLIENT_REVISION: 465596fb7b175e837613fec394777c4615303ffa
        run: |
          set -eu
          mkdir -p .headless-client
@@ -8,7 +8,6 @@ import (
 	"encoding/base64"
 	"encoding/hex"
 	"encoding/json"
-	"net/http"
 	"net/http/httptest"
 	"os"
 	"os/exec"
@@ -38,7 +37,7 @@ func TestHeadlessClientHTTPIntegration(t *testing.T) {
 	_, _ = rand.Read(entropy[:])
 	project := "headless-" + hex.EncodeToString(entropy[:])
 	defer func() {
-		for _, table := range []string{"canvas_model_operations", "canvas_model_identity_map", "canvas_model_requests", "canvas_model_projects", "canvas_asset_bindings", "canvas_subgraphs", "workspaces", "project_accounts"} {
+		for _, table := range []string{"canvas_ai_runs", "canvas_model_operations", "canvas_model_identity_map", "canvas_model_requests", "canvas_model_projects", "canvas_asset_bindings", "canvas_assets", "canvas_subgraphs", "workspaces", "project_accounts"} {
 			_, _ = db.ExecContext(ctx, "DELETE FROM "+table+" WHERE project=$1", project)
 		}
 	}()
@@ -63,22 +62,21 @@ func TestHeadlessClientHTTPIntegration(t *testing.T) {
 	if err = db.QueryRowContext(ctx, `INSERT INTO materials(project,workspace_id,kind,title,created_by,canvas_document) VALUES($1,$2,'board','Root',$3,$4) RETURNING id::text`, project, workspace, account, legacy).Scan(&root); err != nil {
 		t.Fatal(err)
 	}
-	mux := http.NewServeMux()
-	Register(mux, db, map[string]string{project: "https://app.example.com/auth/callback"}, "https://api.example.com", nil)
+	mux := headlessWorkerRoutes(t, db, map[string]string{project: "https://app.example.com/auth/callback"}, project, account)

 	server := httptest.NewServer(mux)
 	defer server.Close()
 	commandContext, cancel := context.WithTimeout(ctx, 2*time.Minute)
 	defer cancel()
 	command := exec.CommandContext(commandContext, "node", filepath.Join(source, "tests/headless/foundry.integration.mjs"))
-	command.Env = append(os.Environ(), "FOUNDRY_HTTP_URL="+server.URL, "FOUNDRY_TEST_PROJECT="+project, "FOUNDRY_TEST_ACCOUNT="+account, "FOUNDRY_TEST_WORKSPACE="+workspace, "FOUNDRY_TEST_ROOT="+root, "FOUNDRY_TEST_TOKEN="+token)
+	command.Env = append(os.Environ(), "FOUNDRY_HTTP_URL="+server.URL, "FOUNDRY_TEST_PROJECT="+project, "FOUNDRY_TEST_ACCOUNT="+account, "FOUNDRY_TEST_WORKSPACE="+workspace, "FOUNDRY_TEST_ROOT="+root, "FOUNDRY_TEST_TOKEN="+token, "FOUNDRY_TEST_AI_WORKER=1")
 	output, err := command.CombinedOutput()
 	// The fixture token is ephemeral, but never print it even on client failures.
 	report := strings.ReplaceAll(string(output), token, "[fixture token redacted]")
 	if err != nil {
 		t.Fatalf("headless HTTP client failed: %v\n%s", err, report)
 	}
-	if !strings.Contains(report, `"status":"ok"`) {
+	if !strings.Contains(report, `"status":"ok"`) || !strings.Contains(report, `"runtimeAI":true`) {
 		t.Fatalf("headless HTTP client did not return success: %s", report)
 	}
 	t.Log(report)
@@ -0,0 +1,155 @@
+package platform
+
+import (
+	"bytes"
+	"database/sql"
+	"encoding/json"
+	"image"
+	"image/color"
+	"image/png"
+	"net"
+	"net/http"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"gitea.entroia.com/Zayden-Jung/foundry-core/internal/codex"
+)
+
+// Only the generation transport is deterministic. Admission, claims, events,
+// canonical writes and action receipts use the real service and isolated DB.
+func headlessWorkerRoutes(t *testing.T, db *sql.DB, callbacks map[string]string, project, account string) http.Handler {
+	t.Helper()
+	type gate struct {
+		entered, release  chan struct{}
+		started, released sync.Once
+	}
+	var mu sync.Mutex
+	var held *gate
+	var executions atomic.Int32
+	var picture bytes.Buffer
+	pixels := image.NewRGBA(image.Rect(0, 0, 3, 2))
+	pixels.Set(0, 0, color.RGBA{B: 255, A: 255})
+	if err := png.Encode(&picture, pixels); err != nil {
+		t.Fatal(err)
+	}
+	socket := filepath.Join(t.TempDir(), "worker.sock")
+	listener, err := net.Listen("unix", socket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	worker := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/canvas/models" {
+			reply(w, 200, response{"models": []any{
+				codex.CanvasModel{ID: "fixture-text", Label: "Fixture text", Available: true, InputModalities: []string{"text", "image"}, OutputModalities: []string{"text"}, MaxInputBytes: 1 << 20, MaxExecutionMS: 90000},
+				codex.CanvasModel{ID: "fixture-image", Label: "Fixture image", Available: true, InputModalities: []string{"text", "image"}, OutputModalities: []string{"image"}, MaxInputBytes: 1 << 20, MaxExecutionMS: 90000},
+			}})
+			return
+		}
+		if r.URL.Path != "/canvas/exec" {
+			reject(w, 404, "not_found")
+			return
+		}
+		var task codex.CanvasTask
+		if json.NewDecoder(r.Body).Decode(&task) != nil {
+			reject(w, 400, "invalid_request")
+			return
+		}
+		executions.Add(1)
+		mu.Lock()
+		current := held
+		mu.Unlock()
+		if current != nil {
+			current.started.Do(func() { close(current.entered) })
+			select {
+			case <-current.release:
+			case <-r.Context().Done():
+				return
+			}
+			mu.Lock()
+			if held == current {
+				held = nil
+			}
+			mu.Unlock()
+		}
+		if task.OutputModality == "image" {
+			reply(w, 200, response{"output": codex.CanvasOutput{Kind: "image", Image: &codex.CanvasImage{MIME: "image/png", Data: picture.Bytes()}}})
+		} else {
+			reply(w, 200, response{"result": "headless fixture output"})
+		}
+	})}
+	go worker.Serve(listener)
+	t.Cleanup(func() {
+		mu.Lock()
+		if held != nil {
+			held.released.Do(func() { close(held.release) })
+		}
+		mu.Unlock()
+		worker.Close()
+	})
+	s := &service{db: db, origin: "https://api.example.com", canvasAI: codex.NewCanvasClient(socket)}
+	mux := http.NewServeMux()
+	prefix := "/projects/{project}/workspaces/{workspace}/materials/{material}/canvas/ai"
+	mux.HandleFunc("GET "+prefix+"/models", s.protected(callbacks, s.canvasAIModels))
+	mux.HandleFunc("POST "+prefix+"/runs", s.protected(callbacks, s.createCanvasAIRun))
+	mux.HandleFunc("GET "+prefix+"/runs/{run}", s.protected(callbacks, s.getCanvasAIRun))
+	mux.HandleFunc("GET "+prefix+"/operations/{request}", s.protected(callbacks, s.getCanvasAIAdmission))
+	mux.HandleFunc("POST "+prefix+"/runs/{run}/cancel", s.protected(callbacks, s.cancelCanvasAIRun))
+	mux.HandleFunc("POST "+prefix+"/runs/{run}/apply", s.protected(callbacks, s.applyCanvasAIOutput))
+	// Private test controls are not registered by production Register().
+	mux.HandleFunc("POST "+prefix+"/fixture/{action}", s.protected(callbacks, func(w http.ResponseWriter, r *http.Request, p, a string) error {
+		if p != project || a != account {
+			reject(w, 403, "forbidden")
+			return nil
+		}
+		switch r.PathValue("action") {
+		case "arm":
+			mu.Lock()
+			held = &gate{entered: make(chan struct{}), release: make(chan struct{})}
+			mu.Unlock()
+		case "wait":
+			mu.Lock()
+			current := held
+			mu.Unlock()
+			if current == nil {
+				reject(w, 409, "worker_not_armed")
+				return nil
+			}
+			select {
+			case <-current.entered:
+			case <-r.Context().Done():
+				return nil
+			}
+		case "release":
+			mu.Lock()
+			if held != nil {
+				held.released.Do(func() { close(held.release) })
+			}
+			mu.Unlock()
+		case "execute":
+			run, err := s.claimCanvasRun(r.Context())
+			if err != nil || run == nil || run.Project != project || run.Account != account || run.Workspace != r.PathValue("workspace") {
+				reject(w, 409, "fixture_run_unavailable")
+				return nil
+			}
+			s.executeCanvasRun(r.Context(), *run)
+		case "stats":
+		default:
+			reject(w, 404, "not_found")
+			return nil
+		}
+		reply(w, 200, response{"executions": executions.Load()})
+		return nil
+	}))
+	regular := http.NewServeMux()
+	Register(regular, db, callbacks, "https://api.example.com", nil)
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if strings.Contains(r.URL.Path, "/canvas/ai/") {
+			mux.ServeHTTP(w, r)
+		} else {
+			regular.ServeHTTP(w, r)
+		}
+	})
+}