fix(ci): restore upgrade tests from RC3 (v1.0.x backport)

go/core/test/upgrade/config_test.go

@@ -1,15 +1,62 @@
 package upgrade
 
 import (
+	"encoding/json"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 
 	"github.com/stretchr/testify/require"
 
+	configv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/config/v1"
+	"github.com/kagent-dev/kagent/go/api/v1alpha3"
 	kagentenv "github.com/kagent-dev/kagent/go/core/pkg/env"
 )
 
+func TestInvokeHarnesses(t *testing.T) {
+	manifest, err := os.ReadFile("../e2e/manifests/lifecycle.yaml.tmpl")
+	require.NoError(t, err)
+	for _, scenario := range []struct {
+		name       string
+		repository string
+	}{
+		{"Given a released controller, when preparing invocation fixtures, assert that they use its published runtimes", "ghcr.io/kagent-dev/kagent"},
+		{"Given an upgraded controller, when preparing invocation fixtures, assert that they use its locally built runtimes", "localhost:5001/kagent-dev/kagent"},
+	} {
+		t.Run(scenario.name, func(t *testing.T) {
+			images := &configv1.HarnessImagesConfig{
+				Kagent: scenario.repository + "/golang-adk@sha256:" + strings.Repeat("a", 64),
+				Claude: scenario.repository + "/claude-harness@sha256:" + strings.Repeat("b", 64),
+				Codex:  scenario.repository + "/codex-harness@sha256:" + strings.Repeat("c", 64),
+			}
+			encoded, err := invokeHarnesses(manifest, images)
+			require.NoError(t, err)
+			var fixtures v1alpha3.HarnessList
+			require.NoError(t, json.Unmarshal(encoded, &fixtures))
+			require.Equal(t, "List", fixtures.Kind)
+			require.Len(t, fixtures.Items, 4)
+			want := map[string]string{
+				"kagent": images.Kagent, "byo-adk-e2e": images.Kagent,
+				"claude-e2e": images.Claude, "codex-e2e": images.Codex,
+			}
+			for _, fixture := range fixtures.Items {
+				require.Equal(t, "Harness", fixture.Kind)
+				require.Contains(t, want, fixture.Name)
+				require.NotNil(t, fixture.Spec.Workload.Image)
+				require.Equal(t, want[fixture.Name], *fixture.Spec.Workload.Image)
+				require.Equal(t, "kagent-default", fixture.Spec.Substrate.WorkerPoolRef.Name)
+				delete(want, fixture.Name)
+			}
+			require.Empty(t, want)
+		})
+	}
+	t.Run("Given no runtime catalog, when preparing invocation fixtures, assert that setup reports the missing images", func(t *testing.T) {
+		_, err := invokeHarnesses(manifest, nil)
+		require.ErrorContains(t, err, "installed controller must provide kagent, claude, and codex runtime images")
+	})
+}
+
 func TestLoadUpgradeEnv(t *testing.T) {
 	repoRoot := t.TempDir()
 	require.NoError(t, os.WriteFile(filepath.Join(repoRoot, "Makefile"), nil, 0o600))

go/core/test/upgrade/invoke_e2e_test.go

@@ -1,8 +1,11 @@
 package upgrade
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
@@ -11,7 +14,12 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/require"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	kubeyaml "k8s.io/apimachinery/pkg/util/yaml"
+	"sigs.k8s.io/yaml"
 
+	configv1 "github.com/kagent-dev/kagent/go/api/gen/kagent/config/v1"
 	kagentenv "github.com/kagent-dev/kagent/go/core/pkg/env"
 )
 
@@ -52,6 +60,7 @@ func runInvokeE2E(t *testing.T, env upgradeEnv, treeGoDir, label string) {
 	t.Helper()
 
 	requireInvokeEnvironment(t, treeGoDir, label)
+	installInvokeHarnesses(t, env, treeGoDir)
 	port, stop := startPortForward(t, env, controllerServiceName, controllerAPIPort)
 	defer stop()
 
@@ -68,6 +77,60 @@ func runInvokeE2E(t *testing.T, env upgradeEnv, treeGoDir, label string) {
 		"[%s] selected invoke test did not run:\n%s", label, string(out))
 }
 
+// Reapply the matching release's fixtures at each invocation stage so upgrades
+// and application rollbacks exercise that installation's runtime images too.
+func installInvokeHarnesses(t *testing.T, env upgradeEnv, treeGoDir string) {
+	t.Helper()
+
+	manifest, err := os.ReadFile(filepath.Join(treeGoDir, "core/test/e2e/manifests/lifecycle.yaml.tmpl"))
+	require.NoError(t, err)
+	controllerYAML := kubectl(t, env, time.Minute, "get", "configmap", controllerServiceName,
+		"-n", env.namespace, "-o", `jsonpath={.data.controller\.yaml}`)
+	var config struct {
+		HarnessImages *configv1.HarnessImagesConfig `json:"harnessImages"`
+	}
+	require.NoError(t, yaml.Unmarshal([]byte(controllerYAML), &config))
+	fixtures, err := invokeHarnesses(manifest, config.HarnessImages)
+	require.NoError(t, err)
+	path := filepath.Join(t.TempDir(), "harnesses.json")
+	require.NoError(t, os.WriteFile(path, fixtures, 0o600))
+	kubectl(t, env, time.Minute, "apply", "-f", path)
+}
+
+func invokeHarnesses(manifest []byte, images *configv1.HarnessImagesConfig) ([]byte, error) {
+	if images.GetKagent() == "" || images.GetClaude() == "" || images.GetCodex() == "" {
+		return nil, fmt.Errorf("installed controller must provide kagent, claude, and codex runtime images")
+	}
+	rendered := strings.NewReplacer(
+		"${KAGENT_E2E_RUNTIME_IMAGE}", images.GetKagent(),
+		"${KAGENT_E2E_CLAUDE_IMAGE}", images.GetClaude(),
+		"${KAGENT_E2E_CODEX_IMAGE}", images.GetCodex(),
+	).Replace(string(manifest))
+	decoder := kubeyaml.NewYAMLOrJSONDecoder(bytes.NewBufferString(rendered), 4096)
+	fixtures := metav1.List{TypeMeta: metav1.TypeMeta{APIVersion: "v1", Kind: "List"}}
+	for {
+		var resource runtime.RawExtension
+		if err := decoder.Decode(&resource); err != nil {
+			if err == io.EOF {
+				break
+			}
+			return nil, fmt.Errorf("decode lifecycle fixture: %w", err)
+		}
+		var metadata metav1.PartialObjectMetadata
+		if err := json.Unmarshal(resource.Raw, &metadata); err != nil {
+			return nil, fmt.Errorf("decode lifecycle fixture metadata: %w", err)
+		}
+		if metadata.Kind != "Harness" {
+			continue
+		}
+		switch metadata.Name {
+		case "kagent", "codex-e2e", "claude-e2e", "byo-adk-e2e":
+			fixtures.Items = append(fixtures.Items, resource)
+		}
+	}
+	return json.Marshal(fixtures)
+}
+
 func upgradeE2EEnv(t *testing.T, env upgradeEnv, port int) []string {
 	t.Helper()
 

go/core/test/upgrade/roundtrip_test.go

@@ -39,12 +39,20 @@ const (
 	postgresRole      = "kagent_owner"
 )
 
-// pgTrackVersion returns the current applied migration version.
+// pgTrackVersion returns the current applied migration version, or zero when
+// the migration source is disabled and its tracking table does not exist.
 func pgTrackVersion(t *testing.T, env upgradeEnv, table string) int {
 	t.Helper()
 
+	// PostgreSQL resolves table references before evaluating CASE, so an absent
+	// optional tracking table must be checked in a separate statement.
+	exists := pgQuery(t, env, fmt.Sprintf("SELECT to_regclass(%s) IS NOT NULL", pgQuote(kagentSchema+"."+table)))
+	if exists == "f" {
+		return 0
+	}
+
 	raw := pgQuery(t, env, fmt.Sprintf(
-		"SELECT CASE WHEN to_regclass('%[1]s.%[2]s') IS NULL THEN 0 ELSE (SELECT COALESCE(MAX(version_id), 0) FROM %[1]s.%[2]s WHERE is_applied) END",
+		"SELECT COALESCE(MAX(version_id), 0) FROM %s.%s WHERE is_applied",
 		kagentSchema, table))
 	return parseInt(t, raw, table+" version")
 }

go/core/test/upgrade/upgrade_test.go

@@ -430,7 +430,9 @@ func pgMigrationState(t *testing.T, env upgradeEnv) postgresMigrationState {
 // pgMigrationStateE is the error-returning core of pgMigrationState, for use
 // inside require.Eventually conditions (see pgQueryE).
 func pgMigrationStateE(t *testing.T, env upgradeEnv) (postgresMigrationState, error) {
-	raw, err := pgQueryE(t, env, fmt.Sprintf("SELECT CASE WHEN to_regclass('%[1]s.schema_migrations') IS NULL THEN 0 ELSE (SELECT COALESCE(MAX(version_id), 0) FROM %[1]s.schema_migrations WHERE is_applied) END", kagentSchema))
+	// Core migrations are required. Before the tracking table exists, return the
+	// query error so readiness polling retries until the controller creates it.
+	raw, err := pgQueryE(t, env, fmt.Sprintf("SELECT COALESCE(MAX(version_id), 0) FROM %s.schema_migrations WHERE is_applied", kagentSchema))
 	if err != nil {
 		return postgresMigrationState{}, err
 	}