test: parallelize substrate e2e suite

.github/workflows/ci.yaml

@@ -48,6 +48,20 @@ jobs:
         uses: actions/checkout@v6
       - name: Initialize Environment
         uses: ./.github/actions/initialize-environment
+      - name: Set up Go
+        uses: actions/setup-go@v7
+        with:
+          go-version-file: go/go.mod
+          cache: false
+      - name: Cache E2E Go dependencies and build
+        uses: actions/cache@v6
+        with:
+          path: |
+            ~/go/pkg/mod
+            ~/.cache/go-build
+          key: e2e-go-${{ runner.os }}-${{ runner.arch }}-${{ hashFiles('go/go.mod', 'go/go.sum') }}
+          restore-keys: |
+            e2e-go-${{ runner.os }}-${{ runner.arch }}-
       - name: Allow unprivileged user namespaces
         # Ubuntu 24.04 (ubuntu-latest) enables AppArmor-based restrictions on
         # unprivileged user namespaces by default, which causes bubblewrap
@@ -110,18 +124,15 @@ jobs:
             --set controller.substrate.atenetRouterURL=http://atenet-router.ate-system.svc:80
             --set controller.substrate.defaultWorkerPool.name=kagent-default
             --set substrateWorkerPool.create=true
-            --set substrateWorkerPool.replicas=1
+            --set substrateWorkerPool.replicas=4
             --set-string substrateWorkerPool.workerImage=ghcr.io/kagent-dev/substrate/ateom-gvisor:v${{ env.SUBSTRATE_VERSION }}
-          DOCKER_BUILD_ARGS: >-
-            --cache-from=type=gha,scope=${{ needs.setup.outputs.cache-key }}-e2e
-            --cache-from=type=gha,scope=${{ env.CACHE_KEY_PREFIX }}-main-e2e
-            --cache-to=type=gha,scope=${{ needs.setup.outputs.cache-key }}-e2e,mode=max
-            --platform=linux/amd64
-            --push
         run: |
           echo "Cache key: ${{ needs.setup.outputs.cache-key }}"
-          jobs=$(nproc)
-          make -j"$(( jobs < 4 ? jobs : 4 ))" build-controller build-golang-adk build-claude-harness build-byo-a2a
+          printf '%s\n' controller golang-adk claude-harness byo-a2a | xargs -P4 -n1 bash -c '
+            image="$1"
+            DOCKER_BUILD_ARGS="--cache-from=type=gha,scope=${{ needs.setup.outputs.cache-key }}-e2e-${image} --cache-from=type=gha,scope=${{ env.CACHE_KEY_PREFIX }}-main-e2e-${image} --cache-to=type=gha,scope=${{ needs.setup.outputs.cache-key }}-e2e-${image},mode=max --platform=linux/amd64 --push" \
+              make GIT_COMMIT=e2e BUILD_DATE=1970-01-01 "build-${image}"
+          ' _
           make helm-install-provider
           kubectl rollout status deployment/kagent-controller -n kagent --timeout=120s
           kubectl wait --for=condition=Ready pod -l app.kubernetes.io/component=controller -n kagent --timeout=120s
@@ -150,7 +161,7 @@ jobs:
           echo "KAGENT_GRPC_URL: $KAGENT_GRPC_URL"
           make core/bin/kagent-local
           export KAGENT_E2E_CLI="$PWD/core/bin/kagent-local"
-          go test -v github.com/kagent-dev/kagent/go/core/test/e2e -failfast -shuffle=on
+          go test -v github.com/kagent-dev/kagent/go/core/test/e2e -failfast -shuffle=on -parallel 4
 
       - name: fail print info
         if: failure()

go/Makefile

@@ -85,7 +85,6 @@ govulncheck: ## Run govulncheck.
 
 .PHONY: core/bin/kagent-local
 core/bin/kagent-local:
-	go test -race ./core/cli/...
 	CGO_ENABLED=0 go build -ldflags "$(LDFLAGS)" -o core/bin/kagent-local ./core/cli/cmd/kagent
 
 core/bin/kagent-linux-amd64:
@@ -147,7 +146,7 @@ e2e: core/bin/kagent-local ## Run end-to-end tests.
 	  exit 1; \
 	}
 	@kind get kubeconfig --name $(KIND_CLUSTER_NAME) > /tmp/kind-config-e2e
-	KAGENT_E2E_CLI=$(CURDIR)/core/bin/kagent-local KUBECONFIG=/tmp/kind-config-e2e go test -v github.com/kagent-dev/kagent/go/core/test/e2e -failfast
+	KAGENT_E2E_CLI=$(CURDIR)/core/bin/kagent-local KUBECONFIG=/tmp/kind-config-e2e go test -v github.com/kagent-dev/kagent/go/core/test/e2e -failfast -parallel 2
 
 ##@ Dependencies
 

go/core/test/e2e/claude_interaction_test.go

@@ -32,6 +32,7 @@ const claudeE2EHarness = "claude-e2e"
 var claudeInteractionMocks embed.FS
 
 func TestE2EClaudeMockInteractionResumeAndPersistence(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	modelURL := reachableServerURL(t, startMockLLMServer(t, claudeInteractionMocks, "mocks/invoke_claude_agent.json"), "")
 	template := createClaudeMockTemplate(t, modelURL)
@@ -63,6 +64,7 @@ func TestE2EClaudeMockInteractionResumeAndPersistence(t *testing.T) {
 }
 
 func TestE2EClaudeMockCheckpointForkAndResume(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	modelURL := reachableServerURL(t, startMockLLMServer(t, claudeInteractionMocks, "mocks/invoke_claude_agent.json"), "")
 	template := createClaudeMockTemplate(t, modelURL)
@@ -140,6 +142,7 @@ func TestE2EClaudeMockCheckpointForkAndResume(t *testing.T) {
 }
 
 func TestE2EClaudeMockActiveTaskCancellation(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	modelURL, started := startBlockingClaudeMock(t)
 	template := createClaudeMockTemplate(t, modelURL)
@@ -148,6 +151,7 @@ func TestE2EClaudeMockActiveTaskCancellation(t *testing.T) {
 }
 
 func TestE2EClaudeMockBuiltinToolEvents(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	modelURL := reachableServerURL(t, startMockLLMServer(t, claudeInteractionMocks, "mocks/invoke_claude_builtin_tools.json"), "")
 	template := createClaudeMockTemplate(t, modelURL)
@@ -163,6 +167,7 @@ func TestE2EClaudeMockBuiltinToolEvents(t *testing.T) {
 }
 
 func TestE2EClaudeMockLocalSubagentRouting(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	modelURL := reachableServerURL(t, startMockLLMServer(t, claudeInteractionMocks, "mocks/invoke_claude_local_subagent.json"), "")
 	kube := interactionKubeClient(t)
@@ -500,6 +505,7 @@ func assertNoClaudeChildInstance(t *testing.T, fixture *interactionFixture, chil
 }
 
 func TestE2EClaudeMockWholeServerMCP(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	mcpURL, mcpMock := startMCPMock(t)
 

go/core/test/e2e/cli_catalog_lifecycle_test.go

@@ -16,9 +16,7 @@ import (
 )
 
 func TestE2ECLIAgentTemplateCatalogAndInstanceLifecycle(t *testing.T) {
-	if os.Getenv("KUBECONFIG") == "" {
-		t.Setenv("KUBECONFIG", clientcmd.RecommendedHomeFile)
-	}
+	t.Parallel()
 	target := interactionTarget(t)
 	templateName := createInteractionTemplate(t, startInteractionMock(t))
 	binary := kagentCLI(t)
@@ -93,6 +91,7 @@ func TestE2ECLIAgentTemplateCatalogAndInstanceLifecycle(t *testing.T) {
 }
 
 func TestE2ECLIAgentInstanceDiscoveryAndInvoke(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	fixture := newInteractionFixture(t, target, startInteractionMock(t))
 	binary := kagentCLI(t)
@@ -158,7 +157,11 @@ func TestE2ECLIAgentInstanceDiscoveryAndInvoke(t *testing.T) {
 func runKagentCLI(t *testing.T, ctx context.Context, binary string, args ...string) string {
 	t.Helper()
 	command := exec.CommandContext(ctx, binary, args...)
-	command.Env = append(os.Environ(), "HOME="+t.TempDir())
+	kubeconfig := os.Getenv(clientcmd.RecommendedConfigPathEnvVar)
+	if kubeconfig == "" {
+		kubeconfig = clientcmd.RecommendedHomeFile
+	}
+	command.Env = append(os.Environ(), "HOME="+t.TempDir(), clientcmd.RecommendedConfigPathEnvVar+"="+kubeconfig)
 	var stdout, stderr bytes.Buffer
 	command.Stdout = &stdout
 	command.Stderr = &stderr

go/core/test/e2e/interaction_test.go

@@ -50,6 +50,7 @@ var interactionMocks embed.FS
 // TestAgentInstanceInteraction verifies the complete public interaction path:
 // gateway routing, Substrate Actor transport, Go ADK execution, and the model call.
 func TestAgentInstanceInteraction(t *testing.T) {
+	t.Parallel()
 	fixture := newInteractionFixture(t, interactionTarget(t), startInteractionMock(t))
 	_, _, task := fixture.send(t, "What is 2+2?")
 	if task.Status.State != a2atype.TaskStateCompleted {
@@ -75,6 +76,7 @@ func TestOpaqueBYOAgentInteraction(t *testing.T) {
 }
 
 func TestAgentInstanceAskUserSurvivesSuspension(t *testing.T) {
+	t.Parallel()
 	fixture := newInteractionFixture(t, interactionTarget(t), startMockLLM(t, "mocks/invoke_golang_hitl_ask_user.json"))
 	fixture.ctx = metadata.AppendToOutgoingContext(fixture.ctx, strings.ToLower(a2atype.SvcParamExtensions), adka2a.HITLExtensionURI)
 	_, _, waiting := fixture.send(t, "Which database should we use for storage?")
@@ -101,6 +103,7 @@ func TestAgentInstanceAskUserSurvivesSuspension(t *testing.T) {
 }
 
 func TestAgentInstanceCheckpoint(t *testing.T) {
+	t.Parallel()
 	fixture := newInteractionFixture(t, interactionTarget(t), startInteractionMock(t))
 	_, _, task := fixture.send(t, "What is 2+2?")
 	created, err := fixture.checkpoints.CreateCheckpoint(fixture.ctx, &apiv1alpha1.CreateCheckpointRequest{
@@ -193,6 +196,7 @@ func TestAgentInstanceCheckpoint(t *testing.T) {
 }
 
 func TestMCPInteraction(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	mcpURL, mcpServer := startMCPMock(t)
 	template := createMCPInteractionTemplate(t, startMockLLM(t, "mocks/invoke_mcp_agent.json"), mcpURL)
@@ -227,6 +231,7 @@ func TestConfiguredBYOMCPInteraction(t *testing.T) {
 }
 
 func TestSharedAgentInteraction(t *testing.T) {
+	t.Parallel()
 	fixture := newSharedInteractionFixture(t, interactionTarget(t))
 	_, _, task := fixture.send(t, "Ask the specialist")
 	if task.Status.State != a2atype.TaskStateCompleted || !strings.Contains(taskText(task), "Answer from the shared specialist.") {
@@ -256,6 +261,7 @@ func TestSharedAgentInteraction(t *testing.T) {
 }
 
 func TestAgentInstanceTaskPersistenceAndIdempotency(t *testing.T) {
+	t.Parallel()
 	fixture := newInteractionFixture(t, interactionTarget(t), startInteractionMock(t))
 	message, request, task := fixture.send(t, "What is 2+2?")
 
@@ -316,6 +322,7 @@ func TestAgentInstanceTaskPersistenceAndIdempotency(t *testing.T) {
 }
 
 func TestAgentInstanceActiveTask(t *testing.T) {
+	t.Parallel()
 	target := interactionTarget(t)
 	modelURL, started := startBlockingInteractionMock(t)
 	fixture := newInteractionFixture(t, target, modelURL)

go/core/test/e2e/lifecycle_test.go

@@ -20,6 +20,7 @@ import (
 // Harness and AgentTemplate fixtures; this test owns only the AgentInstance it
 // creates through the public API.
 func TestAgentInstanceLifecycle(t *testing.T) {
+	t.Parallel()
 	target := os.Getenv("KAGENT_E2E_GRPC_TARGET")
 	if target == "" {
 		target = os.Getenv("KAGENT_GRPC_URL")

go/core/test/e2e/mcp_test.go

@@ -26,6 +26,7 @@ const (
 )
 
 func TestMCPAgentInstanceInteraction(t *testing.T) {
+	t.Parallel()
 	fixture := newInteractionFixture(t, interactionTarget(t), startInteractionMock(t))
 	endpoint := mcpEndpoint(t)
 
@@ -71,6 +72,7 @@ func TestMCPAgentInstanceInteraction(t *testing.T) {
 }
 
 func TestMCPAskUserContinuation(t *testing.T) {
+	t.Parallel()
 	fixture := newInteractionFixture(t, interactionTarget(t), startMockLLM(t, "mocks/invoke_golang_hitl_ask_user.json"))
 	endpoint := mcpEndpoint(t)
 	handle := mcpInvoke(t, endpoint, fixture.instanceID, "Which database should we use for storage?", true)["taskId"].(string)
@@ -95,6 +97,7 @@ func TestMCPAskUserContinuation(t *testing.T) {
 }
 
 func TestMCPCancelTask(t *testing.T) {
+	t.Parallel()
 	modelURL, started := startBlockingInteractionMock(t)
 	fixture := newInteractionFixture(t, interactionTarget(t), modelURL)
 	endpoint := mcpEndpoint(t)
@@ -109,6 +112,7 @@ func TestMCPCancelTask(t *testing.T) {
 }
 
 func TestMCPCheckpointFork(t *testing.T) {
+	t.Parallel()
 	fixture := newInteractionFixture(t, interactionTarget(t), startInteractionMock(t))
 	endpoint := mcpEndpoint(t)
 	if result := mcpInvoke(t, endpoint, fixture.instanceID, "What is 2+2?", false); result["resultType"] != "complete" {