diff --git a/docs/ai-workspace/bottom-up-ai-artifact-deployment-guide.md b/docs/ai-workspace/bottom-up-ai-artifact-deployment-guide.md
index 53b4ea4d64..79c65dcacd 100644
--- a/docs/ai-workspace/bottom-up-ai-artifact-deployment-guide.md
+++ b/docs/ai-workspace/bottom-up-ai-artifact-deployment-guide.md
@@ -297,6 +297,17 @@ You can create artifacts on a gateway while it is disconnected, and they reconci
 
 ---
 
+## Connecting to a new AI Workspace
+
+Whenever a gateway (re)connects, it re-offers **all** of its gateway-created artifacts to the AI Workspace — not only the ones that have never synced. This means a gateway will populate a **new or reset AI Workspace** automatically:
+
+- Point the gateway at a **different AI Workspace**, or re-register it, and all its artifacts sync to the new one.
+- If the AI Workspace database is **purged or restored from an empty state**, the gateway repopulates it on the next connection.
+
+Re-offering already-synced artifacts is safe: the AI Workspace recognizes an unchanged re-push and does **not** create duplicate deployments or overwrite edits you made to runtime-neutral details. A genuinely newer change from the gateway still wins, exactly as during normal syncing.
+
+---
+
 ## Immutable gateways
 
 Some gateways run in **immutable** mode, where artifacts are loaded from on-disk configuration at startup rather than created through the management API (see [Immutable Gateway](../gateway/immutable-gateway.md)).
diff --git a/gateway/gateway-controller/pkg/api/handlers/handlers_test.go b/gateway/gateway-controller/pkg/api/handlers/handlers_test.go
index f9bf138e43..3ba530aed5 100644
--- a/gateway/gateway-controller/pkg/api/handlers/handlers_test.go
+++ b/gateway/gateway-controller/pkg/api/handlers/handlers_test.go
@@ -843,6 +843,16 @@ func (m *MockStorage) GetPendingCPSyncArtifacts() ([]*models.StoredConfig, error
 	return pending, nil
 }
 
+func (m *MockStorage) GetGatewayOriginArtifactsForSync() ([]*models.StoredConfig, error) {
+	var all []*models.StoredConfig
+	for _, config := range m.configs {
+		if config != nil && config.Origin == models.OriginGatewayAPI {
+			all = append(all, config)
+		}
+	}
+	return all, nil
+}
+
 // MockControlPlaneClient implements controlplane.ControlPlaneClient for testing
 type MockControlPlaneClient struct {
 	connected     bool
diff --git a/gateway/gateway-controller/pkg/controlplane/api_deleted_test.go b/gateway/gateway-controller/pkg/controlplane/api_deleted_test.go
index ea9cfc2833..1910015272 100644
--- a/gateway/gateway-controller/pkg/controlplane/api_deleted_test.go
+++ b/gateway/gateway-controller/pkg/controlplane/api_deleted_test.go
@@ -646,6 +646,16 @@ func (m *mockStorageForDeletion) GetPendingCPSyncArtifacts() ([]*models.StoredCo
 	return pending, nil
 }
 
+func (m *mockStorageForDeletion) GetGatewayOriginArtifactsForSync() ([]*models.StoredConfig, error) {
+	var all []*models.StoredConfig
+	for _, config := range m.configs {
+		if config.Origin == models.OriginGatewayAPI {
+			all = append(all, config)
+		}
+	}
+	return all, nil
+}
+
 // Helper to create test API config for deletion tests
 func createTestAPIConfigForDeletion(apiID string) *models.StoredConfig {
 	// Create a complete API configuration so deletion flow can properly process it
diff --git a/gateway/gateway-controller/pkg/controlplane/push_artifacts_test.go b/gateway/gateway-controller/pkg/controlplane/push_artifacts_test.go
index cd03929a2a..911f5ca0dd 100644
--- a/gateway/gateway-controller/pkg/controlplane/push_artifacts_test.go
+++ b/gateway/gateway-controller/pkg/controlplane/push_artifacts_test.go
@@ -199,9 +199,11 @@ func TestPushGatewayArtifacts(t *testing.T) {
 
 	client.apiUtilsService.SetBaseURL(server.URL)
 
-	// Seed a pending gateway-originated artifact (should be pushed), a CP-originated artifact
-	// (skipped: not gateway-originated), and an already-synced gateway artifact (skipped: only
-	// cp_sync_status pending/failed is retried).
+	// Seed a pending gateway-originated artifact (pushed), a CP-originated artifact
+	// (skipped: not gateway-originated), and an already-synced gateway artifact. The full
+	// reconcile re-pushes ALL gateway-origin artifacts regardless of cp_sync_status so a
+	// new/purged control plane (or a re-registered gateway) receives them (#2659); the CP
+	// dedupes replay re-pushes idempotently.
 	if err := client.db.SaveConfig(&models.StoredConfig{
 		UUID:         "gw-artifact-1",
 		Kind:         models.KindRestApi,
@@ -244,7 +246,7 @@ func TestPushGatewayArtifacts(t *testing.T) {
 	}); err != nil {
 		t.Fatalf("seed cp config: %v", err)
 	}
-	// Already-synced gateway artifact: must NOT be re-pushed on reconnect.
+	// Already-synced gateway artifact: with full reconcile it IS re-pushed on (re)connect.
 	if err := client.db.SaveConfig(&models.StoredConfig{
 		UUID:         "gw-artifact-synced",
 		Kind:         models.KindRestApi,
@@ -280,11 +282,19 @@ func TestPushGatewayArtifacts(t *testing.T) {
 
 	mu.Lock()
 	defer mu.Unlock()
-	if len(pushedID) != 1 {
-		t.Fatalf("pushed %d artifacts, want 1 (only the gateway-originated one): %v", len(pushedID), pushedID)
+	// Both gateway-origin artifacts are pushed (pending + already-synced); the CP-origin one is skipped.
+	if len(pushedID) != 2 {
+		t.Fatalf("pushed %d artifacts, want 2 (both gateway-originated): %v", len(pushedID), pushedID)
 	}
-	if pushedID[0] != "gw-artifact-1" {
-		t.Errorf("pushed artifact ID = %q, want gw-artifact-1", pushedID[0])
+	got := map[string]bool{}
+	for _, id := range pushedID {
+		got[id] = true
+	}
+	if !got["gw-artifact-1"] || !got["gw-artifact-synced"] {
+		t.Errorf("pushed IDs = %v, want both gw-artifact-1 and gw-artifact-synced", pushedID)
+	}
+	if got["cp-artifact-1"] {
+		t.Errorf("CP-origin artifact was pushed but should be skipped: %v", pushedID)
 	}
 }
 
diff --git a/gateway/gateway-controller/pkg/controlplane/sync.go b/gateway/gateway-controller/pkg/controlplane/sync.go
index d0615af680..bff0f962b4 100644
--- a/gateway/gateway-controller/pkg/controlplane/sync.go
+++ b/gateway/gateway-controller/pkg/controlplane/sync.go
@@ -631,18 +631,17 @@ func (c *Client) pushGatewayArtifacts() {
 		return
 	}
 
-	// Only artifacts whose control-plane sync is incomplete (cp_sync_status pending/failed)
-	// are pushed. This avoids re-pushing artifacts that have already been successfully synced.
-	configs, err := c.db.GetPendingCPSyncArtifacts()
+	// Push ALL gateway-origin artifacts.
+	configs, err := c.db.GetGatewayOriginArtifactsForSync()
 	if err != nil {
-		c.logger.Error("Failed to list pending gateway-originated artifacts for control plane push", slog.Any("error", err))
+		c.logger.Error("Failed to list gateway-originated artifacts for control plane push", slog.Any("error", err))
 		return
 	}
 	if len(configs) == 0 {
 		return
 	}
 
-	// GetPendingCPSyncArtifacts returns metadata only; load each full config to push.
+	// The query returns metadata only; load each full config to push.
 	full := make([]*models.StoredConfig, 0, len(configs))
 	for _, meta := range configs {
 		cfg, err := c.db.GetConfig(meta.UUID)
@@ -657,9 +656,9 @@ func (c *Client) pushGatewayArtifacts() {
 		return
 	}
 
-	c.logger.Info("Pushing pending gateway-originated artifacts to control plane", slog.Int("count", len(full)))
+	c.logger.Info("Pushing gateway-originated artifacts to control plane", slog.Int("count", len(full)))
 
-	// Push the whole pending set as a single ordered, zipped batch.
+	// Push the whole set as a single ordered, zipped batch.
 	resp, err := c.pushArtifactsWithRetry(full)
 	if err != nil {
 		// Transport-level failure: the artifacts keep their pending/failed cp_sync_status and
diff --git a/gateway/gateway-controller/pkg/secrets/service_test.go b/gateway/gateway-controller/pkg/secrets/service_test.go
index a947c63985..3afd719d77 100644
--- a/gateway/gateway-controller/pkg/secrets/service_test.go
+++ b/gateway/gateway-controller/pkg/secrets/service_test.go
@@ -266,6 +266,9 @@ func (m *minimalStorage) GetPendingBottomUpAPIs() ([]*models.StoredConfig, error
 func (m *minimalStorage) GetPendingCPSyncArtifacts() ([]*models.StoredConfig, error) {
 	return nil, nil
 }
+func (m *minimalStorage) GetGatewayOriginArtifactsForSync() ([]*models.StoredConfig, error) {
+	return nil, nil
+}
 func (m *minimalStorage) UpdateCPSyncStatus(uuid, cpArtifactID string, status models.CPSyncStatus, reason string) error {
 	return nil
 }
diff --git a/gateway/gateway-controller/pkg/storage/interface.go b/gateway/gateway-controller/pkg/storage/interface.go
index 33b4e2806f..ac0e90f5f2 100644
--- a/gateway/gateway-controller/pkg/storage/interface.go
+++ b/gateway/gateway-controller/pkg/storage/interface.go
@@ -156,6 +156,10 @@ type Storage interface {
 	// is incomplete (cp_sync_status IN ('pending', 'failed')).
 	GetPendingCPSyncArtifacts() ([]*models.StoredConfig, error)
 
+	// GetGatewayOriginArtifactsForSync returns artifact metadata (Configuration nil) for every
+	// gateway-originated artifact across all artifacts-table kinds, regardless of cp_sync_status.
+	GetGatewayOriginArtifactsForSync() ([]*models.StoredConfig, error)
+
 	// ========================================
 	// LLM Provider Template Methods
 	// ========================================
diff --git a/gateway/gateway-controller/pkg/storage/llm_provider_template_artifacts_test.go b/gateway/gateway-controller/pkg/storage/llm_provider_template_artifacts_test.go
index c08ce9b462..e0972cd1b2 100644
--- a/gateway/gateway-controller/pkg/storage/llm_provider_template_artifacts_test.go
+++ b/gateway/gateway-controller/pkg/storage/llm_provider_template_artifacts_test.go
@@ -181,3 +181,32 @@ func TestSaveLLMProviderTemplate_DuplicateHandleConflicts(t *testing.T) {
 	_, err = store.GetConfig(dup.UUID)
 	assert.Assert(t, errors.Is(err, ErrNotFound))
 }
+
+// TestGetGatewayOriginArtifactsForSync_IncludesSuccess verifies the full-reconcile query returns
+// ALL gateway-origin artifacts regardless of cp_sync_status — including already-"success" ones —
+// whereas GetPendingCPSyncArtifacts excludes them. This is what lets a reconnect re-sync artifacts
+// to a new/purged control plane (#2659).
+func TestGetGatewayOriginArtifactsForSync_IncludesSuccess(t *testing.T) {
+	store := setupTestStorage(t)
+
+	// A synced (success) template and a not-yet-synced (pending) one.
+	synced := createTestLLMProviderTemplate()
+	pendingTmpl := createTestLLMProviderTemplate()
+	assert.NilError(t, store.SaveLLMProviderTemplate(synced))
+	assert.NilError(t, store.SaveLLMProviderTemplate(pendingTmpl))
+
+	// Mark the first as successfully synced.
+	assert.NilError(t, store.UpdateCPSyncStatus(synced.UUID, "cp-uuid-1", models.CPSyncStatusSuccess, ""))
+
+	// GetPendingCPSyncArtifacts excludes the success one.
+	pending, err := store.GetPendingCPSyncArtifacts()
+	assert.NilError(t, err)
+	assert.Assert(t, !pendingContains(pending, synced.UUID), "success artifact must NOT be pending")
+	assert.Assert(t, pendingContains(pending, pendingTmpl.UUID), "pending artifact must be pending")
+
+	// The full-reconcile query includes BOTH.
+	all, err := store.GetGatewayOriginArtifactsForSync()
+	assert.NilError(t, err)
+	assert.Assert(t, pendingContains(all, synced.UUID), "full reconcile must include the success artifact")
+	assert.Assert(t, pendingContains(all, pendingTmpl.UUID), "full reconcile must include the pending artifact")
+}
diff --git a/gateway/gateway-controller/pkg/storage/sql_store.go b/gateway/gateway-controller/pkg/storage/sql_store.go
index 8167a0d70e..656c338b67 100644
--- a/gateway/gateway-controller/pkg/storage/sql_store.go
+++ b/gateway/gateway-controller/pkg/storage/sql_store.go
@@ -1191,6 +1191,27 @@ func (s *sqlStore) GetPendingCPSyncArtifacts() ([]*models.StoredConfig, error) {
 	return scanArtifactMetadataRows(rows)
 }
 
+// GetGatewayOriginArtifactsForSync returns artifacts-table metadata for all gateway-originated
+// artifacts regardless of cp_sync_status.
+func (s *sqlStore) GetGatewayOriginArtifactsForSync() ([]*models.StoredConfig, error) {
+	query := `
+		SELECT uuid, kind, handle, display_name, version, data_version, desired_state,
+			deployment_id, origin, created_at, updated_at, deployed_at,
+			cp_sync_status, cp_sync_info, cp_artifact_id
+		FROM artifacts
+		WHERE gateway_id = ? AND origin = ?
+		ORDER BY created_at DESC
+	`
+
+	rows, err := s.query(query, s.gatewayId, string(models.OriginGatewayAPI))
+	if err != nil {
+		return nil, fmt.Errorf("failed to query gateway-origin artifacts for sync: %w", err)
+	}
+	defer rows.Close()
+
+	return scanArtifactMetadataRows(rows)
+}
+
 // scanArtifactMetadataRows scans rows from a query selecting the artifacts-table metadata
 // columns (no resource-table JOIN, so Configuration is left nil): uuid, kind, handle,
 // display_name, version, desired_state, deployment_id, origin, created_at, updated_at,
diff --git a/gateway/gateway-controller/pkg/subscriptionxds/subscription_snapshot_test.go b/gateway/gateway-controller/pkg/subscriptionxds/subscription_snapshot_test.go
index 3c1d6b4903..9ecbc82ff3 100644
--- a/gateway/gateway-controller/pkg/subscriptionxds/subscription_snapshot_test.go
+++ b/gateway/gateway-controller/pkg/subscriptionxds/subscription_snapshot_test.go
@@ -48,6 +48,10 @@ func (m *MockStorage) GetPendingCPSyncArtifacts() ([]*models.StoredConfig, error
 	return nil, nil
 }
 
+func (m *MockStorage) GetGatewayOriginArtifactsForSync() ([]*models.StoredConfig, error) {
+	return nil, nil
+}
+
 // UpdateCPSyncStatus implements [storage.Storage].
 func (m *MockStorage) UpdateCPSyncStatus(uuid, cpArtifactID string, status models.CPSyncStatus, reason string) error {
 	return nil
diff --git a/gateway/gateway-controller/pkg/utils/api_utils.go b/gateway/gateway-controller/pkg/utils/api_utils.go
index 53f0e73836..1ea2c1b9d0 100644
--- a/gateway/gateway-controller/pkg/utils/api_utils.go
+++ b/gateway/gateway-controller/pkg/utils/api_utils.go
@@ -966,8 +966,12 @@ type ImportArtifactRequest struct {
 	CreatedAt     time.Time              `json:"createdAt"`
 	UpdatedAt     time.Time              `json:"updatedAt"`
 	DeployedAt    *time.Time             `json:"deployedAt,omitempty"`
+	Properties    map[string]interface{} `json:"properties,omitempty"`
 }
 
+// deploymentRevisionProperty is the Properties key carrying the per-deployment revision.
+const deploymentRevisionProperty = "deploymentRevision"
+
 // ImportArtifactsResponse is the control plane's reply to the bulk DP->CP push: per-artifact
 // results keyed by the artifact's data-plane UUID (dpid), plus aggregate counts.
 type ImportArtifactsResponse struct {
@@ -1217,6 +1221,12 @@ func (s *APIUtilsService) buildImportArtifactRequest(artifact *models.StoredConf
 		utc := artifact.UpdatedAt.UTC()
 		deployedAt = &utc
 	}
+	var properties map[string]interface{}
+	if deployedAt != nil {
+		properties = map[string]interface{}{
+			deploymentRevisionProperty: deployedAt.Format(time.RFC3339Nano),
+		}
+	}
 	return ImportArtifactRequest{
 		DPID:          artifact.UUID,
 		Configuration: configuration,
@@ -1224,6 +1234,7 @@ func (s *APIUtilsService) buildImportArtifactRequest(artifact *models.StoredConf
 		CreatedAt:     artifact.CreatedAt.UTC(),
 		UpdatedAt:     artifact.UpdatedAt.UTC(),
 		DeployedAt:    deployedAt,
+		Properties:    properties,
 	}, nil
 }
 
diff --git a/gateway/gateway-controller/pkg/utils/api_utils_test.go b/gateway/gateway-controller/pkg/utils/api_utils_test.go
index 7cc845be77..0b6573b046 100644
--- a/gateway/gateway-controller/pkg/utils/api_utils_test.go
+++ b/gateway/gateway-controller/pkg/utils/api_utils_test.go
@@ -361,6 +361,33 @@ func TestAPIUtilsService_PushArtifact(t *testing.T) {
 			"template deployedAt should equal UpdatedAt; got %v want %v", got.DeployedAt, cfg.UpdatedAt.UTC())
 	})
 
+	t.Run("Deployable push carries a deploymentRevision derived from deployedAt", func(t *testing.T) {
+		// The control plane uses deploymentRevision to dedupe replay re-pushes (reconnect /
+		// full reconcile) so they do not create duplicate deployments. It is derived from the
+		// deployment time, so it is stable across re-pushes yet distinct across deployments.
+		var got ImportArtifactRequest
+		server := newHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			reqs := readPushedArtifacts(t, r)
+			require.Len(t, reqs, 1)
+			got = reqs[0]
+			w.WriteHeader(http.StatusOK)
+			w.Write([]byte(`{"total":1,"success":1,"failed":0,"artifacts":{"0000-test-api-0000-000000000000":{"id":"cp-rest-1","origin":"gateway_api","status":"deployed"}}}`))
+		}))
+		defer server.Close()
+
+		cfg := createTestStoredConfig("RestApi")
+		deployedAt := time.Date(2026, 5, 6, 7, 8, 9, 123456789, time.UTC)
+		cfg.DeployedAt = &deployedAt
+
+		svc := NewAPIUtilsService(PlatformAPIConfig{BaseURL: server.URL, Token: "test-token"}, logger)
+		_, err := svc.PushArtifact(cfg.UUID, cfg, "")
+		require.NoError(t, err)
+
+		require.NotNil(t, got.Properties, "deployable push must carry Properties")
+		assert.Equal(t, deployedAt.Format(time.RFC3339Nano), got.Properties[deploymentRevisionProperty],
+			"deploymentRevision must equal the deployedAt in RFC3339Nano")
+	})
+
 	t.Run("HTTP error response", func(t *testing.T) {
 		server := newHTTPTestServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 			w.WriteHeader(http.StatusInternalServerError)
diff --git a/gateway/gateway-controller/pkg/utils/mock_db_test.go b/gateway/gateway-controller/pkg/utils/mock_db_test.go
index ae70b18052..7698423eff 100644
--- a/gateway/gateway-controller/pkg/utils/mock_db_test.go
+++ b/gateway/gateway-controller/pkg/utils/mock_db_test.go
@@ -281,6 +281,16 @@ func (m *testMockDB) GetPendingCPSyncArtifacts() ([]*models.StoredConfig, error)
 	return pending, nil
 }
 
+func (m *testMockDB) GetGatewayOriginArtifactsForSync() ([]*models.StoredConfig, error) {
+	var all []*models.StoredConfig
+	for _, config := range m.configs {
+		if config.Origin == models.OriginGatewayAPI {
+			all = append(all, config)
+		}
+	}
+	return all, nil
+}
+
 func (m *testMockDB) SaveWebhookSecret(secret *models.WebhookSecret) error { return nil }
 func (m *testMockDB) GetWebhookSecretsByArtifact(artifactUUID string) ([]*models.WebhookSecret, error) {
 	return nil, nil
diff --git a/platform-api/internal/repository/api_deployments_test.go b/platform-api/internal/repository/api_deployments_test.go
index 463624e816..d9fb3dd046 100644
--- a/platform-api/internal/repository/api_deployments_test.go
+++ b/platform-api/internal/repository/api_deployments_test.go
@@ -603,3 +603,75 @@ func TestMain(m *testing.M) {
 	code := m.Run()
 	os.Exit(code)
 }
+
+// TestGetLatestDeploymentRevision verifies the per-gateway revision lookup used by the DP->CP
+// import flow to dedupe replay re-pushes.
+func TestGetLatestDeploymentRevision(t *testing.T) {
+	db, cleanup := setupTestDB(t)
+	defer cleanup()
+
+	const (
+		orgUUID = "org-rev-001"
+		apiUUID = "api-rev-001"
+		gwUUID  = "gw-rev-001"
+	)
+	createTestAPI(t, db, apiUUID, orgUUID)
+	createTestGateway(t, db, gwUUID, orgUUID)
+
+	repo := NewDeploymentRepo(db, NewArtifactTableRegistry())
+
+	// No deployment yet → empty revision.
+	rev, err := repo.GetLatestDeploymentRevision(apiUUID, gwUUID, orgUUID)
+	if err != nil {
+		t.Fatalf("GetLatestDeploymentRevision (none): %v", err)
+	}
+	if rev != "" {
+		t.Errorf("revision with no deployment = %q, want empty", rev)
+	}
+
+	deployed := model.DeploymentStatusDeployed
+	base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC)
+
+	// One deployment carrying a revision.
+	if err := repo.CreateWithLimitEnforcement(&model.Deployment{
+		DeploymentID:   "dep-rev-1",
+		Name:           "dep-1",
+		ArtifactID:     apiUUID,
+		GatewayID:      gwUUID,
+		OrganizationID: orgUUID,
+		Content:        []byte("c1"),
+		Status:         &deployed,
+		CreatedAt:      base,
+		Metadata:       map[string]any{"gatewayRevision": "rev-A"},
+	}, 100); err != nil {
+		t.Fatalf("create dep-1: %v", err)
+	}
+	if rev, err = repo.GetLatestDeploymentRevision(apiUUID, gwUUID, orgUUID); err != nil || rev != "rev-A" {
+		t.Errorf("revision after dep-1 = %q (err %v), want rev-A", rev, err)
+	}
+
+	// A newer deployment with a different revision wins (latest by created_at).
+	if err := repo.CreateWithLimitEnforcement(&model.Deployment{
+		DeploymentID:   "dep-rev-2",
+		Name:           "dep-2",
+		ArtifactID:     apiUUID,
+		GatewayID:      gwUUID,
+		OrganizationID: orgUUID,
+		Content:        []byte("c2"),
+		Status:         &deployed,
+		CreatedAt:      base.Add(time.Hour),
+		Metadata:       map[string]any{"gatewayRevision": "rev-B"},
+	}, 100); err != nil {
+		t.Fatalf("create dep-2: %v", err)
+	}
+	if rev, err = repo.GetLatestDeploymentRevision(apiUUID, gwUUID, orgUUID); err != nil || rev != "rev-B" {
+		t.Errorf("revision after dep-2 = %q (err %v), want rev-B", rev, err)
+	}
+
+	// A different gateway has no deployment → empty (per-gateway scoping).
+	const gw2 = "gw-rev-002"
+	createTestGateway(t, db, gw2, orgUUID)
+	if rev, err = repo.GetLatestDeploymentRevision(apiUUID, gw2, orgUUID); err != nil || rev != "" {
+		t.Errorf("revision for other gateway = %q (err %v), want empty", rev, err)
+	}
+}
diff --git a/platform-api/internal/repository/deployment.go b/platform-api/internal/repository/deployment.go
index 65feb41fe7..718c32a759 100644
--- a/platform-api/internal/repository/deployment.go
+++ b/platform-api/internal/repository/deployment.go
@@ -768,6 +768,37 @@ func (r *DeploymentRepo) GetLatestDeploymentTime(artifactUUID, orgUUID string) (
 	return &latest, nil
 }
 
+// GetLatestDeploymentRevision returns the gatewayRevision stored in the metadata of the most
+// recent deployment for the given (artifact, gateway, org), or "" when there is no deployment or
+// it carries no revision.
+func (r *DeploymentRepo) GetLatestDeploymentRevision(artifactUUID, gatewayUUID, orgUUID string) (string, error) {
+	query := `
+		SELECT metadata FROM deployments
+		WHERE artifact_uuid = ? AND gateway_uuid = ? AND organization_uuid = ?
+		ORDER BY created_at DESC
+		` + r.db.FetchFirstClause(1)
+	var metadataBytes []byte
+	if err := r.db.QueryRow(r.db.Rebind(query), artifactUUID, gatewayUUID, orgUUID).Scan(&metadataBytes); err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return "", nil
+		}
+		return "", err
+	}
+	if len(metadataBytes) == 0 {
+		return "", nil
+	}
+	var meta map[string]any
+	if err := json.Unmarshal(metadataBytes, &meta); err != nil {
+		// Unparseable metadata is treated as "no revision" rather than a hard error: the caller
+		// falls back to inserting a fresh deployment, which is safe (never drops a real deploy).
+		return "", nil
+	}
+	if rev, ok := meta["gatewayRevision"].(string); ok {
+		return rev, nil
+	}
+	return "", nil
+}
+
 // GetDeployedGatewayIDs returns the gateway IDs that have an active deployment status
 // (DEPLOYED or UNDEPLOYED) for the given artifact. Since the deployment_status table
 // only holds rows for those two states, a plain SELECT is sufficient.
diff --git a/platform-api/internal/repository/interfaces.go b/platform-api/internal/repository/interfaces.go
index e09724ac67..e256ddc210 100644
--- a/platform-api/internal/repository/interfaces.go
+++ b/platform-api/internal/repository/interfaces.go
@@ -146,6 +146,7 @@ type DeploymentRepository interface {
 	GetDeployedGatewayIDs(artifactUUID, orgUUID string) ([]string, error)
 	HasActiveDeployment(artifactUUID, orgUUID string) (bool, error)
 	GetLatestDeploymentTime(artifactUUID, orgUUID string) (*time.Time, error)
+	GetLatestDeploymentRevision(artifactUUID, gatewayUUID, orgUUID string) (string, error)
 
 	// Gateway deployment methods
 	GetControlPlaneDeploymentsByGateway(gatewayID, orgUUID string, since *time.Time) ([]*model.DeploymentInfo, error)
diff --git a/platform-api/internal/service/artifact_import.go b/platform-api/internal/service/artifact_import.go
index 8e45bdc2e4..a45b249236 100644
--- a/platform-api/internal/service/artifact_import.go
+++ b/platform-api/internal/service/artifact_import.go
@@ -90,6 +90,10 @@ type GatewayArtifactImporter interface {
 	Import(ctx *ImportContext) (*ImportResult, error)
 }
 
+// deploymentRevisionProperty is the key under which the gateway sends a stable per-deployment
+// revision string (its deployment time in RFC3339Nano) in the import request Properties bag.
+const deploymentRevisionProperty = "deploymentRevision"
+
 // ArtifactImportService resolves the correct importer for a pushed artifact and
 // runs the shared pre/post processing (gateway auth, project resolution,
 // deployment/status persistence).
@@ -357,6 +361,17 @@ func (s *ArtifactImportService) resolveAndImport(
 // writeDeployment records the immutable deployment artifact, upserts the current
 // deployment_status row, and ensures the artifact<->gateway association exists.
 func (s *ArtifactImportService) writeDeployment(ictx *ImportContext, artifactUUID string) error {
+	revision := utils.StringProperty(ictx.Properties, deploymentRevisionProperty)
+	if revision != "" {
+		latestRevision, err := s.deploymentRepo.GetLatestDeploymentRevision(artifactUUID, ictx.GatewayID, ictx.OrgID)
+		if err != nil {
+			return fmt.Errorf("failed to look up latest deployment revision: %w", err)
+		}
+		if latestRevision == revision {
+			return s.ensureGatewayAssociation(artifactUUID, ictx.GatewayID, ictx.OrgID)
+		}
+	}
+
 	content, err := yaml.Marshal(ictx.Configuration)
 	if err != nil {
 		return fmt.Errorf("failed to serialize deployment content: %w", err)
@@ -384,32 +399,33 @@ func (s *ArtifactImportService) writeDeployment(ictx *ImportContext, artifactUUI
 		Content:        content,
 		Status:         &status,
 		CreatedAt:      createdAt,
+		Metadata:       utils.RevisionMetadata(revision),
 	}
 	if err := s.deploymentRepo.CreateWithLimitEnforcement(deployment, hardLimit); err != nil {
 		return fmt.Errorf("failed to create deployment record: %w", err)
 	}
 
-	// Ensure the gateway association exists so the artifact is listed against the
-	// gateway in the control plane. association_mappings is generic across kinds.
-	assocs, err := s.apiRepo.GetAPIAssociations(artifactUUID, constants.AssociationTypeGateway, ictx.OrgID)
+	return s.ensureGatewayAssociation(artifactUUID, ictx.GatewayID, ictx.OrgID)
+}
+
+// ensureGatewayAssociation ensures the artifact<->gateway association exists so the artifact is
+// listed against the gateway in the control plane. association_mappings is generic across kinds.
+func (s *ArtifactImportService) ensureGatewayAssociation(artifactUUID, gatewayID, orgID string) error {
+	assocs, err := s.apiRepo.GetAPIAssociations(artifactUUID, constants.AssociationTypeGateway, orgID)
 	if err != nil {
 		return fmt.Errorf("failed to check artifact-gateway associations: %w", err)
 	}
-	associated := false
 	for _, a := range assocs {
-		if a.GatewayID == ictx.GatewayID {
-			associated = true
-			break
+		if a.GatewayID == gatewayID {
+			return nil
 		}
 	}
-	if !associated {
-		if err := s.apiRepo.CreateAPIAssociation(&model.APIAssociation{
-			ArtifactID:     artifactUUID,
-			OrganizationID: ictx.OrgID,
-			GatewayID:      ictx.GatewayID,
-		}); err != nil {
-			return fmt.Errorf("failed to create artifact-gateway association: %w", err)
-		}
+	if err := s.apiRepo.CreateAPIAssociation(&model.APIAssociation{
+		ArtifactID:     artifactUUID,
+		OrganizationID: orgID,
+		GatewayID:      gatewayID,
+	}); err != nil {
+		return fmt.Errorf("failed to create artifact-gateway association: %w", err)
 	}
 	return nil
 }
diff --git a/platform-api/internal/service/artifact_import_test.go b/platform-api/internal/service/artifact_import_test.go
index f5da028c14..9089049297 100644
--- a/platform-api/internal/service/artifact_import_test.go
+++ b/platform-api/internal/service/artifact_import_test.go
@@ -858,3 +858,143 @@ func TestArtifactImport_GatewayNotFound(t *testing.T) {
 		t.Fatalf("Import() error = %v, want ErrGatewayNotFound", err)
 	}
 }
+
+// withDeployedAtRev sets DeployedAt and the matching deploymentRevision property, mirroring how
+// the gateway builds an import request (revision derived from the deployment time). Distinct
+// deployment times therefore carry distinct revisions; a re-push of the same deployment carries
+// the same revision and is deduped by the control plane.
+func withDeployedAtRev(req dto.ImportGatewayArtifactRequest, t time.Time) dto.ImportGatewayArtifactRequest {
+	req.DeployedAt = &t
+	req.Properties = map[string]interface{}{deploymentRevisionProperty: t.UTC().Format(time.RFC3339Nano)}
+	return req
+}
+
+// countDeployments returns the number of deployment rows for an artifact on a given gateway.
+func countDeployments(t *testing.T, d *importTestDeps, artifactUUID, gatewayID string) int {
+	t.Helper()
+	var n int
+	if err := d.db.QueryRow(
+		`SELECT COUNT(*) FROM deployments WHERE artifact_uuid = ? AND gateway_uuid = ?`,
+		artifactUUID, gatewayID,
+	).Scan(&n); err != nil {
+		t.Fatalf("count deployments: %v", err)
+	}
+	return n
+}
+
+// A replay re-push (reconnect / full reconcile) carries the same deploymentRevision as the current
+// deployment. It must not create a duplicate deployment row, but must keep the artifact and its
+// gateway association intact.
+func TestArtifactImport_ReplayDoesNotDuplicateDeployment(t *testing.T) {
+	d := setupImportTest(t)
+
+	const id = "22222222-2222-2222-2222-222222222222"
+	resp, err := d.svc.Import(importTestOrgID, importTestGatewayID,
+		withDeployedAtRev(restImportRequest(id, "replay-api", "Replay API"), baseDeployedAt))
+	if err != nil {
+		t.Fatalf("initial import: %v", err)
+	}
+	cpID := resp.ID
+	if got := countDeployments(t, d, cpID, importTestGatewayID); got != 1 {
+		t.Fatalf("after initial import: deployments = %d, want 1", got)
+	}
+
+	// Re-push the same artifact twice (same revision) — simulates reconnect/full reconcile.
+	for i := 0; i < 2; i++ {
+		if _, err := d.svc.Import(importTestOrgID, importTestGatewayID,
+			withDeployedAtRev(restImportRequest(id, "replay-api", "Replay API"), baseDeployedAt)); err != nil {
+			t.Fatalf("replay import %d: %v", i, err)
+		}
+	}
+	if got := countDeployments(t, d, cpID, importTestGatewayID); got != 1 {
+		t.Errorf("after replay re-pushes: deployments = %d, want 1 (no duplicates)", got)
+	}
+	// The association must still be present exactly once.
+	assocs, err := d.apiRepo.GetAPIAssociations(cpID, constants.AssociationTypeGateway, importTestOrgID)
+	if err != nil {
+		t.Fatalf("get associations: %v", err)
+	}
+	count := 0
+	for _, a := range assocs {
+		if a.GatewayID == importTestGatewayID {
+			count++
+		}
+	}
+	if count != 1 {
+		t.Errorf("gateway associations = %d, want 1", count)
+	}
+}
+
+// Two genuinely distinct deployments (e.g. a create then an update, even when they reach the CP
+// out of order) carry distinct revisions and must both be recorded, while last-in-wins keeps the
+// newest working copy. This models the create/PUT reordering: the newer PUT lands first.
+func TestArtifactImport_DistinctRevisionsRecordSeparateDeployments(t *testing.T) {
+	d := setupImportTest(t)
+
+	const id = "33333333-3333-3333-3333-333333333333"
+	// PUT (newer) arrives first: creates the artifact with the newer working copy.
+	resp, err := d.svc.Import(importTestOrgID, importTestGatewayID,
+		withDeployedAtRev(restImportRequest(id, "ro-api", "PUT Name"), newerDeployedAt))
+	if err != nil {
+		t.Fatalf("put import: %v", err)
+	}
+	cpID := resp.ID
+
+	// CREATE (older) arrives second: stale for the working copy (SkipWorkingCopy) but a genuinely
+	// distinct deployment — it must add a second deployment row.
+	if _, err := d.svc.Import(importTestOrgID, importTestGatewayID,
+		withDeployedAtRev(restImportRequest(id, "ro-api", "CREATE Name"), baseDeployedAt)); err != nil {
+		t.Fatalf("create import: %v", err)
+	}
+
+	if got := countDeployments(t, d, cpID, importTestGatewayID); got != 2 {
+		t.Errorf("distinct deployments = %d, want 2", got)
+	}
+	// Last-in-wins: the newer PUT metadata must survive the later, older CREATE push.
+	art, _ := d.artifactRepo.GetByUUID(cpID, importTestOrgID)
+	if art == nil || art.Name != "PUT Name" {
+		t.Errorf("working copy = %v, want newer 'PUT Name' to win", art)
+	}
+	// The latest deployment watermark is the newer time.
+	latest, err := d.deployment.GetLatestDeploymentTime(cpID, importTestOrgID)
+	if err != nil || latest == nil || !latest.Equal(newerDeployedAt) {
+		t.Errorf("latest deployment time = %v, want %v", latest, newerDeployedAt)
+	}
+}
+
+// The revision match is scoped per gateway: a second gateway re-pushing the same artifact with the
+// same revision must still record its own deployment (a new gateway registration must be synced).
+func TestArtifactImport_DifferentGatewaySameRevisionRecordsOwnDeployment(t *testing.T) {
+	d := setupImportTest(t)
+
+	const gw2 = "gw-import-002"
+	if _, err := d.db.Exec(`INSERT INTO gateways (uuid, organization_uuid, handle, display_name, description, properties, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))`,
+		gw2, importTestOrgID, "gw2", "Gateway 2", "", "{}"); err != nil {
+		t.Fatalf("seed gateway2: %v", err)
+	}
+	// The second gateway must authenticate; the import resolves the gateway by the ID passed in,
+	// so seeding the row is sufficient for the service to accept it.
+
+	const id = "44444444-4444-4444-4444-444444444444"
+	req := withDeployedAtRev(restImportRequest(id, "mg-api", "Multi GW API"), baseDeployedAt)
+
+	resp, err := d.svc.Import(importTestOrgID, importTestGatewayID, req)
+	if err != nil {
+		t.Fatalf("gw1 import: %v", err)
+	}
+	cpID := resp.ID
+
+	// Same artifact (same handle → same CP UUID) and same revision, but pushed by a different
+	// gateway: must create a deployment for gw2 rather than being deduped as a replay.
+	if _, err := d.svc.Import(importTestOrgID, gw2, req); err != nil {
+		t.Fatalf("gw2 import: %v", err)
+	}
+
+	if got := countDeployments(t, d, cpID, importTestGatewayID); got != 1 {
+		t.Errorf("gw1 deployments = %d, want 1", got)
+	}
+	if got := countDeployments(t, d, cpID, gw2); got != 1 {
+		t.Errorf("gw2 deployments = %d, want 1", got)
+	}
+}
diff --git a/platform-api/internal/utils/import_artifacts.go b/platform-api/internal/utils/import_artifacts.go
index 455266471a..69bdc39b02 100644
--- a/platform-api/internal/utils/import_artifacts.go
+++ b/platform-api/internal/utils/import_artifacts.go
@@ -279,3 +279,24 @@ func DecodeSpec(spec map[string]interface{}, out interface{}) error {
 	}
 	return nil
 }
+
+// StringProperty reads a string value from an import request Properties bag, returning "" when
+// the bag is nil, the key is absent, or the value is not a string.
+func StringProperty(props map[string]interface{}, key string) string {
+	if props == nil {
+		return ""
+	}
+	if v, ok := props[key].(string); ok {
+		return v
+	}
+	return ""
+}
+
+// RevisionMetadata builds the deployment metadata map carrying the gateway revision, or nil when
+// no revision was supplied (so pre-revision pushes store no metadata, as before).
+func RevisionMetadata(revision string) map[string]any {
+	if revision == "" {
+		return nil
+	}
+	return map[string]any{"gatewayRevision": revision}
+}