internal/sessions/fleet_client.go

@@ -0,0 +1,97 @@
+// Package sessions — FleetClient calls the SandboxFleet controller REST API
+// to assign sessions to fleet pods. This is the primary sandbox path;
+// the raw-k8s SandboxManager is the fallback when the fleet controller
+// is unavailable (e.g. during fleet rollout).
+package sessions
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"time"
+)
+
+const (
+	defaultFleetControllerURL = "http://tool-definition-controller.poimen.svc.cluster.local:8788"
+)
+
+// FleetClient calls the SandboxFleet controller REST API.
+type FleetClient struct {
+	baseURL string
+	client  *http.Client
+}
+
+// FleetAssignResult is returned by the controller assign endpoint.
+type FleetAssignResult struct {
+	PodSvcDns string `json:"podSvcDns"`
+	PodName   string `json:"podName"`
+}
+
+// NewFleetClient creates a FleetClient from env or default URL.
+// FLEET_CONTROLLER_URL — default: defaultFleetControllerURL
+func NewFleetClient() *FleetClient {
+	url := os.Getenv("FLEET_CONTROLLER_URL")
+	if url == "" {
+		url = defaultFleetControllerURL
+	}
+	return &FleetClient{
+		baseURL: url,
+		client:  &http.Client{Timeout: 35 * time.Second},
+	}
+}
+
+// Assign asks the fleet controller to assign sessionID to a fleet pod.
+// Returns the pod's ClusterIP service DNS and pod name.
+func (f *FleetClient) Assign(ctx context.Context, sessionID string) (*FleetAssignResult, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost,
+		f.baseURL+"/sessions/"+sessionID+"/assign", nil)
+	if err != nil {
+		return nil, fmt.Errorf("build request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := f.client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("fleet controller unreachable: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode >= 300 {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		return nil, fmt.Errorf("fleet controller returned %d: %s", resp.StatusCode, body)
+	}
+
+	var result FleetAssignResult
+	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
+		return nil, fmt.Errorf("decode: %w", err)
+	}
+	if result.PodSvcDns == "" {
+		return nil, fmt.Errorf("fleet controller returned empty podSvcDns")
+	}
+	return &result, nil
+}
+
+// Release notifies the fleet controller that a session has ended.
+// Cleans up the per-session ServiceAccount (audit trail SA).
+// Non-fatal: fleet controller handles this async.
+func (f *FleetClient) Release(ctx context.Context, sessionID string) error {
+	req, err := http.NewRequestWithContext(ctx, http.MethodDelete,
+		f.baseURL+"/sessions/"+sessionID, bytes.NewReader(nil))
+	if err != nil {
+		return fmt.Errorf("build request: %w", err)
+	}
+	resp, err := f.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("fleet controller unreachable: %w", err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode >= 300 {
+		body, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
+		return fmt.Errorf("fleet controller returned %d: %s", resp.StatusCode, body)
+	}
+	return nil
+}

internal/sessions/handler.go

@@ -32,10 +32,13 @@ const (
 
 // Handler handles session REST endpoints via internal service calls.
 // All traffic (reads and writes) goes to graphqlURL (poimen-graphql).
+// Sandbox creates/deletes pi-session pods via the SandboxFleet controller
+// (primary) with SandboxManager as direct-k8s fallback.
 type Handler struct {
 	graphqlURL string
 	client     *http.Client
 	sandbox    *SandboxManager
+	fleet      *FleetClient
 }
 
 // NewHandler creates a Handler reading upstream URL from env var.
@@ -49,6 +52,7 @@ func NewHandler() *Handler {
 		graphqlURL: gql,
 		client:     &http.Client{Timeout: 30 * time.Second, Transport: tracing.NewTransport(nil)},
 		sandbox:    NewSandboxManager(),
+		fleet:      NewFleetClient(),
 	}
 	h.startIdleReaper()
 	return h
@@ -545,39 +549,72 @@ func (h *Handler) handleSandbox(w http.ResponseWriter, r *http.Request) {
 	}
 }
 
-// POST ?session_id= — create or resume pi-session sandbox.
+// POST ?session_id= — assign session to SandboxFleet pod (primary) or
+// create a dedicated pod via SandboxManager (fallback).
 func (h *Handler) handleSandboxCreate(w http.ResponseWriter, r *http.Request, sessionID string) {
+	// Primary: fleet controller assigns session to a pre-warmed pod.
+	if assign, err := h.fleet.Assign(r.Context(), sessionID); err == nil {
+		svcDns := assign.PodSvcDns + fmt.Sprintf(":%d", sandboxPort)
+		log.Printf("sandbox: fleet assigned session=%s pod=%s dns=%s", sessionID, assign.PodName, svcDns)
+		go h.updateSandboxSummary(r, sessionID, assign.PodName, "", svcDns)
+		writeJSON(w, http.StatusOK, map[string]any{
+			"session_id": sessionID,
+			"pod_name":   assign.PodName,
+			"svc_dns":    svcDns,
+			"resumed":    false,
+			"fleet":      true,
+		})
+		return
+	} else {
+		log.Printf("sandbox: fleet unavailable for session=%s (%v), falling back to direct k8s", sessionID, err)
+	}
+
+	// Fallback: create a dedicated pod via SandboxManager.
 	result, err := h.sandbox.CreateOrResume(r.Context(), sessionID)
 	if err != nil {
 		log.Printf("sandbox create/resume session=%s: %v", sessionID, err)
 		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "sandbox error: " + err.Error()})
 		return
 	}
-
-	// Update session memory_entity with pod/pvc metadata (fire-and-forget — don't fail the response)
-	go func() {
-		current, _ := h.fetchEntitySummary(r, "session-"+sessionID, sessionID, "SESSION")
-		current["pod_name"] = result.PodName
-		current["pvc_name"] = result.PVCName
-		current["svc_dns"] = result.SvcDNS
-		current["last_active_at"] = time.Now().UTC().Format(time.RFC3339)
-		if b, err := json.Marshal(current); err == nil {
-			h.upsertEntity(r, "session-"+sessionID, sessionID, "SESSION", string(b)) //nolint:errcheck
-		}
-	}()
-
-	log.Printf("sandbox: session=%s pod=%s resumed=%v dns=%s", sessionID, result.PodName, result.Resumed, result.SvcDNS)
+	go h.updateSandboxSummary(r, sessionID, result.PodName, result.PVCName, result.SvcDNS)
+	log.Printf("sandbox: direct pod session=%s pod=%s resumed=%v dns=%s", sessionID, result.PodName, result.Resumed, result.SvcDNS)
 	writeJSON(w, http.StatusOK, result)
 }
 
-// DELETE ?session_id= — delete Pod + Service + PVC.
+// updateSandboxSummary merges pod/svc metadata into the session memory_entity (fire-and-forget).
+// Uses context.Background() because r.Context() is cancelled once the HTTP handler returns.
+func (h *Handler) updateSandboxSummary(r *http.Request, sessionID, podName, pvcName, svcDns string) {
+	// Build a detached request carrying just the auth header.
+	detached, _ := http.NewRequestWithContext(context.Background(), r.Method, r.URL.String(), nil)
+	if auth := r.Header.Get("Authorization"); auth != "" {
+		detached.Header.Set("Authorization", auth)
+	}
+	current, _ := h.fetchEntitySummary(detached, "session-"+sessionID, sessionID, "SESSION")
+	if current == nil {
+		current = map[string]any{}
+	}
+	current["pod_name"] = podName
+	current["svc_dns"] = svcDns
+	current["last_active_at"] = time.Now().UTC().Format(time.RFC3339)
+	if pvcName != "" {
+		current["pvc_name"] = pvcName
+	}
+	if b, err := json.Marshal(current); err == nil {
+		h.upsertEntity(r, "session-"+sessionID, sessionID, "SESSION", string(b)) //nolint:errcheck
+	}
+}
+
+// DELETE ?session_id= — release from fleet + delete dedicated pod+svc+pvc if any.
 func (h *Handler) handleSandboxDelete(w http.ResponseWriter, r *http.Request, sessionID string) {
+	// Release from fleet (cleans up per-session SA; non-fatal if session was fleet-based).
+	if err := h.fleet.Release(r.Context(), sessionID); err != nil {
+		log.Printf("sandbox: fleet release session=%s: %v (continuing)", sessionID, err)
+	}
+	// Also attempt direct k8s cleanup (no-ops if resources don't exist).
 	if err := h.sandbox.Delete(r.Context(), sessionID); err != nil {
 		log.Printf("sandbox delete session=%s: %v", sessionID, err)
-		writeJSON(w, http.StatusBadGateway, map[string]string{"error": "sandbox delete error: " + err.Error()})
-		return
 	}
-	log.Printf("sandbox: deleted pod+svc+pvc for session=%s", sessionID)
+	log.Printf("sandbox: released session=%s", sessionID)
 	writeJSON(w, http.StatusOK, map[string]string{"session_id": sessionID, "status": "deleted"})
 }
 

k8s/network-policy.yaml

@@ -122,7 +122,7 @@ spec:
     ports:
     - protocol: TCP
       port: 7233
-  # Allow to Poimen (memory:8080 + graphql:8090 + redis:6379)
+  # Allow to Poimen (memory:8080 + graphql:8090 + redis:6379 + fleet-controller:8788)
   - to:
     - namespaceSelector:
         matchLabels:
@@ -134,6 +134,8 @@ spec:
       port: 8090
     - protocol: TCP
       port: 6379
+    - protocol: TCP
+      port: 8788
   # Allow to Gotify (push notifications)
   - to:
     - namespaceSelector: