Wake placement as soon as capacity or preparation is ready

services/core/internal/execution/runtime_compute.go

@@ -40,6 +40,9 @@ func (r *runtimeLifecycle) saveCompute(ctx context.Context, owner deployment.All
 	next, err := r.deployment.SetCompute(ctx, owner, phase, raw, until)
 	if err == nil {
 		log.Ctx(ctx).Info("Runtime compute phase saved", "allocation_id", owner.ID, "node_id", owner.NodeID, "from", owner.ComputePhase, "to", phase)
+		if phase == "suspended" {
+			r.freedSlot()
+		}
 	}
 	return next, err
 }
@@ -233,6 +236,7 @@ func (r *runtimeLifecycle) releaseLostCompute(ctx context.Context, p sandbox.San
 		return err
 	}
 	r.clearRuntimeState(released)
+	r.freedSlot()
 	return nil
 }
 

services/core/internal/execution/runtime_compute_wake.go

@@ -130,6 +130,9 @@ func (r *runtimeLifecycle) cleanupCompute(ctx context.Context, p sandbox.Sandbox
 		}
 	}
 	_, err := r.deployment.ReleaseAllocation(ctx, owner)
+	if err == nil {
+		r.freedSlot()
+	}
 	return err
 }
 

services/core/internal/execution/runtime_initialization.go

@@ -98,6 +98,9 @@ func (w *Worker) initializeEnvironment(ctx context.Context, owner sessions.Envir
 	err := w.prepareEnvironment(operation, owner, &failure)
 	if err == nil {
 		err = w.dispatcher.sessionExecution.CompleteEnvironmentInitialization(operation, owner)
+		if err == nil {
+			w.wakeScheduler()
+		}
 	}
 	if err != nil {
 		log.Warn(ctx, "Environment preparation failed", "environment_id", owner.EnvironmentID, "session_id", owner.SessionID)

services/core/internal/execution/runtime_lifecycle.go

@@ -68,6 +68,7 @@ type runtimeLifecycle struct {
 	wakeHints       chan struct{}
 	pendingWake     chan struct{}
 	hintDestination func(string)
+	placementWake   chan<- struct{}
 }
 
 func newRuntimeManager(owner Owner, deployments *deployment.Service, deploymentReader deployment.Reader, sessionReader sessions.Reader, registry *runtimegateway.Registry, config *RuntimeProvider) (*runtimeManager, error) {
@@ -440,6 +441,9 @@ func (r *runtimeLifecycle) observe(ctx context.Context, owner deployment.Allocat
 			return nil // Keep scanning unknown creation; absence is not a final receipt.
 		}
 		_, err = r.deployment.ReleaseAllocation(ctx, owner)
+		if err == nil {
+			r.freedSlot()
+		}
 		return err
 	}
 	// A stopped or missing container does not authorize destroying retained

services/core/internal/execution/runtime_manager.go

@@ -109,7 +109,7 @@ func (m *runtimeManager) node(id string) (*runtimeNode, error) {
 			deployment: m.deployment, deployments: m.deploymentService, reader: m.deploymentReader,
 			lease: m.lease, registry: m.registry, config: m.config, nodeID: id,
 			gate: make(chan struct{}, 1), ctx: ctx, stop: stop,
-			connections: make(map[string]*runtimeConnection), connectionWake: make(chan *runtimegateway.Session, 1), wakeHints: make(chan struct{}, 1), pendingWake: make(chan struct{}, 1), hintDestination: m.hintDestination,
+			connections: make(map[string]*runtimeConnection), connectionWake: make(chan *runtimegateway.Session, 1), wakeHints: make(chan struct{}, 1), pendingWake: make(chan struct{}, 1), hintDestination: m.hintDestination, placementWake: m.placementWake,
 		}}
 		m.nodes[id] = n
 	}
@@ -142,6 +142,16 @@ func (m *runtimeManager) hintDestination(id string) {
 	}
 }
 
+// A freed node slot may admit waiting demand; the placement ticker recovers a missed wake.
+func (r *runtimeLifecycle) freedSlot() {
+	if r.nodeID != "" {
+		select {
+		case r.placementWake <- struct{}{}:
+		default:
+		}
+	}
+}
+
 func (m *runtimeManager) hints(id string) chan<- struct{} {
 	m.mu.Lock()
 	defer m.mu.Unlock()

services/core/internal/execution/runtime_pending.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/MiniMax-AI/OpenAgentCore/internal/obs/log"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment"
+	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/deployment/placement"
 	"github.com/MiniMax-AI/OpenAgentCore/services/core/internal/sandbox"
 )
 
@@ -78,6 +79,7 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error {
 		m.placementCursor = deployment.PlacementDemandCursor{}
 		return nil
 	}
+	incomplete := false
 	for _, environment := range rows {
 		if operation.Err() != nil {
 			if ownership := m.lease.CheckOwnership(ctx); ownership != nil {
@@ -100,6 +102,7 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error {
 				return nil
 			}
 			log.Ctx(ctx).Warn("managed Runtime placement incomplete", "environment_id", environment.ID)
+			incomplete = incomplete || errors.Is(err, placement.ErrNodeUnavailable)
 			continue
 		}
 		node, err := m.node(reserved.NodeID)
@@ -121,6 +124,15 @@ func (m *runtimeManager) reservePlacements(ctx context.Context) error {
 	if next.EnvironmentID == "" {
 		m.placementCursor = deployment.PlacementDemandCursor{}
 	}
+	if incomplete {
+		// Demand blocked by full nodes should start suspension or expiry now, not on each lane's next tick.
+		for _, n := range m.snapshotNodes() {
+			select {
+			case n.lifecycle.wakeHints <- struct{}{}:
+			default:
+			}
+		}
+	}
 	return nil
 }
 

services/core/internal/execution/runtime_placement_demand_test.go

@@ -144,3 +144,42 @@ func TestPlacementBudgetEndsBeforeAttemptPreservesCursor(t *testing.T) {
 		t.Fatal("unattempted demand advanced cursor", m.placementCursor)
 	}
 }
+
+func TestCapacityFailedPlacementHintsNodeLanes(t *testing.T) {
+	m := testRuntimeManager(t)
+	m.config.InstallationID = uuid.NewString()
+	node, err := m.node("a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	m.deploymentReader = &strictDeploymentReader{t: t, placementDemand: func(context.Context, deployment.PlacementDemandCursor) ([]deployment.PlacementDemand, deployment.PlacementDemandCursor, error) {
+		return []deployment.PlacementDemand{{UnallocatedEnvironment: deployment.UnallocatedEnvironment{ID: uuid.NewString(), TenantID: uuid.NewString()}, At: time.Now()}}, deployment.PlacementDemandCursor{}, nil
+	}}
+	_, m.deployment = deploymentOperations(t, &strictDeploymentStorage{t: t}, &strictDeploymentReader{t: t}, &strictExecutionStorage{t: t, withReservation: func(context.Context, deployment.AllocationKey, func(sessions.LockedSession, deployment.ReservationTx) error) error {
+		return placement.ErrNodeUnavailable
+	}})
+	if err := m.reservePlacements(t.Context()); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-node.lifecycle.wakeHints:
+	default:
+		t.Fatal("capacity-blocked demand did not hint the node lane")
+	}
+}
+
+func TestFreedSlotWakesPlacement(t *testing.T) {
+	m := testRuntimeManager(t)
+	m.placementWake = make(chan struct{}, 1)
+	node, err := m.node("a")
+	if err != nil {
+		t.Fatal(err)
+	}
+	node.lifecycle.freedSlot()
+	node.lifecycle.freedSlot() // coalesced, never blocks
+	select {
+	case <-m.placementWake:
+	default:
+		t.Fatal("freed slot did not wake placement")
+	}
+}