[codex] Add serverless container wake path

agent/internal/agent/agent.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"techulus/cloud-agent/internal/build"
@@ -50,35 +51,44 @@ type ActualState struct {
 }
 
 type Agent struct {
-	state                       AgentState
-	stateMutex                  sync.RWMutex
-	reconcileRequested          chan struct{}
-	statusReportRequested       chan string
-	refreshMutex                sync.Mutex
-	pendingExpectedStateRefresh bool
-	workMutex                   sync.Mutex
-	activeWorkItem              *agenthttp.WorkQueueItem
-	pendingWorkResults          []agenthttp.CompletedWorkItem
-	deploymentErrorMutex        sync.Mutex
-	pendingDeploymentErrors     []agenthttp.DeploymentError
-	Client                      *agenthttp.Client
-	Reconciler                  *reconcile.Reconciler
-	Config                      *Config
-	PublicIP                    string
-	PrivateIP                   string
-	DataDir                     string
-	expectedState               *agenthttp.ExpectedState
-	processingStart             time.Time
-	LogCollector                *logs.Collector
-	TraefikLogCollector         *logs.TraefikCollector
-	MetricsSender               MetricsSender
-	Builder                     *build.Builder
-	isBuilding                  bool
-	buildMutex                  sync.Mutex
-	currentBuildID              string
-	IsProxy                     bool
-	dnsInSync                   bool
-	DisableDNS                  bool
+	state                        AgentState
+	stateMutex                   sync.RWMutex
+	reconcileRequested           chan struct{}
+	statusReportRequested        chan string
+	refreshMutex                 sync.Mutex
+	pendingExpectedStateRefresh  bool
+	workMutex                    sync.Mutex
+	activeWorkItem               *agenthttp.WorkQueueItem
+	pendingWorkResults           []agenthttp.CompletedWorkItem
+	deploymentErrorMutex         sync.Mutex
+	pendingDeploymentErrors      []agenthttp.DeploymentError
+	deployLockMutex              sync.Mutex
+	deploymentDeployLocks        map[string]*sync.Mutex
+	serverlessMutex              sync.Mutex
+	pendingServerlessTransitions []agenthttp.ServerlessTransition
+	pendingServerlessSleep       map[string]struct{}
+	pendingServerlessWake        map[string]struct{}
+	expectedStateMutex           sync.RWMutex
+	latestExpectedState          *agenthttp.ExpectedState
+	Client                       *agenthttp.Client
+	Reconciler                   *reconcile.Reconciler
+	Config                       *Config
+	PublicIP                     string
+	PrivateIP                    string
+	DataDir                      string
+	expectedState                *agenthttp.ExpectedState
+	processingStart              time.Time
+	LogCollector                 *logs.Collector
+	TraefikLogCollector          *logs.TraefikCollector
+	MetricsSender                MetricsSender
+	Builder                      *build.Builder
+	isBuilding                   bool
+	buildMutex                   sync.Mutex
+	currentBuildID               string
+	IsProxy                      bool
+	serverlessGatewayRunning     atomic.Bool
+	dnsInSync                    bool
+	DisableDNS                   bool
 }
 
 func NewAgent(
@@ -94,21 +104,24 @@ func NewAgent(
 	disableDNS bool,
 ) *Agent {
 	return &Agent{
-		state:                 StateIdle,
-		reconcileRequested:    make(chan struct{}, 1),
-		statusReportRequested: make(chan string, 1),
-		Client:                client,
-		Reconciler:            reconciler,
-		Config:                config,
-		PublicIP:              publicIP,
-		PrivateIP:             privateIP,
-		DataDir:               dataDir,
-		LogCollector:          logCollector,
-		TraefikLogCollector:   traefikLogCollector,
-		MetricsSender:         metricsSender,
-		Builder:               builder,
-		IsProxy:               isProxy,
-		DisableDNS:            disableDNS,
+		state:                  StateIdle,
+		reconcileRequested:     make(chan struct{}, 1),
+		statusReportRequested:  make(chan string, 1),
+		Client:                 client,
+		Reconciler:             reconciler,
+		Config:                 config,
+		PublicIP:               publicIP,
+		PrivateIP:              privateIP,
+		DataDir:                dataDir,
+		LogCollector:           logCollector,
+		TraefikLogCollector:    traefikLogCollector,
+		MetricsSender:          metricsSender,
+		Builder:                builder,
+		IsProxy:                isProxy,
+		DisableDNS:             disableDNS,
+		deploymentDeployLocks:  map[string]*sync.Mutex{},
+		pendingServerlessSleep: map[string]struct{}{},
+		pendingServerlessWake:  map[string]struct{}{},
 	}
 }
 

agent/internal/agent/drift.go

@@ -23,6 +23,7 @@ const (
 	actionStopUnexpectedContainer    reconcileActionKind = "stop_unexpected_container"
 	actionRemoveUnexpectedContainer  reconcileActionKind = "remove_unexpected_container"
 	actionDeployMissingContainer     reconcileActionKind = "deploy_missing_container"
+	actionStopExpectedContainer      reconcileActionKind = "stop_expected_container"
 	actionStartContainer             reconcileActionKind = "start_container"
 	actionRedeployContainer          reconcileActionKind = "redeploy_container"
 	actionUpdateDNS                  reconcileActionKind = "update_dns"
@@ -102,6 +103,8 @@ func (a *Agent) handleIdle() {
 	if fromCache {
 		log.Printf("[idle] using cached state (CP unreachable)")
 	}
+	a.SetLatestExpectedState(expected)
+	a.ReconcilePendingServerlessTransitionsWithExpected(expected, fromCache)
 
 	actual, err := a.getActualState()
 	if err != nil {
@@ -250,6 +253,9 @@ func (a *Agent) planReconcile(expected *agenthttp.ExpectedState, actual *ActualS
 
 	for id, exp := range expectedMap {
 		if _, exists := actualMap[id]; !exists {
+			if desiredContainerState(exp) == "stopped" || a.HasPendingServerlessSleep(id) || a.HasPendingServerlessWake(id) {
+				continue
+			}
 			expectedContainer := exp
 			actions = append(actions, reconcileAction{
 				Kind:         actionDeployMissingContainer,
@@ -264,26 +270,43 @@ func (a *Agent) planReconcile(expected *agenthttp.ExpectedState, actual *ActualS
 		if act, exists := actualMap[id]; exists {
 			expectedContainer := exp
 			actualContainer := act
-			if act.State == "created" || act.State == "exited" {
+
+			if a.HasPendingServerlessWake(id) {
+				continue
+			}
+			if desiredContainerState(exp) == "stopped" || a.HasPendingServerlessSleep(id) {
+				if shouldStopDesiredStoppedContainer(act.State) {
+					actions = append(actions, reconcileAction{
+						Kind:         actionStopExpectedContainer,
+						Description:  fmt.Sprintf("STOP %s (desired state: stopped)", exp.Name),
+						DeploymentID: id,
+						Expected:     &expectedContainer,
+						Actual:       &actualContainer,
+					})
+				}
+				continue
+			}
+
+			if normalizeImage(exp.Image) != normalizeImage(act.Image) {
 				actions = append(actions, reconcileAction{
-					Kind:         actionStartContainer,
-					Description:  fmt.Sprintf("START %s (state: %s)", exp.Name, act.State),
+					Kind:         actionRedeployContainer,
+					Description:  fmt.Sprintf("REDEPLOY %s (image: %s → %s)", exp.Name, act.Image, exp.Image),
 					DeploymentID: id,
 					Expected:     &expectedContainer,
 					Actual:       &actualContainer,
 				})
-			} else if act.State != "running" {
+			} else if act.State == "created" || act.State == "exited" {
 				actions = append(actions, reconcileAction{
-					Kind:         actionRedeployContainer,
-					Description:  fmt.Sprintf("REDEPLOY %s (state: %s)", exp.Name, act.State),
+					Kind:         actionStartContainer,
+					Description:  fmt.Sprintf("START %s (state: %s)", exp.Name, act.State),
 					DeploymentID: id,
 					Expected:     &expectedContainer,
 					Actual:       &actualContainer,
 				})
-			} else if normalizeImage(exp.Image) != normalizeImage(act.Image) {
+			} else if act.State != "running" {
 				actions = append(actions, reconcileAction{
 					Kind:         actionRedeployContainer,
-					Description:  fmt.Sprintf("REDEPLOY %s (image: %s → %s)", exp.Name, act.Image, exp.Image),
+					Description:  fmt.Sprintf("REDEPLOY %s (state: %s)", exp.Name, act.State),
 					DeploymentID: id,
 					Expected:     &expectedContainer,
 					Actual:       &actualContainer,
@@ -389,16 +412,32 @@ func normalizeImage(image string) string {
 	return image + digest
 }
 
+func desiredContainerState(container agenthttp.ExpectedContainer) string {
+	if container.DesiredState == "stopped" {
+		return "stopped"
+	}
+	return "running"
+}
+
+func shouldStopDesiredStoppedContainer(state string) bool {
+	switch state {
+	case "created", "exited", "stopped":
+		return false
+	default:
+		return true
+	}
+}
+
 func (a *Agent) applyReconcileAction(action reconcileAction) error {
 	log.Printf("[reconcile] %s", action.Description)
 
 	switch action.Kind {
-	case actionStopOrphanNoDeploymentID, actionStopUnexpectedContainer:
+	case actionStopOrphanNoDeploymentID, actionStopUnexpectedContainer, actionStopExpectedContainer:
 		if action.Actual == nil {
 			return fmt.Errorf("missing actual container for %s", action.Kind)
 		}
 		if err := container.Stop(action.Actual.ID); err != nil {
-			return fmt.Errorf("failed to stop orphan container: %w", err)
+			return fmt.Errorf("failed to stop container: %w", err)
 		}
 		return nil
 
@@ -433,7 +472,7 @@ func (a *Agent) applyReconcileAction(action reconcileAction) error {
 		if action.Expected == nil {
 			return fmt.Errorf("missing expected container for %s", action.Kind)
 		}
-		if err := a.Reconciler.Deploy(*action.Expected); err != nil {
+		if err := a.DeployExpectedContainer(*action.Expected); err != nil {
 			return fmt.Errorf("failed to deploy container: %w", err)
 		}
 		return nil
@@ -447,7 +486,7 @@ func (a *Agent) applyReconcileAction(action reconcileAction) error {
 			if err := container.Stop(action.Actual.ID); err != nil {
 				log.Printf("[reconcile] warning: failed to stop old container: %v", err)
 			}
-			if err := a.Reconciler.Deploy(*action.Expected); err != nil {
+			if err := a.DeployExpectedContainer(*action.Expected); err != nil {
 				return fmt.Errorf("failed to redeploy container: %w", err)
 			}
 		}
@@ -460,7 +499,7 @@ func (a *Agent) applyReconcileAction(action reconcileAction) error {
 		if err := container.Stop(action.Actual.ID); err != nil {
 			log.Printf("[reconcile] warning: failed to stop old container: %v", err)
 		}
-		if err := a.Reconciler.Deploy(*action.Expected); err != nil {
+		if err := a.DeployExpectedContainer(*action.Expected); err != nil {
 			return fmt.Errorf("failed to redeploy container: %w", err)
 		}
 		return nil

agent/internal/agent/reporting.go

@@ -18,6 +18,8 @@ import (
 
 var Version = "dev"
 
+const serverlessGatewayCapability = "serverless_gateway"
+
 var (
 	agentStartTime    = time.Now()
 	lastHealthCollect time.Time
@@ -51,8 +53,9 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport
 		report.NetworkHealth = health.CollectNetworkHealth("wg0")
 		report.ContainerHealth = health.CollectContainerHealth()
 		report.AgentHealth = &agenthttp.AgentHealth{
-			Version:    Version,
-			UptimeSecs: int64(time.Since(agentStartTime).Seconds()),
+			Version:      Version,
+			UptimeSecs:   int64(time.Since(agentStartTime).Seconds()),
+			Capabilities: a.agentCapabilities(),
 		}
 		lastHealthCollect = time.Now()
 		log.Printf("[health] collected: cpu=%.1f%%, mem=%.1f%%, disk=%.1f%%, network=%v, containers=%d running",
@@ -68,6 +71,9 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport
 			if c.DeploymentID == "" {
 				continue
 			}
+			if a.ShouldSuppressServerlessContainerReport(c.DeploymentID) {
+				continue
+			}
 
 			status := "stopped"
 			if c.State == "running" {
@@ -95,6 +101,13 @@ func (a *Agent) BuildStatusReport(includeResources bool) *agenthttp.StatusReport
 	return report
 }
 
+func (a *Agent) agentCapabilities() []string {
+	if !a.IsProxy || !a.serverlessGatewayRunning.Load() {
+		return nil
+	}
+	return []string{serverlessGatewayCapability}
+}
+
 func (a *Agent) RecordDeploymentError(deploymentID string, err error) {
 	if deploymentID == "" || err == nil {
 		return

agent/internal/agent/run.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"techulus/cloud-agent/internal/container"
+	"techulus/cloud-agent/internal/serverless"
 )
 
 func (a *Agent) Run(ctx context.Context) {
@@ -32,6 +33,15 @@ func (a *Agent) Run(ctx context.Context) {
 		a.TraefikLogCollector.Start()
 	}
 
+	if a.IsProxy {
+		gateway := serverless.NewGateway(a)
+		if err := gateway.Start(ctx); err != nil {
+			log.Printf("[serverless-gateway] failed to start: %v", err)
+		} else {
+			a.serverlessGatewayRunning.Store(true)
+		}
+	}
+
 	var cleanupTickerC <-chan time.Time
 	if a.Builder != nil {
 		cleanupTicker := time.NewTicker(BuildCleanupInterval)
@@ -109,12 +119,14 @@ func (a *Agent) reportStatus(reason string) {
 	report := a.BuildStatusReport(true)
 	reportedDeploymentErrorCount := len(report.DeploymentErrors)
 	completed, active := a.SnapshotWorkStatus()
-	response, err := a.Client.ReportStatus(report, completed, active)
+	serverlessTransitions := a.SnapshotServerlessTransitions()
+	response, err := a.Client.ReportStatus(report, completed, active, serverlessTransitions)
 	if err != nil {
 		log.Printf("[status] failed to report (%s): %v", reason, err)
 		return
 	}
 	a.ClearReportedDeploymentErrors(reportedDeploymentErrorCount)
+	a.ClearReportedServerlessTransitions(len(serverlessTransitions))
 	a.AcknowledgeWorkResults(response.AcceptedWorkItemResults, response.RejectedWorkItemResults)
 	a.LogRejectedActiveWorkItems(response.RejectedActiveWorkItems)
 	a.AcceptLeasedWorkItems(response.WorkItems)