fix(graph): finalize empty community lifecycle and batch selection

src/core/graph/application/communities/summarizer.py

@@ -62,9 +62,27 @@ async def summarize_community(
         # 1. Fetch data for community
         data = await self._fetch_community_data(community_id, tenant_id, generation_id)
         if not data["entities"] and not data["child_summaries"]:
-            logger.warning(
-                f"Community {community_id} has no entities and no child summaries. Skipping."
+            emptied = await self.graph.execute_write(
+                """
+                MATCH (c:Community {id: $id, tenant_id: $tenant_id})
+                SET c.status = c.status
+                WITH c
+                WHERE ($generation_id IS NULL OR c.generation_id = $generation_id)
+                  AND NOT EXISTS { MATCH (:Entity)-[:BELONGS_TO|IN_COMMUNITY]->(c) }
+                  AND NOT EXISTS { MATCH (c)-[:PARENT_OF]->(:Community) }
+                SET c.status = 'empty', c.is_stale = false
+                RETURN c.id AS id
+                """,
+                {
+                    "id": community_id,
+                    "tenant_id": tenant_id,
+                    "generation_id": generation_id,
+                },
             )
+            if emptied:
+                logger.info(f"Community {community_id} is empty")
+            else:
+                logger.info(f"Community {community_id} changed while it was being summarized")
             return {}
 
         # 2. Resolve the target model before rendering so the input has a hard budget.
@@ -175,7 +193,7 @@ async def summarize_all_stale(
         WHERE c.tenant_id = $tenant_id
           AND ($generation_id IS NULL OR c.generation_id = $generation_id)
           AND ($generation_id IS NOT NULL OR coalesce(c.active, true) = true)
-          AND (c.summary IS NULL OR c.is_stale = true)
+          AND (c.is_stale = true OR (c.summary IS NULL AND coalesce(c.status, '') <> 'empty'))
         RETURN c.id as id, coalesce(c.level, 0) as level
         ORDER BY c.level ASC
         """

tests/unit/test_community_services.py

@@ -123,10 +123,40 @@ async def test_summarize_community_success(self, mock_neo4j, mock_factory):
     async def test_summarize_community_no_data(self, mock_neo4j, mock_factory):
         summarizer = CommunitySummarizer(mock_neo4j, mock_factory)
         mock_neo4j.execute_read.return_value = []
+        mock_neo4j.execute_write.return_value = [{"id": "comm_0_empty"}]
 
         with patch("src.shared.kernel.runtime.get_settings"):
-            result = await summarizer.summarize_community("comm_0_empty", "tenant_1")
+            result = await summarizer.summarize_community(
+                "comm_0_empty", "tenant_1", generation_id="generation_1"
+            )
+
         assert result == {}
+        query, params = mock_neo4j.execute_write.await_args.args
+        assert "tenant_id: $tenant_id" in query
+        assert "c.generation_id = $generation_id" in query
+        assert "[:BELONGS_TO|IN_COMMUNITY]" in query
+        assert "[:PARENT_OF]" in query
+        assert "c.status = 'empty', c.is_stale = false" in query
+        assert params == {
+            "id": "comm_0_empty",
+            "tenant_id": "tenant_1",
+            "generation_id": "generation_1",
+        }
+        mock_factory.get_llm_provider.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_summarize_all_stale_excludes_terminal_empty_communities(
+        self, mock_neo4j, mock_factory
+    ):
+        mock_neo4j.execute_read.return_value = []
+
+        await CommunitySummarizer(mock_neo4j, mock_factory).summarize_all_stale("tenant_1")
+
+        query = " ".join(mock_neo4j.execute_read.await_args.args[0].split())
+        assert (
+            "AND (c.is_stale = true OR "
+            "(c.summary IS NULL AND coalesce(c.status, '') <> 'empty'))" in query
+        )
 
     @pytest.mark.asyncio
     async def test_summarize_community_failure_keeps_ready_summary_available(