fix(platform): harden cold-start deploy flywheel (RED L2+L1+I2)

Close the two LOW follow-ups on the cold-start tenant-RBAC fix, plus confirm
the fresh-org fail-closed status. All on top of v1.786.23 (already SHIP).

L2 — git reconciler no longer treats a transient tenant-RBAC delay as TERMINAL
and no longer head-of-line-blocks other orgs. reconcileBuild now does ONE
non-blocking readiness probe (ensureTenantReady: create-namespace-if-absent +
single SelfSubjectAccessReview) instead of the synchronous image path's ~45s
in-line waitForTenantRBAC. If the operator's RoleBinding has not landed, the
deployment stays 'building' and re-drives on the next 10s tick — never a
permanent fail (there is no client to retry a git build) and never a 45s stall
of the shared sequential reconciler. Only the elapsed build deadline fails it
honestly. errTenantProvisioning from applyLive is also caught as transient
(defense in depth). Namespace-create is decomplected into one shared
ensureNamespaceExists; ensureNamespace (sync, blocking) and ensureTenantReady
(async, probing) compose it.

L1 — image deploy path gains a per-org in-flight-deploy cap (inflightGate,
maxConcurrentDeploys, default 8 via CLOUD_PLATFORM_MAX_CONCURRENT_DEPLOYS),
mirroring the git build cap. deployImage acquires a slot before applyLive's
~45s RBAC wait and releases on any return; over-cap is a retryable 429 refused
BEFORE recording an attempt. Bounds request-goroutine pile-up on a wedged
operator; per-org (one org's saturation never throttles another); fail-closed.

I2 — confirmed the truly-fresh-org path fails closed on 503 (RBAC pending),
NOT a raw 502: the namespace IS created (the trigger for the operator's
RoleBinding), then the bounded RBAC wait yields errTenantProvisioning -> 503.

Tests (all -race green): reconciler stays 'building' then goes live on a later
tick once RBAC lands (not 'failed'); over-cap image deploy -> 429 with per-org
isolation + slot-release re-admit; fresh-org deploy -> 503 with namespace
created + no Service CR + honest 'error' deployment recorded.

clients/platform/deploy.go

@@ -22,6 +22,7 @@ import (
 	"fmt"
 	"net/http"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/zap-proto/zip"
@@ -32,6 +33,47 @@ type deployReq struct {
 	Tag    string `json:"tag"`    // image tag to deploy (image-source)
 }
 
+// inflightGate bounds concurrent in-flight SYNCHRONOUS image deploys PER ORG (L1).
+// The git build path is already capped in the cluster (countActiveBuilds →
+// errTooManyBuilds → 429); the image path has no build Job to count, so cloud-api
+// counts in-flight deploys itself. It exists because deployImage's applyLive may
+// park up to ~45s in waitForTenantRBAC on a cold-start / wedged operator, and
+// without a cap a single validated org could pile up that many held request
+// goroutines. Fail-closed and retryable: over-cap refuses with 429, never proceeds
+// unbounded. Process-local (per replica), which is exactly the goroutine-pile-up
+// boundary each replica needs; the map is pruned to zero entries so keys never grow
+// unbounded.
+type inflightGate struct {
+	mu sync.Mutex
+	n  map[string]int
+}
+
+// acquire reserves one in-flight slot for org, or reports false when org is already
+// at max (caller must 429). Balanced by exactly one release on the success path.
+func (g *inflightGate) acquire(org string, max int) bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if g.n == nil {
+		g.n = map[string]int{}
+	}
+	if g.n[org] >= max {
+		return false
+	}
+	g.n[org]++
+	return true
+}
+
+// release returns org's slot; pruning the key at zero keeps the map bounded.
+func (g *inflightGate) release(org string) {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	if g.n[org] <= 1 {
+		delete(g.n, org)
+		return
+	}
+	g.n[org]--
+}
+
 func (s *svc) deploy(c *zip.Ctx) error {
 	org, ok := s.tenant(c)
 	if !ok {
@@ -75,6 +117,16 @@ func (s *svc) deploy(c *zip.Ctx) error {
 // deployImage applies the operator Service CR with the requested image tag and
 // lands the deployment "deploying" (the operator finishes the rollout async).
 func (s *svc) deployImage(c *zip.Ctx, org, project string, a Application, depID string, version int, now int64, body deployReq, clusterErr error) error {
+	// (L1) Bound concurrent in-flight deploys for this org: applyLive below may park
+	// up to ~45s in waitForTenantRBAC on a cold-start / wedged operator, so a cap
+	// prevents one org piling up held request goroutines (mirrors the git build cap).
+	// Over-cap is a retryable THROTTLE, not a deploy failure — refuse with 429 BEFORE
+	// recording any attempt, since nothing was tried cluster-side.
+	if !s.deployGate.acquire(org, s.k8s.limits.maxConcurrentDeploys()) {
+		return zip.Errorf(http.StatusTooManyRequests, "too many concurrent deploys for this org; retry shortly")
+	}
+	defer s.deployGate.release(org)
+
 	tag := firstNonEmpty(strings.TrimSpace(body.Tag), a.ImageTag, "latest")
 	image := a.ImageRepo + ":" + tag
 

clients/platform/http_test.go

@@ -20,10 +20,11 @@ import (
 	clienttesting "k8s.io/client-go/testing"
 )
 
-// mountAppK8s builds a hermetic app over an svc with the GIVEN k8s client, so
-// tests NEVER touch a real cluster (the earlier version resolved the dev
-// kubeconfig and wrote to live DOKS — never again).
-func mountAppK8s(t *testing.T, k *k8sClient) *zip.App {
+// mountSvcK8s builds a hermetic app AND returns the backing svc, so tests that need
+// to inspect/prime process-local state (e.g. the per-org deploy gate) can reach it.
+// It NEVER touches a real cluster (the earlier version resolved the dev kubeconfig
+// and wrote to live DOKS — never again).
+func mountSvcK8s(t *testing.T, k *k8sClient) (*zip.App, *svc) {
 	t.Helper()
 	store, err := openStore(filepath.Join(t.TempDir(), "platform.db"))
 	if err != nil {
@@ -33,6 +34,12 @@ func mountAppK8s(t *testing.T, k *k8sClient) *zip.App {
 	s := &svc{store: store, k8s: k, log: luxlog.New("test"), brand: "hanzo", sitesHost: "hanzo.app"}
 	app := zip.New(zip.Config{Logger: luxlog.New("test")})
 	s.routes(app)
+	return app, s
+}
+
+// mountAppK8s builds a hermetic app over an svc with the GIVEN k8s client.
+func mountAppK8s(t *testing.T, k *k8sClient) *zip.App {
+	app, _ := mountSvcK8s(t, k)
 	return app
 }
 
@@ -82,6 +89,7 @@ func testLimits() resourceLimits {
 	return resourceLimits{
 		maxReplicas:     20,
 		maxBuilds:       3,
+		maxDeploys:      8,
 		quotaCPU:        "20",
 		quotaMemory:     "40Gi",
 		quotaPods:       "50",
@@ -244,6 +252,66 @@ func TestHTTPDeploySucceedsIntoTenantNamespace(t *testing.T) {
 	}
 }
 
+// TestImageDeployOverCapReturns429 proves the L1 per-org in-flight deploy cap: once
+// an org has maxConcurrentDeploys deploys in flight, its next deploy is refused with
+// a RETRYABLE 429 (never a fabricated success, never unbounded goroutine pile-up),
+// the cap is PER-ORG (a saturated org never throttles another), and releasing a slot
+// re-admits the org. The gate is primed directly (deterministic) rather than by
+// racing real ~45s waits.
+func TestImageDeployOverCapReturns429(t *testing.T) {
+	k := fakeK8s()
+	k.limits.maxDeploys = 2 // small cap for a deterministic test
+	app, s := mountSvcK8s(t, k)
+
+	seedApp := func(org string) {
+		do(t, app, http.MethodPost, "/v1/platform/projects", org, map[string]any{"name": "web"})
+		do(t, app, http.MethodPost, "/v1/platform/projects/web/apps", org, map[string]any{
+			"name": "api", "source": "image",
+			"image": map[string]any{"repository": "ghcr.io/hanzoai/nginx", "tag": "1.27"},
+		})
+	}
+	seedApp("maxpower")
+	seedApp("acme")
+
+	// Saturate maxpower's in-flight deploy gate (simulate 2 deploys already parked in
+	// applyLive's RBAC wait).
+	for i := 0; i < 2; i++ {
+		if !s.deployGate.acquire("maxpower", s.k8s.limits.maxConcurrentDeploys()) {
+			t.Fatalf("precondition: acquire maxpower slot %d must succeed", i)
+		}
+	}
+
+	// maxpower's next deploy is over-cap → 429 (retryable), and records NO deployment
+	// (a throttle is not an attempt).
+	code, body := do(t, app, http.MethodPost, "/v1/platform/projects/web/apps/api/deploy", "maxpower", map[string]any{"tag": "1.27"})
+	if code != http.StatusTooManyRequests {
+		t.Fatalf("over-cap deploy want 429, got %d (%s)", code, body)
+	}
+	if _, deps := listDeps(t, app, "maxpower"); len(deps) != 0 {
+		t.Fatalf("a throttled (429) deploy must record NO deployment, got %d", len(deps))
+	}
+
+	// PER-ORG isolation: acme is unaffected while maxpower is saturated → 202.
+	if code, body := do(t, app, http.MethodPost, "/v1/platform/projects/web/apps/api/deploy", "acme", map[string]any{"tag": "1.27"}); code != http.StatusAccepted {
+		t.Fatalf("acme deploy must be unaffected by maxpower's cap, want 202, got %d (%s)", code, body)
+	}
+
+	// Releasing one maxpower slot re-admits maxpower → 202.
+	s.deployGate.release("maxpower")
+	if code, body := do(t, app, http.MethodPost, "/v1/platform/projects/web/apps/api/deploy", "maxpower", map[string]any{"tag": "1.27"}); code != http.StatusAccepted {
+		t.Fatalf("after releasing a slot maxpower deploy want 202, got %d (%s)", code, body)
+	}
+}
+
+// listDeps GETs an org's api deployments (helper for the cap test).
+func listDeps(t *testing.T, app *zip.App, org string) (int, []deploymentView) {
+	t.Helper()
+	code, body := do(t, app, http.MethodGet, "/v1/platform/projects/web/apps/api/deployments", org, nil)
+	var deps []deploymentView
+	_ = json.Unmarshal(body, &deps)
+	return code, deps
+}
+
 // TestHTTPForgeableOrgRefused proves the validated-principal gate (RED HIGH): a
 // request carrying a client X-Org-Id but NO validated principal (X-User-Id empty
 // — the Phase-1 residual a direct-to-pod caller could forge) is refused 403 on

clients/platform/k8s.go

@@ -194,16 +194,14 @@ func (k *k8sClient) buildImageRef(org, app, tag string) string {
 	return fmt.Sprintf("%s/tenant-%s/%s:%s", strings.TrimRight(prefix, "/"), provisioning.SanitizeOrg(org), app, tag)
 }
 
-// ensureNamespace creates tenant-<org> if it does not exist and ALWAYS ensures
-// the tenant's ResourceQuota + LimitRange are present (idempotent). Applying the
-// bounds on every call — not only at first create — means older tenant
-// namespaces are brought under quota too, and a deleted quota is re-created on
-// the next deploy (MED-3). The namespace is labeled with the org so cluster
-// tooling can attribute it.
-func (k *k8sClient) ensureNamespace(ctx context.Context, ns, org string) error {
-	if err := k.ready(); err != nil {
-		return err
-	}
+// ensureNamespaceExists creates tenant-<org> if it does not exist (idempotent).
+// Creating the namespace is what TRIGGERS the operator's tenant-RBAC controller to
+// project cloud-api's `cloud-api-platform` RoleBinding into it — so this is the ONE
+// precondition for tenant RBAC ever becoming ready. Pure namespace mechanism: no
+// RBAC wait, no quota. Both the synchronous image path (ensureNamespace, which then
+// BLOCKS on the RoleBinding) and the async build reconciler (ensureTenantReady,
+// which only PROBES) build on it, so there is exactly one namespace-create rule.
+func (k *k8sClient) ensureNamespaceExists(ctx context.Context, ns, org string) error {
 	_, err := k.dyn.Resource(namespacesGVR).Get(ctx, ns, metav1.GetOptions{})
 	if apierrors.IsNotFound(err) {
 		obj := &unstructured.Unstructured{Object: map[string]any{
@@ -220,7 +218,25 @@ func (k *k8sClient) ensureNamespace(ctx context.Context, ns, org string) error {
 		if _, cErr := k.dyn.Resource(namespacesGVR).Create(ctx, obj, metav1.CreateOptions{}); cErr != nil && !apierrors.IsAlreadyExists(cErr) {
 			return cErr
 		}
-	} else if err != nil {
+		return nil
+	}
+	return err
+}
+
+// ensureNamespace is the SYNCHRONOUS (image-source) tenant preparation: create the
+// namespace, BLOCK up to ~45s for the operator's async RoleBinding to land, then
+// ensure the tenant's ResourceQuota + LimitRange (idempotent). Applying the bounds
+// on every call — not only at first create — means older tenant namespaces are
+// brought under quota too, and a deleted quota is re-created on the next deploy
+// (MED-3). The caller is a single client deploy that must succeed without a manual
+// retry, so the bounded wait is worth it here; the async build reconciler uses the
+// NON-BLOCKING ensureTenantReady instead (its 10s tick is its retry loop, so it must
+// never park mid-reconcile and head-of-line-block other orgs).
+func (k *k8sClient) ensureNamespace(ctx context.Context, ns, org string) error {
+	if err := k.ready(); err != nil {
+		return err
+	}
+	if err := k.ensureNamespaceExists(ctx, ns, org); err != nil {
 		return err
 	}
 	// A BRAND-NEW tenant namespace is created above, but the operator's tenant-RBAC
@@ -246,6 +262,33 @@ func (k *k8sClient) ensureNamespace(ctx context.Context, ns, org string) error {
 	return nil
 }
 
+// ensureTenantReady is the NON-BLOCKING readiness gate for the async build
+// reconciler (reconcile.go). It creates the tenant namespace if absent (idempotent —
+// the trigger for the operator's RoleBinding) and does exactly ONE readiness probe.
+// It NEVER blocks on the operator's async RoleBinding the way the synchronous
+// ensureNamespace does (bounded ~45s waitForTenantRBAC): the reconciler is itself a
+// retry loop on a 10s tick, so parking in a per-deployment wait would head-of-line-
+// block EVERY other org's go-live behind one slow tenant onboarding. Instead the
+// reconciler probes once and, if not ready, leaves the deployment "building" and
+// re-drives next tick.
+//
+//   - ready=true            → the RoleBinding has landed; the caller may applyService
+//     now (its own waitForTenantRBAC resolves on the first probe, no sleep).
+//   - ready=false, err=nil  → namespace exists but RBAC is still provisioning; retry
+//     on a later tick (the namespace now exists, so the operator will project the
+//     RoleBinding before then).
+//   - err!=nil              → a real cluster error (namespace create / probe blip);
+//     the reconciler retries it as transient too, failing only past the deadline.
+func (k *k8sClient) ensureTenantReady(ctx context.Context, ns, org string) (bool, error) {
+	if err := k.ready(); err != nil {
+		return false, err
+	}
+	if err := k.ensureNamespaceExists(ctx, ns, org); err != nil {
+		return false, err
+	}
+	return k.canGetResourceQuotas(ctx, ns)
+}
+
 // ensureBoundObject create-or-updates one namespaced policy object (ResourceQuota
 // / LimitRange) so the tenant's declared bounds always match the operator's
 // current config: it Creates when absent and merge-patches .spec when present

clients/platform/platform.go

@@ -67,15 +67,16 @@ type EnvVarJSON struct {
 }
 
 type svc struct {
-	store     *Store
-	k8s       *k8sClient
-	cancel    context.CancelFunc // stops the build reconciler on Shutdown
-	log       luxlog.Logger
-	brand     string
-	env       string
-	domain    string
-	sitesHost string   // per-tenant apps host suffix; a custom domain must be under <org>.<sitesHost>
-	appLock   appMutex // per-app serialization of apply-CR→finalize-live (applylive.go, RED LOW-1)
+	store      *Store
+	k8s        *k8sClient
+	cancel     context.CancelFunc // stops the build reconciler on Shutdown
+	log        luxlog.Logger
+	brand      string
+	env        string
+	domain     string
+	sitesHost  string       // per-tenant apps host suffix; a custom domain must be under <org>.<sitesHost>
+	appLock    appMutex     // per-app serialization of apply-CR→finalize-live (applylive.go, RED LOW-1)
+	deployGate inflightGate // per-org in-flight synchronous-deploy cap (deploy.go, RED LOW L1)
 }
 
 // mounted is the active service so Shutdown can release the store.

clients/platform/reconcile.go

@@ -62,7 +62,13 @@ func (s *svc) reconcileBuilds(ctx context.Context) {
 }
 
 // reconcileBuild advances one "building" deployment: waits for its Job, then on
-// success applies the Service CR, on failure/deadline records the honest error.
+// success (once the tenant's operator RBAC is ready) applies the Service CR. Every
+// TRANSIENT condition — cluster briefly unreachable, Job not yet finished, or the
+// tenant's RoleBinding still provisioning (errTenantProvisioning) — leaves the
+// deployment "building" and re-drives on the next tick; only a genuine build failure
+// or the elapsed deadline records an honest terminal error. Crucially, a slow tenant
+// onboarding is NOT failed permanently (there is no client to retry a git build) and
+// is NOT waited on in-line (which would head-of-line-block other orgs' go-lives).
 func (s *svc) reconcileBuild(ctx context.Context, d Deployment) {
 	if d.Source != "git" || d.BuildID == "" {
 		return // only git builds pass through "building"
@@ -109,6 +115,30 @@ func (s *svc) reconcileBuild(ctx context.Context, d Deployment) {
 		s.log.Warn("reconcile: get project", "org", d.Org, "dep", d.ID, "err", err)
 		return
 	}
+
+	// The tenant namespace must exist and its operator RBAC must have landed before
+	// the Service CR can be written. Unlike the synchronous image path (which BLOCKS
+	// up to ~45s so a single client deploy succeeds), the reconciler does ONE
+	// non-blocking readiness probe (creating the namespace if absent, which is what
+	// triggers the operator to project cloud-api's RoleBinding). If RBAC has not yet
+	// landed, leave the deployment "building" and re-drive next tick — a slow/wedged
+	// tenant onboarding then never HEAD-OF-LINE-BLOCKS other orgs' go-lives and never
+	// permanently fails the git build. Give up only once the build deadline elapses.
+	ns := tenantNamespace(d.Org)
+	ready, provErr := s.k8s.ensureTenantReady(ctx, ns, d.Org)
+	if provErr != nil {
+		if overdue {
+			s.failBuild(ctx, d, b, "prepare tenant namespace past deadline: "+provErr.Error())
+		}
+		return // transient cluster error — retry next tick
+	}
+	if !ready {
+		if overdue {
+			s.failBuild(ctx, d, b, "tenant RBAC still provisioning past deadline")
+		}
+		return // stay "building"; the operator's RoleBinding lands before the next tick
+	}
+
 	// Build succeeded — write the operator Service CR and advance the app to live
 	// as ONE per-app-serialized, version-monotonic step (shared with deployImage).
 	// Under a per-app lock, applyLive re-checks supersession against the current
@@ -126,6 +156,13 @@ func (s *svc) reconcileBuild(ctx context.Context, d Deployment) {
 		return
 	}
 	if err != nil {
+		// Defense in depth: the pre-check above confirmed RBAC was ready, but if the
+		// apply still surfaced a provisioning error (a readiness flap between probe and
+		// write), treat it as TRANSIENT — leave the deployment "building" and re-drive
+		// next tick, never a permanent fail — unless the deadline has already elapsed.
+		if errors.Is(err, errTenantProvisioning) && !overdue {
+			return
+		}
 		s.failBuild(ctx, d, b, "apply Service CR: "+err.Error())
 		return
 	}

clients/platform/reconcile_test.go

@@ -2,11 +2,15 @@ package platform
 
 import (
 	"context"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	luxlog "github.com/luxfi/log"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	dynamicfake "k8s.io/client-go/dynamic/fake"
 )
 
 // TestJobOutcome locks the ONE terminal-state classifier the build cap and the
@@ -144,6 +148,94 @@ func TestFinalizeLiveIsMonotonic(t *testing.T) {
 	}
 }
 
+// TestBuildReconcilerRetriesUntilTenantRBACReady is the L2 regression guard: a git
+// build whose Job SUCCEEDED but whose tenant is still cold (operator RBAC not yet
+// projected) must NOT be permanently failed and must NOT block in-line — the
+// reconciler probes readiness once, leaves the deployment "building", and on a later
+// tick (once the RoleBinding lands) reconciles it to live. This is the exact
+// terminal-on-transient + head-of-line-block bug RED flagged: a fresh tenant whose
+// RBAC lags could not self-heal because there is no client to retry a git build.
+func TestBuildReconcilerRetriesUntilTenantRBACReady(t *testing.T) {
+	ctx := context.Background()
+	store := newTestStore(t)
+	k := fakeK8s()
+	fastRBAC(k) // shrink timings; the reconciler probe path must not block regardless
+	// Gate RBAC readiness behind a flag the test flips between ticks. A prepended
+	// reactor wins over fakeK8s's default-allow, so this fully controls readiness.
+	fake := k.dyn.(*dynamicfake.FakeDynamicClient)
+	var rbacReady atomic.Bool
+	allowSSAR(fake, func() bool { return rbacReady.Load() })
+	s := &svc{store: store, k8s: k, log: luxlog.New("test")}
+
+	_ = store.CreateProject(ctx, mkProject("acme", "web", "Web"))
+	proj, _ := store.GetProject(ctx, "acme", "web")
+	app := mkApp("acme", proj.ID, "api")
+	app.Source = "git"
+	_ = store.CreateApplication(ctx, app)
+
+	// Seed a git build whose Job has SUCCEEDED, so the ONLY thing gating go-live is
+	// the tenant RBAC. CreatedAt is NOW so the build is not overdue (else the
+	// not-ready path would honestly fail at the deadline instead of retrying).
+	now := time.Now().Unix()
+	depID, bldID, job := "dep_1", "bld_1", "pf-build-acme-api-1"
+	img := "ghcr.io/hanzoai/tenant-acme/api:main"
+	if err := store.InsertBuild(ctx, Build{ID: bldID, Org: "acme", ApplicationID: app.ID, DeploymentID: depID, Status: "building", Image: img, JobName: job, CreatedAt: now, UpdatedAt: now}); err != nil {
+		t.Fatalf("seed build: %v", err)
+	}
+	if err := store.InsertDeployment(ctx, Deployment{ID: depID, Org: "acme", ApplicationID: app.ID, Version: 1, Status: "building", Source: "git", Image: img, BuildID: bldID, CreatedAt: now, UpdatedAt: now}); err != nil {
+		t.Fatalf("seed deployment: %v", err)
+	}
+	jobObj := &unstructured.Unstructured{Object: map[string]any{
+		"apiVersion": "batch/v1", "kind": "Job",
+		"metadata": map[string]any{"name": job, "namespace": k.buildNS},
+		"status":   map[string]any{"succeeded": int64(1)},
+	}}
+	if _, err := k.dyn.Resource(jobsGVR).Namespace(k.buildNS).Create(ctx, jobObj, metav1.CreateOptions{}); err != nil {
+		t.Fatalf("seed job: %v", err)
+	}
+	get := func() Deployment { d, _ := store.GetDeployment(ctx, "acme", app.ID, depID); return d }
+	ns := tenantNamespace("acme")
+
+	// Tick 1 — tenant RBAC NOT ready. The deployment must stay "building" (NOT failed),
+	// the build must NOT be failed, and no Service CR may exist yet…
+	rbacReady.Store(false)
+	s.reconcileBuild(ctx, get())
+	if d := get(); d.Status != "building" {
+		t.Fatalf("tick 1 (RBAC pending) must leave the deployment 'building', got %q (msg=%q)", d.Status, d.Message)
+	}
+	if b, _ := store.GetBuild(ctx, "acme", bldID); b.Status == "failed" {
+		t.Fatal("a transient RBAC-pending must NOT permanently fail the build")
+	}
+	if _, err := k.dyn.Resource(servicesGVR).Namespace(ns).Get(ctx, app.Slug, metav1.GetOptions{}); !apierrors.IsNotFound(err) {
+		t.Fatalf("no Service CR may be written while RBAC is pending, get err=%v", err)
+	}
+	// …but the namespace WAS created (that create is what triggers the operator to
+	// project cloud-api's RoleBinding), so readiness can ever become true.
+	if _, err := k.dyn.Resource(namespacesGVR).Get(ctx, ns, metav1.GetOptions{}); err != nil {
+		t.Fatalf("tick 1 must create the tenant namespace (triggers operator RBAC): %v", err)
+	}
+
+	// Tick 2 — the operator's RoleBinding has landed → the build reconciles to live.
+	rbacReady.Store(true)
+	s.reconcileBuild(ctx, get())
+	if d := get(); d.Status != "deploying" {
+		t.Fatalf("tick 2 (RBAC ready) must advance the deployment to 'deploying', got %q (msg=%q)", d.Status, d.Message)
+	}
+	if a, _ := store.GetApplicationByID(ctx, "acme", app.ID); a.Status != "live" || a.CurrentDeploy != depID || a.ImageTag != "main" {
+		t.Fatalf("app must be live@%s (tag main), got status=%s current=%s tag=%s", depID, a.Status, a.CurrentDeploy, a.ImageTag)
+	}
+	if b, _ := store.GetBuild(ctx, "acme", bldID); b.Status != "succeeded" {
+		t.Fatalf("build must be 'succeeded' after go-live, got %q", b.Status)
+	}
+	obj, err := k.dyn.Resource(servicesGVR).Namespace(ns).Get(ctx, app.Slug, metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("Service CR must be written once RBAC is ready: %v", err)
+	}
+	if tag, _, _ := unstructured.NestedString(obj.Object, "spec", "image", "tag"); tag != "main" {
+		t.Fatalf("live Service CR image tag want 'main', got %q", tag)
+	}
+}
+
 // TestBuildReconcilerVersionMonotonic is the end-to-end MED-1 guard: two git
 // builds for the SAME app whose Jobs finish OUT OF ORDER. The newer build (v2)
 // completes first and goes live; the older build (v1) then finishes LATE and must

clients/platform/tenant_rbac_test.go

@@ -2,6 +2,7 @@ package platform
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"net/http"
 	"sync/atomic"
@@ -143,3 +144,50 @@ func TestExistingTenantDeployHasNoWait(t *testing.T) {
 		t.Fatalf("ready-tenant probe must not sleep, took %s", elapsed)
 	}
 }
+
+// TestFreshOrgDeployFailsClosedOn503NotCreate502 confirms the I2 fresh-org path at
+// the HTTP layer: a deploy into a genuinely fresh org (whose namespace does NOT
+// exist yet) CREATES the namespace — the trigger for the operator's async RoleBinding
+// — and then, when RBAC has not yet landed, fails closed with a graceful, RETRYABLE
+// 503, NOT a raw 502 at namespace-Create. The presence of the created namespace
+// proves the 503 means "RBAC pending", never "could not create the namespace".
+func TestFreshOrgDeployFailsClosedOn503NotCreate502(t *testing.T) {
+	k := fakeK8s()
+	k.rbacReadyTimeout = 30 * time.Millisecond
+	k.rbacPollInitial = 5 * time.Millisecond
+	fake := k.dyn.(*dynamicfake.FakeDynamicClient)
+	allowSSAR(fake, func() bool { return false }) // RBAC never lands
+	app := mountAppK8s(t, k)
+
+	do(t, app, http.MethodPost, "/v1/platform/projects", "freshco", map[string]any{"name": "web"})
+	do(t, app, http.MethodPost, "/v1/platform/projects/web/apps", "freshco", map[string]any{
+		"name": "api", "source": "image",
+		"image": map[string]any{"repository": "ghcr.io/hanzoai/nginx", "tag": "1.27"},
+	})
+
+	// Precondition: the tenant namespace does NOT exist (genuinely fresh org).
+	ns := tenantNamespace("freshco")
+	if _, err := k.dyn.Resource(namespacesGVR).Get(context.Background(), ns, metav1.GetOptions{}); !apierrors.IsNotFound(err) {
+		t.Fatalf("precondition: fresh org namespace must not exist yet, get err=%v", err)
+	}
+
+	code, body := do(t, app, http.MethodPost, "/v1/platform/projects/web/apps/api/deploy", "freshco", map[string]any{"tag": "1.27"})
+	if code != http.StatusServiceUnavailable {
+		t.Fatalf("fresh-org deploy (RBAC pending) want 503, got %d (%s)", code, body)
+	}
+	// The namespace WAS created — so the 503 is RBAC-pending, not a Create failure (502).
+	if _, err := k.dyn.Resource(namespacesGVR).Get(context.Background(), ns, metav1.GetOptions{}); err != nil {
+		t.Fatalf("fresh-org deploy must CREATE the namespace (503 = RBAC pending, not create-502): %v", err)
+	}
+	// No Service CR was written past the readiness gate (fail-closed).
+	if _, err := k.dyn.Resource(servicesGVR).Namespace(ns).Get(context.Background(), "api", metav1.GetOptions{}); !apierrors.IsNotFound(err) {
+		t.Fatalf("no Service CR may exist when RBAC never lands, get err=%v", err)
+	}
+	// The failed attempt is recorded honestly as an 'error' deployment (not fabricated).
+	code, listBody := do(t, app, http.MethodGet, "/v1/platform/projects/web/apps/api/deployments", "freshco", nil)
+	var deps []deploymentView
+	_ = json.Unmarshal(listBody, &deps)
+	if code != http.StatusOK || len(deps) != 1 || deps[0].Status != "error" {
+		t.Fatalf("fresh-org 503 must record one honest 'error' deployment, got code=%d deps=%+v", code, deps)
+	}
+}

clients/platform/validate.go

@@ -206,6 +206,7 @@ const (
 type resourceLimits struct {
 	maxReplicas     int    // per-app replica ceiling
 	maxBuilds       int    // concurrent build Jobs per org (shared build ns)
+	maxDeploys      int    // concurrent in-flight SYNCHRONOUS image deploys per org (L1)
 	quotaCPU        string // ResourceQuota: total requests.cpu / limits.cpu
 	quotaMemory     string // ResourceQuota: total requests.memory / limits.memory
 	quotaPods       string // ResourceQuota: total pods
@@ -225,6 +226,7 @@ func newResourceLimits() resourceLimits {
 	return resourceLimits{
 		maxReplicas:     atoiDefault(getenv("CLOUD_PLATFORM_MAX_REPLICAS", ""), defaultMaxReplicas),
 		maxBuilds:       atoiDefault(getenv("CLOUD_PLATFORM_MAX_CONCURRENT_BUILDS", ""), defaultMaxBuilds),
+		maxDeploys:      atoiDefault(getenv("CLOUD_PLATFORM_MAX_CONCURRENT_DEPLOYS", ""), defaultMaxDeploys),
 		quotaCPU:        getenv("CLOUD_PLATFORM_QUOTA_CPU", "20"),
 		quotaMemory:     getenv("CLOUD_PLATFORM_QUOTA_MEMORY", "40Gi"),
 		quotaPods:       getenv("CLOUD_PLATFORM_QUOTA_PODS", "50"),
@@ -244,6 +246,12 @@ func newResourceLimits() resourceLimits {
 const (
 	defaultMaxReplicas = 20
 	defaultMaxBuilds   = 3
+	// defaultMaxDeploys bounds concurrent SYNCHRONOUS image deploys per org (L1).
+	// Generous enough never to bite a legitimate bursty redeploy, yet bounds the
+	// request goroutines that a wedged operator (tenant RBAC never landing) could
+	// otherwise let one org pile up in waitForTenantRBAC's ~45s wait. Fail-secure:
+	// an unset ceiling falls back here, never to zero or unlimited.
+	defaultMaxDeploys = 8
 )
 
 // clampReplicas bounds a requested replica count to [1, maxReplicas]. A request
@@ -274,6 +282,16 @@ func (r resourceLimits) maxConcurrentBuilds() int {
 	return r.maxBuilds
 }
 
+// maxConcurrentDeploys is the per-org in-flight SYNCHRONOUS image-deploy ceiling
+// (L1), falling back to the safe default when unset (fail-secure). It mirrors
+// maxConcurrentBuilds for the image path, which has no build Job to count.
+func (r resourceLimits) maxConcurrentDeploys() int {
+	if r.maxDeploys <= 0 {
+		return defaultMaxDeploys
+	}
+	return r.maxDeploys
+}
+
 // resourceQuota renders the namespaced ResourceQuota that caps a tenant's TOTAL
 // scheduled footprint. Applied idempotently by ensureNamespace.
 func (r resourceLimits) resourceQuota(ns string) *unstructured.Unstructured {