diff --git a/.gitea/workflows/check.yml b/.gitea/workflows/check.yml index 358f9d4..d59b1a9 100644 --- a/.gitea/workflows/check.yml +++ b/.gitea/workflows/check.yml @@ -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 diff --git a/internal/platform/canvas_headless_client_integration_test.go b/internal/platform/canvas_headless_client_integration_test.go index 8e707ea..de91788 100644 --- a/internal/platform/canvas_headless_client_integration_test.go +++ b/internal/platform/canvas_headless_client_integration_test.go @@ -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) diff --git a/internal/platform/canvas_headless_worker_integration_test.go b/internal/platform/canvas_headless_worker_integration_test.go new file mode 100644 index 0000000..6fe034b --- /dev/null +++ b/internal/platform/canvas_headless_worker_integration_test.go @@ -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) + } + }) +}