package service

import (
	"context"
	"encoding/json"
	"errors"
	"net"
	"net/http"
	"net/http/httptest"
	"net/url"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/alicebob/miniredis/v2"
	"github.com/prometheus/client_golang/prometheus"
	dto "github.com/prometheus/client_model/go"
	"github.com/sandbox0-ai/sandbox0/manager/pkg/apis/sandbox0/v1alpha1"
	"github.com/sandbox0-ai/sandbox0/manager/pkg/controller"
	"github.com/sandbox0-ai/sandbox0/manager/pkg/network"
	"github.com/sandbox0-ai/sandbox0/manager/pkg/startlimiter"
	"github.com/sandbox0-ai/sandbox0/pkg/ctldapi"
	"github.com/sandbox0-ai/sandbox0/pkg/dataplane"
	"github.com/sandbox0-ai/sandbox0/pkg/gateway/spec"
	"github.com/sandbox0-ai/sandbox0/pkg/internalauth"
	"github.com/sandbox0-ai/sandbox0/pkg/naming"
	obsmetrics "github.com/sandbox0-ai/sandbox0/pkg/observability/metrics"
	"github.com/sandbox0-ai/sandbox0/pkg/rediscache"
	"github.com/sandbox0-ai/sandbox0/pkg/sandboxprobe"
	"github.com/sandbox0-ai/sandbox0/pkg/volumeportal"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/kubernetes/fake"
	corelisters "k8s.io/client-go/listers/core/v1"
	k8stesting "k8s.io/client-go/testing"
	"k8s.io/client-go/tools/cache"
)

func TestClaimIdlePodRequiresPodReady(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	notReadyPod := newClaimTestPod("ns-a", "idle-not-ready", "template-a", false)

	client := fake.NewSimpleClientset(notReadyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, notReadyPod),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
	})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod != nil {
		t.Fatalf("claimIdlePod() = %s, want nil for running but not-ready pod", pod.Name)
	}
}

func TestClaimIdlePodClaimsReadyPod(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)

	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, readyPod),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
	})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod == nil {
		t.Fatal("claimIdlePod() = nil, want ready pod")
	}
	if pod.Name != "idle-ready" {
		t.Fatalf("claimIdlePod() selected %q, want %q", pod.Name, "idle-ready")
	}
	if got := pod.Labels[controller.LabelPoolType]; got != controller.PoolTypeActive {
		t.Fatalf("pool-type = %q, want %q", got, controller.PoolTypeActive)
	}
	if got := pod.Annotations[controller.AnnotationClusterAutoscalerSafeToEvict]; got != "false" {
		t.Fatalf("safe-to-evict annotation = %q, want false", got)
	}
}

func TestClaimIdlePodReservationPreventsStaleCacheReuse(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)
	reservations := newIdlePodReservations()

	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient:           client,
		podLister:           newClaimTestPodLister(t, readyPod),
		clock:               systemTime{},
		logger:              zap.NewNop(),
		idlePodReservations: reservations,
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
	})
	if err != nil {
		t.Fatalf("first claimIdlePod() error = %v", err)
	}
	if pod == nil {
		t.Fatal("first claimIdlePod() = nil, want ready pod")
	}

	pod, err = svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
	})
	if err != nil {
		t.Fatalf("second claimIdlePod() error = %v", err)
	}
	if pod != nil {
		t.Fatalf("second claimIdlePod() = %s, want nil while stale idle pod is reserved", pod.Name)
	}
}

func TestClaimIdlePodSkipsReservedPod(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	podA := newClaimTestPod("ns-a", "idle-a", "template-a", true)
	podB := newClaimTestPod("ns-a", "idle-b", "template-a", true)
	reservations := newIdlePodReservations()

	client := fake.NewSimpleClientset(podA.DeepCopy(), podB.DeepCopy())
	svc := &SandboxService{
		k8sClient:           client,
		podLister:           newClaimTestPodLister(t, podA, podB),
		clock:               systemTime{},
		logger:              zap.NewNop(),
		idlePodReservations: reservations,
	}
	if !svc.reserveIdlePod(podA) {
		t.Fatal("reserveIdlePod(podA) = false, want true")
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
	})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod == nil {
		t.Fatal("claimIdlePod() = nil, want unreserved pod")
	}
	if pod.Name != "idle-b" {
		t.Fatalf("claimIdlePod() selected %q, want idle-b", pod.Name)
	}
}

func TestClaimIdlePodAllowsClaimMounts(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)

	svc := &SandboxService{
		k8sClient: fake.NewSimpleClientset(readyPod.DeepCopy()),
		podLister: newClaimTestPodLister(t, readyPod),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
		Mounts: []ClaimMount{{
			SandboxVolumeID: "vol-1",
			MountPoint:      "/workspace/data",
		}},
	})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod == nil {
		t.Fatal("claimIdlePod() = nil, want ready pod")
	}
	if pod.Name != "idle-ready" {
		t.Fatalf("claimIdlePod() selected %q, want %q", pod.Name, "idle-ready")
	}
}

func TestClaimRequestDoesNotAcceptRuntimeMetadataFromJSON(t *testing.T) {
	var req ClaimRequest
	if err := json.Unmarshal([]byte(`{
		"template":"template-a",
		"metadata":{
			"owner_kind":"managed-agent"
		}
	}`), &req); err != nil {
		t.Fatalf("Unmarshal() error = %v", err)
	}
	if req.Metadata != nil {
		t.Fatalf("metadata = %#v, want nil for public JSON input", req.Metadata)
	}
}

func TestClaimRequestAcceptsSnapshotIDFromJSON(t *testing.T) {
	var req ClaimRequest
	if err := json.Unmarshal([]byte(`{
		"template":"template-a",
		"snapshot_id":"rootfs-snapshot-1"
	}`), &req); err != nil {
		t.Fatalf("Unmarshal() error = %v", err)
	}
	if req.SnapshotID != "rootfs-snapshot-1" {
		t.Fatalf("SnapshotID = %q, want rootfs-snapshot-1", req.SnapshotID)
	}
}

func TestClaimIdlePodAppliesRuntimeOwnerMetadata(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)

	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, readyPod),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
		Metadata: &ClaimMetadata{
			OwnerKind: "managed-agent",
		},
	})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	assertClaimOwnerMetadata(t, pod)
}

func TestClaimIdlePodUsesAbsoluteHardExpiresAtOverride(t *testing.T) {
	now := time.Date(2026, time.March, 7, 12, 0, 0, 0, time.UTC)
	hardExpiresAt := now.Add(30 * time.Minute)
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)

	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, readyPod),
		clock:     fixedClock{now: now},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{
		TeamID:        "team-a",
		UserID:        "user-a",
		Config:        &SandboxConfig{HardTTL: int32Ptr(3600)},
		HardExpiresAt: hardExpiresAt,
	})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod == nil {
		t.Fatal("claimIdlePod() = nil, want ready pod")
	}
	if got := pod.Annotations[controller.AnnotationHardExpiresAt]; got != hardExpiresAt.Format(time.RFC3339) {
		t.Fatalf("hard-expires-at annotation = %q, want %q", got, hardExpiresAt.Format(time.RFC3339))
	}

	var cfg SandboxConfig
	if err := json.Unmarshal([]byte(pod.Annotations[controller.AnnotationConfig]), &cfg); err != nil {
		t.Fatalf("unmarshal config: %v", err)
	}
	if cfg.HardTTL == nil || *cfg.HardTTL != 3600 {
		t.Fatalf("config hard_ttl = %v, want 3600", cfg.HardTTL)
	}
}

func TestClaimIdlePodRequiresCurrentTemplateHash(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)
	readyPod.Annotations[controller.AnnotationTemplateSpecHash] = "stale"

	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, readyPod),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod != nil {
		t.Fatalf("claimIdlePod() = %s, want nil for stale template hash", pod.Name)
	}
}

func TestClaimIdlePodSkipsDeletingPod(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)
	deletedAt := metav1.NewTime(time.Now().UTC())
	readyPod.DeletionTimestamp = &deletedAt

	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, readyPod),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod != nil {
		t.Fatalf("claimIdlePod() = %s, want nil for deleting pod", pod.Name)
	}
}

func TestClaimIdlePodFallsBackWhenPodStartsDeletingDuringClaim(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)

	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	client.PrependReactor("update", "pods", func(k8stesting.Action) (bool, runtime.Object, error) {
		return true, nil, &apierrors.StatusError{ErrStatus: metav1.Status{
			Reason:  metav1.StatusReasonInvalid,
			Message: `Pod "idle-ready" is invalid: metadata.finalizers: Forbidden: no new finalizers can be added if the object is being deleted, found new finalizers []string{"sandbox0.ai/sandbox-cleanup"}`,
		}}
	})
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, readyPod),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod != nil {
		t.Fatalf("claimIdlePod() = %s, want nil to trigger cold fallback", pod.Name)
	}
}

func TestClaimIdlePodFallsBackAfterRepeatedUpdateConflicts(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	idlePods := []*corev1.Pod{
		newClaimTestPod("ns-a", "idle-a", "template-a", true),
		newClaimTestPod("ns-a", "idle-b", "template-a", true),
		newClaimTestPod("ns-a", "idle-c", "template-a", true),
	}

	clientObjects := make([]runtime.Object, 0, len(idlePods))
	for _, pod := range idlePods {
		clientObjects = append(clientObjects, pod.DeepCopy())
	}
	client := fake.NewSimpleClientset(clientObjects...)
	updateConflicts := 0
	client.PrependReactor("update", "pods", func(k8stesting.Action) (bool, runtime.Object, error) {
		updateConflicts++
		return true, nil, &apierrors.StatusError{ErrStatus: metav1.Status{
			Reason:  metav1.StatusReasonConflict,
			Message: "pod was claimed by another manager",
		}}
	})
	svc := &SandboxService{
		k8sClient: client,
		podLister: newClaimTestPodLister(t, idlePods...),
		clock:     systemTime{},
		logger:    zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v, want nil fallback", err)
	}
	if pod != nil {
		t.Fatalf("claimIdlePod() = %s, want nil after repeated update conflicts", pod.Name)
	}
	if updateConflicts != claimIdlePodBackoff.Steps {
		t.Fatalf("update conflicts = %d, want %d", updateConflicts, claimIdlePodBackoff.Steps)
	}
}

func TestClaimIdlePodRequiresDataPlaneReadyNode(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)
	readyPod.Spec.NodeName = "node-a"
	readyPod.Spec.NodeSelector = dataplane.DataPlaneReadyNodeSelector()

	node := newClaimTestNode("node-a", "10.0.0.1")
	node.Labels = map[string]string{dataplane.NodeDataPlaneReadyLabel: dataplane.NotReadyLabelValue}
	client := fake.NewSimpleClientset(readyPod.DeepCopy(), node.DeepCopy())
	svc := &SandboxService{
		k8sClient:  client,
		podLister:  newClaimTestPodLister(t, readyPod),
		nodeLister: newClaimTestNodeLister(t, node),
		clock:      systemTime{},
		logger:     zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err != nil {
		t.Fatalf("claimIdlePod() error = %v", err)
	}
	if pod != nil {
		t.Fatalf("claimIdlePod() = %s, want nil for pod on data-plane-not-ready node", pod.Name)
	}
}

func TestEnsureDataPlaneReadyCapacityAllowsColdStartWithoutReadyNode(t *testing.T) {
	svc := &SandboxService{
		nodeLister: newClaimTestNodeLister(t),
		config: SandboxServiceConfig{
			AllowColdStartWithoutReadyDataPlane: true,
		},
	}
	spec := corev1.PodSpec{
		NodeSelector: dataplane.DataPlaneReadyNodeSelector(),
	}

	if err := svc.ensureDataPlaneReadyCapacity(spec); err != nil {
		t.Fatalf("ensureDataPlaneReadyCapacity() error = %v", err)
	}
}

func TestClaimIdlePodRequestsDeleteAfterNetworkApplyFailure(t *testing.T) {
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
	}
	readyPod := newClaimTestPod("ns-a", "idle-ready", "template-a", true)
	applyErr := errors.New("apply failed")
	removed := make([]string, 0, 1)
	client := fake.NewSimpleClientset(readyPod.DeepCopy())
	svc := &SandboxService{
		k8sClient:            client,
		podLister:            newClaimTestPodLister(t, readyPod),
		NetworkPolicyService: NewNetworkPolicyService(zap.NewNop()),
		networkProvider: &assertingNetworkProvider{
			applyErr: applyErr,
			removeFunc: func(namespace, sandboxID string) {
				removed = append(removed, namespace+"/"+sandboxID)
			},
		},
		clock:  systemTime{},
		logger: zap.NewNop(),
	}

	pod, err := svc.claimIdlePod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err == nil {
		t.Fatal("claimIdlePod() error = nil, want network apply failure")
	}
	if !errors.Is(err, applyErr) {
		t.Fatalf("claimIdlePod() error = %v, want wrapped apply failure", err)
	}
	if pod != nil {
		t.Fatalf("claimIdlePod() pod = %s, want nil on failure", pod.Name)
	}
	pods, err := client.CoreV1().Pods("ns-a").List(context.Background(), metav1.ListOptions{})
	if err != nil {
		t.Fatalf("list pods: %v", err)
	}
	if len(pods.Items) != 0 {
		t.Fatalf("pods after failed hot claim = %d, want 0", len(pods.Items))
	}
	if len(removed) != 0 {
		t.Fatalf("network policy removals = %d, want 0; lifecycle controller owns delete cleanup", len(removed))
	}
}

func TestWaitForPodClaimReadyUsesSandboxReadinessWithoutPodReady(t *testing.T) {
	pod := newClaimReadyTestPod("ns-a", "cold-pod", "template-a")
	indexer := newClaimTestPodIndexer(t, pod)

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.Method != http.MethodPost {
			t.Fatalf("method = %s, want POST", r.Method)
		}
		if r.URL.Path != "/api/v1/pods/ns-a/cold-pod/probes/readiness" {
			t.Fatalf("path = %s", r.URL.Path)
		}
		_ = json.NewEncoder(w).Encode(sandboxprobe.Passed(sandboxprobe.KindReadiness, "SandboxProbePassed", "sandbox probe passed", nil))
	}))
	defer server.Close()
	host, port := splitTestServerAddress(t, server)

	svc := &SandboxService{
		k8sClient:  fake.NewSimpleClientset(pod.DeepCopy(), newClaimTestNode("node-a", host)),
		podLister:  corelisters.NewPodLister(indexer),
		ctldClient: NewCtldClient(CtldClientConfig{Timeout: time.Second}),
		config: SandboxServiceConfig{
			CtldPort: port,
		},
		logger: zap.NewNop(),
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	readyPod, err := svc.waitForPodClaimReady(ctx, pod.Namespace, pod.Name)
	if err != nil {
		t.Fatalf("waitForPodClaimReady() error = %v", err)
	}
	if controller.IsPodReady(readyPod) {
		t.Fatal("waitForPodClaimReady() waited for Kubernetes PodReady; want sandbox claim readiness only")
	}
}

func TestWaitForPodNetworkIdentityWaitsForNodeNameAndPodIP(t *testing.T) {
	pod := newClaimTestPod("ns-a", "cold-pod", "template-a", false)
	indexer := newClaimTestPodIndexer(t, pod)
	svc := &SandboxService{
		podLister: corelisters.NewPodLister(indexer),
		logger:    zap.NewNop(),
	}
	handler := svc.PodEventHandler()

	go func() {
		time.Sleep(80 * time.Millisecond)
		nodeOnly := pod.DeepCopy()
		nodeOnly.Spec.NodeName = "node-a"
		if err := indexer.Update(nodeOnly); err != nil {
			t.Errorf("update pod: %v", err)
		}
		handler.UpdateFunc(pod, nodeOnly)

		time.Sleep(80 * time.Millisecond)
		updated := nodeOnly.DeepCopy()
		updated.Status.PodIP = "10.244.0.10"
		if err := indexer.Update(updated); err != nil {
			t.Errorf("update pod: %v", err)
		}
		handler.UpdateFunc(nodeOnly, updated)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	readyPod, err := svc.waitForPodNetworkIdentity(ctx, "template-a", pod.Namespace, pod.Name)
	if err != nil {
		t.Fatalf("waitForPodNetworkIdentity() error = %v", err)
	}
	if readyPod.Spec.NodeName != "node-a" {
		t.Fatalf("node name = %q, want node-a", readyPod.Spec.NodeName)
	}
	if readyPod.Status.PodIP != "10.244.0.10" {
		t.Fatalf("pod IP = %q, want 10.244.0.10", readyPod.Status.PodIP)
	}
}

func TestWaitForPodNetworkIdentityWaitsForNodeNameWhenPodIPArrivesFirst(t *testing.T) {
	pod := newClaimTestPod("ns-a", "cold-pod", "template-a", false)
	indexer := newClaimTestPodIndexer(t, pod)
	svc := &SandboxService{
		podLister: corelisters.NewPodLister(indexer),
		logger:    zap.NewNop(),
	}
	handler := svc.PodEventHandler()

	go func() {
		time.Sleep(80 * time.Millisecond)
		ipOnly := pod.DeepCopy()
		ipOnly.Status.PodIP = "10.244.0.10"
		if err := indexer.Update(ipOnly); err != nil {
			t.Errorf("update pod: %v", err)
		}
		handler.UpdateFunc(pod, ipOnly)

		time.Sleep(80 * time.Millisecond)
		updated := ipOnly.DeepCopy()
		updated.Spec.NodeName = "node-a"
		if err := indexer.Update(updated); err != nil {
			t.Errorf("update pod: %v", err)
		}
		handler.UpdateFunc(ipOnly, updated)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	readyPod, err := svc.waitForPodNetworkIdentity(ctx, "template-a", pod.Namespace, pod.Name)
	if err != nil {
		t.Fatalf("waitForPodNetworkIdentity() error = %v", err)
	}
	if readyPod.Spec.NodeName != "node-a" {
		t.Fatalf("node name = %q, want node-a", readyPod.Spec.NodeName)
	}
	if readyPod.Status.PodIP != "10.244.0.10" {
		t.Fatalf("pod IP = %q, want 10.244.0.10", readyPod.Status.PodIP)
	}
}

func TestWaitForPodNetworkIdentityWaitsForInformerWhenLivePodIsReady(t *testing.T) {
	cachedPod := newClaimTestPod("ns-a", "cold-pod", "template-a", false)
	livePod := cachedPod.DeepCopy()
	livePod.Spec.NodeName = "node-a"
	livePod.Status.PodIP = "10.244.0.10"
	indexer := newClaimTestPodIndexer(t, cachedPod)
	client := fake.NewSimpleClientset(livePod)
	svc := &SandboxService{
		k8sClient: client,
		podLister: corelisters.NewPodLister(indexer),
		logger:    zap.NewNop(),
	}
	handler := svc.PodEventHandler()

	go func() {
		time.Sleep(80 * time.Millisecond)
		if err := indexer.Update(livePod.DeepCopy()); err != nil {
			t.Errorf("update pod: %v", err)
		}
		handler.UpdateFunc(cachedPod, livePod)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	readyPod, err := svc.waitForPodNetworkIdentity(ctx, "template-a", cachedPod.Namespace, cachedPod.Name)
	if err != nil {
		t.Fatalf("waitForPodNetworkIdentity() error = %v", err)
	}
	if readyPod.Spec.NodeName != "node-a" {
		t.Fatalf("node name = %q, want node-a", readyPod.Spec.NodeName)
	}
	if readyPod.Status.PodIP != "10.244.0.10" {
		t.Fatalf("pod IP = %q, want 10.244.0.10", readyPod.Status.PodIP)
	}
	if actions := client.Actions(); len(actions) != 0 {
		t.Fatalf("unexpected live Kubernetes client actions while waiting for network identity: %#v", actions)
	}
}

func TestWaitForPodNetworkIdentityReturnsReadyFromCache(t *testing.T) {
	pod := newClaimTestPod("ns-a", "cold-pod", "template-a", false)
	pod.Spec.NodeName = "node-a"
	pod.Status.PodIP = "10.244.0.10"
	indexer := newClaimTestPodIndexer(t, pod)
	svc := &SandboxService{
		podLister: corelisters.NewPodLister(indexer),
		logger:    zap.NewNop(),
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	readyPod, err := svc.waitForPodNetworkIdentity(ctx, "template-a", pod.Namespace, pod.Name)
	if err != nil {
		t.Fatalf("waitForPodNetworkIdentity() error = %v", err)
	}
	if readyPod.Name != pod.Name {
		t.Fatalf("ready pod = %s, want %s", readyPod.Name, pod.Name)
	}
}

func TestWaitForPodNetworkIdentityReturnsOnDeleteOrTerminal(t *testing.T) {
	tests := []struct {
		name      string
		reason    string
		updatePod func(*testing.T, cache.ResourceEventHandlerFuncs, cache.Indexer, *corev1.Pod)
	}{
		{
			name:   "terminal",
			reason: "terminal",
			updatePod: func(t *testing.T, handler cache.ResourceEventHandlerFuncs, indexer cache.Indexer, pod *corev1.Pod) {
				updated := pod.DeepCopy()
				updated.Status.Phase = corev1.PodFailed
				if err := indexer.Update(updated); err != nil {
					t.Fatalf("update pod: %v", err)
				}
				handler.UpdateFunc(pod, updated)
			},
		},
		{
			name:   "deleted",
			reason: "deleting",
			updatePod: func(t *testing.T, handler cache.ResourceEventHandlerFuncs, indexer cache.Indexer, pod *corev1.Pod) {
				if err := indexer.Delete(pod); err != nil {
					t.Fatalf("delete pod: %v", err)
				}
				handler.DeleteFunc(pod)
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			pod := newClaimTestPod("ns-a", "cold-pod", "template-a", false)
			indexer := newClaimTestPodIndexer(t, pod)
			svc := &SandboxService{
				podLister: corelisters.NewPodLister(indexer),
				logger:    zap.NewNop(),
			}
			handler := svc.PodEventHandler()

			go func() {
				time.Sleep(80 * time.Millisecond)
				tt.updatePod(t, handler, indexer, pod)
			}()

			ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
			defer cancel()
			_, err := svc.waitForPodNetworkIdentity(ctx, "template-a", pod.Namespace, pod.Name)
			if err == nil {
				t.Fatal("waitForPodNetworkIdentity() error = nil, want error")
			}
			if !strings.Contains(err.Error(), tt.reason) {
				t.Fatalf("waitForPodNetworkIdentity() error = %v, want reason containing %q", err, tt.reason)
			}
		})
	}
}

func TestWaitForPodNetworkIdentityTimesOut(t *testing.T) {
	pod := newClaimTestPod("ns-a", "cold-pod", "template-a", false)
	indexer := newClaimTestPodIndexer(t, pod)
	svc := &SandboxService{
		podLister: corelisters.NewPodLister(indexer),
		logger:    zap.NewNop(),
	}

	ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
	defer cancel()
	_, err := svc.waitForPodNetworkIdentity(ctx, "template-a", pod.Namespace, pod.Name)
	if err == nil {
		t.Fatal("waitForPodNetworkIdentity() error = nil, want timeout")
	}
	if !strings.Contains(err.Error(), "network identity not ready") {
		t.Fatalf("waitForPodNetworkIdentity() error = %v, want network identity timeout", err)
	}
}

func TestWaitForPodNetworkIdentityUsesCacheRecheckForMissedEventRace(t *testing.T) {
	notReady := newClaimTestPod("ns-a", "cold-pod", "template-a", false)
	ready := notReady.DeepCopy()
	ready.Spec.NodeName = "node-a"
	ready.Status.PodIP = "10.244.0.10"
	lister := &sequencePodLister{pods: []*corev1.Pod{notReady, ready}}
	svc := &SandboxService{
		podLister: lister,
		logger:    zap.NewNop(),
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	readyPod, err := svc.waitForPodNetworkIdentity(ctx, "template-a", notReady.Namespace, notReady.Name)
	if err != nil {
		t.Fatalf("waitForPodNetworkIdentity() error = %v", err)
	}
	if readyPod.Status.PodIP != "10.244.0.10" {
		t.Fatalf("pod IP = %q, want 10.244.0.10", readyPod.Status.PodIP)
	}
	if got := lister.callCount(); got != 2 {
		t.Fatalf("pod lister Get calls = %d, want 2", got)
	}
}

func TestCreateNewPodDefersNetworkApplyUntilPodHasNetworkIdentity(t *testing.T) {
	withClaimTestPublicKey(t)

	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	applyCalls := 0
	client := fake.NewSimpleClientset()
	svc := &SandboxService{
		k8sClient:            client,
		secretLister:         newClaimTestSecretLister(t),
		NetworkPolicyService: NewNetworkPolicyService(zap.NewNop()),
		networkProvider: &assertingNetworkProvider{applyFunc: func(_ network.SandboxPolicyInput) {
			applyCalls++
		}},
		clock:  systemTime{},
		logger: zap.NewNop(),
	}

	pod, err := svc.createNewPod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err != nil {
		t.Fatalf("createNewPod() error = %v", err)
	}
	if applyCalls != 0 {
		t.Fatalf("network apply calls = %d, want 0 before pod has network identity", applyCalls)
	}
	if pod.Annotations[controller.AnnotationNetworkPolicy] == "" {
		t.Fatal("network policy annotation is empty")
	}
	if pod.Annotations[controller.AnnotationNetworkPolicyHash] == "" {
		t.Fatal("network policy hash annotation is empty")
	}

	pods, err := client.CoreV1().Pods("ns-a").List(context.Background(), metav1.ListOptions{})
	if err != nil {
		t.Fatalf("list pods: %v", err)
	}
	if len(pods.Items) != 1 {
		t.Fatalf("pods after createNewPod() = %d, want 1", len(pods.Items))
	}
}

func TestCreateNewPodAnnotatesClaimStartReservation(t *testing.T) {
	withClaimTestPublicKey(t)

	redisServer := miniredis.RunT(t)
	client := fake.NewSimpleClientset(&corev1.Node{
		ObjectMeta: metav1.ObjectMeta{Name: "sandbox-a"},
		Status: corev1.NodeStatus{Conditions: []corev1.NodeCondition{{
			Type:   corev1.NodeReady,
			Status: corev1.ConditionTrue,
		}}},
	})
	limiter, err := startlimiter.New(context.Background(), startlimiter.Config{
		ClusterID:      "cluster-a",
		K8sClient:      client,
		PerSandboxNode: 1,
		MaxLimit:       1,
		Redis: rediscache.Config{
			URL:       "redis://" + redisServer.Addr() + "/0",
			KeyPrefix: "test",
		},
	})
	if err != nil {
		t.Fatalf("create claim start limiter: %v", err)
	}
	reservation, _, err := limiter.Reserve(context.Background(), startlimiter.ReasonColdCreate, 1)
	if err != nil {
		t.Fatalf("Reserve() error = %v", err)
	}
	defer reservation.Release()

	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	svc := &SandboxService{
		k8sClient:            client,
		secretLister:         newClaimTestSecretLister(t),
		NetworkPolicyService: NewNetworkPolicyService(zap.NewNop()),
		clock:                systemTime{},
		logger:               zap.NewNop(),
	}

	pod, err := svc.createNewPodWithReservation(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"}, reservation)
	if err != nil {
		t.Fatalf("createNewPodWithReservation() error = %v", err)
	}
	if got := pod.Annotations[controller.AnnotationClaimStartReservation]; got != reservation.Token() {
		t.Fatalf("claim start reservation annotation = %q, want %q", got, reservation.Token())
	}
}

func TestClaimSandboxDeletesColdPodAfterNetworkApplyFailure(t *testing.T) {
	withClaimTestPublicKey(t)

	templateNamespace, err := naming.TemplateNamespaceForBuiltin("managed-agent-claude")
	if err != nil {
		t.Fatalf("template namespace: %v", err)
	}
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "managed-agent-claude",
			Namespace: templateNamespace,
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	applyErr := errors.New("apply failed")
	indexer := newClaimTestPodIndexer(t)
	client := fake.NewSimpleClientset()
	scheduleCreatedClaimPodInIndexer(t, client, indexer, nil)
	svc := &SandboxService{
		k8sClient:            client,
		podLister:            corelisters.NewPodLister(indexer),
		secretLister:         newClaimTestSecretLister(t),
		templateLister:       staticTemplateLister{templates: []*v1alpha1.SandboxTemplate{template}},
		NetworkPolicyService: NewNetworkPolicyService(zap.NewNop()),
		networkProvider:      &assertingNetworkProvider{applyErr: applyErr},
		clock:                systemTime{},
		logger:               zap.NewNop(),
	}

	_, err = svc.ClaimSandbox(context.Background(), &ClaimRequest{Template: "managed-agent-claude", TeamID: "team-a", UserID: "user-a"})
	if err == nil {
		t.Fatal("ClaimSandbox() error = nil, want network apply failure")
	}
	if !errors.Is(err, applyErr) {
		t.Fatalf("ClaimSandbox() error = %v, want wrapped apply failure", err)
	}

	pods, err := client.CoreV1().Pods(templateNamespace).List(context.Background(), metav1.ListOptions{})
	if err != nil {
		t.Fatalf("list pods: %v", err)
	}
	if len(pods.Items) != 0 {
		t.Fatalf("pods after failed cold claim = %d, want 0", len(pods.Items))
	}
}

func TestCreateNewPodMarksColdPodNonEvictable(t *testing.T) {
	withClaimTestPublicKey(t)

	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	client := fake.NewSimpleClientset()
	svc := &SandboxService{
		k8sClient:    client,
		secretLister: newClaimTestSecretLister(t),
		clock:        systemTime{},
		logger:       zap.NewNop(),
	}

	pod, err := svc.createNewPod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err != nil {
		t.Fatalf("createNewPod() error = %v", err)
	}
	if got := pod.Annotations[controller.AnnotationClusterAutoscalerSafeToEvict]; got != "false" {
		t.Fatalf("safe-to-evict annotation = %q, want false", got)
	}
}

func TestCreateNewPodPreMountsAllTemplateVolumePortals(t *testing.T) {
	withClaimTestPublicKey(t)

	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
			VolumeMounts: []v1alpha1.VolumeMountSpec{
				{Name: "data", MountPath: "/workspace/data"},
				{Name: "cache", MountPath: "/workspace/cache"},
			},
		},
	}
	client := fake.NewSimpleClientset()
	svc := &SandboxService{
		k8sClient:    client,
		secretLister: newClaimTestSecretLister(t),
		clock:        systemTime{},
		logger:       zap.NewNop(),
	}

	pod, err := svc.createNewPod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
		Mounts: []ClaimMount{{
			SandboxVolumeID: "vol-1",
			MountPoint:      "/workspace/data",
		}},
	})
	if err != nil {
		t.Fatalf("createNewPod() error = %v", err)
	}
	if volume := findClaimTestCSIVolumeByPortal(pod.Spec.Volumes, "data"); volume == nil {
		t.Fatalf("expected data csi volume, got %#v", pod.Spec.Volumes)
	}
	if volume := findClaimTestCSIVolumeByPortal(pod.Spec.Volumes, "cache"); volume == nil {
		t.Fatalf("expected unclaimed cache csi volume for rootfs-backed portal, got %#v", pod.Spec.Volumes)
	}
	if volume := findClaimTestCSIVolumeByPortal(pod.Spec.Volumes, volumeportal.WebhookStatePortalName); volume == nil {
		t.Fatalf("expected webhook state csi volume, got %#v", pod.Spec.Volumes)
	}
}

func TestCreateNewPodAppliesRuntimeOwnerMetadata(t *testing.T) {
	withClaimTestPublicKey(t)

	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	client := fake.NewSimpleClientset()
	svc := &SandboxService{
		k8sClient:    client,
		secretLister: newClaimTestSecretLister(t),
		clock:        systemTime{},
		logger:       zap.NewNop(),
	}

	pod, err := svc.createNewPod(context.Background(), template, &ClaimRequest{
		TeamID: "team-a",
		UserID: "user-a",
		Metadata: &ClaimMetadata{
			OwnerKind: "managed-agent",
		},
	})
	if err != nil {
		t.Fatalf("createNewPod() error = %v", err)
	}
	assertClaimOwnerMetadata(t, pod)
}

func TestCreateNewPodUsesAbsoluteHardExpiresAtOverride(t *testing.T) {
	withClaimTestPublicKey(t)

	now := time.Date(2026, time.March, 7, 12, 0, 0, 0, time.UTC)
	hardExpiresAt := now.Add(30 * time.Minute)
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	client := fake.NewSimpleClientset()
	svc := &SandboxService{
		k8sClient:    client,
		secretLister: newClaimTestSecretLister(t),
		clock:        fixedClock{now: now},
		logger:       zap.NewNop(),
	}

	pod, err := svc.createNewPod(context.Background(), template, &ClaimRequest{
		TeamID:        "team-a",
		UserID:        "user-a",
		Config:        &SandboxConfig{HardTTL: int32Ptr(3600)},
		HardExpiresAt: hardExpiresAt,
	})
	if err != nil {
		t.Fatalf("createNewPod() error = %v", err)
	}
	if got := pod.Annotations[controller.AnnotationHardExpiresAt]; got != hardExpiresAt.Format(time.RFC3339) {
		t.Fatalf("hard-expires-at annotation = %q, want %q", got, hardExpiresAt.Format(time.RFC3339))
	}

	var cfg SandboxConfig
	if err := json.Unmarshal([]byte(pod.Annotations[controller.AnnotationConfig]), &cfg); err != nil {
		t.Fatalf("unmarshal config: %v", err)
	}
	if cfg.HardTTL == nil || *cfg.HardTTL != 3600 {
		t.Fatalf("config hard_ttl = %v, want 3600", cfg.HardTTL)
	}
}

func TestCreateNewPodFailsBeforeCreateWhenDataPlaneNotReady(t *testing.T) {
	withClaimTestManagerConfig(t, `sandbox_pod_placement:
  node_selector:
    sandbox0.ai/data-plane-ready: "true"
`)
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "template-a",
			Namespace: "ns-a",
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	client := fake.NewSimpleClientset()
	svc := &SandboxService{
		k8sClient:  client,
		nodeLister: newClaimTestNodeLister(t),
		clock:      systemTime{},
		logger:     zap.NewNop(),
	}

	_, err := svc.createNewPod(context.Background(), template, &ClaimRequest{TeamID: "team-a", UserID: "user-a"})
	if err == nil {
		t.Fatal("createNewPod() error = nil, want data-plane-not-ready")
	}
	if !errors.Is(err, ErrDataPlaneNotReady) {
		t.Fatalf("createNewPod() error = %v, want ErrDataPlaneNotReady", err)
	}
	pods, err := client.CoreV1().Pods("ns-a").List(context.Background(), metav1.ListOptions{})
	if err != nil {
		t.Fatalf("list pods: %v", err)
	}
	if len(pods.Items) != 0 {
		t.Fatalf("pods after data-plane-not-ready cold claim = %d, want 0", len(pods.Items))
	}
}

func TestClaimSandboxInitializesRootFSFromSnapshotBeforeProcd(t *testing.T) {
	withClaimTestPublicKey(t)

	templateID := "template-a"
	templateNamespace, err := naming.TemplateNamespaceForBuiltin(templateID)
	if err != nil {
		t.Fatalf("template namespace: %v", err)
	}
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      templateID,
			Namespace: templateNamespace,
		},
	}

	var calls []string
	var applyReq ctldapi.ApplyRootFSRequest
	ctld := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/api/v1/rootfs/apply" {
			http.NotFound(w, r)
			return
		}
		if err := json.NewDecoder(r.Body).Decode(&applyReq); err != nil {
			t.Fatalf("decode apply request: %v", err)
		}
		calls = append(calls, "apply")
		_ = json.NewEncoder(w).Encode(ctldapi.ApplyRootFSResponse{Applied: true})
	}))
	defer ctld.Close()
	ctldURL, ctldPort := parsedTestServer(t, ctld.URL)

	procd := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/api/v1/initialize" {
			http.NotFound(w, r)
			return
		}
		if len(calls) != 1 || calls[0] != "apply" {
			t.Fatalf("calls before procd = %v, want [apply]", calls)
		}
		calls = append(calls, "procd")
		if err := spec.WriteSuccess(w, http.StatusOK, InitializeResponse{SandboxID: "sandbox-a", TeamID: "team-a"}); err != nil {
			t.Fatalf("write procd response: %v", err)
		}
	}))
	defer procd.Close()
	procdURL, procdPort := parsedTestServer(t, procd.URL)

	idlePod := newClaimTestPod(templateNamespace, "idle-ready", templateID, true)
	idlePod.Spec.NodeName = "node-a"
	idlePod.Status.HostIP = ctldURL.Hostname()
	idlePod.Status.PodIP = procdURL.Hostname()
	store := &memorySandboxStore{
		records: map[string]*SandboxRecord{},
		rootFSSnapshots: map[string]*RootFSSnapshot{
			"rootfs-snapshot-1": {
				ID:              "rootfs-snapshot-1",
				FilesystemID:    "source-fs",
				TeamID:          "team-a",
				SourceSandboxID: "source-sandbox",
				HeadLayerID:     "layer-v1",
				CreatedAt:       time.Now().UTC(),
			},
		},
	}
	client := fake.NewSimpleClientset(idlePod.DeepCopy())
	svc := &SandboxService{
		k8sClient:              client,
		podLister:              newClaimTestPodLister(t, idlePod),
		secretLister:           newClaimTestSecretLister(t),
		templateLister:         staticTemplateLister{templates: []*v1alpha1.SandboxTemplate{template}},
		sandboxStore:           store,
		ctldClient:             NewCtldClient(CtldClientConfig{Timeout: time.Second}),
		procdClient:            NewProcdClient(ProcdClientConfig{Timeout: time.Second}),
		internalTokenGenerator: staticTokenGenerator{},
		config: SandboxServiceConfig{
			CtldEnabled:      true,
			CtldPort:         ctldPort,
			ProcdPort:        procdPort,
			ProcdInitTimeout: time.Second,
		},
		clock:  systemTime{},
		logger: zap.NewNop(),
	}

	resp, err := svc.ClaimSandbox(context.Background(), &ClaimRequest{
		Template:   templateID,
		SnapshotID: "rootfs-snapshot-1",
		TeamID:     "team-a",
		UserID:     "user-a",
	})
	if err != nil {
		t.Fatalf("ClaimSandbox() error = %v", err)
	}
	if resp == nil || resp.SandboxID == "" {
		t.Fatalf("ClaimSandbox() response = %+v, want sandbox id", resp)
	}
	if len(calls) != 2 || calls[0] != "apply" || calls[1] != "procd" {
		t.Fatalf("calls = %v, want [apply procd]", calls)
	}
	if applyReq.Target.PodName != "idle-ready" {
		t.Fatalf("apply target pod = %q, want idle-ready", applyReq.Target.PodName)
	}
	if applyReq.BaselineLayerID != "layer-v1" {
		t.Fatalf("BaselineLayerID = %q, want layer-v1", applyReq.BaselineLayerID)
	}
	if len(applyReq.Layers) != 1 || applyReq.Layers[0].LayerID != "layer-v1" {
		t.Fatalf("apply layers = %+v, want layer-v1", applyReq.Layers)
	}
	state := store.rootFSStates[resp.SandboxID]
	if state == nil || state.LayerID != "layer-v1" {
		t.Fatalf("rootfs state = %+v, want layer-v1 for claimed sandbox", state)
	}
	record := store.records[resp.SandboxID]
	if record == nil || record.Status != SandboxStatusRunning {
		t.Fatalf("record = %+v, want running claimed sandbox", record)
	}
}

func TestClaimSandboxCleansColdPodWhenClaimReadinessFails(t *testing.T) {
	withClaimTestPublicKey(t)
	templateNamespace, err := naming.TemplateNamespaceForBuiltin("managed-agent-claude")
	if err != nil {
		t.Fatalf("template namespace: %v", err)
	}
	template := &v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "managed-agent-claude",
			Namespace: templateNamespace,
		},
		Spec: v1alpha1.SandboxTemplateSpec{
			MainContainer: v1alpha1.ContainerSpec{Image: "busybox"},
		},
	}
	indexer := newClaimTestPodIndexer(t)
	client := fake.NewSimpleClientset()
	scheduleCreatedClaimPodInIndexer(t, client, indexer, func(pod *corev1.Pod) {
		pod.Status.ContainerStatuses = nil
	})
	ctx, cancel := context.WithCancel(context.Background())
	svc := &SandboxService{
		k8sClient:            client,
		podLister:            corelisters.NewPodLister(indexer),
		secretLister:         newClaimTestSecretLister(t),
		templateLister:       staticTemplateLister{templates: []*v1alpha1.SandboxTemplate{template}},
		NetworkPolicyService: NewNetworkPolicyService(zap.NewNop()),
		networkProvider: &assertingNetworkProvider{applyFunc: func(_ network.SandboxPolicyInput) {
			cancel()
		}},
		clock:  systemTime{},
		logger: zap.NewNop(),
	}

	_, err = svc.ClaimSandbox(ctx, &ClaimRequest{Template: "managed-agent-claude", TeamID: "team-a", UserID: "user-a"})
	if err == nil {
		t.Fatal("ClaimSandbox() error = nil, want claim readiness failure")
	}

	pods, err := client.CoreV1().Pods(templateNamespace).List(context.Background(), metav1.ListOptions{})
	if err != nil {
		t.Fatalf("list pods: %v", err)
	}
	if len(pods.Items) != 0 {
		t.Fatalf("pods after failed cold claim = %d, want 0", len(pods.Items))
	}
}

func TestObserveClaimPhaseRecordsMetric(t *testing.T) {
	registry := prometheus.NewRegistry()
	svc := &SandboxService{metrics: obsmetrics.NewManager(registry)}

	svc.observeClaimPhase("managed-agents", "cold", "wait_for_pod_claim_ready", time.Now().Add(-20*time.Millisecond), nil)

	families, err := registry.Gather()
	if err != nil {
		t.Fatalf("gather metrics: %v", err)
	}
	metric := findMetric(families, "manager_sandbox_claim_phase_duration_seconds", map[string]string{
		"template": "managed-agents",
		"type":     "cold",
		"phase":    "wait_for_pod_claim_ready",
		"status":   "success",
	})
	if metric == nil || metric.GetHistogram() == nil {
		t.Fatal("claim phase histogram metric not found")
	}
	if got := metric.GetHistogram().GetSampleCount(); got != 1 {
		t.Fatalf("claim phase sample count = %d, want 1", got)
	}
	if got := metric.GetHistogram().GetSampleSum(); got <= 0 {
		t.Fatalf("claim phase sample sum = %f, want > 0", got)
	}
}

func TestObservePodNetworkIdentityStageRecordsMetric(t *testing.T) {
	registry := prometheus.NewRegistry()
	svc := &SandboxService{metrics: obsmetrics.NewManager(registry)}

	svc.observePodNetworkIdentityStage("managed-agents", "pod_ip_assigned", "timeout", "pod IP is not assigned", time.Now().Add(-20*time.Millisecond))

	families, err := registry.Gather()
	if err != nil {
		t.Fatalf("gather metrics: %v", err)
	}
	metric := findMetric(families, "manager_pod_network_identity_stage_duration_seconds", map[string]string{
		"template": "managed-agents",
		"stage":    "pod_ip_assigned",
		"status":   "timeout",
		"reason":   "ip_unassigned",
	})
	if metric == nil || metric.GetHistogram() == nil {
		t.Fatal("pod network identity stage histogram metric not found")
	}
	if got := metric.GetHistogram().GetSampleCount(); got != 1 {
		t.Fatalf("pod network identity stage sample count = %d, want 1", got)
	}
	if got := metric.GetHistogram().GetSampleSum(); got <= 0 {
		t.Fatalf("pod network identity stage sample sum = %f, want > 0", got)
	}
}

func TestObserveIdleClaimRecordsMetric(t *testing.T) {
	registry := prometheus.NewRegistry()
	svc := &SandboxService{metrics: obsmetrics.NewManager(registry)}

	svc.observeIdleClaim("managed-agents", "success")

	families, err := registry.Gather()
	if err != nil {
		t.Fatalf("gather metrics: %v", err)
	}
	metric := findMetric(families, "manager_sandbox_idle_claims_total", map[string]string{
		"template": "managed-agents",
		"result":   "success",
	})
	if metric == nil || metric.GetCounter() == nil {
		t.Fatal("idle claim counter metric not found")
	}
	if got := metric.GetCounter().GetValue(); got != 1 {
		t.Fatalf("idle claim counter = %f, want 1", got)
	}
}

func TestWaitForPodClaimReadyWaitsForProcdContainerRunning(t *testing.T) {
	pod := newClaimReadyTestPod("ns-a", "cold-pod", "template-a")
	pod.Status.ContainerStatuses = nil
	indexer := newClaimTestPodIndexer(t, pod)

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		_ = json.NewEncoder(w).Encode(sandboxprobe.Passed(sandboxprobe.KindReadiness, "SandboxProbePassed", "sandbox probe passed", nil))
	}))
	defer server.Close()
	host, port := splitTestServerAddress(t, server)

	svc := &SandboxService{
		k8sClient:  fake.NewSimpleClientset(pod.DeepCopy(), newClaimTestNode("node-a", host)),
		podLister:  corelisters.NewPodLister(indexer),
		ctldClient: NewCtldClient(CtldClientConfig{Timeout: time.Second}),
		config: SandboxServiceConfig{
			CtldPort: port,
		},
		logger: zap.NewNop(),
	}

	handler := svc.PodEventHandler()
	go func() {
		time.Sleep(80 * time.Millisecond)
		updated := pod.DeepCopy()
		updated.Status.ContainerStatuses = []corev1.ContainerStatus{{
			Name:  "procd",
			Ready: true,
			State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{StartedAt: metav1.Now()}},
		}}
		if err := indexer.Update(updated); err != nil {
			t.Errorf("update pod: %v", err)
			return
		}
		handler.UpdateFunc(pod, updated)
	}()

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	readyPod, err := svc.waitForPodClaimReady(ctx, pod.Namespace, pod.Name)
	if err != nil {
		t.Fatalf("waitForPodClaimReady() error = %v", err)
	}
	if !podContainerRunning(readyPod, "procd") {
		t.Fatal("waitForPodClaimReady() returned before procd container was running")
	}
}

func TestWaitForPodClaimReadyReprobesOnlyAfterInformerEvent(t *testing.T) {
	pod := newClaimReadyTestPod("ns-a", "cold-pod", "template-a")
	indexer := newClaimTestPodIndexer(t, pod)

	probeCalls := make(chan struct{}, 8)
	var probeMu sync.Mutex
	probeReady := false
	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
		probeMu.Lock()
		ready := probeReady
		probeMu.Unlock()

		probeCalls <- struct{}{}
		if ready {
			_ = json.NewEncoder(w).Encode(sandboxprobe.Passed(sandboxprobe.KindReadiness, "SandboxProbePassed", "sandbox probe passed", nil))
			return
		}
		_ = json.NewEncoder(w).Encode(sandboxprobe.Failed(sandboxprobe.KindReadiness, "SandboxProbePending", "sandbox probe pending", nil))
	}))
	defer server.Close()
	host, port := splitTestServerAddress(t, server)

	svc := &SandboxService{
		k8sClient:  fake.NewSimpleClientset(pod.DeepCopy(), newClaimTestNode("node-a", host)),
		podLister:  corelisters.NewPodLister(indexer),
		ctldClient: NewCtldClient(CtldClientConfig{Timeout: time.Second}),
		config: SandboxServiceConfig{
			CtldPort: port,
		},
		logger: zap.NewNop(),
	}

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	resultCh := make(chan error, 1)
	go func() {
		_, err := svc.waitForPodClaimReady(ctx, pod.Namespace, pod.Name)
		resultCh <- err
	}()

	for i := 0; i < 2; i++ {
		select {
		case <-probeCalls:
		case <-time.After(time.Second):
			t.Fatalf("probe call %d did not happen", i+1)
		}
	}

	select {
	case <-probeCalls:
		t.Fatal("unexpected readiness probe before informer event")
	case err := <-resultCh:
		t.Fatalf("waitForPodClaimReady returned before informer event: %v", err)
	case <-time.After(150 * time.Millisecond):
	}

	probeMu.Lock()
	probeReady = true
	probeMu.Unlock()

	updated := pod.DeepCopy()
	updated.ResourceVersion = "2"
	if err := indexer.Update(updated); err != nil {
		t.Fatalf("update pod: %v", err)
	}
	svc.PodEventHandler().UpdateFunc(pod, updated)

	select {
	case err := <-resultCh:
		if err != nil {
			t.Fatalf("waitForPodClaimReady() error = %v", err)
		}
	case <-time.After(time.Second):
		t.Fatal("waitForPodClaimReady did not return after informer event")
	}
}

func TestProbeSandboxPodReadinessRequiresPublishedVolumePortals(t *testing.T) {
	pod := newClaimReadyTestPod("ns-a", "pod-a", "template-a")
	pod.UID = types.UID("pod-uid")
	pod.Spec.Volumes = []corev1.Volume{{
		Name: "workspace",
		VolumeSource: corev1.VolumeSource{
			CSI: &corev1.CSIVolumeSource{
				Driver: volumeportal.DriverName,
				VolumeAttributes: map[string]string{
					volumeportal.AttributePortalName: "workspace",
					volumeportal.AttributeMountPath:  "/workspace",
				},
			},
		},
	}}

	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		w.Header().Set("Content-Type", "application/json")
		switch r.URL.Path {
		case "/api/v1/pods/ns-a/pod-a/probes/readiness":
			_ = json.NewEncoder(w).Encode(sandboxprobe.Passed(sandboxprobe.KindReadiness, "SandboxProbePassed", "sandbox probe passed", nil))
		case "/api/v1/volume-portals/check":
			var req ctldapi.CheckVolumePortalsRequest
			if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
				t.Fatalf("decode check request: %v", err)
			}
			if req.PodUID != "pod-uid" {
				t.Fatalf("check pod UID = %q, want pod-uid", req.PodUID)
			}
			if len(req.Portals) != 1 || req.Portals[0].PortalName != "workspace" {
				t.Fatalf("check portals = %+v, want workspace", req.Portals)
			}
			_ = json.NewEncoder(w).Encode(ctldapi.CheckVolumePortalsResponse{
				Ready:   false,
				Missing: []string{"workspace"},
			})
		default:
			http.NotFound(w, r)
		}
	}))
	defer server.Close()
	host, port := splitTestServerAddress(t, server)
	svc := &SandboxService{
		k8sClient:  fake.NewSimpleClientset(newClaimTestNode("node-a", host)),
		ctldClient: NewCtldClient(CtldClientConfig{}),
		config:     SandboxServiceConfig{CtldPort: port},
	}

	result, err := svc.ProbeSandboxPod(context.Background(), pod, sandboxprobe.KindReadiness)
	if err != nil {
		t.Fatalf("ProbeSandboxPod() error = %v", err)
	}
	if result == nil {
		t.Fatal("ProbeSandboxPod() result = nil")
	}
	if result.Status != sandboxprobe.StatusFailed {
		t.Fatalf("ProbeSandboxPod() status = %q, want failed", result.Status)
	}
	if result.Reason != "VolumePortalsNotReady" {
		t.Fatalf("ProbeSandboxPod() reason = %q, want VolumePortalsNotReady", result.Reason)
	}
}

func TestValidateClaimMountsRejectsDuplicateVolume(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{
			{SandboxVolumeID: "vol-1", MountPoint: "/workspace/a"},
			{SandboxVolumeID: "vol-1", MountPoint: "/workspace/b"},
		},
	}

	err := validateClaimMounts(req)
	if err == nil {
		t.Fatal("expected duplicate volume validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}
}

func TestNormalizeSandboxConfigRejectsTTLGreaterThanHardTTL(t *testing.T) {
	cfg := &SandboxConfig{
		TTL:     int32Ptr(1000),
		HardTTL: int32Ptr(10),
	}

	err := normalizeSandboxConfigForPersistence(cfg)
	if err == nil {
		t.Fatal("expected lifecycle validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}
}

func TestValidateClaimMountsRejectsDuplicateMountPoint(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{
			{SandboxVolumeID: "vol-1", MountPoint: "/workspace/data"},
			{SandboxVolumeID: "vol-2", MountPoint: "/workspace/data"},
		},
	}

	err := validateClaimMounts(req)
	if err == nil {
		t.Fatal("expected duplicate mount point validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}
}

func TestValidateClaimMountsNormalizesMountPoint(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{{SandboxVolumeID: "vol-1", MountPoint: "/workspace/project/../data"}},
	}

	if err := validateClaimMounts(req); err != nil {
		t.Fatalf("validateClaimMounts() error = %v", err)
	}
	if got := req.Mounts[0].MountPoint; got != "/workspace/data" {
		t.Fatalf("mount point = %q, want %q", got, "/workspace/data")
	}
}

func TestMountsAnnotationRoundTrip(t *testing.T) {
	annotations := map[string]string{}
	mounts := []ClaimMount{{SandboxVolumeID: "vol-1", MountPoint: "/workspace/project/../data"}}
	normalized, err := normalizeClaimMounts(mounts)
	if err != nil {
		t.Fatalf("normalizeClaimMounts() error = %v", err)
	}
	if err := setMountsAnnotation(annotations, normalized); err != nil {
		t.Fatalf("setMountsAnnotation() error = %v", err)
	}
	got := parseClaimMounts(annotations[controller.AnnotationMounts])
	if len(got) != 1 {
		t.Fatalf("parseClaimMounts() returned %d mounts, want 1", len(got))
	}
	if got[0].SandboxVolumeID != "vol-1" || got[0].MountPoint != "/workspace/data" {
		t.Fatalf("mount = %+v, want vol-1 mounted at /workspace/data", got[0])
	}
}

func TestValidateClaimMountsRejectsWebhookStatePath(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{{SandboxVolumeID: "vol-1", MountPoint: webhookStateMountPoint + "/custom"}},
	}

	err := validateClaimMounts(req)
	if err == nil {
		t.Fatal("expected reserved mount point validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}
}

func TestValidateClaimMountsForTemplateRequiresDeclaredMountPoint(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{{SandboxVolumeID: "vol-1", MountPoint: "/workspace/data"}},
	}
	template := &v1alpha1.SandboxTemplate{
		Spec: v1alpha1.SandboxTemplateSpec{
			VolumeMounts: []v1alpha1.VolumeMountSpec{{Name: "cache", MountPath: "/workspace/cache"}},
		},
	}

	err := validateClaimMountsForTemplate(req, template)
	if err == nil {
		t.Fatal("expected undeclared mount point validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}
}

func TestValidateClaimMountsForTemplateRejectsMountWhenTemplateDeclaresNone(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{{SandboxVolumeID: "vol-1", MountPoint: "/workspace/data"}},
	}
	template := &v1alpha1.SandboxTemplate{}

	err := validateClaimMountsForTemplate(req, template)
	if err == nil {
		t.Fatal("expected undeclared mount point validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}
}

func TestValidateClaimMountsForTemplateAllowsOmittedDeclaredMountPoints(t *testing.T) {
	req := &ClaimRequest{}
	template := &v1alpha1.SandboxTemplate{
		Spec: v1alpha1.SandboxTemplateSpec{
			VolumeMounts: []v1alpha1.VolumeMountSpec{{Name: "data", MountPath: "/workspace/data"}},
		},
	}

	if err := validateClaimMountsForTemplate(req, template); err != nil {
		t.Fatalf("validateClaimMountsForTemplate() error = %v", err)
	}
}

func TestValidateClaimMountsForTemplateAllowsPartialDeclaredMountPoints(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{{SandboxVolumeID: "vol-1", MountPoint: "/workspace/data"}},
	}
	template := &v1alpha1.SandboxTemplate{
		Spec: v1alpha1.SandboxTemplateSpec{
			VolumeMounts: []v1alpha1.VolumeMountSpec{
				{Name: "data", MountPath: "/workspace/data"},
				{Name: "cache", MountPath: "/workspace/cache"},
			},
		},
	}

	if err := validateClaimMountsForTemplate(req, template); err != nil {
		t.Fatalf("validateClaimMountsForTemplate() error = %v", err)
	}
}

func TestValidateClaimMountsForTemplateAllowsDeclaredMountPoint(t *testing.T) {
	req := &ClaimRequest{
		Mounts: []ClaimMount{{SandboxVolumeID: "vol-1", MountPoint: "/workspace/project/../data"}},
	}
	if err := validateClaimMounts(req); err != nil {
		t.Fatalf("validateClaimMounts() error = %v", err)
	}
	template := &v1alpha1.SandboxTemplate{
		Spec: v1alpha1.SandboxTemplateSpec{
			VolumeMounts: []v1alpha1.VolumeMountSpec{{Name: "data", MountPath: "/workspace/data"}},
		},
	}

	if err := validateClaimMountsForTemplate(req, template); err != nil {
		t.Fatalf("validateClaimMountsForTemplate() error = %v", err)
	}
}

func TestValidateVolumePortalAccessRejectsRWXNodeLocalPortal(t *testing.T) {
	svc := &SandboxService{volumeMetadata: fakeVolumeMetadataClient{accessMode: "RWX"}}

	err := svc.validateVolumePortalAccess(context.Background(), "team-a", "user-a", "vol-1", v1alpha1.VolumeMountSpec{
		Name:      "data",
		MountPath: "/workspace/data",
	})
	if err == nil {
		t.Fatal("expected RWX access mode validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}
}

func TestValidateVolumePortalAccessAllowsROXOnlyForReadOnlyTemplateMount(t *testing.T) {
	svc := &SandboxService{volumeMetadata: fakeVolumeMetadataClient{accessMode: "ROX"}}

	err := svc.validateVolumePortalAccess(context.Background(), "team-a", "user-a", "vol-1", v1alpha1.VolumeMountSpec{
		Name:      "data",
		MountPath: "/workspace/data",
	})
	if err == nil {
		t.Fatal("expected read-write ROX validation error")
	}
	if !errors.Is(err, ErrInvalidClaimRequest) {
		t.Fatalf("expected ErrInvalidClaimRequest, got %v", err)
	}

	if err := svc.validateVolumePortalAccess(context.Background(), "team-a", "user-a", "vol-1", v1alpha1.VolumeMountSpec{
		Name:      "data",
		MountPath: "/workspace/data",
		ReadOnly:  true,
	}); err != nil {
		t.Fatalf("validateVolumePortalAccess() read-only error = %v", err)
	}
}

type fakeVolumeMetadataClient struct {
	accessMode string
	err        error
	prepared   []string
	prepareErr error
}

func (c fakeVolumeMetadataClient) Get(_ context.Context, teamID, userID, volumeID string) (*SandboxVolumeInfo, error) {
	if c.err != nil {
		return nil, c.err
	}
	return &SandboxVolumeInfo{
		ID:         volumeID,
		TeamID:     teamID,
		UserID:     userID,
		AccessMode: c.accessMode,
	}, nil
}

func (c *fakeVolumeMetadataClient) PrepareForVolumePortalBind(_ context.Context, req PrepareVolumePortalBindRequest) error {
	if c == nil {
		return nil
	}
	c.prepared = append(c.prepared, req.TeamID+":"+req.UserID+":"+req.VolumeID+":"+req.PodUID)
	return c.prepareErr
}

func TestPrepareVolumePortalBindUsesPreparationClientWhenAvailable(t *testing.T) {
	metadata := &fakeVolumeMetadataClient{}
	svc := &SandboxService{volumeMetadata: metadata}

	if err := svc.prepareVolumePortalBind(context.Background(), PrepareVolumePortalBindRequest{
		TeamID:   "team-a",
		UserID:   "user-a",
		VolumeID: "vol-1",
		PodUID:   "pod-uid",
	}); err != nil {
		t.Fatalf("prepareVolumePortalBind() error = %v", err)
	}
	if len(metadata.prepared) != 1 {
		t.Fatalf("prepared calls = %d, want 1", len(metadata.prepared))
	}
	if metadata.prepared[0] != "team-a:user-a:vol-1:pod-uid" {
		t.Fatalf("prepared call = %q, want %q", metadata.prepared[0], "team-a:user-a:vol-1:pod-uid")
	}
}

func TestBindVolumePortalTreatsPreparationConflictAsClaimConflict(t *testing.T) {
	metadata := &fakeVolumeMetadataClient{prepareErr: ErrVolumePortalBindConflict}
	client := fake.NewSimpleClientset(&corev1.Node{
		ObjectMeta: metav1.ObjectMeta{Name: "node-a"},
		Status: corev1.NodeStatus{
			Addresses: []corev1.NodeAddress{{
				Type:    corev1.NodeInternalIP,
				Address: "10.0.0.8",
			}},
		},
	})
	svc := &SandboxService{
		k8sClient:      client,
		ctldClient:     &CtldClient{},
		volumeMetadata: metadata,
		config:         SandboxServiceConfig{CtldPort: 8095},
	}
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{Name: "sandbox-a", Namespace: "team-a", UID: "pod-uid"},
		Spec:       corev1.PodSpec{NodeName: "node-a"},
	}

	_, err := svc.bindVolumePortal(context.Background(), pod, "team-a", "user-a", "team-a", "vol-1", "/workspace/data", "data")
	if err == nil {
		t.Fatal("bindVolumePortal() error = nil, want claim conflict")
	}
	if !errors.Is(err, ErrClaimConflict) {
		t.Fatalf("bindVolumePortal() error = %v, want ErrClaimConflict", err)
	}
}

func TestBindVolumePortalRetriesWhilePortalPublicationIsPending(t *testing.T) {
	var bindCalls int
	ctld := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/api/v1/volume-portals/bind" {
			http.NotFound(w, r)
			return
		}
		bindCalls++
		w.Header().Set("Content-Type", "application/json")
		if bindCalls < 3 {
			w.WriteHeader(http.StatusBadRequest)
			_ = json.NewEncoder(w).Encode(ctldapi.BindVolumePortalResponse{
				Error: "volume portal data for pod pod-uid is not published",
			})
			return
		}
		_ = json.NewEncoder(w).Encode(ctldapi.BindVolumePortalResponse{
			SandboxVolumeID: "vol-1",
			MountPoint:      "/workspace/data",
		})
	}))
	defer ctld.Close()

	ctldURL, err := url.Parse(ctld.URL)
	if err != nil {
		t.Fatalf("parse ctld url: %v", err)
	}
	ctldPort, err := strconv.Atoi(ctldURL.Port())
	if err != nil {
		t.Fatalf("parse ctld port: %v", err)
	}

	client := fake.NewSimpleClientset(&corev1.Node{
		ObjectMeta: metav1.ObjectMeta{Name: "node-a"},
		Status: corev1.NodeStatus{
			Addresses: []corev1.NodeAddress{{
				Type:    corev1.NodeInternalIP,
				Address: ctldURL.Hostname(),
			}},
		},
	})
	svc := &SandboxService{
		k8sClient:  client,
		ctldClient: NewCtldClient(CtldClientConfig{Timeout: time.Second}),
		config:     SandboxServiceConfig{CtldPort: ctldPort},
	}
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{Name: "sandbox-a", Namespace: "team-a", UID: "pod-uid"},
		Spec:       corev1.PodSpec{NodeName: "node-a"},
	}

	resp, err := svc.bindVolumePortal(context.Background(), pod, "team-a", "user-a", "team-a", "vol-1", "/workspace/data", "data")
	if err != nil {
		t.Fatalf("bindVolumePortal() error = %v", err)
	}
	if resp == nil || resp.SandboxVolumeID != "vol-1" {
		t.Fatalf("bindVolumePortal() response = %+v, want bound vol-1", resp)
	}
	if bindCalls != 3 {
		t.Fatalf("bind calls = %d, want 3", bindCalls)
	}
}

func TestBindVolumePortalTreatsCtldConflictAsClaimConflict(t *testing.T) {
	ctld := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Path != "/api/v1/volume-portals/bind" {
			http.NotFound(w, r)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		w.WriteHeader(http.StatusConflict)
		_ = json.NewEncoder(w).Encode(ctldapi.BindVolumePortalResponse{
			Error: "volume vol-1 is already bound to /workspace",
		})
	}))
	defer ctld.Close()

	ctldURL, err := url.Parse(ctld.URL)
	if err != nil {
		t.Fatalf("parse ctld url: %v", err)
	}
	ctldPort, err := strconv.Atoi(ctldURL.Port())
	if err != nil {
		t.Fatalf("parse ctld port: %v", err)
	}

	client := fake.NewSimpleClientset(&corev1.Node{
		ObjectMeta: metav1.ObjectMeta{Name: "node-a"},
		Status: corev1.NodeStatus{
			Addresses: []corev1.NodeAddress{{
				Type:    corev1.NodeInternalIP,
				Address: ctldURL.Hostname(),
			}},
		},
	})
	svc := &SandboxService{
		k8sClient:  client,
		ctldClient: NewCtldClient(CtldClientConfig{Timeout: time.Second}),
		config:     SandboxServiceConfig{CtldPort: ctldPort},
	}
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{Name: "sandbox-a", Namespace: "team-a", UID: "pod-uid"},
		Spec:       corev1.PodSpec{NodeName: "node-a"},
	}

	_, err = svc.bindVolumePortal(context.Background(), pod, "team-a", "user-a", "team-a", "vol-1", "/workspace/data", "data")
	if err == nil {
		t.Fatal("bindVolumePortal() error = nil, want claim conflict")
	}
	if !errors.Is(err, ErrClaimConflict) {
		t.Fatalf("bindVolumePortal() error = %v, want ErrClaimConflict", err)
	}
}

func newClaimTestPodLister(t *testing.T, pods ...*corev1.Pod) corelisters.PodLister {
	t.Helper()
	return corelisters.NewPodLister(newClaimTestPodIndexer(t, pods...))
}

func findClaimTestCSIVolumeByPortal(volumes []corev1.Volume, portalName string) *corev1.Volume {
	for i := range volumes {
		if volumes[i].CSI == nil {
			continue
		}
		if volumes[i].CSI.VolumeAttributes[volumeportal.AttributePortalName] == portalName {
			return &volumes[i]
		}
	}
	return nil
}

func withClaimTestPublicKey(t *testing.T) {
	t.Helper()
	keyPath := filepath.Join(t.TempDir(), "internal_jwt_public.key")
	if err := os.WriteFile(keyPath, []byte("test-public-key"), 0o600); err != nil {
		t.Fatalf("write public key: %v", err)
	}
	previousKeyPath := internalauth.DefaultInternalJWTPublicKeyPath
	internalauth.DefaultInternalJWTPublicKeyPath = keyPath
	t.Cleanup(func() { internalauth.DefaultInternalJWTPublicKeyPath = previousKeyPath })
}

func withClaimTestManagerConfig(t *testing.T, content string) {
	t.Helper()
	path := filepath.Join(t.TempDir(), "manager-config.yaml")
	if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
		t.Fatalf("write manager config: %v", err)
	}
	previousPath, hadPath := os.LookupEnv("CONFIG_PATH")
	if err := os.Setenv("CONFIG_PATH", path); err != nil {
		t.Fatalf("set CONFIG_PATH: %v", err)
	}
	t.Cleanup(func() {
		if hadPath {
			_ = os.Setenv("CONFIG_PATH", previousPath)
			return
		}
		_ = os.Unsetenv("CONFIG_PATH")
	})
}

func newClaimTestSecretLister(t *testing.T, secrets ...*corev1.Secret) corelisters.SecretLister {
	t.Helper()
	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{
		cache.NamespaceIndex: cache.MetaNamespaceIndexFunc,
	})
	for _, secret := range secrets {
		if err := indexer.Add(secret); err != nil {
			t.Fatalf("add secret: %v", err)
		}
	}
	return corelisters.NewSecretLister(indexer)
}

func newClaimTestNodeLister(t *testing.T, nodes ...*corev1.Node) corelisters.NodeLister {
	t.Helper()
	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{})
	for _, node := range nodes {
		if err := indexer.Add(node); err != nil {
			t.Fatalf("add node: %v", err)
		}
	}
	return corelisters.NewNodeLister(indexer)
}

func newClaimTestPodIndexer(t *testing.T, pods ...*corev1.Pod) cache.Indexer {
	t.Helper()
	indexer := cache.NewIndexer(cache.MetaNamespaceKeyFunc, cache.Indexers{
		cache.NamespaceIndex: cache.MetaNamespaceIndexFunc,
	})
	for _, pod := range pods {
		if err := indexer.Add(pod); err != nil {
			t.Fatalf("add pod: %v", err)
		}
	}
	return indexer
}

func scheduleCreatedClaimPodInIndexer(t *testing.T, client *fake.Clientset, indexer cache.Indexer, mutate func(*corev1.Pod)) {
	t.Helper()
	client.PrependReactor("create", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
		createAction, ok := action.(k8stesting.CreateAction)
		if !ok {
			return false, nil, nil
		}
		pod, ok := createAction.GetObject().(*corev1.Pod)
		if !ok || pod == nil {
			return false, nil, nil
		}
		indexedPod := pod.DeepCopy()
		indexedPod.ResourceVersion = "2"
		indexedPod.Spec.NodeName = "node-a"
		indexedPod.Status.Phase = corev1.PodRunning
		indexedPod.Status.PodIP = "10.244.0.10"
		if mutate != nil {
			mutate(indexedPod)
		}
		if err := indexer.Add(indexedPod); err != nil {
			return true, nil, err
		}
		return false, nil, nil
	})
}

func newClaimTestPod(namespace, name, templateID string, ready bool) *corev1.Pod {
	status := corev1.ConditionFalse
	if ready {
		status = corev1.ConditionTrue
	}
	now := time.Now().UTC()
	return &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: namespace,
			Name:      name,
			Labels: map[string]string{
				controller.LabelTemplateID: templateID,
				controller.LabelPoolType:   controller.PoolTypeIdle,
			},
			Annotations: map[string]string{
				controller.AnnotationTemplateSpecHash: claimTestTemplateHash(templateID),
			},
			ResourceVersion: "1",
		},
		Status: corev1.PodStatus{
			Phase: corev1.PodRunning,
			Conditions: []corev1.PodCondition{
				{
					Type:               corev1.PodReady,
					Status:             status,
					LastTransitionTime: metav1.NewTime(now),
				},
			},
		},
	}
}

func assertClaimOwnerMetadata(t *testing.T, pod *corev1.Pod) {
	t.Helper()
	if pod == nil {
		t.Fatal("pod is nil")
	}
	if got := pod.Labels[controller.LabelOwnerKind]; got != "managed-agent" {
		t.Fatalf("owner kind label = %q, want managed-agent", got)
	}
	if got := pod.Annotations[controller.AnnotationOwnerKind]; got != "managed-agent" {
		t.Fatalf("owner kind annotation = %q, want managed-agent", got)
	}
}

func claimTestTemplateHash(templateID string) string {
	hash, _ := controller.TemplateSpecHash(&v1alpha1.SandboxTemplate{
		ObjectMeta: metav1.ObjectMeta{Name: templateID},
	})
	return hash
}

func newClaimReadyTestPod(namespace, name, templateID string) *corev1.Pod {
	pod := newClaimTestPod(namespace, name, templateID, false)
	pod.Spec.NodeName = "node-a"
	pod.Spec.ReadinessGates = []corev1.PodReadinessGate{{ConditionType: v1alpha1.SandboxPodReadinessConditionType}}
	pod.Status.PodIP = "10.244.0.10"
	pod.Status.Conditions = append(pod.Status.Conditions,
		corev1.PodCondition{Type: corev1.ContainersReady, Status: corev1.ConditionTrue},
		corev1.PodCondition{Type: v1alpha1.SandboxPodReadinessConditionType, Status: corev1.ConditionFalse},
	)
	pod.Status.ContainerStatuses = []corev1.ContainerStatus{{
		Name:  "procd",
		Ready: true,
		State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{StartedAt: metav1.Now()}},
	}}
	return pod
}

func newClaimTestNode(name, internalIP string) *corev1.Node {
	return &corev1.Node{
		ObjectMeta: metav1.ObjectMeta{Name: name},
		Status: corev1.NodeStatus{Addresses: []corev1.NodeAddress{{
			Type:    corev1.NodeInternalIP,
			Address: internalIP,
		}}},
	}
}

func splitTestServerAddress(t *testing.T, server *httptest.Server) (string, int) {
	t.Helper()
	parsed, err := url.Parse(server.URL)
	if err != nil {
		t.Fatalf("parse server URL: %v", err)
	}
	host, portString, err := net.SplitHostPort(parsed.Host)
	if err != nil {
		t.Fatalf("split server host: %v", err)
	}
	port, err := strconv.Atoi(portString)
	if err != nil {
		t.Fatalf("parse server port: %v", err)
	}
	return host, port
}

func findMetric(families []*dto.MetricFamily, name string, labels map[string]string) *dto.Metric {
	for _, family := range families {
		if family.GetName() != name {
			continue
		}
		for _, metric := range family.GetMetric() {
			if metricLabelsMatch(metric, labels) {
				return metric
			}
		}
	}
	return nil
}

func metricLabelsMatch(metric *dto.Metric, labels map[string]string) bool {
	if metric == nil {
		return false
	}
	for wantName, wantValue := range labels {
		found := false
		for _, label := range metric.GetLabel() {
			if label.GetName() == wantName && label.GetValue() == wantValue {
				found = true
				break
			}
		}
		if !found {
			return false
		}
	}
	return true
}

type sequencePodLister struct {
	mu    sync.Mutex
	pods  []*corev1.Pod
	calls int
}

func (l *sequencePodLister) List(_ labels.Selector) ([]*corev1.Pod, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if len(l.pods) == 0 {
		return nil, nil
	}
	return []*corev1.Pod{l.pods[len(l.pods)-1].DeepCopy()}, nil
}

func (l *sequencePodLister) Pods(namespace string) corelisters.PodNamespaceLister {
	return &sequencePodNamespaceLister{parent: l, namespace: namespace}
}

func (l *sequencePodLister) callCount() int {
	l.mu.Lock()
	defer l.mu.Unlock()
	return l.calls
}

type sequencePodNamespaceLister struct {
	parent    *sequencePodLister
	namespace string
}

func (l *sequencePodNamespaceLister) List(_ labels.Selector) ([]*corev1.Pod, error) {
	pods, err := l.parent.List(labels.Everything())
	if err != nil {
		return nil, err
	}
	out := make([]*corev1.Pod, 0, len(pods))
	for _, pod := range pods {
		if pod.Namespace == l.namespace {
			out = append(out, pod)
		}
	}
	return out, nil
}

func (l *sequencePodNamespaceLister) Get(name string) (*corev1.Pod, error) {
	l.parent.mu.Lock()
	defer l.parent.mu.Unlock()
	if len(l.parent.pods) == 0 {
		return nil, apierrors.NewNotFound(corev1.Resource("pods"), name)
	}
	index := l.parent.calls
	if index >= len(l.parent.pods) {
		index = len(l.parent.pods) - 1
	}
	l.parent.calls++
	pod := l.parent.pods[index]
	if pod == nil || pod.Namespace != l.namespace || pod.Name != name {
		return nil, apierrors.NewNotFound(corev1.Resource("pods"), name)
	}
	return pod.DeepCopy(), nil
}
