diff --git a/pkg/sync/expand/drop_stats_test.go b/pkg/sync/expand/drop_stats_test.go
index 9cee3fa8c..10cae9fc5 100644
--- a/pkg/sync/expand/drop_stats_test.go
+++ b/pkg/sync/expand/drop_stats_test.go
@@ -78,4 +78,25 @@ func TestTopologicalSourceMissingRecordedOnDropStats(t *testing.T) {
 	defer stats.mu.Unlock()
 	require.Equal(t, int64(1), stats.sourceMissing, "the dropped source edge must land on the aggregate")
 	require.Contains(t, stats.seen, source.GetId())
+	require.Contains(t, graph.DanglingEntitlementIDs, source.GetId(),
+		"the topological entitlement load must record the missing source")
+}
+
+func TestTopologicalDestinationMissingRecordsDanglingID(t *testing.T) {
+	ctx := context.Background()
+	store := NewMockExpanderStore()
+
+	group := makeResource("group", "org")
+	source := makeEntitlement("ent:source", group)
+	dest := makeEntitlement("ent:dest", group)
+	store.AddEntitlement(source)
+
+	graph := NewEntitlementGraph(ctx)
+	graph.AddEntitlementID(source.GetId())
+	graph.AddEntitlementID(dest.GetId())
+	require.NoError(t, graph.AddEdge(ctx, source.GetId(), dest.GetId(), false, nil))
+
+	require.NoError(t, NewExpander(store, graph).RunTopologicalMergeStreaming(ctx))
+	require.Contains(t, graph.DanglingEntitlementIDs, dest.GetId(),
+		"the production topological path must record the missing destination")
 }
diff --git a/pkg/sync/expand/expander.go b/pkg/sync/expand/expander.go
index 034a1d142..bca9682a3 100644
--- a/pkg/sync/expand/expander.go
+++ b/pkg/sync/expand/expander.go
@@ -433,6 +433,8 @@ func (e *Expander) runAction(ctx context.Context, action *EntitlementGraphAction
 			// the per-sync report. The batch shares one source, but every
 			// destination edge drops — count them all.
 			e.dropStats.RecordSourceMissingEdges(action.SourceEntitlementID, len(dests))
+			// The edges are deleted below, so a later seed could not reach them.
+			e.graph.NoteUnrecoverableDangling()
 			l.Debug("runAction: source entitlement not found, dropping batch edges",
 				zap.String("source_entitlement_id", action.SourceEntitlementID))
 			for _, d := range dests {
@@ -469,6 +471,8 @@ func (e *Expander) runAction(ctx context.Context, action *EntitlementGraphAction
 				// A single missing descendant drops only its own edge; the rest
 				// of the batch still expands. Debug + aggregate, as above.
 				e.dropStats.RecordDestinationMissing(d.EntitlementID)
+				// The edge is deleted below, so a later seed could not reach it.
+				e.graph.NoteUnrecoverableDangling()
 				l.Debug("runAction: descendant entitlement not found, dropping edge",
 					zap.String("descendant_entitlement_id", d.EntitlementID))
 				_ = e.graph.DeleteEdge(ctx, action.SourceEntitlementID, d.EntitlementID)
diff --git a/pkg/sync/expand/graph.go b/pkg/sync/expand/graph.go
index cf741dcdf..142fedb9f 100644
--- a/pkg/sync/expand/graph.go
+++ b/pkg/sync/expand/graph.go
@@ -91,6 +91,24 @@ type EntitlementGraph struct {
 	HasNoCycles           bool                      `json:"has_no_cycles"`
 	ExpansionPlan         *EntitlementGraphPlan     `json:"plan,omitempty"`
 	ExpansionMetrics      *EntitlementGraphMetrics  `json:"metrics,omitempty"`
+	// DanglingEntitlementIDs are edge endpoints this expansion skipped because
+	// they had no entitlement row in the store. If a later sync supplies one of
+	// those rows it adds no edge and no grant — the only two things incremental
+	// expansion seeds from — so nothing would seed the walk and the fast path
+	// would miss grants full expansion writes. Persisted so the next compaction
+	// can precheck these ids: any that now resolve seed expansion, while those
+	// still missing remain recorded without seeding a walk.
+	//
+	// Scoped deliberately. Most dangling references are permanent (connector
+	// magic-id bugs, disabled-by-default resource types — see DroppedEdgeStats),
+	// so a graph-wide "something was dangling" flag would re-arm on every full
+	// expansion and disable the fast path forever for those tenants.
+	DanglingEntitlementIDs map[string]struct{} `json:"dangling_entitlement_ids,omitempty"`
+	// DanglingOverflow means the dangling set hit its cap, or an evaluator
+	// dropped the edge outright so seeding cannot recover it. Either way the id
+	// list is no longer a complete description of what was skipped, so the next
+	// compaction must decline rather than seed.
+	DanglingOverflow bool `json:"dangling_overflow,omitempty"`
 }
 
 type EntitlementGraphMetrics struct {
@@ -112,6 +130,43 @@ func NewEntitlementGraph(_ context.Context) *EntitlementGraph {
 	}
 }
 
+// maxDanglingEntitlementIDs caps the persisted dangling set so a pathologically
+// broken connector cannot bloat the sidecar. Past the cap the graph records
+// overflow and the next compaction declines instead of seeding. A var so tests
+// can lower it.
+var maxDanglingEntitlementIDs = 4096
+
+// NoteDanglingReference records that an expansion skipped entitlementID because
+// it has no row in the store. Nil-safe: the legacy expander reaches it through
+// a graph that may not be installed yet.
+func (g *EntitlementGraph) NoteDanglingReference(entitlementID string) {
+	if g == nil || entitlementID == "" {
+		return
+	}
+	if g.DanglingEntitlementIDs == nil {
+		g.DanglingEntitlementIDs = make(map[string]struct{})
+	}
+	if _, ok := g.DanglingEntitlementIDs[entitlementID]; ok {
+		return
+	}
+	if len(g.DanglingEntitlementIDs) >= maxDanglingEntitlementIDs {
+		g.DanglingOverflow = true
+		return
+	}
+	g.DanglingEntitlementIDs[entitlementID] = struct{}{}
+}
+
+// NoteUnrecoverableDangling records a skipped endpoint whose edge was dropped
+// from the graph, so seeding the id on a later run could not reach it. Only the
+// source-batched evaluator deletes edges, and it never runs on Pebble, but the
+// graph must still refuse to be reused if it ever does.
+func (g *EntitlementGraph) NoteUnrecoverableDangling() {
+	if g == nil {
+		return
+	}
+	g.DanglingOverflow = true
+}
+
 // isNodeExpanded - is every outgoing edge from this node Expanded?
 func (g *EntitlementGraph) isNodeExpanded(nodeID int) bool {
 	for _, edgeID := range g.SourcesToDestinations[nodeID] {
diff --git a/pkg/sync/expand/graph_blob.go b/pkg/sync/expand/graph_blob.go
index 7e394fc6b..1854afffd 100644
--- a/pkg/sync/expand/graph_blob.go
+++ b/pkg/sync/expand/graph_blob.go
@@ -19,7 +19,11 @@ type graphBlobEnvelope struct {
 	Graph         *EntitlementGraph               `json:"graph"`
 }
 
-const graphBlobFormatVersion uint32 = 2
+// Version 3 adds EntitlementGraph.DanglingEntitlementIDs and DanglingOverflow.
+// A v2 blob predates that bookkeeping, so what it skipped is unknowable and it
+// must not be trusted as an incremental base: the version check below rejects
+// it and the reader falls back to full expansion.
+const graphBlobFormatVersion uint32 = 3
 
 // MarshalGraphBlob serializes a legacy, unbound graph blob for compatibility
 // tests. Transient state is stripped first (a reload rebuilds it).
diff --git a/pkg/sync/expand/graph_blob_test.go b/pkg/sync/expand/graph_blob_test.go
index 3a9ecc942..5335f8cd7 100644
--- a/pkg/sync/expand/graph_blob_test.go
+++ b/pkg/sync/expand/graph_blob_test.go
@@ -246,3 +246,61 @@ func BenchmarkMarshalGraphBlob(b *testing.B) {
 func entName(i int) string {
 	return fmt.Sprintf("group:g%06d:member", i)
 }
+
+// TestUnmarshalGraphBlob_RejectsPreDanglingVersion: a v2 blob predates
+// dangling bookkeeping, so what it skipped is unknowable and it
+// must not be reused as an incremental base. Rejection is silent — (nil, nil)
+// — so the reader falls back to full expansion rather than failing.
+func TestUnmarshalGraphBlob_RejectsPreDanglingVersion(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.AddEntitlementID("ent-a")
+
+	v2Blob, err := json.Marshal(map[string]any{
+		"format_version": 2,
+		"sync_id":        "sync-1",
+		"graph":          g,
+	})
+	require.NoError(t, err)
+
+	got, digest, err := UnmarshalGraphBlobWithGrantDigest(v2Blob, "sync-1")
+	require.NoError(t, err)
+	require.Nil(t, got)
+	require.Nil(t, digest)
+}
+
+// TestGraphBlob_RoundTripsDanglingState: the recorded ids are what the next run
+// seeds from, so losing them in serialization would silently reintroduce the
+// divergence this whole mechanism exists to prevent.
+func TestGraphBlob_RoundTripsDanglingState(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.AddEntitlementID("ent-a")
+	g.NoteDanglingReference("missing:one")
+	g.NoteDanglingReference("missing:two")
+
+	data, err := MarshalGraphBlob("sync-1", g)
+	require.NoError(t, err)
+	got, err := UnmarshalGraphBlob(data, "sync-1")
+	require.NoError(t, err)
+	require.NotNil(t, got)
+	require.Contains(t, got.DanglingEntitlementIDs, "missing:one")
+	require.Contains(t, got.DanglingEntitlementIDs, "missing:two")
+	require.False(t, got.DanglingOverflow)
+}
+
+// TestGraphBlob_RoundTripsDanglingOverflow: overflow is the flag that makes the
+// compactor decline outright, so it must survive too.
+func TestGraphBlob_RoundTripsDanglingOverflow(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.AddEntitlementID("ent-a")
+	g.NoteUnrecoverableDangling()
+
+	data, err := MarshalGraphBlob("sync-1", g)
+	require.NoError(t, err)
+	got, err := UnmarshalGraphBlob(data, "sync-1")
+	require.NoError(t, err)
+	require.NotNil(t, got)
+	require.True(t, got.DanglingOverflow)
+}
diff --git a/pkg/sync/expand/incremental.go b/pkg/sync/expand/incremental.go
index f5db117c7..37b1b385c 100644
--- a/pkg/sync/expand/incremental.go
+++ b/pkg/sync/expand/incremental.go
@@ -45,6 +45,13 @@ func (g *EntitlementGraph) Clone() (*EntitlementGraph, error) {
 		Loaded:                g.Loaded,
 		Depth:                 g.Depth,
 		HasNoCycles:           g.HasNoCycles,
+		DanglingOverflow:      g.DanglingOverflow,
+	}
+	if g.DanglingEntitlementIDs != nil {
+		out.DanglingEntitlementIDs = make(map[string]struct{}, len(g.DanglingEntitlementIDs))
+		for id := range g.DanglingEntitlementIDs {
+			out.DanglingEntitlementIDs[id] = struct{}{}
+		}
 	}
 	for id, node := range g.Nodes {
 		node.EntitlementIDs = append([]string(nil), node.EntitlementIDs...)
@@ -110,6 +117,9 @@ func (g *EntitlementGraph) reinitMaps() {
 	if g.Edges == nil {
 		g.Edges = map[int]Edge{}
 	}
+	if g.DanglingEntitlementIDs == nil {
+		g.DanglingEntitlementIDs = map[string]struct{}{}
+	}
 }
 
 // ErrIncrementalFallback means a new edge closed a cycle; the caller should
@@ -127,6 +137,14 @@ var ErrIncrementalRevocationDecline = errors.New("incremental expansion: revocat
 // that normal full expansion is the safer bounded-cost path.
 var ErrIncrementalDenseChangeDecline = errors.New("incremental expansion: dense affected graph, fall back to full expansion")
 
+// ErrIncrementalDanglingReferenceDecline means the base graph can no longer
+// describe what it skipped — the dangling set overflowed its cap, or an
+// evaluator dropped the edges outright — so prechecking the recorded ids cannot
+// make the fast path agree with full expansion. Ordinary dangling references do
+// NOT reach this: still-missing ids remain recorded without seeding a walk, and
+// ids that now resolve seed their affected closure.
+var ErrIncrementalDanglingReferenceDecline = errors.New("incremental expansion: base graph dangling set is not seedable, fall back to full expansion")
+
 const (
 	incrementalDenseGraphMinNodes = 1000
 	incrementalMaxAffectedPercent = 10
@@ -184,8 +202,35 @@ func NewIncrementalExpander(store ExpanderStore, graph *EntitlementGraph) *Incre
 //
 // The walk reads current membership from the store, so changed members
 // (already merged in) propagate without being passed in. Returns
-// ErrIncrementalFallback if a new edge closes a cycle.
+// ErrIncrementalFallback if a new edge closes a cycle. This method mutates the
+// graph while applying changes and may do so before returning an error; callers
+// that intend to retry must discard or restore the graph first.
 func (ie *IncrementalExpander) ExpandChanges(ctx context.Context, newEdges []NewEdge, changedEntitlementIDs []string) (*IncrementalResult, error) {
+	if ie.graph.DanglingOverflow {
+		return nil, ErrIncrementalDanglingReferenceDecline
+	}
+	resolvedDanglingIDs, err := ie.precheckDanglingEntitlements(ctx)
+	if err != nil {
+		return nil, err
+	}
+	if len(resolvedDanglingIDs) > 0 {
+		// Keep the exported method from appending into or sorting the caller's
+		// backing array while combining the internally discovered seeds.
+		changedEntitlementIDs = append([]string(nil), changedEntitlementIDs...)
+		seen := make(map[string]struct{}, len(changedEntitlementIDs)+len(resolvedDanglingIDs))
+		for _, id := range changedEntitlementIDs {
+			seen[id] = struct{}{}
+		}
+		for _, id := range resolvedDanglingIDs {
+			if _, ok := seen[id]; ok {
+				continue
+			}
+			seen[id] = struct{}{}
+			changedEntitlementIDs = append(changedEntitlementIDs, id)
+		}
+		sort.Strings(changedEntitlementIDs)
+	}
+
 	if len(newEdges) == 0 && len(changedEntitlementIDs) == 0 {
 		return &IncrementalResult{}, nil
 	}
@@ -258,6 +303,45 @@ func (ie *IncrementalExpander) ExpandChanges(ctx context.Context, newEdges []New
 	return result, nil
 }
 
+// precheckDanglingEntitlements separates recorded endpoints that are still
+// missing from those that now resolve. Still-missing ids stay persisted but do
+// not seed the walk, avoiding a useless recomputation of their entire forward
+// closure on every run. Resolved ids become normal changed-entitlement seeds.
+//
+// Build the replacement set locally and publish it only after every lookup
+// succeeds, so a transient store error does not partially mutate the graph.
+func (ie *IncrementalExpander) precheckDanglingEntitlements(ctx context.Context) ([]string, error) {
+	if len(ie.graph.DanglingEntitlementIDs) == 0 {
+		return nil, nil
+	}
+
+	ids := make([]string, 0, len(ie.graph.DanglingEntitlementIDs))
+	for id := range ie.graph.DanglingEntitlementIDs {
+		ids = append(ids, id)
+	}
+	sort.Strings(ids)
+
+	stillMissing := make(map[string]struct{}, len(ids))
+	resolved := make([]string, 0, len(ids))
+	for _, id := range ids {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		entitlement, err := ie.getEntitlement(ctx, id)
+		if err != nil {
+			return nil, err
+		}
+		if entitlement == nil {
+			stillMissing[id] = struct{}{}
+			continue
+		}
+		resolved = append(resolved, id)
+	}
+
+	ie.graph.DanglingEntitlementIDs = stillMissing
+	return resolved, nil
+}
+
 func topologicalAffectedNodeOrder(g *EntitlementGraph, affected map[int]struct{}) ([]int, error) {
 	inDegree := make(map[int]int, len(affected))
 	for nodeID := range affected {
@@ -376,7 +460,9 @@ func (ie *IncrementalExpander) recomputeDestination(ctx context.Context, nodeID
 	}
 	if destEnt == nil {
 		// Dangling ref: skip-with-warn, matching the full evaluator (don't
-		// error into a fallback).
+		// error into a fallback). Record it on the persisted graph so a later
+		// incremental run can precheck it and seed the walk if it resolves.
+		ie.graph.NoteDanglingReference(destEntitlementID)
 		ctxzap.Extract(ctx).Warn("incremental expansion: destination entitlement not in store; skipping",
 			zap.String("entitlement_id", destEntitlementID))
 		return 0, nil
@@ -406,6 +492,7 @@ func (ie *IncrementalExpander) recomputeDestination(ctx context.Context, nodeID
 				return 0, err
 			}
 			if sourceEnt == nil {
+				ie.graph.NoteDanglingReference(sourceEntitlementID)
 				ctxzap.Extract(ctx).Warn("incremental expansion: source entitlement not in store; skipping",
 					zap.String("entitlement_id", sourceEntitlementID))
 				continue
diff --git a/pkg/sync/expand/incremental_test.go b/pkg/sync/expand/incremental_test.go
index b4ed2e8c3..01642c524 100644
--- a/pkg/sync/expand/incremental_test.go
+++ b/pkg/sync/expand/incremental_test.go
@@ -8,6 +8,8 @@ import (
 	v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2"
 	reader_v2 "github.com/conductorone/baton-sdk/pb/c1/reader/v2"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 // buildExpandedChain constructs an already-expanded linear chain
@@ -29,7 +31,7 @@ func buildExpandedChain(t *testing.T, ctx context.Context, store *MockExpanderSt
 	}
 	// Run the real expander so the base carries realistic expanded grants and
 	// provenance (sources maps), not hand-seeded approximations.
-	require.NoError(t, NewExpander(store, g).Run(ctx))
+	require.NoError(t, NewExpander(notFoundExpanderStore{store}, g).Run(ctx))
 	return g
 }
 
@@ -508,3 +510,252 @@ func itoa(i int) string {
 	}
 	return string(b)
 }
+
+// TestIncremental_DanglingDestinationRecordsID: skipping an endpoint with no
+// entitlement row must record that endpoint's id on the graph. The sidecar this
+// run persists then prechecks that id on the NEXT run, which is the only place
+// the divergence is catchable — an increment that supplies the missing record
+// changes neither the edge set nor any grant.
+func TestIncremental_DanglingDestinationRecordsID(t *testing.T) {
+	ctx := context.Background()
+	store := NewMockExpanderStore()
+	g := buildExpandedChain(t, ctx, store, "alice", "eng:member", "github:access")
+	require.Empty(t, g.DanglingEntitlementIDs)
+
+	g.AddEntitlementID("missing:ent")
+	require.NoError(t, g.AddEdge(ctx, "github:access", "missing:ent", false, nil))
+
+	ie := NewIncrementalExpander(store, g)
+	res, err := ie.ExpandChanges(ctx, nil, []string{"eng:member"})
+	require.NoError(t, err) // skip-with-warn, not an error
+	require.Equal(t, 0, res.GrantsWritten)
+	require.Contains(t, g.DanglingEntitlementIDs, "missing:ent")
+	require.False(t, g.DanglingOverflow)
+}
+
+// TestIncremental_DanglingSourceRecordsID: same contract for a missing source.
+func TestIncremental_DanglingSourceRecordsID(t *testing.T) {
+	ctx := context.Background()
+	store := NewMockExpanderStore()
+	g := buildExpandedChain(t, ctx, store, "alice", "eng:member", "github:access")
+
+	g.AddEntitlementID("missing:src")
+	require.NoError(t, g.AddEdge(ctx, "missing:src", "github:access", false, nil))
+
+	ie := NewIncrementalExpander(store, g)
+	_, err := ie.ExpandChanges(ctx, nil, []string{"missing:src"})
+	require.NoError(t, err)
+	require.Contains(t, g.DanglingEntitlementIDs, "missing:src")
+}
+
+// TestIncremental_ResolvedDanglingSeedExpands is the whole point of recording
+// ids rather than a graph-wide flag: seeding a previously-dangling endpoint that
+// has since gained an entitlement row must expand it, matching what full
+// expansion would produce.
+func TestIncremental_ResolvedDanglingSeedExpands(t *testing.T) {
+	ctx := context.Background()
+	store := NewMockExpanderStore()
+	g := buildExpandedChain(t, ctx, store, "alice", "eng:member", "github:access")
+
+	// An edge to an entitlement that had no row when the base expanded.
+	g.AddEntitlementID("late:ent")
+	require.NoError(t, g.AddEdge(ctx, "github:access", "late:ent", false, nil))
+	ie := NewIncrementalExpander(store, g)
+	_, err := ie.ExpandChanges(ctx, nil, []string{"eng:member"})
+	require.NoError(t, err)
+	require.Contains(t, g.DanglingEntitlementIDs, "late:ent")
+	require.Empty(t, principalsOn(t, ctx, store, "late:ent"))
+
+	// The row arrives. No new edge, no new grant — seeding the recorded id is
+	// the only thing that can reach it. The caller supplies no changed ids: the
+	// dangling precheck must discover that the row resolved and seed it itself.
+	store.AddEntitlement(makeEntitlement("late:ent", makeResource("group", "late:ent")))
+	ie2 := NewIncrementalExpander(store, g)
+	res, err := ie2.ExpandChanges(ctx, nil, nil)
+	require.NoError(t, err)
+	require.Greater(t, res.GrantsWritten, 0, "the resolved endpoint must expand")
+	require.Contains(t, principalsOn(t, ctx, store, "late:ent"), "alice")
+
+	// And it drops out of the set, so later runs stop re-walking its closure.
+	require.NotContains(t, g.DanglingEntitlementIDs, "late:ent")
+}
+
+func TestIncremental_ResolvedDanglingDoesNotMutateChangedIDs(t *testing.T) {
+	ctx := context.Background()
+	store := NewMockExpanderStore()
+	g := buildExpandedChain(t, ctx, store, "alice", "eng:member", "github:access")
+
+	g.AddEntitlementID("late:ent")
+	require.NoError(t, g.AddEdge(ctx, "github:access", "late:ent", false, nil))
+	g.NoteDanglingReference("late:ent")
+	store.AddEntitlement(makeEntitlement("late:ent", makeResource("group", "late:ent")))
+
+	// Keep a sentinel in the caller's spare capacity. Appending or sorting the
+	// input slice in place would overwrite it or reorder the visible elements.
+	backing := []string{"github:access", "eng:member", "sentinel"}
+	changed := backing[:2]
+	want := append([]string(nil), backing...)
+
+	_, err := NewIncrementalExpander(store, g).ExpandChanges(ctx, nil, changed)
+	require.NoError(t, err)
+	require.Equal(t, want, backing)
+}
+
+func TestIncremental_DanglingPrecheckHonorsCanceledContext(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	store := NewMockExpanderStore()
+	g := NewEntitlementGraph(ctx)
+	g.NoteDanglingReference("still:missing")
+
+	_, err := NewIncrementalExpander(store, g).ExpandChanges(ctx, nil, nil)
+	require.ErrorIs(t, err, context.Canceled)
+	require.Contains(t, g.DanglingEntitlementIDs, "still:missing")
+}
+
+func TestIncremental_DanglingOverflowDeclines(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.NoteUnrecoverableDangling()
+
+	result, err := NewIncrementalExpander(NewMockExpanderStore(), g).ExpandChanges(ctx, nil, nil)
+	require.Nil(t, result)
+	require.ErrorIs(t, err, ErrIncrementalDanglingReferenceDecline)
+}
+
+// TestIncremental_StillMissingDanglingIsPreservedWithoutWalking: a permanent
+// dangling source must cost only its precheck lookup. Seeding it would put its
+// entire forward closure in the affected set before the walk discovers that
+// the source is still missing.
+func TestIncremental_StillMissingDanglingIsPreservedWithoutWalking(t *testing.T) {
+	ctx := context.Background()
+	store := NewMockExpanderStore()
+	g := buildExpandedChain(t, ctx, store, "alice", "eng:member", "github:access")
+	g.AddEntitlementID("missing:src")
+	require.NoError(t, g.AddEdge(ctx, "missing:src", "eng:member", false, nil))
+	g.DanglingEntitlementIDs = map[string]struct{}{"missing:src": {}}
+
+	ie := NewIncrementalExpander(store, g)
+	res, err := ie.ExpandChanges(ctx, nil, nil)
+	require.NoError(t, err)
+	require.Empty(t, res.EntitlementsWalked,
+		"a still-missing seed must not make its descendant closure affected")
+	require.Contains(t, g.DanglingEntitlementIDs, "missing:src",
+		"still-missing ids must survive so a later run can detect resolution")
+}
+
+// TestNoteDanglingReference_OverflowStopsRecording: past the cap the graph stops
+// collecting ids and records overflow instead, so the sidecar cannot be bloated
+// by a pathologically broken connector. Overflow is what makes the compactor
+// decline; ordinary dangling references do not.
+func TestNoteDanglingReference_OverflowStopsRecording(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+
+	original := maxDanglingEntitlementIDs
+	maxDanglingEntitlementIDs = 3
+	defer func() { maxDanglingEntitlementIDs = original }()
+
+	for _, id := range []string{"a", "b", "c"} {
+		g.NoteDanglingReference(id)
+	}
+	require.Len(t, g.DanglingEntitlementIDs, 3)
+	require.False(t, g.DanglingOverflow)
+
+	g.NoteDanglingReference("d")
+	require.Len(t, g.DanglingEntitlementIDs, 3, "the cap must hold")
+	require.True(t, g.DanglingOverflow)
+	require.NotContains(t, g.DanglingEntitlementIDs, "d")
+
+	// A repeat of an id already recorded must not trip overflow.
+	g2 := NewEntitlementGraph(ctx)
+	g2.NoteDanglingReference("a")
+	g2.NoteDanglingReference("a")
+	require.Len(t, g2.DanglingEntitlementIDs, 1)
+	require.False(t, g2.DanglingOverflow)
+}
+
+// TestClone_PreservesDanglingState: Clone is what the compactor hands the
+// expander, so state lost here would silently change which ids get seeded.
+func TestClone_PreservesDanglingState(t *testing.T) {
+	ctx := context.Background()
+	g := NewEntitlementGraph(ctx)
+	g.AddEntitlementID("a")
+	g.NoteDanglingReference("missing:one")
+	g.NoteUnrecoverableDangling()
+
+	clone, err := g.Clone()
+	require.NoError(t, err)
+	require.Contains(t, clone.DanglingEntitlementIDs, "missing:one")
+	require.True(t, clone.DanglingOverflow)
+
+	// Deep copy: mutating the clone must not reach the original.
+	clone.NoteDanglingReference("missing:two")
+	require.NotContains(t, g.DanglingEntitlementIDs, "missing:two")
+}
+
+// notFoundExpanderStore makes an unknown entitlement return a real NotFound
+// error. MockExpanderStore returns (nil, nil) instead, which the topological
+// evaluator tolerates (fetchEntitlement checks resp == nil) but the
+// source-batched one does not — it only skips on a NotFound code. Real stores
+// return NotFound, so this wrapper is what lets the legacy path be tested at
+// all.
+type notFoundExpanderStore struct {
+	*MockExpanderStore
+}
+
+func (s notFoundExpanderStore) GetEntitlement(
+	ctx context.Context,
+	req *reader_v2.EntitlementsReaderServiceGetEntitlementRequest,
+) (*reader_v2.EntitlementsReaderServiceGetEntitlementResponse, error) {
+	resp, err := s.MockExpanderStore.GetEntitlement(ctx, req)
+	if err == nil && resp == nil {
+		return nil, status.Errorf(codes.NotFound, "entitlement %s not found", req.GetEntitlementId())
+	}
+	return resp, err
+}
+
+// TestLegacyExpander_DanglingSetsOverflow pins the one place the scoped design
+// does not apply. The source-batched evaluator — the path taken when the store
+// cannot guarantee principal-sorted grants — DELETES the edge when an endpoint
+// is missing, so seeding that id on a later run could not reach it. Those sites
+// must set overflow (a hard decline) rather than record a seedable id.
+//
+// Reachable here because MockExpanderStore reports
+// GrantsForEntitlementPrincipalSorted() == false, which is what selects this
+// evaluator; on Pebble the topological one runs instead.
+func TestLegacyExpander_DanglingSetsOverflow(t *testing.T) {
+	t.Run("missing destination", func(t *testing.T) {
+		ctx := context.Background()
+		store := NewMockExpanderStore()
+		g := NewEntitlementGraph(ctx)
+		for _, e := range []string{"eng:member", "missing:dst"} {
+			g.AddEntitlementID(e)
+		}
+		store.AddEntitlement(makeEntitlement("eng:member", makeResource("group", "eng:member")))
+		store.AddGrant(directGrant("eng:member", makeResource("user", "alice")))
+		require.NoError(t, g.AddEdge(ctx, "eng:member", "missing:dst", false, nil))
+
+		require.NoError(t, NewExpander(notFoundExpanderStore{store}, g).Run(ctx))
+		require.True(t, g.DanglingOverflow,
+			"a dropped edge is not recoverable by seeding, so it must decline")
+		require.NotContains(t, g.DanglingEntitlementIDs, "missing:dst",
+			"the id must not be offered as a seed when its edge is gone")
+	})
+
+	t.Run("missing source", func(t *testing.T) {
+		ctx := context.Background()
+		store := NewMockExpanderStore()
+		g := NewEntitlementGraph(ctx)
+		for _, e := range []string{"missing:src", "eng:member"} {
+			g.AddEntitlementID(e)
+		}
+		store.AddEntitlement(makeEntitlement("eng:member", makeResource("group", "eng:member")))
+		require.NoError(t, g.AddEdge(ctx, "missing:src", "eng:member", false, nil))
+
+		require.NoError(t, NewExpander(notFoundExpanderStore{store}, g).Run(ctx))
+		require.True(t, g.DanglingOverflow)
+		require.NotContains(t, g.DanglingEntitlementIDs, "missing:src")
+	})
+}
diff --git a/pkg/sync/expand/topological_merge.go b/pkg/sync/expand/topological_merge.go
index 37eee0201..880e892a9 100644
--- a/pkg/sync/expand/topological_merge.go
+++ b/pkg/sync/expand/topological_merge.go
@@ -227,7 +227,14 @@ func (e *Expander) loadExpansionEntitlements(ctx context.Context) (map[string]*v
 		}
 		if ok {
 			entitlements[entID] = ent
+			continue
 		}
+		// Single resolution point for this evaluator: every downstream "not in
+		// store" skip is a consequence of this absence, so recording here is
+		// provably complete over the graph. The id is persisted with the graph
+		// so the next incremental run can precheck it: if the row has appeared it
+		// seeds expansion, otherwise it remains recorded without seeding a walk.
+		e.graph.NoteDanglingReference(entID)
 	}
 	return entitlements, nil
 }
diff --git a/pkg/synccompactor/compactor.go b/pkg/synccompactor/compactor.go
index 5c98ca139..9ed60ff8b 100644
--- a/pkg/synccompactor/compactor.go
+++ b/pkg/synccompactor/compactor.go
@@ -23,6 +23,8 @@ import (
 	"github.com/conductorone/baton-sdk/pkg/tempdir"
 	"github.com/grpc-ecosystem/go-grpc-middleware/logging/zap/ctxzap"
 	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/metric"
 	"go.uber.org/zap"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -30,7 +32,20 @@ import (
 	"github.com/conductorone/baton-sdk/pkg/uotel"
 )
 
-var tracer = otel.Tracer("baton-sdk/pkg.synccompactor")
+var (
+	tracer = otel.Tracer("baton-sdk/pkg.synccompactor")
+	meter  = otel.Meter("baton-sdk/pkg.synccompactor")
+
+	// One increment per compaction that considered incremental expansion.
+	// Both attributes are closed sets (see logIncrementalOutcome), so the
+	// cardinality is bounded — this is the counterpart of the
+	// incremental_expansion_outcome log field, for alerting on a fast path
+	// that has quietly stopped engaging.
+	incrementalExpansionOutcomeCounter, _ = meter.Int64Counter(
+		"compactor_incremental_expansion_total",
+		metric.WithDescription("Incremental grant expansion attempts by result. Attributes: outcome (not_attempted|fell_back|declined|succeeded|failed), reason."),
+	)
+)
 
 type CompactorType string
 
@@ -653,6 +668,14 @@ func (c *Compactor) expandGrantsIncremental(ctx context.Context, newSyncId strin
 	if base.HasCollapsedCycles() {
 		return false, expand.ErrIncrementalFallback
 	}
+	// The expander prechecks the base's dangling endpoints. Still-missing ids
+	// remain recorded without seeding a walk; ids that now resolve seed the
+	// affected closure. Overflow is the exception: the recorded set no longer
+	// describes everything that was skipped, so incremental expansion cannot
+	// safely agree with full expansion.
+	if base.DanglingOverflow {
+		return false, expand.ErrIncrementalDanglingReferenceDecline
+	}
 
 	// Bound the walk by the remaining run duration; the walk polls ctx.Err().
 	// Finalization uses detached contexts, so an expired walk deadline never
@@ -821,8 +844,7 @@ func (c *Compactor) expandGrantsIncremental(ctx context.Context, newSyncId strin
 		}
 		return false, err
 	}
-
-	if len(newEdges) == 0 && len(changedEntitlementIDs) == 0 {
+	if len(newEdges) == 0 && len(changedEntitlementIDs) == 0 && len(base.DanglingEntitlementIDs) == 0 {
 		// Nothing changed relative to the base — its grants were already merged in.
 		base, err = base.Clone()
 		if err != nil {
@@ -1264,6 +1286,11 @@ func (c *Compactor) expandGrants(ctx context.Context, newSyncId string, compacti
 			logIncrementalOutcome(ctx, "declined", "revocation")
 		case errors.Is(err, expand.ErrIncrementalDenseChangeDecline):
 			logIncrementalOutcome(ctx, "declined", "dense_change")
+		case errors.Is(err, expand.ErrIncrementalDanglingReferenceDecline):
+			// Overflow means the bounded id set no longer completely describes
+			// the endpoints skipped by the base expansion, so incremental cannot
+			// safely agree with full expansion.
+			logIncrementalOutcome(ctx, "declined", "dangling_overflow")
 		case errors.Is(err, expand.ErrIncrementalFallback):
 			// New edge closed a cycle: full expansion handles cycles correctly.
 			logIncrementalOutcome(ctx, "declined", "cycle")
@@ -1351,12 +1378,19 @@ func (c *Compactor) expandGrants(ctx context.Context, newSyncId string, compacti
 	return nil
 }
 
+// logIncrementalOutcome is the single reporting site for how an incremental
+// expansion attempt ended. Every caller passes literal outcome/reason strings,
+// so both are safe as metric attributes.
 func logIncrementalOutcome(ctx context.Context, outcome, reason string, fields ...zap.Field) {
 	fields = append([]zap.Field{
 		zap.String("incremental_expansion_outcome", outcome),
 		zap.String("incremental_expansion_reason", reason),
 	}, fields...)
 	ctxzap.Extract(ctx).Info("incremental grant expansion outcome", fields...)
+	incrementalExpansionOutcomeCounter.Add(ctx, 1, metric.WithAttributes(
+		attribute.String("outcome", outcome),
+		attribute.String("reason", reason),
+	))
 }
 
 func (c *Compactor) loadIncrementalBaseGraph(ctx context.Context) (*expand.EntitlementGraph, error) {
diff --git a/pkg/synccompactor/incremental_expansion_test.go b/pkg/synccompactor/incremental_expansion_test.go
index 7829fc347..6741b7a1d 100644
--- a/pkg/synccompactor/incremental_expansion_test.go
+++ b/pkg/synccompactor/incremental_expansion_test.go
@@ -304,6 +304,111 @@ func baseGraphForFixtures(t testing.TB, ctx context.Context) *expand.Entitlement
 	return g
 }
 
+func buildResolvedDanglingFixtures(t *testing.T, ctx context.Context, dir string, missingSource bool) []*CompactableSync {
+	t.Helper()
+
+	sourceGroup, destGroup := grp("source"), grp("dest")
+	alice := usr("alice")
+	sourceEnt, destEnt := ent("ent-source", sourceGroup), ent("ent-dest", destGroup)
+	userRT := v2.ResourceType_builder{Id: "user", DisplayName: "User"}.Build()
+	groupRT := v2.ResourceType_builder{Id: "group", DisplayName: "Group"}.Build()
+
+	basePath := filepath.Join(dir, "base-dangling.c1z")
+	base, err := dotc1z.NewStore(ctx, basePath, dotc1z.WithEngine(c1zstore.EnginePebble))
+	require.NoError(t, err)
+	baseSyncID, err := base.StartNewSync(ctx, connectorstore.SyncTypeFull, "")
+	require.NoError(t, err)
+	require.NoError(t, base.PutResourceTypes(ctx, userRT, groupRT))
+	require.NoError(t, base.PutResources(ctx, sourceGroup, destGroup, alice))
+	if missingSource {
+		require.NoError(t, base.PutEntitlements(ctx, destEnt))
+	} else {
+		require.NoError(t, base.PutEntitlements(ctx, sourceEnt))
+	}
+	require.NoError(t, base.PutGrants(ctx,
+		memberGrant(sourceEnt, alice),
+		ruleGrant(destEnt, sourceGroup, sourceEnt.GetId()),
+	))
+	require.NoError(t, base.EndSync(ctx))
+
+	graph := expand.NewEntitlementGraph(ctx)
+	graph.AddEntitlementID(sourceEnt.GetId())
+	graph.AddEntitlementID(destEnt.GetId())
+	require.NoError(t, graph.AddEdge(ctx, sourceEnt.GetId(), destEnt.GetId(), false, nil))
+	graph.MarkEdgeExpanded(sourceEnt.GetId(), destEnt.GetId())
+	graph.Loaded = true
+	graph.HasNoCycles = true
+	missingID := destEnt.GetId()
+	if missingSource {
+		missingID = sourceEnt.GetId()
+	}
+	graph.NoteDanglingReference(missingID)
+	persistFixtureGraph(t, ctx, base, baseSyncID, graph)
+	require.NoError(t, base.Close(ctx))
+
+	incPath := filepath.Join(dir, "resolved-dangling.c1z")
+	inc, err := dotc1z.NewStore(ctx, incPath, dotc1z.WithEngine(c1zstore.EnginePebble))
+	require.NoError(t, err)
+	incSyncID, err := inc.StartNewSync(ctx, connectorstore.SyncTypePartial, "")
+	require.NoError(t, err)
+	if missingSource {
+		require.NoError(t, inc.PutEntitlements(ctx, sourceEnt))
+	} else {
+		require.NoError(t, inc.PutEntitlements(ctx, destEnt))
+	}
+	require.NoError(t, inc.EndSync(ctx))
+	require.NoError(t, inc.Close(ctx))
+
+	return []*CompactableSync{
+		{FilePath: basePath, SyncID: baseSyncID},
+		{FilePath: incPath, SyncID: incSyncID},
+	}
+}
+
+func TestCompactor_ResolvedDanglingEntitlementSeedsIncrementalWalk(t *testing.T) {
+	tests := []struct {
+		name          string
+		missingSource bool
+		missingID     string
+	}{
+		{name: "destination", missingID: "ent-dest"},
+		{name: "source", missingSource: true, missingID: "ent-source"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctx := context.Background()
+			incrementalEntries := buildResolvedDanglingFixtures(t, ctx, t.TempDir(), tc.missingSource)
+			incremental, incrementalCleanup, err := NewCompactor(ctx, t.TempDir(), incrementalEntries,
+				WithTmpDir(t.TempDir()), WithEngine(c1zstore.EnginePebble), WithIncrementalExpansion())
+			require.NoError(t, err)
+			defer func() { require.NoError(t, incrementalCleanup()) }()
+
+			incrementalOut, err := incremental.Compact(ctx)
+			require.NoError(t, err)
+			require.True(t, incremental.incrementalExpansionRan,
+				"a row-only dangling resolution must use the incremental path")
+
+			fullEntries := buildResolvedDanglingFixtures(t, ctx, t.TempDir(), tc.missingSource)
+			full, fullCleanup, err := NewCompactor(ctx, t.TempDir(), fullEntries,
+				WithTmpDir(t.TempDir()), WithEngine(c1zstore.EnginePebble))
+			require.NoError(t, err)
+			defer func() { require.NoError(t, fullCleanup()) }()
+
+			fullOut, err := full.Compact(ctx)
+			require.NoError(t, err)
+			incrementalGrants := grantOutcome(t, ctx, incrementalOut.FilePath, incrementalOut.SyncID)
+			fullGrants := grantOutcome(t, ctx, fullOut.FilePath, fullOut.SyncID)
+			hasGrant(t, fullGrants, "ent-dest|user|alice")
+			require.Equal(t, fullGrants, incrementalGrants,
+				"incremental grants and provenance must match full expansion")
+
+			graph := artifactGraph(t, ctx, incrementalOut.FilePath, incrementalOut.SyncID)
+			require.NotContains(t, graph.DanglingEntitlementIDs, tc.missingID,
+				"a resolved id must not persist into the next generation")
+		})
+	}
+}
+
 // grantOutcome reads every grant from a compacted c1z and returns the set of
 // full-row keys "entitlement|principalType|principalResource|sources=..." —
 // INCLUDING the sources/provenance map, so the differential also pins that
diff --git a/pkg/synccompactor/incremental_hardening_test.go b/pkg/synccompactor/incremental_hardening_test.go
index e844a2b38..320ee6a46 100644
--- a/pkg/synccompactor/incremental_hardening_test.go
+++ b/pkg/synccompactor/incremental_hardening_test.go
@@ -66,6 +66,7 @@ func TestIncrementalExpansionOutcomeLogging(t *testing.T) {
 		options     []Option
 		wantOutcome string
 		wantReason  string
+		checkOutput func(*testing.T, context.Context, *CompactableSync)
 	}{
 		{
 			name: "success", build: func(t *testing.T, ctx context.Context, dir string) []*CompactableSync {
@@ -107,6 +108,46 @@ func TestIncrementalExpansionOutcomeLogging(t *testing.T) {
 			options:     []Option{WithEngine(c1zstore.EnginePebble), WithIncrementalExpansion()},
 			wantOutcome: "fell_back", wantReason: "base_graph_error",
 		},
+		{
+			// Overflow means the recorded ids no longer describe what was
+			// skipped, so seeding them cannot make this agree with full
+			// expansion. Only this case declines.
+			name: "dangling overflow decline",
+			build: func(t *testing.T, ctx context.Context, dir string) []*CompactableSync {
+				entries := buildIncrementalFixtures(t, ctx, dir)
+				graph := baseGraphForFixtures(t, ctx)
+				graph.NoteUnrecoverableDangling()
+				store, err := dotc1z.NewStore(ctx, entries[0].FilePath, dotc1z.WithTmpDir(t.TempDir()))
+				require.NoError(t, err)
+				persistFixtureGraph(t, ctx, store, entries[0].SyncID, graph)
+				require.NoError(t, store.Close(ctx))
+				return entries
+			},
+			options:     []Option{WithEngine(c1zstore.EnginePebble), WithIncrementalExpansion()},
+			wantOutcome: "declined", wantReason: "dangling_overflow",
+		},
+		{
+			// Ordinary dangling ids must NOT decline. The expander prechecks them:
+			// still-missing ids remain recorded without seeding a walk, while ids
+			// that now resolve seed their affected closure.
+			name: "recorded dangling ids still take the fast path",
+			build: func(t *testing.T, ctx context.Context, dir string) []*CompactableSync {
+				entries := buildIncrementalFixtures(t, ctx, dir)
+				graph := baseGraphForFixtures(t, ctx)
+				graph.NoteDanglingReference("never:resolves")
+				store, err := dotc1z.NewStore(ctx, entries[0].FilePath, dotc1z.WithTmpDir(t.TempDir()))
+				require.NoError(t, err)
+				persistFixtureGraph(t, ctx, store, entries[0].SyncID, graph)
+				require.NoError(t, store.Close(ctx))
+				return entries
+			},
+			options:     []Option{WithEngine(c1zstore.EnginePebble), WithIncrementalExpansion()},
+			wantOutcome: "succeeded", wantReason: "none",
+			checkOutput: func(t *testing.T, ctx context.Context, out *CompactableSync) {
+				graph := artifactGraph(t, ctx, out.FilePath, out.SyncID)
+				require.Contains(t, graph.DanglingEntitlementIDs, "never:resolves")
+			},
+		},
 		{
 			name: "unsupported engine", build: func(t *testing.T, ctx context.Context, dir string) []*CompactableSync {
 				return buildIncrementalFixturesEngine(t, ctx, dir, c1zstore.EngineSQLite)
@@ -127,8 +168,11 @@ func TestIncrementalExpansionOutcomeLogging(t *testing.T) {
 			compactor, cleanup, err := NewCompactor(ctx, t.TempDir(), entries, options...)
 			require.NoError(t, err)
 			defer func() { require.NoError(t, cleanup()) }()
-			_, err = compactor.Compact(ctx)
+			out, err := compactor.Compact(ctx)
 			require.NoError(t, err)
+			if tc.checkOutput != nil {
+				tc.checkOutput(t, ctx, out)
+			}
 
 			found := false
 			for _, line := range strings.Split(strings.TrimSpace(logs.String()), "\n") {