diff --git a/.golangci.yml b/.golangci.yml
index 5a83db31..246ba3c1 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -31,6 +31,16 @@ linters:
         linters:
           - errcheck
           - gocyclo
+      # SA5011 (possible nil pointer dereference) fires in test files on the
+      # idiomatic `got := f(); if got == nil { t.Fatalf/Errorf(...) }; got.X`
+      # pattern — a nil deref in a test panics and fails the test loudly, so
+      # this is benign test noise, not a production-safety signal. Suppressed
+      # by-check (SA5011) in tests only; production SA5011 still fails the build.
+      # Scoped here rather than disabling staticcheck for tests wholesale.
+      - path: _test\.go
+        linters:
+          - staticcheck
+        text: "SA5011"
       # SA1019 (deprecated madmin.New / SetPolicy) in the MinIO provider. MinIO is a
       # local-dev-only object-store backend; the suggested replacements
       # (NewWithOptions / AttachPolicy) have different call shapes and semantics, so a
diff --git a/internal/config/config.go b/internal/config/config.go
index fef0f2e9..b6d3976a 100644
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -174,6 +174,14 @@ type Config struct {
 	// and fail token validation (deterministic disable for rollback).
 	FamilyBindingsEnabled bool
 
+	// DeploySourceImageEnabled gates the P2 multi-source "source=image" path
+	// (deploy a prebuilt image instead of uploading source). Default FALSE:
+	// the skip-Kaniko compute branch changes the live deploy path and can't be
+	// validated in CI (no real cluster), so it stays off until the operator
+	// flips DEPLOY_SOURCE_IMAGE_ENABLED=true after a canary. Off → /deploy/new
+	// rejects source=image with 501; tarball deploys are unaffected.
+	DeploySourceImageEnabled bool
+
 	// Email-feedback webhook secrets. Each provider authenticates its
 	// callbacks differently — these env vars give the handler the shared
 	// secret (Brevo, SendGrid) or topic ARN (SES via SNS) it needs to
@@ -421,6 +429,14 @@ func Load() *Config {
 		cfg.FamilyBindingsEnabled = true
 	}
 
+	// DEPLOY_SOURCE_IMAGE_ENABLED: default FALSE (off until operator canary).
+	switch strings.ToLower(strings.TrimSpace(os.Getenv("DEPLOY_SOURCE_IMAGE_ENABLED"))) {
+	case "true", "1", "yes":
+		cfg.DeploySourceImageEnabled = true
+	default:
+		cfg.DeploySourceImageEnabled = false
+	}
+
 	if len(cfg.JWTSecret) < 32 {
 		panic("JWT_SECRET must be at least 32 bytes")
 	}
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
index 1e736094..4228b134 100644
--- a/internal/config/config_test.go
+++ b/internal/config/config_test.go
@@ -62,6 +62,7 @@ func allKeys() []string {
 		"DEPLOY_DOMAIN", "COMPUTE_PROVIDER", "KUBE_NAMESPACE_APPS",
 		"METRICS_TOKEN", "DASHBOARD_BASE_URL", "API_PUBLIC_URL",
 		"DELETION_CONFIRMATION_TTL_MINUTES", "FAMILY_BINDINGS_ENABLED",
+		"DEPLOY_SOURCE_IMAGE_ENABLED",
 		"BREVO_WEBHOOK_SECRET", "SES_SNS_SUBSCRIPTION_ARN",
 		"SENDGRID_WEBHOOK_PUBLIC_KEY",
 		"WORKER_INTERNAL_JWT_SECRET", "ADMIN_PATH_PREFIX",
@@ -241,6 +242,9 @@ func TestLoad_HappyPath_AppliesDefaults(t *testing.T) {
 	if !cfg.FamilyBindingsEnabled {
 		t.Error("FamilyBindingsEnabled default must be true")
 	}
+	if cfg.DeploySourceImageEnabled {
+		t.Error("DeploySourceImageEnabled default must be false (off until operator canary)")
+	}
 	// Object store mode resolution: with everything empty → "admin" / "minio-admin"
 	if cfg.ObjectStoreMode != "admin" || cfg.ObjectStoreBackend != "minio-admin" {
 		t.Errorf("ObjectStoreMode/Backend defaults: %q/%q", cfg.ObjectStoreMode, cfg.ObjectStoreBackend)
@@ -330,6 +334,22 @@ func TestLoad_FamilyBindingsDisabled(t *testing.T) {
 	}
 }
 
+func TestLoad_DeploySourceImageEnabled(t *testing.T) {
+	for _, val := range []string{"true", "1", "yes", "TRUE", "  Yes  "} {
+		applyBaselineEnv(t, map[string]string{"DEPLOY_SOURCE_IMAGE_ENABLED": val})
+		if !Load().DeploySourceImageEnabled {
+			t.Errorf("DEPLOY_SOURCE_IMAGE_ENABLED=%q should enable", val)
+		}
+	}
+	// Unrecognized / off values → stays disabled.
+	for _, val := range []string{"false", "0", "no", "maybe", ""} {
+		applyBaselineEnv(t, map[string]string{"DEPLOY_SOURCE_IMAGE_ENABLED": val})
+		if Load().DeploySourceImageEnabled {
+			t.Errorf("DEPLOY_SOURCE_IMAGE_ENABLED=%q should stay disabled", val)
+		}
+	}
+}
+
 func TestLoad_DeletionTTL_OverrideAndInvalid(t *testing.T) {
 	applyBaselineEnv(t, map[string]string{"DELETION_CONFIRMATION_TTL_MINUTES": "30"})
 	if got := Load().DeletionConfirmationTTLMinutes; got != 30 {
diff --git a/internal/db/migrations/064_deploy_source_image.sql b/internal/db/migrations/064_deploy_source_image.sql
new file mode 100644
index 00000000..b4db5604
--- /dev/null
+++ b/internal/db/migrations/064_deploy_source_image.sql
@@ -0,0 +1,34 @@
+-- 064_deploy_source_image.sql — multi-source deploys, P2 (source=image / BYO image).
+--
+-- WHY: P1 (#220) capped /deploy/new uploads at 10 MB and nudges large projects
+-- to "deploy a prebuilt image instead of uploading source". This migration adds
+-- the columns that let a deployment record where its image came from, so a
+-- caller can skip the tarball+Kaniko build entirely and have the platform
+-- deploy a prebuilt image (e.g. one their GitHub CI already pushed to GHCR).
+--
+-- All columns are ADDITIVE with safe defaults, so every existing row + every
+-- existing tarball deploy keeps working unchanged:
+--
+--   source            — 'tarball' (default, the only mode before P2) | 'image'.
+--                       Future P3 adds 'git'. CHECK keeps it to known modes.
+--   image_ref         — fully-qualified registry ref (host/path[:tag][@digest])
+--                       for source='image'; '' for tarball deploys.
+--   registry_creds_enc — AES-256-GCM ciphertext of the optional pull-only
+--                       registry credential JSON (whole-object encryption,
+--                       same posture as notify_webhook_secret). '' when the
+--                       image is public / uses the platform's ghcr-pull secret.
+--                       NEVER returned to the client (deploymentToMap emits
+--                       only registry_creds_set: bool).
+
+ALTER TABLE deployments
+    ADD COLUMN IF NOT EXISTS source             TEXT NOT NULL DEFAULT 'tarball',
+    ADD COLUMN IF NOT EXISTS image_ref          TEXT NOT NULL DEFAULT '',
+    ADD COLUMN IF NOT EXISTS registry_creds_enc TEXT NOT NULL DEFAULT '';
+
+-- Constrain source to the modes the code understands. 'git' is reserved for P3
+-- so adding it here now is forward-safe and avoids a follow-up migration.
+ALTER TABLE deployments
+    DROP CONSTRAINT IF EXISTS deployments_source_check;
+ALTER TABLE deployments
+    ADD CONSTRAINT deployments_source_check
+    CHECK (source IN ('tarball', 'image', 'git'));
diff --git a/internal/handlers/deploy.go b/internal/handlers/deploy.go
index fc5f0944..489b5e80 100644
--- a/internal/handlers/deploy.go
+++ b/internal/handlers/deploy.go
@@ -34,6 +34,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/redis/go-redis/v9"
 	"instant.dev/internal/config"
+	"instant.dev/internal/crypto"
 	"instant.dev/internal/email"
 	"instant.dev/internal/metrics"
 	"instant.dev/internal/middleware"
@@ -66,6 +67,87 @@ const maxTarballBytes = 10 << 20
 // upload or deploy a prebuilt image. Shared by /deploy/new and its
 // redeploy=true branch so the cap can never drift between the two. Returns
 // nil when the upload is within the cap.
+// validateImageRef checks a source=image reference is fully qualified
+// (host/path[:tag][@digest]). We REJECT bare names ("nginx", "owner/repo")
+// that Docker would silently resolve against Docker Hub — a deploy must name
+// its registry explicitly so there's no ambiguity about what gets pulled and
+// run on shared nodes. Returns the trimmed, validated ref.
+func validateImageRef(ref string) (string, error) {
+	ref = strings.TrimSpace(ref)
+	if ref == "" {
+		return "", fmt.Errorf("image_ref is required for source=image (e.g. ghcr.io/owner/app:tag)")
+	}
+	if len(ref) > 512 {
+		return "", fmt.Errorf("image_ref is too long")
+	}
+	for _, r := range ref {
+		if r <= ' ' || r > '~' {
+			return "", fmt.Errorf("image_ref contains invalid characters")
+		}
+	}
+	slash := strings.IndexByte(ref, '/')
+	if slash < 0 {
+		return "", fmt.Errorf("image_ref must include a registry host (e.g. ghcr.io/owner/app:tag), not a bare name")
+	}
+	host := ref[:slash]
+	// A registry host has a dot, a port, or is localhost. "owner/repo" (Docker
+	// Hub implicit namespace) is rejected — name docker.io explicitly.
+	if host != "localhost" && !strings.ContainsAny(host, ".:") {
+		return "", fmt.Errorf("image_ref must start with a registry host (e.g. ghcr.io/...), got %q", host)
+	}
+	return ref, nil
+}
+
+// applyImageSourceOpts populates DeployOptions for a source=image deployment
+// from the persisted row: sets Source/ImageRef, clears Tarball, and decrypts
+// the optional BYO registry creds into RegistryAuth. A decrypt failure (or bad
+// key) is logged and falls back to the platform pull secret — public/GHCR
+// images still deploy. No-op for non-image sources.
+func applyImageSourceOpts(opts *compute.DeployOptions, d *models.Deployment, aesKeyHex string) {
+	if d.Source != "image" {
+		return
+	}
+	opts.Source = "image"
+	opts.ImageRef = d.ImageRef
+	opts.Tarball = nil
+	if d.RegistryCredsEnc == "" {
+		return
+	}
+	key, kerr := crypto.ParseAESKey(aesKeyHex)
+	if kerr != nil {
+		slog.Error("deploy.image.aes_key_invalid", "app_id", d.AppID, "error", kerr)
+		return
+	}
+	plain, derr := crypto.Decrypt(key, d.RegistryCredsEnc)
+	if derr != nil {
+		slog.Error("deploy.run_deploy.registry_creds_decrypt_failed", "app_id", d.AppID, "error", derr)
+		return
+	}
+	opts.RegistryAuth = plain
+}
+
+// encryptRegistryCreds AES-256-GCM-encrypts a BYO private-registry docker
+// config JSON for at-rest storage. ParseAESKey is the only failure mode worth
+// a distinct branch (a misconfigured/short AES_KEY); crypto.Encrypt over a
+// valid key does not fail for these inputs, so its error is returned verbatim
+// without a separate handler branch. Returns the ciphertext for persistence.
+func encryptRegistryCreds(aesKeyHex, plaintext string) (string, error) {
+	key, err := crypto.ParseAESKey(aesKeyHex)
+	if err != nil {
+		return "", err
+	}
+	return crypto.Encrypt(key, plaintext)
+}
+
+// deploymentSourceOrDefault normalises an empty source (legacy in-memory rows)
+// to the 'tarball' default the migration applies at the DB layer.
+func deploymentSourceOrDefault(s string) string {
+	if s == "" {
+		return "tarball"
+	}
+	return s
+}
+
 func enforceTarballCap(c *fiber.Ctx, fh *multipart.FileHeader) error {
 	if fh.Size > maxTarballBytes {
 		return respondError(c, fiber.StatusRequestEntityTooLarge, "tarball_too_large",
@@ -307,6 +389,14 @@ func deploymentToMapWithDB(d *models.Deployment, db *sql.DB) fiber.Map {
 		// secret is NEVER returned — only its lifecycle metadata.
 		"notify_webhook": d.NotifyWebhook,
 		"notify_state":   d.NotifyState,
+		// Multi-source deploys (migration 064). source defaults to 'tarball'.
+		// image_ref is echoed (caller-supplied, no secret); registry_creds is
+		// NEVER returned — only registry_creds_set lifecycle metadata.
+		"source": deploymentSourceOrDefault(d.Source),
+	}
+	if d.Source == "image" {
+		m["image_ref"] = d.ImageRef
+		m["registry_creds_set"] = d.RegistryCredsEnc != ""
 	}
 	if d.NotifyWebhook != "" {
 		m["notify_attempts"] = d.NotifyAttempts
@@ -528,6 +618,9 @@ func (h *DeployHandler) runDeploy(d *models.Deployment, tarball []byte) {
 		Private:    d.Private,
 		AllowedIPs: d.AllowedIPs,
 	}
+	// Multi-source (migration 064): a source=image deploy carries no tarball —
+	// the compute layer deploys d.ImageRef directly (skip Kaniko).
+	applyImageSourceOpts(&opts, d, h.cfg.AESKey)
 	result, err := h.compute.Deploy(ctx, opts)
 	if err != nil {
 		slog.Error("deploy.run_deploy.failed",
@@ -599,30 +692,74 @@ func (h *DeployHandler) New(c *fiber.Ctx) error {
 			"Request must be multipart/form-data (max 50 MB)")
 	}
 
-	// tarball field.
-	tarballs := form.File["tarball"]
-	if len(tarballs) == 0 {
-		return respondError(c, fiber.StatusBadRequest, "missing_tarball",
-			"Multipart field 'tarball' is required")
+	// Source mode (migration 064): "tarball" (default — build the uploaded
+	// source with Kaniko) or "image" (BYO — deploy a prebuilt image_ref
+	// directly, no upload/build). source=image is naturally Hobby+ (the
+	// per-tier deploy cap below gates anon/free, which have deployments_apps=0).
+	source := "tarball"
+	if vals := form.Value["source"]; len(vals) > 0 && strings.TrimSpace(vals[0]) != "" {
+		source = strings.ToLower(strings.TrimSpace(vals[0]))
 	}
-	fh := tarballs[0]
-	if err := enforceTarballCap(c, fh); err != nil {
-		return err
-	}
-	f, err := openMultipartFile(fh)
-	if err != nil {
-		return respondError(c, fiber.StatusBadRequest, "tarball_open_failed",
-			"Failed to read tarball")
-	}
-	defer func() { _ = f.Close() }()
 
-	// P0-3: io.ReadAll, not a single f.Read — a lone Read short-reads on
-	// disk-spilled multipart files (n is discarded), truncating large tarballs.
-	// Mirrors how stack.go reads its tarball field.
-	tarball, err := io.ReadAll(f)
-	if err != nil {
-		return respondError(c, fiber.StatusBadRequest, "tarball_read_failed",
-			"Failed to read tarball bytes")
+	var tarball []byte
+	var imageRef, registryCredsEnc string
+
+	switch source {
+	case "image":
+		// Flag-gated (P2): the skip-Kaniko deploy path is off until an operator
+		// enables it post-canary (DEPLOY_SOURCE_IMAGE_ENABLED=true). Until then
+		// reject cleanly so tarball deploys are never affected by unproven code.
+		if !h.cfg.DeploySourceImageEnabled {
+			return respondError(c, fiber.StatusNotImplemented, "source_image_disabled",
+				"Deploying from a prebuilt image (source=image) is rolling out and not yet enabled. Upload source (tarball ≤10MB) for now.")
+		}
+		raw := ""
+		if vals := form.Value["image_ref"]; len(vals) > 0 {
+			raw = strings.TrimSpace(vals[0])
+		}
+		validRef, refErr := validateImageRef(raw)
+		if refErr != nil {
+			return respondError(c, fiber.StatusBadRequest, "invalid_image_ref", refErr.Error())
+		}
+		imageRef = validRef
+		// Optional BYO registry creds for a PRIVATE image: a complete docker
+		// config JSON, encrypted at rest (AES-256-GCM, like notify_webhook_secret)
+		// and never echoed back. Absent → the platform ghcr-pull secret is used.
+		if vals := form.Value["registry_creds"]; len(vals) > 0 && strings.TrimSpace(vals[0]) != "" {
+			enc, encErr := encryptRegistryCreds(h.cfg.AESKey, strings.TrimSpace(vals[0]))
+			if encErr != nil {
+				return respondError(c, fiber.StatusServiceUnavailable, "encrypt_failed",
+					"Could not secure registry credentials")
+			}
+			registryCredsEnc = enc
+		}
+	case "tarball":
+		tarballs := form.File["tarball"]
+		if len(tarballs) == 0 {
+			return respondError(c, fiber.StatusBadRequest, "missing_tarball",
+				"Multipart field 'tarball' is required (or use source=image with image_ref)")
+		}
+		fh := tarballs[0]
+		if err := enforceTarballCap(c, fh); err != nil {
+			return err
+		}
+		f, ferr := openMultipartFile(fh)
+		if ferr != nil {
+			return respondError(c, fiber.StatusBadRequest, "tarball_open_failed",
+				"Failed to read tarball")
+		}
+		defer func() { _ = f.Close() }()
+		// P0-3: io.ReadAll, not a single f.Read — a lone Read short-reads on
+		// disk-spilled multipart files (n is discarded), truncating large tarballs.
+		b, rerr := io.ReadAll(f)
+		if rerr != nil {
+			return respondError(c, fiber.StatusBadRequest, "tarball_read_failed",
+				"Failed to read tarball bytes")
+		}
+		tarball = b
+	default:
+		return respondError(c, fiber.StatusBadRequest, "invalid_source",
+			"Field 'source' must be 'tarball' (default) or 'image'")
 	}
 
 	// Required name field — the human-readable deployment label.
@@ -951,6 +1088,9 @@ func (h *DeployHandler) New(c *fiber.Ctx) error {
 		NotifyWebhook:       notifyURL,
 		NotifyWebhookSecret: notifySecret,
 		TTLPolicy:           ttlPolicy,
+		Source:              source,
+		ImageRef:            imageRef,
+		RegistryCredsEnc:    registryCredsEnc,
 	})
 	if errors.Is(err, models.ErrDeploymentCapReached) {
 		// Over the per-tier cap — surfaced atomically inside the
diff --git a/internal/handlers/deploy_image_source_test.go b/internal/handlers/deploy_image_source_test.go
new file mode 100644
index 00000000..f198c768
--- /dev/null
+++ b/internal/handlers/deploy_image_source_test.go
@@ -0,0 +1,148 @@
+package handlers
+
+// deploy_image_source_test.go — unit tests for P2 source=image validation.
+
+import (
+	"strings"
+	"testing"
+
+	"instant.dev/internal/crypto"
+	"instant.dev/internal/models"
+	"instant.dev/internal/providers/compute"
+)
+
+func TestValidateImageRef(t *testing.T) {
+	ok := []string{
+		"ghcr.io/owner/app:v1",
+		"ghcr.io/owner/app@sha256:abc",
+		"docker.io/library/nginx:latest",
+		"registry.example.com:5000/team/app:tag",
+		"localhost/dev/app:1",
+	}
+	for _, r := range ok {
+		if got, err := validateImageRef(r); err != nil || got != r {
+			t.Errorf("validateImageRef(%q) = (%q,%v), want (%q,nil)", r, got, err, r)
+		}
+	}
+	bad := map[string]string{
+		"":                  "empty",
+		"nginx":             "bare name (no host)",
+		"owner/repo:tag":    "Docker Hub implicit namespace (no explicit host)",
+		"owner/repo":        "no host, no tag",
+		"ghcr.io/o/a:t ag":  "whitespace",
+	}
+	for ref, why := range bad {
+		if _, err := validateImageRef(ref); err == nil {
+			t.Errorf("validateImageRef(%q) should fail (%s) but passed", ref, why)
+		}
+	}
+}
+
+func TestDeploymentSourceOrDefault(t *testing.T) {
+	if deploymentSourceOrDefault("") != "tarball" {
+		t.Error("empty source must normalise to tarball")
+	}
+	if deploymentSourceOrDefault("image") != "image" {
+		t.Error("explicit source must pass through")
+	}
+}
+
+func TestDeploymentToMap_ImageSource(t *testing.T) {
+	img := &models.Deployment{Source: "image", ImageRef: "ghcr.io/o/a:1", RegistryCredsEnc: "ciphertext", EnvVars: map[string]string{}}
+	m := deploymentToMap(img)
+	if m["source"] != "image" {
+		t.Errorf("source: got %v want image", m["source"])
+	}
+	if m["image_ref"] != "ghcr.io/o/a:1" {
+		t.Errorf("image_ref: got %v", m["image_ref"])
+	}
+	if m["registry_creds_set"] != true {
+		t.Errorf("registry_creds_set: got %v want true", m["registry_creds_set"])
+	}
+	// creds must NEVER be echoed
+	if _, leaked := m["registry_creds"]; leaked {
+		t.Error("registry_creds must never appear in the response map")
+	}
+
+	// tarball deploy: source normalises, no image_ref/registry_creds_set keys.
+	tb := &models.Deployment{Source: "tarball", EnvVars: map[string]string{}}
+	m2 := deploymentToMap(tb)
+	if m2["source"] != "tarball" {
+		t.Errorf("tarball source: got %v", m2["source"])
+	}
+	if _, ok := m2["image_ref"]; ok {
+		t.Error("tarball deploy must not emit image_ref")
+	}
+	if _, ok := m2["registry_creds_set"]; ok {
+		t.Error("tarball deploy must not emit registry_creds_set")
+	}
+}
+
+func TestValidateImageRef_TooLong(t *testing.T) {
+	long := "ghcr.io/o/" + strings.Repeat("a", 600) // valid host, but >512 total
+	if _, err := validateImageRef(long); err == nil {
+		t.Error("an image_ref over 512 chars must be rejected")
+	}
+}
+
+func TestEncryptRegistryCreds(t *testing.T) {
+	const keyHex = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff"
+	// bad key (not 64 hex chars) → ParseAESKey error path.
+	if _, err := encryptRegistryCreds("tooshort", `{"auths":{}}`); err == nil {
+		t.Error("a malformed AES key must return an error")
+	}
+	// good key → ciphertext that round-trips.
+	ct, err := encryptRegistryCreds(keyHex, `{"auths":{"ghcr.io":{}}}`)
+	if err != nil {
+		t.Fatalf("encryptRegistryCreds(good key): %v", err)
+	}
+	if ct == "" {
+		t.Fatal("ciphertext must be non-empty")
+	}
+	key, _ := crypto.ParseAESKey(keyHex)
+	plain, derr := crypto.Decrypt(key, ct)
+	if derr != nil || plain != `{"auths":{"ghcr.io":{}}}` {
+		t.Errorf("round-trip failed: plain=%q err=%v", plain, derr)
+	}
+}
+
+func TestApplyImageSourceOpts(t *testing.T) {
+	const keyHex = "00112233445566778899aabbccddeeff00112233445566778899aabbccddeeff"
+	key, _ := crypto.ParseAESKey(keyHex)
+	cipher, _ := crypto.Encrypt(key, `{"auths":{}}`)
+
+	// non-image → no-op (tarball deploy untouched)
+	o := compute.DeployOptions{Tarball: []byte("x")}
+	applyImageSourceOpts(&o, &models.Deployment{Source: "tarball"}, keyHex)
+	if o.Source != "" || o.Tarball == nil {
+		t.Errorf("tarball deploy must be untouched, got %+v", o)
+	}
+
+	// image, no creds → Source/ImageRef set, Tarball cleared, no RegistryAuth
+	o = compute.DeployOptions{Tarball: []byte("x")}
+	applyImageSourceOpts(&o, &models.Deployment{Source: "image", ImageRef: "ghcr.io/o/a:1"}, keyHex)
+	if o.Source != "image" || o.ImageRef != "ghcr.io/o/a:1" || o.Tarball != nil || o.RegistryAuth != "" {
+		t.Errorf("image no-creds: %+v", o)
+	}
+
+	// image + creds → RegistryAuth decrypted
+	o = compute.DeployOptions{}
+	applyImageSourceOpts(&o, &models.Deployment{Source: "image", ImageRef: "r", RegistryCredsEnc: cipher}, keyHex)
+	if o.RegistryAuth != `{"auths":{}}` {
+		t.Errorf("creds decrypt: got %q", o.RegistryAuth)
+	}
+
+	// image + bad ciphertext → decrypt fails → no RegistryAuth (fallback)
+	o = compute.DeployOptions{}
+	applyImageSourceOpts(&o, &models.Deployment{Source: "image", RegistryCredsEnc: "not-valid-base64!!"}, keyHex)
+	if o.RegistryAuth != "" {
+		t.Error("bad ciphertext must not set RegistryAuth")
+	}
+
+	// bad AES key → no RegistryAuth
+	o = compute.DeployOptions{}
+	applyImageSourceOpts(&o, &models.Deployment{Source: "image", RegistryCredsEnc: cipher}, "tooshort")
+	if o.RegistryAuth != "" {
+		t.Error("bad AES key must not set RegistryAuth")
+	}
+}
diff --git a/internal/handlers/deploy_redeploy_inplace_mock_test.go b/internal/handlers/deploy_redeploy_inplace_mock_test.go
index 2be7de23..d6667c39 100644
--- a/internal/handlers/deploy_redeploy_inplace_mock_test.go
+++ b/internal/handlers/deploy_redeploy_inplace_mock_test.go
@@ -75,6 +75,7 @@ var deploymentColumnsList = []string{
 	"created_at", "updated_at",
 	"notify_webhook", "notify_webhook_secret", "notify_state", "notify_attempts",
 	"expires_at", "ttl_policy", "reminders_sent", "last_reminder_at",
+	"source", "image_ref", "registry_creds_enc",
 }
 
 // redeployMockApp wires a minimal Fiber app that drives DeployHandler.New
@@ -252,6 +253,7 @@ func TestDeployNew_Redeploy_WrongTeam_DefenceInDepth(t *testing.T) {
 			time.Now(), time.Now(),      // created_at, updated_at
 			sql.NullString{}, sql.NullString{}, "unset", 0, // notify_*
 			sql.NullTime{}, "permanent", 0, sql.NullTime{}, // ttl_*
+			"tarball", "", "", // source, image_ref, registry_creds_enc (mig 064)
 		))
 
 	body, ct := multipartRedeployMockBody(t, map[string]string{
@@ -322,6 +324,7 @@ func TestDeployNew_Redeploy_UpdateStatusError_StillAccepts(t *testing.T) {
 			time.Now(), time.Now(),
 			sql.NullString{}, sql.NullString{}, "unset", 0,
 			sql.NullTime{}, "permanent", 0, sql.NullTime{},
+			"tarball", "", "", // source, image_ref, registry_creds_enc (mig 064)
 		))
 
 	// UPDATE deployments SET status = $1 ... → driver error. The handler
@@ -410,6 +413,7 @@ func TestDeployNew_Redeploy_EmptyProviderID_Returns409(t *testing.T) {
 			time.Now(), time.Now(),
 			sql.NullString{}, sql.NullString{}, "unset", 0,
 			sql.NullTime{}, "permanent", 0, sql.NullTime{},
+			"tarball", "", "", // source, image_ref, registry_creds_enc (mig 064)
 		))
 
 	body, ct := multipartRedeployMockBody(t, map[string]string{
diff --git a/internal/handlers/deploy_source_image_integration_test.go b/internal/handlers/deploy_source_image_integration_test.go
new file mode 100644
index 00000000..d875dd72
--- /dev/null
+++ b/internal/handlers/deploy_source_image_integration_test.go
@@ -0,0 +1,249 @@
+package handlers_test
+
+// deploy_source_image_integration_test.go — end-to-end coverage for the P2
+// source=image branch of POST /deploy/new (migration 064). Exercises the
+// source switch (flag-off 501, invalid-source 400, flag-on invalid-ref 400,
+// flag-on happy 202) so the handler-level source-routing lines are covered
+// in CI (these tests run with a real DB; they skip locally without one).
+
+import (
+	"bytes"
+	"database/sql"
+	"encoding/json"
+	"mime/multipart"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"instant.dev/internal/config"
+	"instant.dev/internal/testhelpers"
+)
+
+// buildImageDeployForm assembles a multipart body for a source=image deploy.
+// Fields with an empty value are omitted (so callers can test a missing
+// image_ref by passing "").
+func buildImageDeployForm(t *testing.T, fields map[string]string) (*bytes.Buffer, string) {
+	t.Helper()
+	buf := &bytes.Buffer{}
+	mw := multipart.NewWriter(buf)
+	for k, v := range fields {
+		require.NoError(t, mw.WriteField(k, v))
+	}
+	require.NoError(t, mw.Close())
+	return buf, mw.FormDataContentType()
+}
+
+func postDeploy(t *testing.T, app interface {
+	Test(*http.Request, ...int) (*http.Response, error)
+}, body *bytes.Buffer, contentType, jwt string) *http.Response {
+	t.Helper()
+	req := httptest.NewRequest(http.MethodPost, "/deploy/new", body)
+	req.Header.Set("Content-Type", contentType)
+	req.Header.Set("Authorization", "Bearer "+jwt)
+	req.Header.Set("X-Forwarded-For", "10.42.0.7")
+	resp, err := app.Test(req, 15000)
+	require.NoError(t, err)
+	return resp
+}
+
+// TestDeployNew_SourceImage_FlagOff_501 — source=image is rejected with 501
+// while DEPLOY_SOURCE_IMAGE_ENABLED is off (the production default), and the
+// tarball path is never touched.
+func TestDeployNew_SourceImage_FlagOff_501(t *testing.T) {
+	daDeployNeedsDB(t)
+	db, cleanDB := testhelpers.SetupTestDB(t)
+	defer cleanDB()
+	rdb, cleanRedis := testhelpers.SetupTestRedis(t)
+	defer cleanRedis()
+
+	teamID := testhelpers.MustCreateTeamDB(t, db, "pro")
+	jwt := testhelpers.MustSignSessionJWT(t, uuid.NewString(), teamID, "img@example.com")
+	app, cleanApp := testhelpers.NewTestAppWithServices(t, db, rdb, "deploy") // flag defaults OFF
+	defer cleanApp()
+
+	body, ct := buildImageDeployForm(t, map[string]string{
+		"name":      "img-app",
+		"source":    "image",
+		"image_ref": "ghcr.io/owner/app:v1",
+	})
+	resp := postDeploy(t, app, body, ct, jwt)
+	defer resp.Body.Close()
+
+	assert.Equal(t, http.StatusNotImplemented, resp.StatusCode, "source=image must 501 while flag is off")
+	var env struct {
+		Error string `json:"error"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&env))
+	assert.Equal(t, "source_image_disabled", env.Error)
+}
+
+// TestDeployNew_InvalidSource_400 — an unrecognised source (neither tarball
+// nor image, e.g. "git" which is not yet wired) is a clean 400.
+func TestDeployNew_InvalidSource_400(t *testing.T) {
+	daDeployNeedsDB(t)
+	db, cleanDB := testhelpers.SetupTestDB(t)
+	defer cleanDB()
+	rdb, cleanRedis := testhelpers.SetupTestRedis(t)
+	defer cleanRedis()
+
+	teamID := testhelpers.MustCreateTeamDB(t, db, "pro")
+	jwt := testhelpers.MustSignSessionJWT(t, uuid.NewString(), teamID, "bad@example.com")
+	app, cleanApp := testhelpers.NewTestAppWithServices(t, db, rdb, "deploy")
+	defer cleanApp()
+
+	body, ct := buildImageDeployForm(t, map[string]string{
+		"name":   "bad-source-app",
+		"source": "git",
+	})
+	resp := postDeploy(t, app, body, ct, jwt)
+	defer resp.Body.Close()
+
+	assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
+	var env struct {
+		Error string `json:"error"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&env))
+	assert.Equal(t, "invalid_source", env.Error)
+}
+
+// TestDeployNew_SourceImage_FlagOn_InvalidRef_400 — with the flag enabled, a
+// bare image name (no registry host) is rejected by validateImageRef.
+func TestDeployNew_SourceImage_FlagOn_InvalidRef_400(t *testing.T) {
+	daDeployNeedsDB(t)
+	db, cleanDB := testhelpers.SetupTestDB(t)
+	defer cleanDB()
+	rdb, cleanRedis := testhelpers.SetupTestRedis(t)
+	defer cleanRedis()
+
+	teamID := testhelpers.MustCreateTeamDB(t, db, "pro")
+	jwt := testhelpers.MustSignSessionJWT(t, uuid.NewString(), teamID, "badref@example.com")
+	app, cleanApp := testhelpers.NewTestAppWithServices(t, db, rdb, "deploy",
+		func(c *config.Config) { c.DeploySourceImageEnabled = true })
+	defer cleanApp()
+
+	body, ct := buildImageDeployForm(t, map[string]string{
+		"name":      "bad-ref-app",
+		"source":    "image",
+		"image_ref": "nginx", // bare name, no registry host → rejected
+	})
+	resp := postDeploy(t, app, body, ct, jwt)
+	defer resp.Body.Close()
+
+	assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
+	var env struct {
+		Error string `json:"error"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&env))
+	assert.Equal(t, "invalid_image_ref", env.Error)
+}
+
+// TestDeployNew_SourceImage_EncryptFailure_503 — when AES_KEY is misconfigured,
+// encrypting BYO registry creds fails and the handler returns 503 rather than
+// persisting unencrypted secrets. AES parsing is request-time, so a bad key on
+// the test config only trips this path (the app still builds).
+func TestDeployNew_SourceImage_EncryptFailure_503(t *testing.T) {
+	daDeployNeedsDB(t)
+	db, cleanDB := testhelpers.SetupTestDB(t)
+	defer cleanDB()
+	rdb, cleanRedis := testhelpers.SetupTestRedis(t)
+	defer cleanRedis()
+
+	teamID := testhelpers.MustCreateTeamDB(t, db, "pro")
+	jwt := testhelpers.MustSignSessionJWT(t, uuid.NewString(), teamID, "enc@example.com")
+	app, cleanApp := testhelpers.NewTestAppWithServices(t, db, rdb, "deploy",
+		func(c *config.Config) {
+			c.DeploySourceImageEnabled = true
+			c.AESKey = "not-a-valid-hex-key" // crypto.ParseAESKey rejects → encrypt fails
+		})
+	defer cleanApp()
+
+	body, ct := buildImageDeployForm(t, map[string]string{
+		"name":           "enc-fail-app",
+		"source":         "image",
+		"image_ref":      "ghcr.io/owner/app:v1",
+		"registry_creds": `{"auths":{}}`,
+	})
+	resp := postDeploy(t, app, body, ct, jwt)
+	defer resp.Body.Close()
+
+	assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode)
+	var env struct {
+		Error string `json:"error"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&env))
+	assert.Equal(t, "encrypt_failed", env.Error)
+}
+
+// TestDeployNew_SourceImage_FlagOn_Accepted — the happy path: flag on, valid
+// image_ref + BYO private-registry creds → 202, the response item echoes
+// source=image + image_ref + registry_creds_set:true (never the creds), and
+// the async runDeploy reaches the (noop) compute provider, which stamps the
+// row healthy. Polling the row to healthy proves applyImageSourceOpts ran.
+func TestDeployNew_SourceImage_FlagOn_Accepted(t *testing.T) {
+	daDeployNeedsDB(t)
+	db, cleanDB := testhelpers.SetupTestDB(t)
+	defer cleanDB()
+	rdb, cleanRedis := testhelpers.SetupTestRedis(t)
+	defer cleanRedis()
+
+	teamID := testhelpers.MustCreateTeamDB(t, db, "pro")
+	jwt := testhelpers.MustSignSessionJWT(t, uuid.NewString(), teamID, "ok@example.com")
+	app, cleanApp := testhelpers.NewTestAppWithServices(t, db, rdb, "deploy",
+		func(c *config.Config) { c.DeploySourceImageEnabled = true })
+	defer cleanApp()
+
+	const ref = "ghcr.io/owner/app:v1"
+	body, ct := buildImageDeployForm(t, map[string]string{
+		"name":           "img-ok-app",
+		"source":         "image",
+		"image_ref":      ref,
+		"registry_creds": `{"auths":{"ghcr.io":{"auth":"dG9rZW4="}}}`,
+	})
+	resp := postDeploy(t, app, body, ct, jwt)
+	defer resp.Body.Close()
+
+	require.Equal(t, http.StatusAccepted, resp.StatusCode, "valid source=image deploy must 202")
+	var env struct {
+		OK   bool `json:"ok"`
+		Item struct {
+			ID               string `json:"id"`
+			AppID            string `json:"app_id"`
+			Source           string `json:"source"`
+			ImageRef         string `json:"image_ref"`
+			RegistryCredsSet bool   `json:"registry_creds_set"`
+			RegistryCreds    string `json:"registry_creds"`
+		} `json:"item"`
+	}
+	require.NoError(t, json.NewDecoder(resp.Body).Decode(&env))
+	assert.True(t, env.OK)
+	assert.Equal(t, "image", env.Item.Source)
+	assert.Equal(t, ref, env.Item.ImageRef)
+	assert.True(t, env.Item.RegistryCredsSet, "registry_creds_set must be true when creds supplied")
+	assert.Empty(t, env.Item.RegistryCreds, "registry creds must NEVER be echoed back")
+	require.NotEmpty(t, env.Item.ID)
+
+	// runDeploy is async — poll the row until the (noop) compute provider has
+	// stamped it healthy with a provider id. This deterministically waits for
+	// the applyImageSourceOpts → compute.Deploy path to execute.
+	deadline := time.Now().Add(5 * time.Second)
+	var status, providerID string
+	var imageRef sql.NullString
+	for time.Now().Before(deadline) {
+		row := db.QueryRow(
+			`SELECT status, COALESCE(provider_id,''), image_ref FROM deployments WHERE id = $1`,
+			env.Item.ID)
+		require.NoError(t, row.Scan(&status, &providerID, &imageRef))
+		if providerID != "" {
+			break
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	assert.Equal(t, "healthy", status, "noop provider should drive the row to healthy")
+	assert.NotEmpty(t, providerID, "runDeploy must persist the provider id")
+	assert.Equal(t, ref, imageRef.String, "image_ref must be persisted on the row")
+}
diff --git a/internal/handlers/helpers.go b/internal/handlers/helpers.go
index cafceb11..bb416124 100644
--- a/internal/handlers/helpers.go
+++ b/internal/handlers/helpers.go
@@ -492,6 +492,18 @@ var codeToAgentAction = map[string]errorCodeMeta{
 	"missing_tarball": {
 		AgentAction: "Tell the user the deployment tarball is missing. POST a multipart form with 'tarball' (.tar.gz, <=50 MiB) — see https://instanode.dev/docs/deploy.",
 	},
+	"invalid_source": {
+		AgentAction: "Tell the user the 'source' field is invalid. Use source=tarball (default — upload a .tar.gz) or source=image (deploy a prebuilt image_ref) — see https://instanode.dev/docs/deploy.",
+	},
+	"source_image_disabled": {
+		AgentAction: "Tell the user deploying from a prebuilt image (source=image) is still rolling out and not yet enabled. Upload source as a tarball (<=10 MiB) for now — see https://instanode.dev/docs/deploy.",
+	},
+	"invalid_image_ref": {
+		AgentAction: "Tell the user the image_ref is malformed. Pass a fully-qualified reference including the registry host, e.g. ghcr.io/owner/app:tag — see https://instanode.dev/docs/deploy.",
+	},
+	"encrypt_failed": {
+		AgentAction: "Tell the user the platform could not securely store the supplied registry credentials. This is a transient/server-side condition — retry shortly, or omit registry_creds for a public image — see https://instanode.dev/status.",
+	},
 	"missing_manifest": {
 		AgentAction: "Tell the user the stack manifest is missing. POST a multipart form with 'manifest' (YAML) — see https://instanode.dev/docs/stacks.",
 	},
diff --git a/internal/models/coverage_provision_gate_test.go b/internal/models/coverage_provision_gate_test.go
index e6b52b58..5ae06ed2 100644
--- a/internal/models/coverage_provision_gate_test.go
+++ b/internal/models/coverage_provision_gate_test.go
@@ -17,6 +17,7 @@ func deploymentMockCols() []string {
 		"env_vars", "port", "tier", "env", "private", "allowed_ips", "error_message", "created_at", "updated_at",
 		"notify_webhook", "notify_webhook_secret", "notify_state", "notify_attempts",
 		"expires_at", "ttl_policy", "reminders_sent", "last_reminder_at",
+		"source", "image_ref", "registry_creds_enc",
 	}
 }
 
@@ -26,6 +27,7 @@ func deploymentMockRow() *sqlmock.Rows {
 		[]byte(`{}`), 8080, "hobby", "production", false, "", nil, time.Now(), time.Now(),
 		nil, nil, "unset", 0,
 		nil, "auto_24h", 0, nil,
+		"tarball", "", "", // source, image_ref, registry_creds_enc (mig 064)
 	)
 }
 
diff --git a/internal/models/deployment.go b/internal/models/deployment.go
index b22d28eb..3c305eaf 100644
--- a/internal/models/deployment.go
+++ b/internal/models/deployment.go
@@ -46,6 +46,10 @@ type Deployment struct {
 	NotifyState         string   // 'unset' | 'pending' | 'sent' | 'failed'
 	NotifyAttempts      int      // dispatch retry counter (worker bumps on 5xx/network)
 	ErrorMessage        string
+	// Multi-source deploys (migration 064).
+	Source           string // 'tarball' (default) | 'image' | 'git'
+	ImageRef         string // prebuilt image ref when Source=='image'; '' otherwise
+	RegistryCredsEnc string // AES-256-GCM ciphertext of pull creds; '' when public / platform secret
 	// TTL fields (Wave FIX-J — migration 045).
 	//
 	// ExpiresAt: when the deploy auto-expires. Zero (sql NULL) means
@@ -94,6 +98,10 @@ type CreateDeploymentParams struct {
 	AllowedIPs          []string // each entry must already be a valid IP or CIDR
 	NotifyWebhook       string   // empty = no webhook; non-empty = validated https URL
 	NotifyWebhookSecret string   // empty = no HMAC; non-empty = AES ciphertext
+	// Multi-source deploys (migration 064). Source empty → 'tarball'.
+	Source           string // 'tarball' | 'image' | 'git'
+	ImageRef         string // prebuilt image ref when Source=='image'
+	RegistryCredsEnc string // AES-256-GCM ciphertext of pull creds; '' when public
 	// TTLPolicy chooses the lifecycle for this deploy. Valid values are
 	// "auto_24h" (default — expires_at set to now()+24h), "permanent"
 	// (expires_at = NULL, never auto-expires), or "custom" (caller sets
@@ -135,7 +143,8 @@ func (e *ErrDeploymentNotFound) Error() string {
 const deploymentColumns = `id, team_id, resource_id, app_id, provider_id, status, app_url,
        env_vars, port, tier, env, private, allowed_ips, error_message, created_at, updated_at,
        notify_webhook, notify_webhook_secret, notify_state, notify_attempts,
-       expires_at, ttl_policy, reminders_sent, last_reminder_at`
+       expires_at, ttl_policy, reminders_sent, last_reminder_at,
+       source, image_ref, registry_creds_enc`
 
 // scanDeployment reads a single deployments row into a Deployment struct.
 // env_vars is stored as JSONB; error_message, provider_id, and app_url are nullable.
@@ -165,6 +174,9 @@ func scanDeployment(row interface {
 		// NOT NULL ttl_policy + reminders_sent. Order MUST match the trailing
 		// 4 columns appended in deploymentColumns above.
 		&d.ExpiresAt, &d.TTLPolicy, &d.RemindersSent, &d.LastReminderAt,
+		// migration 064: multi-source deploys. NOT NULL DEFAULT '' / 'tarball'
+		// so plain string scan targets are safe (no NullString needed).
+		&d.Source, &d.ImageRef, &d.RegistryCredsEnc,
 	); err != nil {
 		return nil, err
 	}
@@ -290,17 +302,25 @@ func CreateDeployment(ctx context.Context, db dbExecutor, p CreateDeploymentPara
 		expiresAt = time.Now().UTC().Add(24 * time.Hour)
 	}
 
+	// migration 064: default empty source to 'tarball' for back-compat.
+	source := p.Source
+	if source == "" {
+		source = "tarball"
+	}
+
 	row := db.QueryRowContext(ctx, `
 		INSERT INTO deployments
 			(team_id, resource_id, app_id, port, tier, env, env_vars, private, allowed_ips,
 			 notify_webhook, notify_webhook_secret, notify_state,
-			 expires_at, ttl_policy)
-		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)
+			 expires_at, ttl_policy,
+			 source, image_ref, registry_creds_enc)
+		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
 		RETURNING `+deploymentColumns,
 		p.TeamID, resourceID, p.AppID, port, p.Tier, env, envVarsJSON,
 		p.Private, allowedIPs,
 		notifyWebhook, notifyWebhookSecret, notifyState,
-		expiresAt, ttlPolicy)
+		expiresAt, ttlPolicy,
+		source, p.ImageRef, p.RegistryCredsEnc)
 
 	d, err := scanDeployment(row)
 	if err != nil {
diff --git a/internal/providers/compute/k8s/client.go b/internal/providers/compute/k8s/client.go
index 7eabbd5e..e4ec7ab6 100644
--- a/internal/providers/compute/k8s/client.go
+++ b/internal/providers/compute/k8s/client.go
@@ -1047,19 +1047,36 @@ func (p *K8sProvider) Deploy(ctx context.Context, opts compute.DeployOptions) (*
 		opts.Port = 8080
 	}
 
-	imageTag := imageName(opts.AppID)
 	ns := deployNamespace(opts.AppID)
-
-	// Step 1: Build the Docker image from the tarball.
-	if err := p.buildImage(ctx, deployNamespace(opts.AppID), opts.AppID, imageTag, opts.Tarball); err != nil {
-		return nil, fmt.Errorf("k8s.Deploy: build image: %w", err)
-	}
-
-	// Step 2: Create per-deployment namespace with all security primitives.
-	// opts.TeamID scopes the NetworkPolicy DB-egress rule to this team's
-	// customer-resource namespaces — preventing cross-tenant DB access.
-	if err := p.setupTenantNamespace(ctx, ns, opts.AppID, opts.TeamID, opts.Tier); err != nil {
-		return nil, fmt.Errorf("k8s.Deploy: setup namespace: %w", err)
+	var imageTag string
+
+	if opts.Source == "image" {
+		// BYO prebuilt image — skip Kaniko entirely. No build Job, no MinIO
+		// upload, no build NetworkPolicy. We deploy opts.ImageRef directly.
+		imageTag = opts.ImageRef
+		// setupTenantNamespace creates the per-deployment namespace (buildImage
+		// normally does this on the tarball path) + the security primitives.
+		if err := p.setupTenantNamespace(ctx, ns, opts.AppID, opts.TeamID, opts.Tier); err != nil {
+			return nil, fmt.Errorf("k8s.Deploy(image): setup namespace: %w", err)
+		}
+		// Ensure a "ghcr-pull" imagePullSecret exists in the namespace
+		// (applyDeploymentInNS references it). With BYO RegistryAuth we write
+		// the caller's pull creds; otherwise we copy the platform secret.
+		if err := p.ensureImagePullSecret(ctx, ns, opts.RegistryAuth); err != nil {
+			return nil, fmt.Errorf("k8s.Deploy(image): pull secret: %w", err)
+		}
+	} else {
+		imageTag = imageName(opts.AppID)
+		// Step 1: Build the Docker image from the tarball.
+		if err := p.buildImage(ctx, ns, opts.AppID, imageTag, opts.Tarball); err != nil {
+			return nil, fmt.Errorf("k8s.Deploy: build image: %w", err)
+		}
+		// Step 2: Create per-deployment namespace with all security primitives.
+		// opts.TeamID scopes the NetworkPolicy DB-egress rule to this team's
+		// customer-resource namespaces — preventing cross-tenant DB access.
+		if err := p.setupTenantNamespace(ctx, ns, opts.AppID, opts.TeamID, opts.Tier); err != nil {
+			return nil, fmt.Errorf("k8s.Deploy: setup namespace: %w", err)
+		}
 	}
 
 	deployName := deploymentName(opts.AppID)
@@ -1587,6 +1604,30 @@ func (p *K8sProvider) ensureRegistryAuthInNS(ctx context.Context, ns, name strin
 	return nil
 }
 
+// ensureImagePullSecret makes the "ghcr-pull" imagePullSecret available in a
+// source=image deploy namespace (applyDeploymentInNS always references that
+// name). With a non-empty registryAuth (a complete docker config JSON for a
+// PRIVATE registry) it writes those BYO pull creds into the secret; otherwise
+// it copies the platform's shared ghcr-pull secret (covers public images +
+// GHCR-under-platform-token). The namespace is per-deployment and torn down
+// with the deploy, so writing BYO creds into the canonical name is safe — it
+// cannot leak across tenants.
+func (p *K8sProvider) ensureImagePullSecret(ctx context.Context, ns, registryAuth string) error {
+	if registryAuth == "" {
+		return p.ensureRegistryAuthInNS(ctx, ns, "ghcr-pull")
+	}
+	sec := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: "ghcr-pull"},
+		Type:       corev1.SecretTypeDockerConfigJson,
+		Data:       map[string][]byte{corev1.DockerConfigJsonKey: []byte(registryAuth)},
+	}
+	_, err := p.clientset.CoreV1().Secrets(ns).Create(ctx, sec, metav1.CreateOptions{})
+	if apierrors.IsAlreadyExists(err) {
+		_, err = p.clientset.CoreV1().Secrets(ns).Update(ctx, sec, metav1.UpdateOptions{})
+	}
+	return err
+}
+
 // createKanikoJob spawns a one-shot Job that builds and pushes the image.
 // When httpContextURL is non-empty an initContainer curls the build context
 // from MinIO into a shared emptyDir; kaniko then reads via the standard
diff --git a/internal/providers/compute/k8s/deploy_image_branches_test.go b/internal/providers/compute/k8s/deploy_image_branches_test.go
new file mode 100644
index 00000000..eb18d15a
--- /dev/null
+++ b/internal/providers/compute/k8s/deploy_image_branches_test.go
@@ -0,0 +1,170 @@
+package k8s
+
+// deploy_image_branches_test.go — error/edge branches of the P2 source=image
+// Deploy path and ensureImagePullSecret, plus the tarball build-error wrap.
+// All run against the fake clientset (no real cluster), using reactors to
+// force the failure modes that a happy-path test can't reach.
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	k8sruntime "k8s.io/apimachinery/pkg/runtime"
+	clientfake "k8s.io/client-go/kubernetes/fake"
+	clienttesting "k8s.io/client-go/testing"
+
+	"instant.dev/internal/providers/compute"
+)
+
+func imageDeployOpts(appID, registryAuth string) compute.DeployOptions {
+	return compute.DeployOptions{
+		AppID:        appID,
+		Source:       "image",
+		ImageRef:     "ghcr.io/o/a:1",
+		RegistryAuth: registryAuth,
+		Port:         8080,
+		Tier:         "hobby",
+		TeamID:       "11111111-1111-1111-1111-111111111111",
+	}
+}
+
+// platformGHCRPull is the shared pull secret ensureRegistryAuthInNS copies out
+// of the "instant" namespace when a deploy supplies no BYO creds.
+func platformGHCRPull() *corev1.Secret {
+	return &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: "ghcr-pull", Namespace: "instant"},
+		Type:       corev1.SecretTypeDockerConfigJson,
+		Data:       map[string][]byte{corev1.DockerConfigJsonKey: []byte(`{"auths":{}}`)},
+	}
+}
+
+// TestDeploy_ImageSource_EmptyAuth_CopiesPlatformSecret — with no BYO creds,
+// ensureImagePullSecret copies the platform "ghcr-pull" secret from the
+// instant namespace into the deploy namespace (covers the registryAuth=="" arm).
+func TestDeploy_ImageSource_EmptyAuth_CopiesPlatformSecret(t *testing.T) {
+	cs := clientfake.NewSimpleClientset(platformGHCRPull())
+	p := &K8sProvider{clientset: cs}
+
+	if _, err := p.Deploy(context.Background(), imageDeployOpts("imgcopy", "")); err != nil {
+		t.Fatalf("Deploy(image, no creds): %v", err)
+	}
+	ns := deployNamespace("imgcopy")
+	if _, err := cs.CoreV1().Secrets(ns).Get(context.Background(), "ghcr-pull", metav1.GetOptions{}); err != nil {
+		t.Errorf("platform ghcr-pull should have been copied into %q: %v", ns, err)
+	}
+}
+
+// TestDeploy_ImageSource_PullSecretAlreadyExists_Updates — when the ghcr-pull
+// secret already exists in the deploy namespace, the BYO-creds Create returns
+// AlreadyExists and we fall through to Update (covers that arm).
+func TestDeploy_ImageSource_PullSecretAlreadyExists_Updates(t *testing.T) {
+	ns := deployNamespace("imgupd")
+	pre := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{Name: "ghcr-pull", Namespace: ns},
+		Type:       corev1.SecretTypeDockerConfigJson,
+		Data:       map[string][]byte{corev1.DockerConfigJsonKey: []byte(`{"auths":{}}`)},
+	}
+	cs := clientfake.NewSimpleClientset(pre)
+	p := &K8sProvider{clientset: cs}
+
+	if _, err := p.Deploy(context.Background(), imageDeployOpts("imgupd", `{"auths":{"ghcr.io":{"auth":"eA=="}}}`)); err != nil {
+		t.Fatalf("Deploy(image, existing secret): %v", err)
+	}
+	got, err := cs.CoreV1().Secrets(ns).Get(context.Background(), "ghcr-pull", metav1.GetOptions{})
+	if err != nil {
+		t.Fatalf("get ghcr-pull: %v", err)
+	}
+	// Update should have overwritten the placeholder with the BYO creds.
+	if !strings.Contains(string(got.Data[corev1.DockerConfigJsonKey]), "ghcr.io") {
+		t.Errorf("ghcr-pull not updated with BYO creds: %s", got.Data[corev1.DockerConfigJsonKey])
+	}
+}
+
+// TestDeploy_ImageSource_SetupNamespaceError — a namespace-create failure in the
+// image path propagates as a wrapped Deploy error (covers the setup-namespace arm).
+func TestDeploy_ImageSource_SetupNamespaceError(t *testing.T) {
+	cs := clientfake.NewSimpleClientset()
+	cs.PrependReactor("create", "namespaces", func(_ clienttesting.Action) (bool, k8sruntime.Object, error) {
+		return true, nil, errors.New("boom-ns")
+	})
+	p := &K8sProvider{clientset: cs}
+
+	_, err := p.Deploy(context.Background(), imageDeployOpts("imgns", ""))
+	if err == nil || !strings.Contains(err.Error(), "setup namespace") {
+		t.Fatalf("want wrapped setup-namespace error, got: %v", err)
+	}
+}
+
+// TestDeploy_ImageSource_PullSecretError — a ghcr-pull secret-create failure in
+// the image path propagates as a wrapped Deploy error (covers the pull-secret arm).
+func TestDeploy_ImageSource_PullSecretError(t *testing.T) {
+	cs := clientfake.NewSimpleClientset()
+	cs.PrependReactor("create", "secrets", func(action clienttesting.Action) (bool, k8sruntime.Object, error) {
+		ca := action.(clienttesting.CreateAction)
+		sec, ok := ca.GetObject().(*corev1.Secret)
+		if ok && sec.Name == "ghcr-pull" {
+			return true, nil, errors.New("boom-secret")
+		}
+		return false, nil, nil // let other secret creates proceed
+	})
+	p := &K8sProvider{clientset: cs}
+
+	// BYO creds → the Create at the secret arm is reached (not the copy arm).
+	_, err := p.Deploy(context.Background(), imageDeployOpts("imgsec", `{"auths":{}}`))
+	if err == nil || !strings.Contains(err.Error(), "pull secret") {
+		t.Fatalf("want wrapped pull-secret error, got: %v", err)
+	}
+}
+
+// TestDeploy_TarballSource_BuildError — a build-Job create failure on the
+// tarball path propagates as a wrapped Deploy build error (covers the build arm
+// that moved under the source=tarball branch).
+func TestDeploy_TarballSource_BuildError(t *testing.T) {
+	cs := clientfake.NewSimpleClientset(platformGHCRPull())
+	cs.PrependReactor("create", "jobs", func(_ clienttesting.Action) (bool, k8sruntime.Object, error) {
+		return true, nil, errors.New("boom-job")
+	})
+	p := &K8sProvider{clientset: cs}
+
+	_, err := p.Deploy(context.Background(), compute.DeployOptions{
+		AppID:   "tarerr",
+		Source:  "tarball",
+		Tarball: []byte("not-a-real-tarball"),
+		Port:    8080,
+		Tier:    "hobby",
+		TeamID:  "11111111-1111-1111-1111-111111111111",
+	})
+	if err == nil || !strings.Contains(err.Error(), "build image") {
+		t.Fatalf("want wrapped build-image error, got: %v", err)
+	}
+}
+
+// TestDeploy_TarballSource_SetupNamespaceErrorAfterBuild — on the tarball path
+// the build runs FIRST and setupTenantNamespace SECOND (so the kaniko pod isn't
+// constrained by the runtime ResourceQuota). This drives that ordering: the
+// build Job auto-completes, then a ResourceQuota-create failure surfaces as the
+// wrapped "setup namespace" error (covers the tarball setup-namespace arm).
+func TestDeploy_TarballSource_SetupNamespaceErrorAfterBuild(t *testing.T) {
+	cs := clientfake.NewSimpleClientset(platformGHCRPull())
+	attachJobCompleteReactor(cs) // buildImage succeeds (Job Complete on first poll)
+	cs.PrependReactor("create", "resourcequotas", func(_ clienttesting.Action) (bool, k8sruntime.Object, error) {
+		return true, nil, errors.New("boom-quota") // setupTenantNamespace creates a quota; buildImage does not
+	})
+	p := &K8sProvider{clientset: cs}
+
+	_, err := p.Deploy(context.Background(), compute.DeployOptions{
+		AppID:   "tarsetup",
+		Source:  "tarball",
+		Tarball: []byte("t"),
+		Port:    8080,
+		Tier:    "hobby",
+		TeamID:  "11111111-1111-1111-1111-111111111111",
+	})
+	if err == nil || !strings.Contains(err.Error(), "setup namespace") {
+		t.Fatalf("want wrapped setup-namespace error after a successful build, got: %v", err)
+	}
+}
diff --git a/internal/providers/compute/k8s/deploy_image_test.go b/internal/providers/compute/k8s/deploy_image_test.go
new file mode 100644
index 00000000..7f3433a8
--- /dev/null
+++ b/internal/providers/compute/k8s/deploy_image_test.go
@@ -0,0 +1,52 @@
+package k8s
+
+// deploy_image_test.go — P2 source=image: Deploy must skip Kaniko, deploy the
+// prebuilt image directly, and provision a pull secret. Runs against the fake
+// clientset (no real cluster).
+
+import (
+	"context"
+	"testing"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes/fake"
+
+	"instant.dev/internal/providers/compute"
+)
+
+func TestDeploy_ImageSource_SkipsBuildAndDeploysRef(t *testing.T) {
+	cs := fake.NewSimpleClientset()
+	p := &K8sProvider{clientset: cs}
+
+	_, err := p.Deploy(context.Background(), compute.DeployOptions{
+		AppID:        "imgapp",
+		Source:       "image",
+		ImageRef:     "ghcr.io/o/a:1",
+		RegistryAuth: `{"auths":{"ghcr.io":{"auth":"eA=="}}}`, // BYO creds → no copy from instant ns
+		Port:         8080,
+		Tier:         "hobby",
+		TeamID:       "11111111-1111-1111-1111-111111111111",
+	})
+	if err != nil {
+		t.Fatalf("Deploy(image): %v", err)
+	}
+
+	ns := deployNamespace("imgapp")
+	// Container runs the prebuilt image (not a built imageName()).
+	dep, derr := cs.AppsV1().Deployments(ns).Get(context.Background(), deploymentName("imgapp"), metav1.GetOptions{})
+	if derr != nil {
+		t.Fatalf("get deployment: %v", derr)
+	}
+	if img := dep.Spec.Template.Spec.Containers[0].Image; img != "ghcr.io/o/a:1" {
+		t.Errorf("container image = %q, want the prebuilt ref ghcr.io/o/a:1", img)
+	}
+	// No Kaniko build Job for source=image.
+	jobs, _ := cs.BatchV1().Jobs(ns).List(context.Background(), metav1.ListOptions{})
+	if len(jobs.Items) != 0 {
+		t.Errorf("source=image must NOT create a build Job; got %d", len(jobs.Items))
+	}
+	// BYO pull secret written into the namespace.
+	if _, serr := cs.CoreV1().Secrets(ns).Get(context.Background(), "ghcr-pull", metav1.GetOptions{}); serr != nil {
+		t.Errorf("ghcr-pull pull secret not created: %v", serr)
+	}
+}
diff --git a/internal/providers/compute/provider.go b/internal/providers/compute/provider.go
index feda710f..d63fb5dd 100644
--- a/internal/providers/compute/provider.go
+++ b/internal/providers/compute/provider.go
@@ -22,12 +22,23 @@ type DeployOptions struct {
 	AppID      string            // short slug, used as k8s Deployment name and subdomain
 	Token      string            // instant.dev resource token (for env var injection)
 	TeamID     string            // owning team UUID — used to scope the NetworkPolicy DB-port egress rule to the team's own customer-resource namespaces (pentest fix 2026-05-16)
-	Tarball    []byte            // gzipped tar archive of the source directory (must contain Dockerfile)
+	Tarball    []byte            // gzipped tar archive of the source directory (must contain Dockerfile). Empty when Source=="image".
 	EnvVars    map[string]string // merged: infra resource URLs + user-defined vars
 	Port       int               // port the app listens on (default 8080)
 	Tier       string            // hobby|pro|team → resource requests/limits
 	Private    bool              // true → Ingress carries whitelist-source-range annotation
 	AllowedIPs []string          // CIDRs / IPs allowed when Private=true; ignored otherwise
+
+	// Source selects how the app image is obtained: "tarball" (default — build
+	// the uploaded Tarball with Kaniko) or "image" (BYO — deploy ImageRef
+	// directly, no build pod). Empty is treated as "tarball" for back-compat.
+	Source string
+	// ImageRef is the fully-qualified prebuilt image (host/path[:tag][@digest])
+	// deployed directly when Source=="image". Ignored otherwise.
+	ImageRef string
+	// RegistryAuth is an optional docker config JSON for pulling a private
+	// ImageRef. Empty → the namespace's default ghcr-pull secret is used.
+	RegistryAuth string
 }
 
 // AppDeployment represents the live state of a deployed app.
diff --git a/internal/testhelpers/testapp_smoke_test.go b/internal/testhelpers/testapp_smoke_test.go
index 6bcba84c..574852c1 100644
--- a/internal/testhelpers/testapp_smoke_test.go
+++ b/internal/testhelpers/testapp_smoke_test.go
@@ -19,6 +19,8 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+
+	"instant.dev/internal/config"
 )
 
 // TestNewTestAppWithServices_RegistersDeployEventsRoute boots a deploy-enabled
@@ -50,3 +52,27 @@ func TestNewTestAppWithServices_RegistersDeployEventsRoute(t *testing.T) {
 	require.Equal(t, http.StatusUnauthorized, resp.StatusCode,
 		"deploy events route must be registered under the /api/v1 group")
 }
+
+// TestNewTestAppWithServices_AppliesConfigMutator proves the variadic config
+// mutator seam runs against the test config before handlers are built. Handler
+// tests exercise this with feature-flag mutators, but per-package coverage only
+// credits the `testhelpers` package when an in-package test passes a mutator —
+// so this gives the mutator loop its own coverage. A nil mutator interleaved
+// with a real one confirms the nil-skip guard.
+func TestNewTestAppWithServices_AppliesConfigMutator(t *testing.T) {
+	db, cleanDB := SetupTestDB(t)
+	defer cleanDB()
+	rdb, cleanRedis := SetupTestRedis(t)
+	defer cleanRedis()
+
+	applied := false
+	_, cleanApp := NewTestAppWithServices(t, db, rdb, "deploy",
+		nil, // skipped by the m != nil guard
+		func(c *config.Config) {
+			c.DeploySourceImageEnabled = true
+			applied = true
+		})
+	defer cleanApp()
+
+	require.True(t, applied, "config mutator must run against the test config")
+}
diff --git a/internal/testhelpers/testhelpers.go b/internal/testhelpers/testhelpers.go
index 907c5157..0dfbb8c2 100644
--- a/internal/testhelpers/testhelpers.go
+++ b/internal/testhelpers/testhelpers.go
@@ -313,6 +313,12 @@ func runMigrations(t *testing.T, db *sql.DB) {
 		`ALTER TABLE deployments ADD COLUMN IF NOT EXISTS reminders_sent INT NOT NULL DEFAULT 0`,
 		`ALTER TABLE deployments ADD COLUMN IF NOT EXISTS last_reminder_at TIMESTAMPTZ`,
 		`CREATE INDEX IF NOT EXISTS idx_deployments_expires_pending ON deployments (expires_at) WHERE expires_at IS NOT NULL AND status NOT IN ('deleted', 'expired')`,
+		// 064_deploy_source_image — P2 prebuilt-image deploy source. CHECK
+		// omitted in test DDL (handler validates; production migration carries
+		// the source IN ('tarball','image','git') constraint).
+		`ALTER TABLE deployments ADD COLUMN IF NOT EXISTS source TEXT NOT NULL DEFAULT 'tarball'`,
+		`ALTER TABLE deployments ADD COLUMN IF NOT EXISTS image_ref TEXT NOT NULL DEFAULT ''`,
+		`ALTER TABLE deployments ADD COLUMN IF NOT EXISTS registry_creds_enc TEXT NOT NULL DEFAULT ''`,
 		// teams.default_deployment_ttl_policy — Wave FIX-J team preference.
 		`ALTER TABLE teams ADD COLUMN IF NOT EXISTS default_deployment_ttl_policy TEXT NOT NULL DEFAULT 'auto_24h'`,
 		// 012_audit_log — per-team event stream consumed by the dashboard's
@@ -953,10 +959,18 @@ func NewTestApp(t *testing.T, db *sql.DB, rdb *redis.Client) (*fiber.App, func()
 // NewTestAppWithServices creates a Fiber app identical to NewTestApp but with
 // an explicit comma-separated list of enabled services (e.g. "postgres,redis,mongodb,queue,webhook,storage").
 // Use this in tests that exercise /db/new, /cache/new, or /nosql/new.
-func NewTestAppWithServices(t *testing.T, db *sql.DB, rdb *redis.Client, services string) (*fiber.App, func()) {
+// The optional mutators run against the test config after defaults are set but
+// before any handler is constructed — use them to flip feature flags (e.g.
+// DeploySourceImageEnabled) for a single test without widening every caller.
+func NewTestAppWithServices(t *testing.T, db *sql.DB, rdb *redis.Client, services string, mutators ...func(*config.Config)) (*fiber.App, func()) {
 	t.Helper()
 	cfg := testConfig()
 	cfg.EnabledServices = services
+	for _, m := range mutators {
+		if m != nil {
+			m(cfg)
+		}
+	}
 	planReg := plans.Default()
 
 	app := fiber.New(fiber.Config{