CosmicClash/server/supervisor/supervisor.go

// Package supervisor contains the small PID-1 lifecycle boundary around an
// allocated Godot process. The Agones client is HTTP-only so local/Compose
// execution remains independent of the cloud SDK.
package supervisor

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"time"
)

type GameServer struct {
	// ObjectMeta.Annotations carries per-allocation data the agones package
	// requests on the GameServerAllocation (server/agones/allocation.go) --
	// currently cosmic-clash.io/match-id, cosmic-clash.io/allocation-id, and
	// allocator-selected compatibility fields.
	// This is the only channel for match-specific config to reach an
	// already-Ready pod: env vars are fixed at pod creation, before Agones
	// assigns a match to it. NOTE: the exact JSON key for this field
	// (object_meta vs objectMeta) is not independently verified against a
	// live Agones SDK sidecar from this sandbox; if it turns out wrong,
	// annotationMatchID simply returns "" and callers fall back to whatever
	// was explicitly configured, so this degrades safely either way.
	ObjectMeta struct {
		Annotations map[string]string `json:"annotations"`
	} `json:"object_meta"`
	Status struct {
		Address string `json:"address"`
		Ports   []struct {
			Name string `json:"name"`
			Port int    `json:"port"`
		} `json:"ports"`
	} `json:"status"`
}

type Config struct {
	Command      []string
	Environment  []string
	SDKBaseURL   string
	ReadyURL     string
	Transport    string
	DrainURL     string
	AdmissionURL string
	DrainToken   string
	ReadyTimeout time.Duration
	PollInterval time.Duration
	HTTPClient   *http.Client

	// ControlPlaneURL, when set, opts into reporting process-ready to the
	// matchmaking control plane (multiplayer-next.md task 8.28) once Agones
	// Ready succeeds. Leaving it empty preserves every existing behavior
	// exactly -- direct/Compose mode and allocated-without-control-plane mode
	// are both unaffected. WorkloadTokenPath, if set, is read fresh on every
	// call rather than cached, matching how a Kubernetes projected service
	// account token is rotated in place by kubelet before it expires -- this
	// is for a future Kubernetes-JWT-based WorkloadVerify (server/workload/
	// jwt.go), not yet wired server-side. Today the control plane instead
	// verifies a self-issued signed token (server/workload/signed_token.go),
	// which reaches this process via the cosmic-clash.io/workload-token
	// annotation Agones applies to the allocated GameServer (see
	// server/agones.Client.Allocate) -- see workloadToken() for the
	// precedence between the two sources. ServerID and ImageDigest are
	// expected to be populated from the pod spec (Downward API / mounted
	// build metadata). MatchID may be left empty here and is then read from
	// the allocated GameServer's own annotations (see GameServer.ObjectMeta
	// above) -- an explicit value here always wins.
	ControlPlaneURL   string
	WorkloadTokenPath string
	ServerID          string
	MatchID           string
	ProtocolVersion   int
	ImageDigest       string

	// AssignmentReadyAttempts/AssignmentReadyBackoff bound the retry loop for
	// reporting assignment-ready once process-ready has already succeeded.
	// The control-plane's own durable gate (every participant already
	// holding a live, unexpired assignment -- see
	// AdvanceServerRegistrationSQL) may not be satisfied on the very first
	// attempt if the signed roster is still propagating, and that is
	// expected during the bounded retries. Exhaustion is fatal because player
	// assignments remain hidden until this transition. Default 5 attempts, 2s apart.
	AssignmentReadyAttempts int
	AssignmentReadyBackoff  time.Duration
	// RosterPath is an operator-mounted writable path where the supervisor
	// materializes the workload-authenticated signed roster before starting
	// Godot. It is deliberately separate from WorkloadTokenPath: the former
	// contains match join envelopes, the latter contains a bearer credential.
	RosterPath string
}

type Supervisor struct {
	config         Config
	client         *http.Client
	cmd            *exec.Cmd
	lastGameServer GameServer
}

const (
	ChildControlPlaneURLEnv = "COSMIC_CLASH_CONTROL_PLANE_URL"
	ChildWorkloadTokenEnv   = "COSMIC_CLASH_WORKLOAD_TOKEN"
	ChildAdmissionSignalEnv = "COSMIC_CLASH_INITIAL_CONNECT_SIGNAL_REQUIRED"
)

const (
	DefaultDrainGrace  = 285 * time.Second
	DefaultHTTPTimeout = 10 * time.Second
)

func New(config Config) (*Supervisor, error) {
	if len(config.Command) == 0 || config.Command[0] == "" {
		return nil, fmt.Errorf("supervisor command is required")
	}
	if config.ReadyTimeout <= 0 {
		config.ReadyTimeout = 30 * time.Second
	}
	if config.PollInterval <= 0 {
		config.PollInterval = 100 * time.Millisecond
	}
	if config.AssignmentReadyAttempts <= 0 {
		config.AssignmentReadyAttempts = 5
	}
	if config.AssignmentReadyBackoff <= 0 {
		config.AssignmentReadyBackoff = 2 * time.Second
	}
	if config.Transport == "" {
		config.Transport = "enet"
	}
	if config.Transport != "enet" && config.Transport != "steam_sdr" {
		return nil, fmt.Errorf("unsupported transport %q", config.Transport)
	}
	if config.HTTPClient == nil {
		config.HTTPClient = &http.Client{Timeout: DefaultHTTPTimeout}
	}
	if (config.DrainURL == "") != (config.DrainToken == "") {
		return nil, fmt.Errorf("drain URL and token must be configured together")
	}
	if config.DrainURL != "" {
		if err := validateLocalDrainURL(config.DrainURL); err != nil {
			return nil, err
		}
	}
	if config.AdmissionURL != "" {
		if config.DrainToken == "" {
			return nil, fmt.Errorf("initial-connect admission URL requires a control token")
		}
		if err := validateLocalDrainURL(config.AdmissionURL); err != nil {
			return nil, fmt.Errorf("invalid initial-connect admission URL: %w", err)
		}
	}
	if config.ControlPlaneURL != "" && (config.ServerID == "" || config.ProtocolVersion < 1 || config.ImageDigest == "") {
		return nil, fmt.Errorf("control-plane registration requires a server ID, protocol version and image digest")
	}
	if config.ControlPlaneURL != "" {
		parsed, err := url.Parse(config.ControlPlaneURL)
		if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" || parsed.Path != "" {
			return nil, fmt.Errorf("control-plane URL must be an HTTP(S) origin")
		}
	}
	if config.RosterPath != "" && config.ControlPlaneURL == "" {
		return nil, fmt.Errorf("roster path requires control-plane URL")
	}
	// Neither MatchID nor WorkloadTokenPath is required here: both can
	// instead be resolved at Start time from the allocated GameServer's own
	// annotations (see registerControlPlane/workloadToken/matchID). They are
	// validated to actually be resolvable there, not silently skipped.
	return &Supervisor{config: config, client: config.HTTPClient}, nil
}

func validateLocalDrainURL(raw string) error {
	parsed, err := url.Parse(raw)
	if err != nil || (parsed.Scheme != "http" && parsed.Scheme != "https") || parsed.Host == "" || parsed.Path == "" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" {
		return fmt.Errorf("drain URL must be a loopback HTTP endpoint")
	}
	host := parsed.Hostname()
	if host != "localhost" {
		ip := net.ParseIP(host)
		if ip == nil || !ip.IsLoopback() {
			return fmt.Errorf("drain URL must be a loopback HTTP endpoint")
		}
	}
	return nil
}

// Start launches the process and marks Agones Ready only after the explicit
// readiness probe succeeds. No stdout/log scraping is used. With no SDK URL,
// this is direct/Compose mode and the command is simply started.
func (s *Supervisor) Start(ctx context.Context) error {
	env := append([]string(nil), os.Environ()...)
	env = append(env, s.config.Environment...)
	if s.config.SDKBaseURL != "" {
		port, address, err := s.waitAssignedEndpoint(ctx)
		if err != nil {
			return err
		}
		if s.config.Transport == "steam_sdr" {
			env = append(env, "SDR_LISTEN_PORT="+strconv.Itoa(port), "SDR_IP="+address+":"+strconv.Itoa(port))
		}
		rosterExpiry, err := s.fetchRoster(ctx)
		if err != nil {
			return err
		}
		childControlPlaneEnv, err := s.controlPlaneChildEnvironment()
		if err != nil {
			return err
		}
		env = append(env, childControlPlaneEnv...)
		command := withAllocatedConfig(s.config.Command, s.matchID(), s.config.ServerID, s.config.ImageDigest, rosterExpiry)
		command, err = withAllocatedCompatibility(command, s.lastGameServer)
		if err != nil {
			return err
		}
		command = withPort(command, port)
		s.cmd = exec.CommandContext(ctx, command[0], command[1:]...)
	} else {
		s.cmd = exec.CommandContext(ctx, s.config.Command[0], s.config.Command[1:]...)
	}
	s.cmd.Env = env
	// A server's structured stdout/stderr is its operational interface. The
	// zero value for exec.Cmd streams is /dev/null, which would make a child
	// startup failure invisible to Docker, Kubernetes, and the Compose
	// readiness harness while the supervisor can report only "exit status 1".
	s.cmd.Stdout = os.Stdout
	s.cmd.Stderr = os.Stderr
	if err := s.cmd.Start(); err != nil {
		return err
	}
	if s.config.SDKBaseURL == "" {
		return nil
	}
	if err := s.waitReady(ctx); err != nil {
		_ = s.cmd.Process.Kill()
		return err
	}
	if err := s.sdkPost(ctx, "/ready"); err != nil {
		return err
	}
	if err := s.registerControlPlane(ctx, false); err != nil {
		// Unlike a bare Agones Ready, this failure leaves the match's durable
		// control-plane record stuck at ALLOCATING with no way for the
		// matcher/allocator to learn this process is actually listening --
		// players would wait indefinitely for a server that Agones considers
		// healthy. Kill the child so Kubernetes reschedules rather than
		// leaving that silent split-brain running.
		_ = s.cmd.Process.Kill()
		return err
	}
	if err := s.reportAssignmentReady(ctx); err != nil {
		// Player assignments remain deliberately hidden until this durable
		// transition succeeds. Do not leave an Agones-Ready process accepting
		// connections for a match the control plane cannot expose.
		_ = s.cmd.Process.Kill()
		return err
	}
	if err := s.signalInitialConnectReady(ctx); err != nil {
		_ = s.cmd.Process.Kill()
		return err
	}
	return nil
}

// waitAssignedEndpoint covers the short interval between the SDK sidecar
// accepting requests and the GameServer controller populating status.address
// and status.ports. Treating the sidecar's first incomplete response as fatal
// creates a restart loop precisely while Agones is finishing normal startup.
func (s *Supervisor) waitAssignedEndpoint(ctx context.Context) (int, string, error) {
	deadline := time.NewTimer(s.config.ReadyTimeout)
	defer deadline.Stop()
	var lastErr error
	for {
		port, address, err := s.assignedEndpoint(ctx)
		if err == nil {
			return port, address, nil
		}
		lastErr = err
		select {
		case <-ctx.Done():
			return 0, "", ctx.Err()
		case <-deadline.C:
			return 0, "", fmt.Errorf("assigned endpoint timed out: %w", lastErr)
		case <-time.After(s.config.PollInterval):
		}
	}
}

func (s *Supervisor) signalInitialConnectReady(ctx context.Context) error {
	if s.config.AdmissionURL == "" {
		return nil
	}
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.config.AdmissionURL, nil)
	if err != nil {
		return err
	}
	request.Header.Set("Authorization", "Bearer "+s.config.DrainToken)
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("initial-connect admission control returned %s", response.Status)
	}
	return nil
}

func (s *Supervisor) controlPlaneChildEnvironment() ([]string, error) {
	if s.config.ControlPlaneURL == "" {
		return nil, nil
	}
	token, err := s.workloadToken()
	if err != nil {
		return nil, err
	}
	if strings.ContainsRune(token, '\x00') {
		return nil, fmt.Errorf("workload token contains an invalid environment byte")
	}
	environment := []string{
		ChildControlPlaneURLEnv + "=" + strings.TrimRight(s.config.ControlPlaneURL, "/"),
		ChildWorkloadTokenEnv + "=" + token,
	}
	if s.config.AdmissionURL != "" {
		environment = append(environment, ChildAdmissionSignalEnv+"=1")
	}
	return environment, nil
}

func (s *Supervisor) fetchRoster(ctx context.Context) (time.Time, error) {
	if s.config.RosterPath == "" {
		return time.Time{}, nil
	}
	matchID := s.matchID()
	if matchID == "" {
		return time.Time{}, fmt.Errorf("roster fetch has no match ID")
	}
	token, err := s.workloadToken()
	if err != nil {
		return time.Time{}, err
	}
	rosterURL := strings.TrimRight(s.config.ControlPlaneURL, "/") + "/v1/servers/" + url.PathEscape(s.config.ServerID) + "/roster"
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, rosterURL, nil)
	if err != nil {
		return time.Time{}, err
	}
	request.Header.Set("Authorization", "Bearer "+token)
	response, err := s.client.Do(request)
	if err != nil {
		return time.Time{}, err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return time.Time{}, fmt.Errorf("control-plane roster returned %s", response.Status)
	}
	var roster []json.RawMessage
	if err := json.NewDecoder(io.LimitReader(response.Body, 1<<20)).Decode(&roster); err != nil || len(roster) == 0 {
		if err == nil {
			err = fmt.Errorf("empty roster")
		}
		return time.Time{}, fmt.Errorf("decode control-plane roster: %w", err)
	}
	var expiry time.Time
	for _, envelope := range roster {
		if len(envelope) == 0 || string(envelope) == "null" {
			return time.Time{}, fmt.Errorf("control-plane roster contains an invalid envelope")
		}
		var decoded struct {
			Authorisation struct {
				ExpiresAt time.Time `json:"expires_at"`
			} `json:"authorisation"`
		}
		if err := json.Unmarshal(envelope, &decoded); err != nil || decoded.Authorisation.ExpiresAt.IsZero() {
			return time.Time{}, fmt.Errorf("control-plane roster contains an envelope without expiry")
		}
		if expiry.IsZero() || decoded.Authorisation.ExpiresAt.Before(expiry) {
			expiry = decoded.Authorisation.ExpiresAt
		}
	}
	contents, err := json.Marshal(roster)
	if err != nil {
		return time.Time{}, fmt.Errorf("encode roster: %w", err)
	}
	directory := filepath.Dir(s.config.RosterPath)
	temporary, err := os.CreateTemp(directory, ".cosmic-clash-roster-*")
	if err != nil {
		return time.Time{}, fmt.Errorf("create roster file: %w", err)
	}
	temporaryName := temporary.Name()
	defer os.Remove(temporaryName)
	if err := temporary.Chmod(0600); err != nil {
		_ = temporary.Close()
		return time.Time{}, fmt.Errorf("secure roster file: %w", err)
	}
	_, err = temporary.Write(contents)
	if closeErr := temporary.Close(); err == nil {
		err = closeErr
	}
	if err != nil {
		return time.Time{}, fmt.Errorf("write roster file: %w", err)
	}
	if err := os.Rename(temporaryName, s.config.RosterPath); err != nil {
		return time.Time{}, fmt.Errorf("install roster file: %w", err)
	}
	return expiry, nil
}

func withAllocatedConfig(command []string, matchID, serverID, imageDigest string, rosterExpiry time.Time) []string {
	result := append([]string(nil), command...)
	values := map[string]string{
		"match-id":            matchID,
		"server-id":           serverID,
		"server-image-digest": imageDigest,
	}
	if !rosterExpiry.IsZero() {
		values["assignment-expiry-unix"] = strconv.FormatInt(rosterExpiry.Unix(), 10)
	}
	for key, value := range values {
		if value == "" {
			continue
		}
		prefix := "--" + key + "="
		replaced := false
		for i, arg := range result {
			if strings.HasPrefix(arg, prefix) {
				result[i] = prefix + value
				replaced = true
				break
			}
		}
		if !replaced {
			result = append(result, prefix+value)
		}
	}
	return result
}

// withAllocatedCompatibility overlays fields selected by the allocator onto
// child flags. These values arrive through Agones allocation annotations after
// the pod was created, so static Fleet defaults must never win over them.
func withAllocatedCompatibility(command []string, gameServer GameServer) ([]string, error) {
	values := map[string]string{}
	annotations := gameServer.ObjectMeta.Annotations
	if annotations == nil {
		return command, nil
	}
	for annotation, flag := range map[string]string{
		"cosmic-clash.io/arena-path": "arena-path",
		"cosmic-clash.io/playlist":   "playlist",
		"cosmic-clash.io/region":     "region",
		"cosmic-clash.io/build":      "client-build",
		"cosmic-clash.io/protocol":   "protocol-version",
		"cosmic-clash.io/transport":  "transport",
	} {
		value := annotations[annotation]
		if value == "" {
			continue
		}
		if strings.ContainsAny(value, "\r\n\t") {
			return nil, fmt.Errorf("allocated annotation %q contains control characters", annotation)
		}
		values[flag] = value
	}
	return withAllocatedValues(command, values), nil
}

func withAllocatedValues(command []string, values map[string]string) []string {
	result := append([]string(nil), command...)
	for key, value := range values {
		if value == "" {
			continue
		}
		prefix := "--" + key + "="
		replaced := false
		for i, arg := range result {
			if strings.HasPrefix(arg, prefix) {
				result[i] = prefix + value
				replaced = true
				break
			}
		}
		if !replaced {
			result = append(result, prefix+value)
		}
	}
	return result
}

// reportAssignmentReady retries the durable gate that makes player
// assignments visible. A process without this transition is not usable even
// when Agones and the local readiness probe consider it healthy.
func (s *Supervisor) reportAssignmentReady(ctx context.Context) error {
	if s.config.ControlPlaneURL == "" {
		return nil
	}
	var lastErr error
	for attempt := 0; attempt < s.config.AssignmentReadyAttempts; attempt++ {
		if attempt > 0 {
			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-time.After(s.config.AssignmentReadyBackoff):
			}
		}
		if lastErr = s.registerControlPlane(ctx, true); lastErr == nil {
			return nil
		}
	}
	return fmt.Errorf("assignment-ready registration did not succeed after %d attempts: %w", s.config.AssignmentReadyAttempts, lastErr)
}

// registerControlPlane reports the allocated process's readiness to the
// matchmaking control plane (POST /v1/servers/{id}/register). It is a no-op
// whenever ControlPlaneURL is unset, which is the default and preserves
// every existing direct/Compose/allocated-only behavior exactly. The
// workload token is read fresh from disk on every call rather than cached --
// a Kubernetes projected service account token is rotated in place by
// kubelet before it expires, so caching it risks presenting a stale one on a
// long-lived process.
// matchID resolves the match ID for control-plane registration: an
// explicitly configured value always wins, otherwise it falls back to the
// cosmic-clash.io/match-id annotation Agones applied to this GameServer at
// allocation time (see server/agones.Client.Allocate). Empty if neither is
// available.
func (s *Supervisor) matchID() string {
	if s.config.MatchID != "" {
		return s.config.MatchID
	}
	return s.lastGameServer.ObjectMeta.Annotations["cosmic-clash.io/match-id"]
}

// workloadToken resolves the bearer credential for control-plane
// registration. WorkloadTokenPath, when configured, always wins -- it is
// for a future Kubernetes-projected-JWT WorkloadVerify path (see the Config
// field's doc comment) and an operator who explicitly set it presumably
// wants it used. Otherwise it falls back to the cosmic-clash.io/workload-
// token annotation Agones applied to this GameServer at allocation time
// (server/agones.Client.Allocate, verified by
// api.WorkloadVerifierFromSignedToken today) -- the same annotation-fallback
// pattern matchID already uses for cosmic-clash.io/match-id.
func (s *Supervisor) workloadToken() (string, error) {
	if s.config.WorkloadTokenPath != "" {
		tokenBytes, err := os.ReadFile(s.config.WorkloadTokenPath)
		if err != nil {
			return "", fmt.Errorf("read workload token: %w", err)
		}
		token := strings.TrimSpace(string(tokenBytes))
		if token == "" {
			return "", fmt.Errorf("workload token file %q is empty", s.config.WorkloadTokenPath)
		}
		return token, nil
	}
	token := s.lastGameServer.ObjectMeta.Annotations["cosmic-clash.io/workload-token"]
	if token == "" {
		return "", fmt.Errorf("control-plane registration has no workload token: no --workload-token-path configured, and no cosmic-clash.io/workload-token annotation was present on the allocated GameServer")
	}
	return token, nil
}

func (s *Supervisor) registerControlPlane(ctx context.Context, assignmentReady bool) error {
	if s.config.ControlPlaneURL == "" {
		return nil
	}
	matchID := s.matchID()
	if matchID == "" {
		return fmt.Errorf("control-plane registration has no match ID: not configured, and no cosmic-clash.io/match-id annotation was present on the allocated GameServer")
	}
	token, err := s.workloadToken()
	if err != nil {
		return err
	}
	body, err := json.Marshal(struct {
		MatchID         string `json:"match_id"`
		ProtocolVersion int    `json:"protocol_version"`
		ImageDigest     string `json:"image_digest"`
		AssignmentReady bool   `json:"assignment_ready"`
	}{matchID, s.config.ProtocolVersion, s.config.ImageDigest, assignmentReady})
	if err != nil {
		return err
	}
	endpoint := strings.TrimRight(s.config.ControlPlaneURL, "/") + "/v1/servers/" + url.PathEscape(s.config.ServerID) + "/register"
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
	if err != nil {
		return err
	}
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("Authorization", "Bearer "+token)
	// Idempotent per (server, match, readiness stage): a supervisor restart
	// or a dropped response retrying this exact call must replay, not
	// conflict. The API enforces a 16-128 byte key; ServerID and MatchID are
	// both already required non-empty by this point.
	key := "supervisor-register-" + s.config.ServerID + "-" + matchID + "-" + strconv.FormatBool(assignmentReady)
	if len(key) > 128 {
		key = key[:128]
	}
	request.Header.Set("Idempotency-Key", key)
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("control-plane register returned %s", response.Status)
	}
	return nil
}

func withPort(command []string, port int) []string {
	result := append([]string(nil), command...)
	for i, arg := range result {
		if strings.HasPrefix(arg, "--port=") {
			result[i] = "--port=" + strconv.Itoa(port)
			return result
		}
	}
	return append(result, "--port="+strconv.Itoa(port))
}

func (s *Supervisor) Wait() error {
	if s.cmd == nil {
		return fmt.Errorf("supervisor has not started")
	}
	return s.cmd.Wait()
}

// Run owns the PID-1 termination sequence. The child gets its own context so
// cancellation of the supervisor does not kill it before the authenticated
// drain request has had a chance to stop new admissions. A non-responsive
// child is force-killed after drainGrace; a drain failure is recorded only by
// the returned error if the child exits cleanly, while the deadline still
// prevents a stuck process from hanging termination forever.
func (s *Supervisor) Run(ctx context.Context, drainGrace time.Duration) error {
	if s == nil || ctx == nil {
		return fmt.Errorf("supervisor context is required")
	}
	if drainGrace <= 0 {
		drainGrace = DefaultDrainGrace
	}
	processCtx, cancel := context.WithCancel(context.Background())
	defer cancel()
	if err := s.Start(processCtx); err != nil {
		return err
	}
	wait := make(chan error, 1)
	go func() { wait <- s.Wait() }()
	select {
	case err := <-wait:
		return err
	case <-ctx.Done():
	}

	var drainErr error
	if s.config.DrainURL != "" {
		drainCtx, drainCancel := context.WithTimeout(context.Background(), 5*time.Second)
		drainErr = s.Drain(drainCtx)
		drainCancel()
	}
	var shutdownErr error
	if s.config.ControlPlaneURL != "" && drainErr == nil {
		shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second)
		shutdownErr = s.acknowledgeShutdown(shutdownCtx, "server_draining")
		shutdownCancel()
	}
	timer := time.NewTimer(drainGrace)
	defer timer.Stop()
	select {
	case err := <-wait:
		if drainErr != nil {
			return fmt.Errorf("child exited after drain failure: %w", drainErr)
		}
		if shutdownErr != nil {
			return fmt.Errorf("child exited after shutdown acknowledgement failure: %w", shutdownErr)
		}
		return err
	case <-timer.C:
		if s.cmd != nil && s.cmd.Process != nil {
			_ = s.cmd.Process.Kill()
		}
		<-wait
		if drainErr != nil {
			return fmt.Errorf("drain failed and child was force-killed: %w", drainErr)
		}
		if shutdownErr != nil {
			return fmt.Errorf("shutdown acknowledgement failed and child was force-killed: %w", shutdownErr)
		}
		return fmt.Errorf("child force-killed after drain deadline")
	}
}

// Drain asks the allocated Godot process to stop accepting new work. The
// token is sent only over the configured localhost control endpoint and is
// never placed in command arguments or logs.
func (s *Supervisor) Drain(ctx context.Context) error {
	if s.config.DrainURL == "" || s.config.DrainToken == "" {
		return fmt.Errorf("authenticated drain endpoint is required")
	}
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, s.config.DrainURL, nil)
	if err != nil {
		return err
	}
	request.Header.Set("Authorization", "Bearer "+s.config.DrainToken)
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("drain endpoint returned %s", response.Status)
	}
	return nil
}

// acknowledgeShutdown records the supervisor's planned termination after the
// local game process has been told to drain. It uses the same bound workload
// credential as registration. The idempotency key makes a repeated call safe.
func (s *Supervisor) acknowledgeShutdown(ctx context.Context, reason string) error {
	if s.config.ControlPlaneURL == "" {
		return nil
	}
	matchID := s.matchID()
	if matchID == "" {
		return fmt.Errorf("shutdown acknowledgement has no match ID")
	}
	token, err := s.workloadToken()
	if err != nil {
		return err
	}
	body, err := json.Marshal(struct {
		Reason string `json:"reason"`
	}{reason})
	if err != nil {
		return err
	}
	endpoint := strings.TrimRight(s.config.ControlPlaneURL, "/") + "/v1/servers/" + url.PathEscape(s.config.ServerID) + "/shutdown"
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
	if err != nil {
		return err
	}
	request.Header.Set("Content-Type", "application/json")
	request.Header.Set("Authorization", "Bearer "+token)
	key := "supervisor-shutdown-" + s.config.ServerID + "-" + matchID + "-" + reason
	if len(key) > 128 {
		key = key[:128]
	}
	request.Header.Set("Idempotency-Key", key)
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("control-plane shutdown returned %s", response.Status)
	}
	return nil
}

func (s *Supervisor) assignedEndpoint(ctx context.Context) (int, string, error) {
	var server GameServer
	if err := s.sdkGet(ctx, "/gameserver", &server); err != nil {
		return 0, "", err
	}
	s.lastGameServer = server
	if len(server.Status.Ports) == 0 || strings.TrimSpace(server.Status.Address) == "" || strings.ContainsAny(server.Status.Address, " \t\r\n") {
		return 0, "", fmt.Errorf("Agones returned no assigned endpoint")
	}
	for _, port := range server.Status.Ports {
		if port.Port > 0 && port.Port <= 65535 && (port.Name == "game" || len(server.Status.Ports) == 1) {
			return port.Port, server.Status.Address, nil
		}
	}
	return 0, "", fmt.Errorf("Agones returned no usable game port")
}

func (s *Supervisor) waitReady(ctx context.Context) error {
	if s.config.ReadyURL == "" {
		return fmt.Errorf("allocated mode requires an explicit readiness URL")
	}
	deadline := time.NewTimer(s.config.ReadyTimeout)
	defer deadline.Stop()
	for {
		request, err := http.NewRequestWithContext(ctx, http.MethodGet, s.config.ReadyURL, nil)
		if err == nil {
			response, requestErr := s.client.Do(request)
			if requestErr == nil {
				_ = response.Body.Close()
				if response.StatusCode >= 200 && response.StatusCode < 300 {
					return nil
				}
			}
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-deadline.C:
			return fmt.Errorf("process-ready probe timed out")
		case <-time.After(s.config.PollInterval):
		}
	}
}

func (s *Supervisor) sdkGet(ctx context.Context, path string, target any) error {
	request, err := http.NewRequestWithContext(ctx, http.MethodGet, strings.TrimRight(s.config.SDKBaseURL, "/")+path, nil)
	if err != nil {
		return err
	}
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("Agones GET %s returned %s", path, response.Status)
	}
	return json.NewDecoder(response.Body).Decode(target)
}

func (s *Supervisor) sdkPost(ctx context.Context, path string) error {
	request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(s.config.SDKBaseURL, "/")+path, nil)
	if err != nil {
		return err
	}
	response, err := s.client.Do(request)
	if err != nil {
		return err
	}
	defer response.Body.Close()
	if response.StatusCode/100 != 2 {
		return fmt.Errorf("Agones POST %s returned %s", path, response.Status)
	}
	return nil
}