fix(graph): exempt deleted rows from the orphan-registration sweep

robosystems/operations/graph/infrastructure.py

@@ -407,10 +407,13 @@ def cleanup_stale_graphs(self) -> CleanupResult:
     when the instance reappears. Reconciling from on-disk truth is the
     follow-up, not this sweep.
 
-    Shared repositories are exempt from the orphan check — their rows are
-    bookkeeping, not routing, and their master is deliberately parked to zero
-    between ingestion runs. An exempt row that still carries a stamp from
-    before this rule is cleared on the next sweep.
+    Shared repositories and rows already marked ``deleted`` are exempt from
+    the orphan check — neither is routing. A shared repository's master is
+    deliberately parked to zero between ingestion runs; a deleted graph's
+    instance is recycled long before the row ages out, so counting it means
+    every fleet replacement pages for ``STALE_GRAPH_DAYS`` and then heals
+    itself. An exempt row that still carries a stamp from before these rules
+    is cleared on the next sweep.
     """
     logger.info("Starting graph registry cleanup")
 
@@ -481,9 +484,16 @@ def cleanup_stale_graphs(self) -> CleanupResult:
         # legitimately absent most of the day; counting that as drift pages
         # every night for a healthy fleet. Same predicate the router uses, so
         # the two can never disagree about which graphs this applies to.
+        # A row already marked deleted is not routing either: it is waiting
+        # out ``STALE_GRAPH_DAYS`` above, and the instance it names was
+        # released with the graph and recycled by the ASG well inside that
+        # window. Counting it turns every fleet replacement into a page that
+        # clears itself a week later, on an alarm whose whole meaning is
+        # "a live graph's routing is stale".
         instance_missing = (
           bool(instance_id)
           and instance_id not in valid_instances
+          and status != "deleted"
           and not is_shared_repository_or_subgraph(graph_id)
         )
         already_marked = item.get("instance_missing_since") is not None

tests/operations/graph/test_infrastructure.py

@@ -553,6 +553,95 @@ def test_user_graph_is_still_orphaned_when_a_shared_repo_is_present(self, monito
     metric = monitor._cloudwatch.put_metric_data.call_args.kwargs
     assert metric["MetricData"][0]["Value"] == 1
 
+  @pytest.mark.unit
+  def test_recently_deleted_graph_is_not_orphaned(self, monitor):
+    """A deleted row inside the retention window is not routing: its instance
+    was released with the graph and recycled by the ASG. Counting it made
+    every fleet replacement page for a week and then heal itself."""
+    graph_table = _make_dynamo_table(
+      items=[
+        {
+          "graph_id": "kg_deleted",
+          "status": "deleted",
+          "deleted_at": (datetime.now(UTC) - timedelta(days=2)).isoformat(),
+          "instance_id": "i-recycled000000001",
+        },
+      ]
+    )
+    instance_table = _make_dynamo_table(items=[])
+    monitor._dynamodb.Table.side_effect = lambda name: (
+      graph_table if name == "test-graph" else instance_table
+    )
+
+    result = monitor.cleanup_stale_graphs()
+
+    assert result.orphaned_count == 0
+    assert result.removed_count == 0
+    graph_table.update_item.assert_not_called()
+    graph_table.delete_item.assert_not_called()
+    metric = monitor._cloudwatch.put_metric_data.call_args.kwargs
+    assert metric["MetricData"][0]["Value"] == 0
+
+  @pytest.mark.unit
+  def test_deleted_graph_marker_from_before_the_exemption_is_cleared(self, monitor):
+    """A row stamped while it was still live, then deleted, drops the marker
+    rather than carrying it until the row ages out."""
+    graph_table = _make_dynamo_table(
+      items=[
+        {
+          "graph_id": "kg_deleted",
+          "status": "deleted",
+          "deleted_at": (datetime.now(UTC) - timedelta(days=2)).isoformat(),
+          "instance_id": "i-recycled000000001",
+          "instance_missing_since": "2026-09-02T03:01:26+00:00",
+        },
+      ]
+    )
+    instance_table = _make_dynamo_table(items=[])
+    monitor._dynamodb.Table.side_effect = lambda name: (
+      graph_table if name == "test-graph" else instance_table
+    )
+
+    result = monitor.cleanup_stale_graphs()
+
+    assert result.orphaned_count == 0
+    assert result.updated_count == 1
+    update = graph_table.update_item.call_args.kwargs
+    assert update["Key"] == {"graph_id": "kg_deleted"}
+    assert update["UpdateExpression"] == "REMOVE instance_missing_since"
+
+  @pytest.mark.unit
+  def test_live_graph_is_still_orphaned_alongside_a_deleted_row(self, monitor):
+    """The exemption is scoped to deleted rows — a live graph swept at the
+    same time is still marked and counted."""
+    graph_table = _make_dynamo_table(
+      items=[
+        {
+          "graph_id": "kg_deleted",
+          "status": "deleted",
+          "deleted_at": (datetime.now(UTC) - timedelta(days=2)).isoformat(),
+          "instance_id": "i-recycled000000001",
+        },
+        {
+          "graph_id": "kg_orphan",
+          "status": "active",
+          "instance_id": "i-doesnotexist00001",
+        },
+      ]
+    )
+    instance_table = _make_dynamo_table(items=[])
+    monitor._dynamodb.Table.side_effect = lambda name: (
+      graph_table if name == "test-graph" else instance_table
+    )
+
+    result = monitor.cleanup_stale_graphs()
+
+    assert result.orphaned_count == 1
+    update = graph_table.update_item.call_args.kwargs
+    assert update["Key"] == {"graph_id": "kg_orphan"}
+    metric = monitor._cloudwatch.put_metric_data.call_args.kwargs
+    assert metric["MetricData"][0]["Value"] == 1
+
   @pytest.mark.unit
   def test_exception_sets_error_message(self, monitor):
     """Top-level exception is captured in error_message."""