fix(provisioning): address local review findings

Author: onutc

api/auth.go

@@ -119,7 +119,7 @@ func newAuthConfig() authConfig {
 		bearerTeamsPaths:          splitListOrDefault(os.Getenv("SPRITZ_AUTH_BEARER_TEAMS_PATHS"), nil),
 		bearerTypePaths:           splitListOrDefault(os.Getenv("SPRITZ_AUTH_BEARER_TYPE_PATHS"), nil),
 		bearerScopesPaths:         splitListOrDefault(os.Getenv("SPRITZ_AUTH_BEARER_SCOPES_PATHS"), []string{"scope", "scopes", "scp"}),
-		bearerDefaultType:         normalizePrincipalType(envOrDefault("SPRITZ_AUTH_BEARER_DEFAULT_TYPE", string(principalTypeHuman)), principalTypeHuman),
+		bearerDefaultType:         normalizePrincipalType(envOrDefault("SPRITZ_AUTH_BEARER_DEFAULT_TYPE", string(principalTypeService)), principalTypeService),
 		bearerAuthorizationHeader: envOrDefault("SPRITZ_AUTH_BEARER_HEADER", "Authorization"),
 		bearerJWKSURL:             strings.TrimSpace(os.Getenv("SPRITZ_AUTH_BEARER_JWKS_URL")),
 		bearerJWKSIssuer:          strings.TrimSpace(os.Getenv("SPRITZ_AUTH_BEARER_ISSUER")),

api/auth_middleware_test.go

@@ -178,3 +178,49 @@ func TestBearerAuthParsesSpaceDelimitedScopes(t *testing.T) {
 		t.Fatalf("expected two scopes, got %#v", payload["scopes"])
 	}
 }
+
+func TestBearerAuthDefaultsToServiceTypeWithoutTypeClaim(t *testing.T) {
+	introspection := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"sub":   "zenobot",
+			"scope": "spritz.instances.create spritz.instances.assign_owner",
+		})
+	}))
+	defer introspection.Close()
+
+	t.Setenv("SPRITZ_AUTH_MODE", "auto")
+	t.Setenv("SPRITZ_AUTH_BEARER_INTROSPECTION_URL", introspection.URL)
+	t.Setenv("SPRITZ_AUTH_BEARER_ID_PATHS", "sub")
+	t.Setenv("SPRITZ_AUTH_BEARER_SCOPES_PATHS", "scope")
+
+	s := &server{auth: newAuthConfig()}
+	e := echo.New()
+	secured := e.Group("", s.authMiddleware())
+	secured.GET("/api/spritzes", func(c echo.Context) error {
+		p, ok := principalFromContext(c)
+		if !ok {
+			return c.JSON(http.StatusInternalServerError, map[string]string{"error": "missing principal"})
+		}
+		return c.JSON(http.StatusOK, map[string]any{
+			"type":   p.Type,
+			"scopes": p.Scopes,
+		})
+	})
+
+	req := httptest.NewRequest(http.MethodGet, "/api/spritzes", nil)
+	req.Header.Set("Authorization", "Bearer test-token")
+	rec := httptest.NewRecorder()
+	e.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected 200, got %d: %s", rec.Code, rec.Body.String())
+	}
+
+	payload := map[string]any{}
+	if err := json.Unmarshal(rec.Body.Bytes(), &payload); err != nil {
+		t.Fatalf("failed to decode response: %v", err)
+	}
+	if payload["type"] != string(principalTypeService) {
+		t.Fatalf("expected default bearer principal type to be service, got %#v", payload["type"])
+	}
+}

api/main.go

@@ -29,26 +29,27 @@ import (
 )
 
 type server struct {
-	client            client.Client
-	clientset         *kubernetes.Clientset
-	restConfig        *rest.Config
-	scheme            *runtime.Scheme
-	namespace         string
-	auth              authConfig
-	internalAuth      internalAuthConfig
-	ingressDefaults   ingressDefaults
-	terminal          terminalConfig
-	sshGateway        sshGatewayConfig
-	sshDefaults       sshDefaults
-	sshMintLimiter    *sshMintLimiter
-	acp               acpConfig
-	presets           presetCatalog
-	provisioners      provisionerPolicy
-	defaultMetadata   map[string]string
-	sharedMounts      sharedMountsConfig
-	sharedMountsStore *sharedMountsStore
-	sharedMountsLive  *sharedMountsLatestNotifier
-	userConfigPolicy  userConfigPolicy
+	client               client.Client
+	clientset            *kubernetes.Clientset
+	restConfig           *rest.Config
+	scheme               *runtime.Scheme
+	namespace            string
+	auth                 authConfig
+	internalAuth         internalAuthConfig
+	ingressDefaults      ingressDefaults
+	terminal             terminalConfig
+	sshGateway           sshGatewayConfig
+	sshDefaults          sshDefaults
+	sshMintLimiter       *sshMintLimiter
+	acp                  acpConfig
+	presets              presetCatalog
+	provisioners         provisionerPolicy
+	defaultMetadata      map[string]string
+	sharedMounts         sharedMountsConfig
+	sharedMountsStore    *sharedMountsStore
+	sharedMountsLive     *sharedMountsLatestNotifier
+	userConfigPolicy     userConfigPolicy
+	nameGeneratorFactory func(context.Context, string, string) (func() string, error)
 }
 
 func main() {
@@ -534,6 +535,9 @@ func (s *server) createSpritz(c echo.Context) error {
 				if existing != nil && strings.TrimSpace(existing.Annotations[idempotencyHashAnnotationKey]) == strings.TrimSpace(annotations[idempotencyHashAnnotationKey]) {
 					return writeJSON(c, http.StatusOK, summarizeCreateResponse(existing, principal, body.PresetID, provisionerSource(&body), body.IdempotencyKey, true))
 				}
+				if !nameProvided {
+					continue
+				}
 				return writeError(c, http.StatusConflict, "idempotencyKey already used with a different request")
 			}
 			if !nameProvided && apierrors.IsAlreadyExists(err) {

api/main_create_owner_test.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"bytes"
+	"context"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
@@ -11,7 +12,10 @@ import (
 
 	"github.com/labstack/echo/v4"
 	corev1 "k8s.io/api/core/v1"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 
 	spritzv1 "spritz.sh/operator/api/v1"
@@ -49,6 +53,20 @@ func newCreateSpritzTestServer(t *testing.T) *server {
 	}
 }
 
+type createInterceptClient struct {
+	client.Client
+	onCreate func(context.Context, client.Object) error
+}
+
+func (c *createInterceptClient) Create(ctx context.Context, obj client.Object, opts ...client.CreateOption) error {
+	if c.onCreate != nil {
+		if err := c.onCreate(ctx, obj); err != nil {
+			return err
+		}
+	}
+	return c.Client.Create(ctx, obj, opts...)
+}
+
 func configureProvisionerTestServer(s *server) {
 	s.presets = presetCatalog{
 		byID: []runtimePreset{{
@@ -392,3 +410,63 @@ func TestCreateSpritzReplaysIdempotentProvisionerRequestBeforeQuotaCheck(t *test
 		t.Fatalf("expected replay status 200, got %d: %s", rec2.Code, rec2.Body.String())
 	}
 }
+
+func TestCreateSpritzRetriesGeneratedServiceNameAfterAlreadyExists(t *testing.T) {
+	s := newCreateSpritzTestServer(t)
+	configureProvisionerTestServer(s)
+	baseClient := s.client
+	s.client = &createInterceptClient{
+		Client: baseClient,
+		onCreate: func(_ context.Context, obj client.Object) error {
+			spritz, ok := obj.(*spritzv1.Spritz)
+			if !ok {
+				return nil
+			}
+			if spritz.Name == "openclaw-first" {
+				return apierrors.NewAlreadyExists(schema.GroupResource{
+					Group:    spritzv1.GroupVersion.Group,
+					Resource: "spritzes",
+				}, spritz.Name)
+			}
+			return nil
+		},
+	}
+	s.nameGeneratorFactory = func(context.Context, string, string) (func() string, error) {
+		names := []string{"openclaw-first", "openclaw-second"}
+		index := 0
+		return func() string {
+			name := names[index]
+			if index < len(names)-1 {
+				index++
+			}
+			return name
+		}, nil
+	}
+
+	e := echo.New()
+	secured := e.Group("", s.authMiddleware())
+	secured.POST("/api/spritzes", s.createSpritz)
+
+	body := []byte(`{"presetId":"openclaw","ownerId":"user-123","idempotencyKey":"discord-race"}`)
+	req := httptest.NewRequest(http.MethodPost, "/api/spritzes", bytes.NewReader(body))
+	req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
+	req.Header.Set("X-Spritz-User-Id", "zenobot")
+	req.Header.Set("X-Spritz-Principal-Type", "service")
+	req.Header.Set("X-Spritz-Principal-Scopes", "spritz.instances.create,spritz.instances.assign_owner")
+	rec := httptest.NewRecorder()
+
+	e.ServeHTTP(rec, req)
+
+	if rec.Code != http.StatusCreated {
+		t.Fatalf("expected status 201 after autogenerated name retry, got %d: %s", rec.Code, rec.Body.String())
+	}
+
+	var payload map[string]any
+	if err := json.Unmarshal(rec.Body.Bytes(), &payload); err != nil {
+		t.Fatalf("failed to decode response: %v", err)
+	}
+	name := payload["data"].(map[string]any)["spritz"].(map[string]any)["metadata"].(map[string]any)["name"]
+	if name != "openclaw-second" {
+		t.Fatalf("expected second generated name after race, got %#v", name)
+	}
+}

api/random_name.go

@@ -296,6 +296,9 @@ func randomSuffix(length int) string {
 }
 
 func (s *server) newSpritzNameGenerator(ctx context.Context, namespace string, prefix string) (func() string, error) {
+	if s.nameGeneratorFactory != nil {
+		return s.nameGeneratorFactory(ctx, namespace, prefix)
+	}
 	list := &spritzv1.SpritzList{}
 	opts := []client.ListOption{client.InNamespace(namespace)}
 	if err := s.client.List(ctx, list, opts...); err != nil {

api/terminal.go

@@ -165,7 +165,7 @@ func (s *server) openTerminal(c echo.Context) error {
 	if usingZmx {
 		log.Printf("spritz terminal: zmx attach name=%s namespace=%s session=%s user_id=%s", name, namespace, resolvedSession, principal.ID)
 	}
-	if err := s.streamTerminal(c.Request().Context(), pod, conn, command); err != nil {
+	if err := s.streamTerminal(c.Request().Context(), namespace, name, pod, conn, command); err != nil {
 		if errors.Is(err, context.Canceled) {
 			return nil
 		}
@@ -202,7 +202,7 @@ func (s *server) findRunningPod(ctx context.Context, namespace, name, container
 	return nil, fmt.Errorf("spritz not ready")
 }
 
-func (s *server) streamTerminal(ctx context.Context, pod *corev1.Pod, conn *websocket.Conn, command []string) error {
+func (s *server) streamTerminal(ctx context.Context, namespace, name string, pod *corev1.Pod, conn *websocket.Conn, command []string) error {
 	if len(command) == 0 {
 		return errors.New("terminal command missing")
 	}
@@ -236,7 +236,13 @@ func (s *server) streamTerminal(ctx context.Context, pod *corev1.Pod, conn *webs
 
 	readErr := make(chan error, 1)
 	go func() {
-		readErr <- readTerminalInput(ctx, conn, stdinWriter, sizeQueue)
+		readErr <- readTerminalInput(ctx, conn, stdinWriter, sizeQueue, func() {
+			refreshCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+			defer cancel()
+			if err := s.markSpritzActivity(refreshCtx, namespace, name, time.Now()); err != nil {
+				log.Printf("spritz terminal: failed to refresh activity name=%s namespace=%s pod=%s err=%v", name, namespace, pod.Name, err)
+			}
+		})
 	}()
 
 	streamErr := executor.StreamWithContext(ctx, remotecommand.StreamOptions{
@@ -268,7 +274,7 @@ type resizeMessage struct {
 	Rows int    `json:"rows"`
 }
 
-func readTerminalInput(ctx context.Context, conn *websocket.Conn, stdin *io.PipeWriter, sizeQueue *terminalSizeQueue) error {
+func readTerminalInput(ctx context.Context, conn *websocket.Conn, stdin *io.PipeWriter, sizeQueue *terminalSizeQueue, onInput func()) error {
 	for {
 		select {
 		case <-ctx.Done():
@@ -290,6 +296,9 @@ func readTerminalInput(ctx context.Context, conn *websocket.Conn, stdin *io.Pipe
 			if _, err := stdin.Write(payload); err != nil {
 				return err
 			}
+			if onInput != nil {
+				onInput()
+			}
 		}
 	}
 }

api/terminal_test.go

@@ -0,0 +1,87 @@
+package main
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gorilla/websocket"
+)
+
+func TestReadTerminalInputInvokesActivityCallbackOnInput(t *testing.T) {
+	upgrader := websocket.Upgrader{}
+	serverConn := make(chan *websocket.Conn, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			t.Fatalf("upgrade failed: %v", err)
+		}
+		serverConn <- conn
+	}))
+	defer srv.Close()
+
+	wsURL, err := url.Parse(srv.URL)
+	if err != nil {
+		t.Fatalf("failed to parse server url: %v", err)
+	}
+	wsURL.Scheme = "ws"
+	clientConn, _, err := websocket.DefaultDialer.Dial(wsURL.String(), nil)
+	if err != nil {
+		t.Fatalf("failed to dial websocket: %v", err)
+	}
+	defer clientConn.Close()
+
+	conn := <-serverConn
+	defer conn.Close()
+
+	reader, writer := io.Pipe()
+	defer reader.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var callbacks atomic.Int32
+	done := make(chan error, 1)
+	go func() {
+		done <- readTerminalInput(ctx, conn, writer, newTerminalSizeQueue(), func() {
+			callbacks.Add(1)
+		})
+	}()
+
+	if err := clientConn.WriteMessage(websocket.TextMessage, []byte(`{"type":"resize","cols":80,"rows":24}`)); err != nil {
+		t.Fatalf("failed to send resize message: %v", err)
+	}
+	if callbacks.Load() != 0 {
+		t.Fatalf("expected resize message to skip activity callback, got %d", callbacks.Load())
+	}
+
+	if err := clientConn.WriteMessage(websocket.TextMessage, []byte("ls\n")); err != nil {
+		t.Fatalf("failed to send terminal input: %v", err)
+	}
+
+	buf := make([]byte, 3)
+	if _, err := io.ReadFull(reader, buf); err != nil {
+		t.Fatalf("failed to read stdin payload: %v", err)
+	}
+	deadline := time.Now().Add(2 * time.Second)
+	for callbacks.Load() != 1 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if callbacks.Load() != 1 {
+		t.Fatalf("expected one activity callback for terminal input, got %d", callbacks.Load())
+	}
+
+	cancel()
+	_ = clientConn.Close()
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for terminal reader to exit")
+	}
+}