diff --git a/CHANGELOG.md b/CHANGELOG.md
index a5182671a2..c3081d25c3 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,8 @@
 ## main / (unreleased)
 
+* [CHANGE] limit: `alertmanager_alerts_limited_total` now has a `state` label (`firing`/`resolved`) to distinguish dropped firing alerts from dropped resolved notifications. Dashboards/alerts using this metric must be updated.
+* [CHANGE] limit: With `--alerts.per-alertname-limit` set, a resolved notification is only forwarded if its firing counterpart was previously admitted (which frees its slot); resolved notifications with no admitted firing alert are now dropped.
+
 ## 0.34.0 / 2026-08-16
 
 * [CHANGE] notify: The `reason` label on `alertmanager_notifications_failed_total` now distinguishes `authError` (HTTP 401/403) and `rateLimited` (HTTP 429) from the generic `clientError`. Dashboards/alerts matching `reason="clientError"` for these codes must be updated. #5332
diff --git a/docs/alertmanager.md b/docs/alertmanager.md
index 9da4ed4e43..fab3688abc 100644
--- a/docs/alertmanager.md
+++ b/docs/alertmanager.md
@@ -78,11 +78,19 @@ It's important not to load balance traffic between Prometheus and its Alertmanag
 Alertmanager supports configuration to limit the number of active alerts per alertname.
 This can be configured using the [--alerts.per-alertname-limit] flag.
 
-When the limit is reached any new alerts are dropped, heartbeats from already know alerts are processed.
-The known alert (fingerprint) automatically expire to make room for new alerts.
+When the limit is reached any new firing alerts are dropped, while heartbeats from
+already known alerts are still processed. Known alerts (fingerprints) automatically
+expire to make room for new alerts.
+
+Resolved notifications are handled specially. A resolved notification is only
+forwarded if its firing counterpart was previously admitted (its fingerprint is
+still tracked); forwarding it frees the slot the firing alert was holding. A
+resolved notification with no previously admitted firing alert is dropped as
+noise, since nothing downstream ever received a firing alert for it.
 
 This feature is useful when an unexpected high number of instances of the same alert are sent to Alertmanager.
 Limiting the number of alerts per alertname can prevent reliability issues and avoid alert receivers from being flooded.
 
 The `alertmanager_alerts_limited_total` metric shows the total number of alerts that were dropped due to per alert name limit.
+The `state` label distinguishes dropped `firing` alerts from dropped `resolved` notifications.
 Enabling the `alert-names-in-metrics` feature flag will add the `alertname` label to the metric.
\ No newline at end of file
diff --git a/docs/configuration.md b/docs/configuration.md
index be9a73f841..ef057f6e9a 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -29,7 +29,13 @@ use the `--silences.max-silences` flag.
 You can limit the maximum size of individual silences with `--silences.max-silence-size-bytes`,
 where the unit is in bytes.
 
-Both limits are disabled by default.
+To limit the maximum number of active alerts per alertname, use the
+`--alerts.per-alertname-limit` flag. When the limit is reached, new firing alerts
+for that alertname are dropped. Resolved notifications are only forwarded when
+their firing counterpart was previously admitted; otherwise they are dropped. See
+[Alert limits](alertmanager.md#alert-limits-optional) for details.
+
+All limits are disabled by default.
 
 ## Configuration file introduction
 
diff --git a/limit/bucket.go b/limit/bucket.go
index bccc91bcc5..61cd830231 100644
--- a/limit/bucket.go
+++ b/limit/bucket.go
@@ -97,6 +97,21 @@ func NewBucket[V comparable](capacity int) *Bucket[V] {
 	}
 }
 
+// Remove deletes the value from the bucket, freeing its slot.
+// It returns true if the value was present.
+func (b *Bucket[V]) Remove(value V) bool {
+	b.mtx.Lock()
+	defer b.mtx.Unlock()
+
+	item, ok := b.index[value]
+	if !ok {
+		return false
+	}
+	heap.Remove(&b.items, item.index)
+	delete(b.index, value)
+	return true
+}
+
 // IsStale returns true if the latest item in the bucket is expired.
 func (b *Bucket[V]) IsStale() (stale bool) {
 	b.mtx.Lock()
diff --git a/limit/bucket_test.go b/limit/bucket_test.go
index 93901aef26..16eff05810 100644
--- a/limit/bucket_test.go
+++ b/limit/bucket_test.go
@@ -238,6 +238,116 @@ func TestBucketAddEdgeCases(t *testing.T) {
 	})
 }
 
+func TestBucketRemove(t *testing.T) {
+	t.Run("Remove present fingerprint returns true and frees its slot", func(t *testing.T) {
+		bucket := NewBucket[model.Fingerprint](2)
+		alert := model.Alert{Labels: model.LabelSet{"alertname": "Alert1"}, EndsAt: time.Now().Add(1 * time.Hour)}
+
+		require.True(t, bucket.Upsert(alert.Fingerprint(), alert.EndsAt), "alert should be added")
+
+		require.True(t, bucket.Remove(alert.Fingerprint()), "removing a present fingerprint should return true")
+		require.Empty(t, bucket.index, "index should be empty after removal")
+		require.Zero(t, bucket.items.Len(), "heap should be empty after removal")
+	})
+
+	t.Run("Remove absent fingerprint returns false", func(t *testing.T) {
+		bucket := NewBucket[model.Fingerprint](2)
+		present := model.Alert{Labels: model.LabelSet{"alertname": "Present"}, EndsAt: time.Now().Add(1 * time.Hour)}
+		bucket.Upsert(present.Fingerprint(), present.EndsAt)
+
+		absent := model.Alert{Labels: model.LabelSet{"alertname": "Absent"}}
+		require.False(t, bucket.Remove(absent.Fingerprint()), "removing an absent fingerprint should return false")
+		require.Len(t, bucket.index, 1, "present item should remain")
+		require.Equal(t, 1, bucket.items.Len(), "present item should remain in heap")
+	})
+
+	t.Run("Remove from empty bucket returns false", func(t *testing.T) {
+		bucket := NewBucket[model.Fingerprint](2)
+		alert := model.Alert{Labels: model.LabelSet{"alertname": "Alert1"}}
+		require.False(t, bucket.Remove(alert.Fingerprint()), "removing from an empty bucket should return false")
+	})
+
+	t.Run("Removing twice returns false the second time", func(t *testing.T) {
+		bucket := NewBucket[model.Fingerprint](2)
+		alert := model.Alert{Labels: model.LabelSet{"alertname": "Alert1"}, EndsAt: time.Now().Add(1 * time.Hour)}
+		bucket.Upsert(alert.Fingerprint(), alert.EndsAt)
+
+		require.True(t, bucket.Remove(alert.Fingerprint()), "first removal should return true")
+		require.False(t, bucket.Remove(alert.Fingerprint()), "second removal should return false")
+	})
+
+	t.Run("Remove frees a slot in a full bucket so a new alert is admitted", func(t *testing.T) {
+		bucket := NewBucket[model.Fingerprint](2)
+		a := model.Alert{Labels: model.LabelSet{"alertname": "A"}, EndsAt: time.Now().Add(1 * time.Hour)}
+		b := model.Alert{Labels: model.LabelSet{"alertname": "B"}, EndsAt: time.Now().Add(1 * time.Hour)}
+		c := model.Alert{Labels: model.LabelSet{"alertname": "C"}, EndsAt: time.Now().Add(1 * time.Hour)}
+
+		require.True(t, bucket.Upsert(a.Fingerprint(), a.EndsAt), "A should be added")
+		require.True(t, bucket.Upsert(b.Fingerprint(), b.EndsAt), "B should be added")
+		require.False(t, bucket.Upsert(c.Fingerprint(), c.EndsAt), "C should be rejected while bucket is full of active items")
+
+		require.True(t, bucket.Remove(a.Fingerprint()), "A should be removed")
+		require.True(t, bucket.Upsert(c.Fingerprint(), c.EndsAt), "C should be admitted after A freed a slot")
+
+		_, hasA := bucket.index[a.Fingerprint()]
+		_, hasC := bucket.index[c.Fingerprint()]
+		require.False(t, hasA, "A should no longer be present")
+		require.True(t, hasC, "C should now be present")
+		require.Len(t, bucket.index, 2, "bucket should hold B and C")
+	})
+
+	t.Run("Remove keeps heap eviction order intact", func(t *testing.T) {
+		bucket := NewBucket[model.Fingerprint](3)
+		oldest := model.Alert{Labels: model.LabelSet{"alertname": "Oldest"}, EndsAt: time.Now().Add(-2 * time.Hour)}
+		middle := model.Alert{Labels: model.LabelSet{"alertname": "Middle"}, EndsAt: time.Now().Add(-1 * time.Hour)}
+		newest := model.Alert{Labels: model.LabelSet{"alertname": "Newest"}, EndsAt: time.Now().Add(1 * time.Hour)}
+
+		bucket.Upsert(oldest.Fingerprint(), oldest.EndsAt)
+		bucket.Upsert(middle.Fingerprint(), middle.EndsAt)
+		bucket.Upsert(newest.Fingerprint(), newest.EndsAt)
+
+		// Remove the current heap root (oldest); the next-oldest expired item
+		// must then sit at the root and be the one evicted when full.
+		require.True(t, bucket.Remove(oldest.Fingerprint()), "oldest should be removed")
+		require.Equal(t, middle.Fingerprint(), bucket.items[0].value, "middle should be the new heap root")
+
+		// Bucket has room for one more; fill it, then force an eviction.
+		other := model.Alert{Labels: model.LabelSet{"alertname": "Other"}, EndsAt: time.Now().Add(2 * time.Hour)}
+		bucket.Upsert(other.Fingerprint(), other.EndsAt)
+
+		evictor := model.Alert{Labels: model.LabelSet{"alertname": "Evictor"}, EndsAt: time.Now().Add(3 * time.Hour)}
+		require.True(t, bucket.Upsert(evictor.Fingerprint(), evictor.EndsAt), "evictor should replace the expired middle item")
+
+		_, hasMiddle := bucket.index[middle.Fingerprint()]
+		require.False(t, hasMiddle, "expired middle item should have been evicted")
+		require.Len(t, bucket.index, 3, "index and heap should stay consistent")
+		require.Equal(t, 3, bucket.items.Len(), "index and heap should stay consistent")
+	})
+}
+
+func TestBucketRemoveConcurrency(t *testing.T) {
+	bucket := NewBucket[model.Fingerprint](2)
+	alert1 := model.Alert{Labels: model.LabelSet{"alertname": "Alert1"}, EndsAt: time.Now().Add(1 * time.Hour)}
+	alert2 := model.Alert{Labels: model.LabelSet{"alertname": "Alert2"}, EndsAt: time.Now().Add(1 * time.Hour)}
+	bucket.Upsert(alert1.Fingerprint(), alert1.EndsAt)
+	bucket.Upsert(alert2.Fingerprint(), alert2.EndsAt)
+
+	done := make(chan bool, 2)
+	go func() {
+		bucket.Remove(alert1.Fingerprint())
+		done <- true
+	}()
+	go func() {
+		bucket.Remove(alert2.Fingerprint())
+		done <- true
+	}()
+	<-done
+	<-done
+
+	require.Empty(t, bucket.index, "both alerts should be removed after concurrent removes")
+	require.Zero(t, bucket.items.Len(), "heap should be empty after concurrent removes")
+}
+
 // Benchmark tests for Bucket.Upsert() performance.
 func BenchmarkBucketUpsert(b *testing.B) {
 	b.Run("EmptyBucket", func(b *testing.B) {
diff --git a/provider/mem/mem.go b/provider/mem/mem.go
index 7646eb9fdf..4b2178b67b 100644
--- a/provider/mem/mem.go
+++ b/provider/mem/mem.go
@@ -28,13 +28,13 @@ import (
 	"go.opentelemetry.io/otel/propagation"
 	"go.opentelemetry.io/otel/trace"
 
+	"github.com/prometheus/alertmanager/alert"
 	"github.com/prometheus/alertmanager/eventrecorder"
 	"github.com/prometheus/alertmanager/eventrecorder/eventrecorderpb"
 	"github.com/prometheus/alertmanager/featurecontrol"
 	"github.com/prometheus/alertmanager/provider"
 	"github.com/prometheus/alertmanager/store"
 	"github.com/prometheus/alertmanager/tracing"
-	"github.com/prometheus/alertmanager/types"
 )
 
 const alertChannelLength = 200
@@ -69,13 +69,13 @@ type AlertStoreCallback interface {
 	// alert is not stored.
 	// Existing flag indicates whether alert has existed before (and is only updated) or not.
 	// If alert has existed before, then alert passed to PreStore is result of merging existing alert with new alert.
-	PreStore(alert *types.Alert, existing bool) error
+	PreStore(alert *alert.Alert, existing bool) error
 
 	// PostStore is called after alert has been put into store.
-	PostStore(alert *types.Alert, existing bool)
+	PostStore(alert *alert.Alert, existing bool)
 
 	// PostDelete is called after alert have been removed from the store due to alert garbage collection.
-	PostDelete(alert *types.Alert)
+	PostDelete(alert *alert.Alert)
 
 	// PostGC is called after alerts have been removed from the store due to alert garbage collection.
 	PostGC(fingerprints model.Fingerprints)
@@ -97,6 +97,7 @@ func (a *Alerts) registerMetrics(r prometheus.Registerer) {
 	if a.flagger.EnableAlertNamesInMetrics() {
 		labels = append(labels, "alertname")
 	}
+	labels = append(labels, "state")
 	a.alertsLimitedTotal = promauto.With(r).NewCounterVec(
 		prometheus.CounterOpts{
 			Name: "alertmanager_alerts_limited_total",
@@ -194,7 +195,7 @@ func (a *Alerts) gc() {
 	a.callback.PostGC(ff)
 }
 
-func (a *Alerts) gcAlerts() []*types.Alert {
+func (a *Alerts) gcAlerts() []*alert.Alert {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
 	return a.alerts.GC()
@@ -247,7 +248,7 @@ func (a *Alerts) Subscribe(name string) provider.AlertIterator {
 	return provider.NewAlertIterator(ch, done, nil)
 }
 
-func (a *Alerts) SlurpAndSubscribe(name string) ([]*types.Alert, provider.AlertIterator) {
+func (a *Alerts) SlurpAndSubscribe(name string) ([]*alert.Alert, provider.AlertIterator) {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
 
@@ -292,14 +293,14 @@ func (a *Alerts) GetPending() provider.AlertIterator {
 }
 
 // Get returns the alert for a given fingerprint.
-func (a *Alerts) Get(fp model.Fingerprint) (*types.Alert, error) {
+func (a *Alerts) Get(fp model.Fingerprint) (*alert.Alert, error) {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
 	return a.alerts.Get(fp)
 }
 
 // Put adds the given alert to the set.
-func (a *Alerts) Put(ctx context.Context, alerts ...*types.Alert) error {
+func (a *Alerts) Put(ctx context.Context, alerts ...*alert.Alert) error {
 	a.mtx.Lock()
 	defer a.mtx.Unlock()
 
@@ -340,6 +341,11 @@ func (a *Alerts) Put(ctx context.Context, alerts ...*types.Alert) error {
 				if a.flagger.EnableAlertNamesInMetrics() {
 					labels = append(labels, alert.Name())
 				}
+				state := "firing"
+				if alert.Resolved() {
+					state = "resolved"
+				}
+				labels = append(labels, state)
 				a.alertsLimitedTotal.WithLabelValues(labels...).Inc()
 			}
 			continue
@@ -374,7 +380,7 @@ func (a *Alerts) Put(ctx context.Context, alerts ...*types.Alert) error {
 
 type noopCallback struct{}
 
-func (n noopCallback) PreStore(_ *types.Alert, _ bool) error { return nil }
-func (n noopCallback) PostStore(_ *types.Alert, _ bool)      {}
-func (n noopCallback) PostDelete(_ *types.Alert)             {}
+func (n noopCallback) PreStore(_ *alert.Alert, _ bool) error { return nil }
+func (n noopCallback) PostStore(_ *alert.Alert, _ bool)      {}
+func (n noopCallback) PostDelete(_ *alert.Alert)             {}
 func (n noopCallback) PostGC(_ model.Fingerprints)           {}
diff --git a/provider/mem/mem_test.go b/provider/mem/mem_test.go
index 9db6ac365b..753a0b318e 100644
--- a/provider/mem/mem_test.go
+++ b/provider/mem/mem_test.go
@@ -29,16 +29,16 @@ import (
 	"github.com/prometheus/common/promslog"
 	"github.com/stretchr/testify/require"
 
+	"github.com/prometheus/alertmanager/alert"
 	"github.com/prometheus/alertmanager/eventrecorder"
 	"github.com/prometheus/alertmanager/store"
-	"github.com/prometheus/alertmanager/types"
 )
 
 var (
 	t0 = time.Now()
 	t1 = t0.Add(100 * time.Millisecond)
 
-	alert1 = &types.Alert{
+	alert1 = &alert.Alert{
 		Alert: model.Alert{
 			Labels:       model.LabelSet{"bar": "foo"},
 			Annotations:  model.LabelSet{"foo": "bar"},
@@ -50,7 +50,7 @@ var (
 		Timeout:   false,
 	}
 
-	alert2 = &types.Alert{
+	alert2 = &alert.Alert{
 		Alert: model.Alert{
 			Labels:       model.LabelSet{"bar": "foo2"},
 			Annotations:  model.LabelSet{"foo": "bar2"},
@@ -62,7 +62,7 @@ var (
 		Timeout:   false,
 	}
 
-	alert3 = &types.Alert{
+	alert3 = &alert.Alert{
 		Alert: model.Alert{
 			Labels:       model.LabelSet{"bar": "foo3"},
 			Annotations:  model.LabelSet{"foo": "bar3"},
@@ -89,10 +89,10 @@ func TestAlertsSubscribePutStarvation(t *testing.T) {
 
 	iterator := alerts.Subscribe("test")
 
-	alertsToInsert := []*types.Alert{}
+	alertsToInsert := []*alert.Alert{}
 	// Exhaust alert channel
 	for i := range alertChannelLength + 1 {
-		alertsToInsert = append(alertsToInsert, &types.Alert{
+		alertsToInsert = append(alertsToInsert, &alert.Alert{
 			Alert: model.Alert{
 				// Make sure the fingerprints differ
 				Labels:       model.LabelSet{"iteration": model.LabelValue(strconv.Itoa(i))},
@@ -140,9 +140,9 @@ func TestDeadLock(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	alertsToInsert := []*types.Alert{}
+	alertsToInsert := []*alert.Alert{}
 	for i := range 200 + 1 {
-		alertsToInsert = append(alertsToInsert, &types.Alert{
+		alertsToInsert = append(alertsToInsert, &alert.Alert{
 			Alert: model.Alert{
 				// Make sure the fingerprints differ
 				Labels:       model.LabelSet{"iteration": model.LabelValue(strconv.Itoa(i))},
@@ -193,7 +193,7 @@ func TestAlertsPut(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	insert := []*types.Alert{alert1, alert2, alert3}
+	insert := []*alert.Alert{alert1, alert2, alert3}
 
 	if err := alerts.Put(context.Background(), insert...); err != nil {
 		t.Fatalf("Insert failed: %s", err)
@@ -220,7 +220,7 @@ func TestAlertsSubscribe(t *testing.T) {
 		t.Fatalf("Insert failed: %s", err)
 	}
 
-	expectedAlerts := map[model.Fingerprint]*types.Alert{
+	expectedAlerts := map[model.Fingerprint]*alert.Alert{
 		alert1.Fingerprint(): alert1,
 		alert2.Fingerprint(): alert2,
 		alert3.Fingerprint(): alert3,
@@ -296,7 +296,7 @@ func TestAlertsGetPending(t *testing.T) {
 		t.Fatalf("Insert failed: %s", err)
 	}
 
-	expectedAlerts := map[model.Fingerprint]*types.Alert{
+	expectedAlerts := map[model.Fingerprint]*alert.Alert{
 		alert1.Fingerprint(): alert1,
 		alert2.Fingerprint(): alert2,
 	}
@@ -310,7 +310,7 @@ func TestAlertsGetPending(t *testing.T) {
 		t.Fatalf("Insert failed: %s", err)
 	}
 
-	expectedAlerts = map[model.Fingerprint]*types.Alert{
+	expectedAlerts = map[model.Fingerprint]*alert.Alert{
 		alert1.Fingerprint(): alert1,
 		alert2.Fingerprint(): alert2,
 		alert3.Fingerprint(): alert3,
@@ -328,7 +328,7 @@ func TestAlertsGC(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	insert := []*types.Alert{alert1, alert2, alert3}
+	insert := []*alert.Alert{alert1, alert2, alert3}
 
 	if err := alerts.Put(context.Background(), insert...); err != nil {
 		t.Fatalf("Insert failed: %s", err)
@@ -363,7 +363,7 @@ func TestAlertsStoreCallback(t *testing.T) {
 	alert1Mod := *alert1
 	alert1Mod.Annotations = model.LabelSet{"foo": "bar", "new": "test"} // Update annotations for alert1
 
-	alert4 := &types.Alert{
+	alert4 := &alert.Alert{
 		Alert: model.Alert{
 			Labels:       model.LabelSet{"bar4": "foo4"},
 			Annotations:  model.LabelSet{"foo4": "bar4"},
@@ -405,7 +405,7 @@ func TestAlertsStoreCallback(t *testing.T) {
 	}
 }
 
-func alertDiff(left, right *types.Alert) error {
+func alertDiff(left, right *alert.Alert) error {
 	if left == nil || right == nil {
 		return errors.New("should not be nil")
 	}
@@ -440,7 +440,7 @@ type limitCountCallback struct {
 
 var errTooManyAlerts = fmt.Errorf("too many alerts")
 
-func (l *limitCountCallback) PreStore(_ *types.Alert, existing bool) error {
+func (l *limitCountCallback) PreStore(_ *alert.Alert, existing bool) error {
 	if existing {
 		return nil
 	}
@@ -452,14 +452,14 @@ func (l *limitCountCallback) PreStore(_ *types.Alert, existing bool) error {
 	return nil
 }
 
-func (l *limitCountCallback) PostStore(_ *types.Alert, existing bool) {
+func (l *limitCountCallback) PostStore(_ *alert.Alert, existing bool) {
 	if !existing {
 		l.alerts.Add(1)
 		l.gcCount.Add(1)
 	}
 }
 
-func (l *limitCountCallback) PostDelete(_ *types.Alert) {
+func (l *limitCountCallback) PostDelete(_ *alert.Alert) {
 	l.alerts.Add(-1)
 }
 
@@ -492,7 +492,7 @@ func TestAlertsConcurrently(t *testing.T) {
 				default:
 				}
 				now := time.Now()
-				err := a.Put(context.Background(), &types.Alert{
+				err := a.Put(context.Background(), &alert.Alert{
 					Alert: model.Alert{
 						Labels:   model.LabelSet{"bar": model.LabelValue(strconv.Itoa(j))},
 						StartsAt: now,
@@ -566,7 +566,7 @@ func TestSubscriberChannelMetrics(t *testing.T) {
 
 	// Put some alerts
 	now := time.Now()
-	alertsToSend := []*types.Alert{
+	alertsToSend := []*alert.Alert{
 		{
 			Alert: model.Alert{
 				Labels:       model.LabelSet{"test": "1"},
diff --git a/provider/provider.go b/provider/provider.go
index dcc40f89fe..0412845ff7 100644
--- a/provider/provider.go
+++ b/provider/provider.go
@@ -19,7 +19,7 @@ import (
 
 	"github.com/prometheus/common/model"
 
-	"github.com/prometheus/alertmanager/types"
+	"github.com/prometheus/alertmanager/alert"
 )
 
 // ErrNotFound is returned if a provider cannot find a requested item.
@@ -28,7 +28,7 @@ var ErrNotFound = fmt.Errorf("item not found")
 type Alert struct {
 	// Header contains metadata, for example propagated tracing information.
 	Header map[string]string
-	Data   *types.Alert
+	Data   *alert.Alert
 }
 
 // Iterator provides the functions common to all iterators. To be useful, a
@@ -93,13 +93,13 @@ type Alerts interface {
 	// Implementation of SlurpAndSubcribe is optional - providers may choose to
 	// return an empty list for the first return value and the result of Subscribe
 	// for the second return value.
-	SlurpAndSubscribe(name string) ([]*types.Alert, AlertIterator)
+	SlurpAndSubscribe(name string) ([]*alert.Alert, AlertIterator)
 
 	// GetPending returns an iterator over all alerts that have
 	// pending notifications.
 	GetPending() AlertIterator
 	// Get returns the alert for a given fingerprint.
-	Get(model.Fingerprint) (*types.Alert, error)
+	Get(model.Fingerprint) (*alert.Alert, error)
 	// Put adds the given set of alerts to the set.
-	Put(ctx context.Context, alerts ...*types.Alert) error
+	Put(ctx context.Context, alerts ...*alert.Alert) error
 }
diff --git a/store/store.go b/store/store.go
index ba6b718401..3aae128dbd 100644
--- a/store/store.go
+++ b/store/store.go
@@ -21,8 +21,8 @@ import (
 
 	"github.com/prometheus/common/model"
 
+	"github.com/prometheus/alertmanager/alert"
 	"github.com/prometheus/alertmanager/limit"
-	"github.com/prometheus/alertmanager/types"
 )
 
 // ErrLimited is returned if a Store has reached the per-alert limit.
@@ -40,8 +40,8 @@ var ErrDestroyed = errors.New("alert store destroyed")
 // resolved alerts that have been removed.
 type Alerts struct {
 	sync.Mutex
-	alerts        map[model.Fingerprint]*types.Alert
-	gcCallback    func([]*types.Alert)
+	alerts        map[model.Fingerprint]*alert.Alert
+	gcCallback    func([]*alert.Alert)
 	limits        map[string]*limit.Bucket[model.Fingerprint]
 	perAlertLimit int
 	destroyed     bool
@@ -50,8 +50,8 @@ type Alerts struct {
 // NewAlerts returns a new Alerts struct.
 func NewAlerts() *Alerts {
 	a := &Alerts{
-		alerts:        make(map[model.Fingerprint]*types.Alert),
-		gcCallback:    func(_ []*types.Alert) {},
+		alerts:        make(map[model.Fingerprint]*alert.Alert),
+		gcCallback:    func(_ []*alert.Alert) {},
 		perAlertLimit: 0,
 	}
 
@@ -70,7 +70,7 @@ func (a *Alerts) WithPerAlertLimit(lim int) *Alerts {
 }
 
 // SetGCCallback sets a GC callback to be executed after each GC.
-func (a *Alerts) SetGCCallback(cb func([]*types.Alert)) {
+func (a *Alerts) SetGCCallback(cb func([]*alert.Alert)) {
 	a.Lock()
 	defer a.Unlock()
 
@@ -93,7 +93,7 @@ func (a *Alerts) Run(ctx context.Context, interval time.Duration) {
 }
 
 // GC deletes resolved alerts and returns them.
-func (a *Alerts) GC() (deleted []*types.Alert) {
+func (a *Alerts) GC() (deleted []*alert.Alert) {
 	// Remove stale alert limit buckets.
 	a.gcLimitBuckets()
 
@@ -109,7 +109,7 @@ func (a *Alerts) GC() (deleted []*types.Alert) {
 }
 
 // gcAlerts deletes resolved alerts and returns a copy of them.
-func (a *Alerts) gcAlerts() (deleted []*types.Alert) {
+func (a *Alerts) gcAlerts() (deleted []*alert.Alert) {
 	a.Lock()
 	defer a.Unlock()
 	for fp, alert := range a.alerts {
@@ -135,7 +135,7 @@ func (a *Alerts) gcLimitBuckets() {
 
 // Get returns the Alert with the matching fingerprint, or an error if it is
 // not found.
-func (a *Alerts) Get(fp model.Fingerprint) (*types.Alert, error) {
+func (a *Alerts) Get(fp model.Fingerprint) (*alert.Alert, error) {
 	a.Lock()
 	defer a.Unlock()
 
@@ -147,7 +147,7 @@ func (a *Alerts) Get(fp model.Fingerprint) (*types.Alert, error) {
 }
 
 // Set unconditionally sets the alert in memory.
-func (a *Alerts) Set(alert *types.Alert) error {
+func (a *Alerts) Set(alert *alert.Alert) error {
 	a.Lock()
 	defer a.Unlock()
 
@@ -158,15 +158,28 @@ func (a *Alerts) Set(alert *types.Alert) error {
 	fp := alert.Fingerprint()
 	name := alert.Name()
 
-	// Apply per alert limits if necessary
+	// Apply per alert limits if necessary.
 	if a.perAlertLimit > 0 {
-		bucket, ok := a.limits[name]
-		if !ok {
-			bucket = limit.NewBucket[model.Fingerprint](a.perAlertLimit)
-			a.limits[name] = bucket
-		}
-		if !bucket.Upsert(fp, alert.EndsAt) {
-			return ErrLimited
+		if alert.Resolved() {
+			// Forward the resolution only if we admitted its firing counterpart,
+			// which is still tracked in the bucket. Removing frees the slot the
+			// firing alert was holding. If we never admitted the firing alert, drop
+			// the resolution as noise, since nothing downstream ever saw a firing
+			// for it from us.
+			bucket, ok := a.limits[name]
+			if !ok || !bucket.Remove(fp) {
+				return ErrLimited
+			}
+		} else {
+			// Firing alert: apply the per-alert limit.
+			bucket, ok := a.limits[name]
+			if !ok {
+				bucket = limit.NewBucket[model.Fingerprint](a.perAlertLimit)
+				a.limits[name] = bucket
+			}
+			if !bucket.Upsert(fp, alert.EndsAt) {
+				return ErrLimited
+			}
 		}
 	}
 
@@ -176,7 +189,7 @@ func (a *Alerts) Set(alert *types.Alert) error {
 
 // DeleteIfNotModified deletes the slice of Alerts from the store if not
 // modified.
-func (a *Alerts) DeleteIfNotModified(alerts types.AlertSlice, destroyIfEmpty bool) error {
+func (a *Alerts) DeleteIfNotModified(alerts alert.AlertSlice, destroyIfEmpty bool) error {
 	a.Lock()
 	defer a.Unlock()
 	for _, alert := range alerts {
@@ -195,11 +208,11 @@ func (a *Alerts) DeleteIfNotModified(alerts types.AlertSlice, destroyIfEmpty boo
 }
 
 // List returns a slice of Alerts currently held in memory.
-func (a *Alerts) List() []*types.Alert {
+func (a *Alerts) List() []*alert.Alert {
 	a.Lock()
 	defer a.Unlock()
 
-	alerts := make([]*types.Alert, 0, len(a.alerts))
+	alerts := make([]*alert.Alert, 0, len(a.alerts))
 	for _, alert := range a.alerts {
 		alerts = append(alerts, alert)
 	}
diff --git a/store/store_test.go b/store/store_test.go
index c50a2c6c28..29f20acee3 100644
--- a/store/store_test.go
+++ b/store/store_test.go
@@ -21,12 +21,12 @@ import (
 	"github.com/prometheus/common/model"
 	"github.com/stretchr/testify/require"
 
-	"github.com/prometheus/alertmanager/types"
+	"github.com/prometheus/alertmanager/alert"
 )
 
 func TestSetGet(t *testing.T) {
 	a := NewAlerts()
-	alert := &types.Alert{
+	alert := &alert.Alert{
 		UpdatedAt: time.Now(),
 	}
 	require.NoError(t, a.Set(alert))
@@ -40,7 +40,7 @@ func TestSetGet(t *testing.T) {
 func TestDeleteIfNotModified(t *testing.T) {
 	t.Run("unmodified alert should be deleted", func(t *testing.T) {
 		a := NewAlerts()
-		a1 := &types.Alert{
+		a1 := &alert.Alert{
 			Alert: model.Alert{
 				Labels: model.LabelSet{
 					"foo": "bar",
@@ -51,7 +51,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 		require.NoError(t, a.Set(a1))
 
 		// a1 should be deleted as it has not been modified.
-		a.DeleteIfNotModified(types.AlertSlice{a1}, false)
+		a.DeleteIfNotModified(alert.AlertSlice{a1}, false)
 		got, err := a.Get(a1.Fingerprint())
 		require.Equal(t, ErrNotFound, err)
 		require.Nil(t, got)
@@ -59,7 +59,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 
 	t.Run("modified alert should not be deleted", func(t *testing.T) {
 		a := NewAlerts()
-		a1 := &types.Alert{
+		a1 := &alert.Alert{
 			Alert: model.Alert{
 				Labels: model.LabelSet{
 					"foo": "bar",
@@ -71,7 +71,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 
 		// Make a copy of a1 that is older, but do not put it.
 		// We want to make sure a1 is not deleted.
-		a2 := &types.Alert{
+		a2 := &alert.Alert{
 			Alert: model.Alert{
 				Labels: model.LabelSet{
 					"foo": "bar",
@@ -80,7 +80,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 			UpdatedAt: time.Now().Add(-time.Second),
 		}
 		require.True(t, a2.UpdatedAt.Before(a1.UpdatedAt))
-		a.DeleteIfNotModified(types.AlertSlice{a2}, false)
+		a.DeleteIfNotModified(alert.AlertSlice{a2}, false)
 		// a1 should not be deleted.
 		got, err := a.Get(a1.Fingerprint())
 		require.NoError(t, err)
@@ -88,7 +88,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 
 		// Make another copy of a1 that is older, but do not put it.
 		// We want to make sure a2 is not deleted here either.
-		a3 := &types.Alert{
+		a3 := &alert.Alert{
 			Alert: model.Alert{
 				Labels: model.LabelSet{
 					"foo": "bar",
@@ -97,7 +97,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 			UpdatedAt: time.Now().Add(time.Second),
 		}
 		require.True(t, a3.UpdatedAt.After(a1.UpdatedAt))
-		a.DeleteIfNotModified(types.AlertSlice{a3}, false)
+		a.DeleteIfNotModified(alert.AlertSlice{a3}, false)
 		// a1 should not be deleted.
 		got, err = a.Get(a1.Fingerprint())
 		require.NoError(t, err)
@@ -106,7 +106,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 
 	t.Run("should not delete other alerts", func(t *testing.T) {
 		a := NewAlerts()
-		a1 := &types.Alert{
+		a1 := &alert.Alert{
 			Alert: model.Alert{
 				Labels: model.LabelSet{
 					"foo": "bar",
@@ -114,7 +114,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 			},
 			UpdatedAt: time.Now(),
 		}
-		a2 := &types.Alert{
+		a2 := &alert.Alert{
 			Alert: model.Alert{
 				Labels: model.LabelSet{
 					"bar": "baz",
@@ -126,7 +126,7 @@ func TestDeleteIfNotModified(t *testing.T) {
 		require.NoError(t, a.Set(a2))
 
 		// Deleting a1 should not delete a2.
-		require.NoError(t, a.DeleteIfNotModified(types.AlertSlice{a1}, true))
+		require.NoError(t, a.DeleteIfNotModified(alert.AlertSlice{a1}, true))
 		// a1 should be deleted.
 		got, err := a.Get(a1.Fingerprint())
 		require.Equal(t, ErrNotFound, err)
@@ -139,10 +139,59 @@ func TestDeleteIfNotModified(t *testing.T) {
 	})
 }
 
+func TestPerAlertLimitResolved(t *testing.T) {
+	newAlert := func(instance string, resolved bool) *alert.Alert {
+		end := time.Now().Add(time.Hour)
+		if resolved {
+			end = time.Now().Add(-time.Hour)
+		}
+		return &alert.Alert{
+			Alert: model.Alert{
+				Labels:   model.LabelSet{"alertname": "Test", "instance": model.LabelValue(instance)},
+				StartsAt: time.Now().Add(-2 * time.Hour),
+				EndsAt:   end,
+			},
+			UpdatedAt: time.Now(),
+		}
+	}
+
+	t.Run("resolved frees the slot held by an admitted firing alert", func(t *testing.T) {
+		a := NewAlerts().WithPerAlertLimit(1)
+
+		// The first firing alert is admitted and fills the bucket.
+		require.NoError(t, a.Set(newAlert("server1", false)))
+
+		// A second firing alert is limited while the bucket is full.
+		require.ErrorIs(t, a.Set(newAlert("server2", false)), ErrLimited)
+
+		// Resolving the admitted alert frees its slot and is stored.
+		require.NoError(t, a.Set(newAlert("server1", true)))
+
+		// The freed slot now admits another firing alert.
+		require.NoError(t, a.Set(newAlert("server2", false)))
+	})
+
+	t.Run("resolved without an admitted firing counterpart is dropped", func(t *testing.T) {
+		a := NewAlerts().WithPerAlertLimit(1)
+
+		require.ErrorIs(t, a.Set(newAlert("ghost", true)), ErrLimited)
+		require.Zero(t, a.Len())
+	})
+
+	t.Run("resolving twice drops the duplicate resolution", func(t *testing.T) {
+		a := NewAlerts().WithPerAlertLimit(1)
+
+		require.NoError(t, a.Set(newAlert("server1", false)))
+		require.NoError(t, a.Set(newAlert("server1", true)))
+		// The slot was already freed by the first resolution.
+		require.ErrorIs(t, a.Set(newAlert("server1", true)), ErrLimited)
+	})
+}
+
 func TestGC(t *testing.T) {
 	now := time.Now()
-	newAlert := func(key string, start, end time.Duration) *types.Alert {
-		return &types.Alert{
+	newAlert := func(key string, start, end time.Duration) *alert.Alert {
+		return &alert.Alert{
 			Alert: model.Alert{
 				Labels:   model.LabelSet{model.LabelName(key): "b"},
 				StartsAt: now.Add(start * time.Minute),
@@ -150,11 +199,11 @@ func TestGC(t *testing.T) {
 			},
 		}
 	}
-	active := []*types.Alert{
+	active := []*alert.Alert{
 		newAlert("b", 10, 20),
 		newAlert("c", -10, 10),
 	}
-	resolved := []*types.Alert{
+	resolved := []*alert.Alert{
 		newAlert("a", -10, -5),
 		newAlert("d", -10, -1),
 	}
@@ -164,7 +213,7 @@ func TestGC(t *testing.T) {
 		done        = make(chan struct{})
 		ctx, cancel = context.WithCancel(context.Background())
 	)
-	s.SetGCCallback(func(a []*types.Alert) {
+	s.SetGCCallback(func(a []*alert.Alert) {
 		n += len(a)
 		if n >= len(resolved) {
 			cancel()