fix(lattice): compile durable fragment tombstones

apps/predbat/lattice_fragment_adapters.py

@@ -20,6 +20,7 @@
 import hashlib
 import threading
 from dataclasses import dataclass
+from functools import partial
 from types import MappingProxyType
 from typing import Optional, Protocol
 
@@ -110,6 +111,43 @@ def __post_init__(self):
         object.__setattr__(self, "provider_id", provider_id)
 
 
+def _compiler_fragment_snapshot(adapter):
+    """Fresh-read one compiler input, translating only durable removals."""
+    state = adapter.read_state()
+    if not isinstance(state, FragmentAdapterState):
+        raise FragmentAdapterReadError("adapter read_state must return FragmentAdapterState")
+    try:
+        state.__post_init__()
+    except ValueError as exc:
+        raise FragmentAdapterReadError(str(exc)) from exc
+    provider_id = _validate_provider_id(getattr(adapter, "provider_id", None))
+    if state.provider_id != provider_id:
+        raise FragmentAdapterReadError("adapter state belongs to provider {}".format(state.provider_id))
+    if not state.removed:
+        return state.snapshot
+    return ProviderSnapshot(
+        provider_id=state.provider_id,
+        generation=state.generation,
+        health=ProviderHealth.HEALTHY,
+        topology_fragment={
+            "topologyVersion": "0.3.0",
+            "scope": "fragment",
+            "docVersion": state.generation,
+            "producer": {
+                "name": "PredBat fragment tombstone",
+                "provider": state.provider_id,
+                "authority": 0,
+            },
+            "nodes": [],
+            "relationships": [],
+        },
+        aliases=(),
+        identity_aliases=(),
+        role_assignments=(),
+        config_projections=(),
+    )
+
+
 class FragmentAdapterStateStore:
     """Required atomic durable-store protocol owned by one integration."""
 
@@ -484,7 +522,13 @@ def create_compiler(self, state_store, override_reader=None):
                 raise RuntimeError("cannot create a compiler without fragment adapters")
             for adapter in self._adapters.values():
                 self._validate_adapter(adapter)
-            readers = {provider_id: adapter.read_snapshot for provider_id, adapter in self._adapters.items()}
+            readers = {
+                provider_id: partial(
+                    _compiler_fragment_snapshot,
+                    adapter,
+                )
+                for provider_id, adapter in self._adapters.items()
+            }
             compiler = CompiledLatticeCompiler(
                 readers,
                 state_store=state_store,

apps/predbat/tests/test_lattice_fragment_adapters.py

@@ -401,8 +401,8 @@ def test_other_provider_invalidation_exposes_reader_failure_fail_closed(self):
             {issue.code for issue in run.issues},
         )
 
-    def test_removal_invalidates_and_preserves_last_known_good(self):
-        """A durable removal triggers a bounded fail-closed recompile."""
+    def test_removal_invalidates_and_clears_prior_provider_contribution(self):
+        """A durable removal compiles as an acknowledged empty fragment."""
         adapter, _store, _initial = publisher("gateway")
         _registry, compiler, _compiled_store, _ids = compiled_registry(adapter)
         first = compiler.drain()
@@ -411,13 +411,16 @@ def test_removal_invalidates_and_preserves_last_known_good(self):
         run = compiler.drain()
 
         self.assertEqual(run.attempts, 1)
-        self.assertEqual(run.status, CompileStatus.STALE)
-        self.assertTrue(run.pending)
-        self.assertIs(run.publication, first.publication)
-        self.assertIn(
-            "provider_read_failed",
-            {issue.code for issue in run.issues},
+        self.assertEqual(run.status, CompileStatus.FRESH)
+        self.assertFalse(run.pending)
+        self.assertTrue(run.published)
+        self.assertEqual(run.publication.lattice_version, 2)
+        self.assertEqual(
+            dict(run.publication.provider_generations),
+            {"gateway": 2},
         )
+        self.assertEqual(run.plan.topology["nodes"], ())
+        self.assertNotEqual(run.publication, first.publication)
 
     def test_restart_restores_adapter_and_compiler_cursor_protection(self):
         """Both durable layers reject reuse after a complete process restart."""
@@ -482,7 +485,8 @@ def test_invalidation_during_read_gets_one_bounded_follow_up(self):
         adapter, _store, initial = publisher("gateway")
         registry = FragmentAdapterRegistry(enabled=True)
         registry.register(adapter)
-        original_reader = adapter.read_snapshot
+        compiler = registry.create_compiler(InMemoryCompiledLatticeStateStore())
+        original_reader = adapter.read_state
         fired = [False]
 
         def invalidating_reader():
@@ -496,8 +500,7 @@ def invalidating_reader():
                 )
             return value
 
-        adapter.read_snapshot = invalidating_reader
-        compiler = registry.create_compiler(InMemoryCompiledLatticeStateStore())
+        adapter.read_state = invalidating_reader
         run = compiler.drain()
 
         self.assertEqual(run.attempts, 2)

apps/predbat/tests/test_lattice_fragment_tombstone_compilation.py

@@ -0,0 +1,258 @@
+"""Focused coverage for compiler-only fragment tombstone translation."""
+
+# cspell:ignore autoconfig
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
+
+from lattice_autoconfig import CompileStatus, ProviderHealth  # noqa: E402
+from lattice_compiled_publication import (  # noqa: E402
+    InMemoryCompiledLatticeStateStore,
+)
+from lattice_fragment_adapters import (  # noqa: E402
+    DurableFragmentAdapter,
+    FragmentAdapterState,
+    FragmentAdapterRegistry,
+    FragmentAdapterRemoved,
+    InMemoryFragmentAdapterStateStore,
+    _compiler_fragment_snapshot,
+)
+from tests.test_lattice_autoconfig import snapshot  # noqa: E402
+
+
+def publisher(provider_id, node_id):
+    """Create one seeded integration-owned fragment publisher."""
+    store = InMemoryFragmentAdapterStateStore()
+    adapter = DurableFragmentAdapter(provider_id, store)
+    initial = snapshot(
+        provider_id,
+        generation=1,
+        node_id=node_id,
+    )
+    adapter.publish(initial, "initial fragment")
+    return adapter, store, initial
+
+
+def compiler_for(store, *adapters):
+    """Create one explicitly enabled compiler over fixed membership."""
+    registry = FragmentAdapterRegistry(enabled=True)
+    for adapter in adapters:
+        registry.register(adapter)
+    return registry, registry.create_compiler(store)
+
+
+def structurally_corrupt_state(
+    provider_id,
+    generation,
+    semantic_fingerprint,
+    provider_snapshot,
+    removed=False,
+):
+    """Bypass construction validation to model a corrupt structural reader."""
+    state = object.__new__(FragmentAdapterState)
+    object.__setattr__(state, "provider_id", provider_id)
+    object.__setattr__(state, "generation", generation)
+    object.__setattr__(
+        state,
+        "semantic_fingerprint",
+        semantic_fingerprint,
+    )
+    object.__setattr__(state, "snapshot", provider_snapshot)
+    object.__setattr__(state, "removed", removed)
+    return state
+
+
+class TestCompilerFragmentTombstones(unittest.TestCase):
+    """Removal is empty compiler input without weakening adapter reads."""
+
+    def test_live_snapshot_is_exact_and_removed_snapshot_is_empty(self):
+        """Translation changes only a durable removal state."""
+        adapter, _store, initial = publisher("gateway", "GW-INV")
+
+        self.assertIs(_compiler_fragment_snapshot(adapter), initial)
+        self.assertTrue(adapter.remove(2, "integration removed"))
+
+        with self.assertRaises(FragmentAdapterRemoved):
+            adapter.read_snapshot()
+        tombstone = _compiler_fragment_snapshot(adapter)
+
+        self.assertEqual(tombstone.provider_id, "gateway")
+        self.assertEqual(tombstone.generation, 2)
+        self.assertEqual(tombstone.health, ProviderHealth.HEALTHY)
+        self.assertEqual(tombstone.topology_fragment["nodes"], ())
+        self.assertEqual(tombstone.topology_fragment["relationships"], ())
+        self.assertEqual(tombstone.aliases, ())
+        self.assertEqual(tombstone.identity_aliases, ())
+        self.assertEqual(tombstone.role_assignments, ())
+        self.assertEqual(tombstone.config_projections, ())
+
+    def test_all_removed_publishes_deterministic_empty_plan_and_restarts(self):
+        """All tombstones settle, persist, and restore as one empty plan."""
+        alpha, alpha_store, _alpha = publisher("alpha", "ALPHA-INV")
+        beta, beta_store, _beta = publisher("beta", "BETA-INV")
+        compiled_store = InMemoryCompiledLatticeStateStore()
+        _registry, compiler = compiler_for(compiled_store, alpha, beta)
+        baseline = compiler.drain()
+
+        self.assertTrue(beta.remove(2, "beta removed"))
+        self.assertTrue(alpha.remove(2, "alpha removed"))
+        removed = compiler.drain()
+
+        self.assertEqual(removed.status, CompileStatus.FRESH)
+        self.assertTrue(removed.published)
+        self.assertFalse(removed.pending)
+        self.assertEqual(removed.plan.topology["nodes"], ())
+        self.assertEqual(removed.plan.aliases, ())
+        self.assertEqual(
+            dict(removed.publication.provider_generations),
+            {"alpha": 2, "beta": 2},
+        )
+        self.assertEqual(
+            dict(removed.publication.provider_requested_generations),
+            {"alpha": 2, "beta": 2},
+        )
+        self.assertNotEqual(
+            removed.publication.digest,
+            baseline.publication.digest,
+        )
+
+        restarted_alpha = DurableFragmentAdapter("alpha", alpha_store)
+        restarted_beta = DurableFragmentAdapter("beta", beta_store)
+        _registry, restarted = compiler_for(
+            compiled_store,
+            restarted_beta,
+            restarted_alpha,
+        )
+        settled = restarted.drain()
+
+        self.assertFalse(settled.published)
+        self.assertFalse(settled.pending)
+        self.assertEqual(settled.publication, removed.publication)
+        self.assertEqual(settled.plan.digest, removed.plan.digest)
+
+    def test_removed_payload_cannot_reappear_at_the_same_generation(self):
+        """A tombstone never leaks its payload and cursor reuse remains unsafe."""
+        adapter, _store, initial = publisher("cloud", "CLOUD-INV")
+        self.assertTrue(adapter.remove(2, "cloud removed"))
+
+        empty = _compiler_fragment_snapshot(adapter)
+        self.assertEqual(empty.topology_fragment["nodes"], ())
+        with self.assertRaisesRegex(ValueError, "reused"):
+            adapter.publish(
+                snapshot(
+                    "cloud",
+                    generation=2,
+                    node_id=initial.topology_fragment["nodes"][0]["id"],
+                ),
+                "attempted resurrection",
+            )
+        self.assertEqual(
+            _compiler_fragment_snapshot(adapter).topology_fragment["nodes"],
+            (),
+        )
+
+    def test_fresh_read_revalidates_semantic_fingerprint_and_keeps_lkg(self):
+        """Post-registration state corruption cannot reach publication."""
+        adapter, _store, _initial = publisher("gateway", "GW-INV")
+        compiled_store = InMemoryCompiledLatticeStateStore()
+        _registry, compiler = compiler_for(compiled_store, adapter)
+        baseline = compiler.drain()
+        corrupt = structurally_corrupt_state(
+            "gateway",
+            2,
+            "0" * 64,
+            snapshot(
+                "gateway",
+                generation=2,
+                node_id="GW-INV",
+            ),
+        )
+        adapter.read_state = lambda: corrupt
+
+        self.assertTrue(
+            compiler.invalidate(
+                "gateway",
+                2,
+                "corrupt structural read",
+            )
+        )
+        failed = compiler.drain()
+
+        self.assertEqual(failed.status, CompileStatus.STALE)
+        self.assertTrue(failed.pending)
+        self.assertFalse(failed.published)
+        self.assertIs(failed.publication, baseline.publication)
+        self.assertEqual(compiled_store.writes, 1)
+        self.assertIn(
+            "provider_read_failed",
+            {issue.code for issue in failed.issues},
+        )
+
+    def test_fresh_read_revalidates_snapshot_cursor_bindings(self):
+        """Provider and generation binding corruption both fail closed."""
+        cases = (
+            (
+                "provider",
+                snapshot(
+                    "other-provider",
+                    generation=2,
+                    node_id="GW-INV",
+                ),
+            ),
+            (
+                "generation",
+                snapshot(
+                    "gateway",
+                    generation=3,
+                    node_id="GW-INV",
+                ),
+            ),
+        )
+        for label, corrupt_snapshot in cases:
+            with self.subTest(binding=label):
+                adapter, _store, _initial = publisher(
+                    "gateway",
+                    "GW-INV",
+                )
+                compiled_store = InMemoryCompiledLatticeStateStore()
+                _registry, compiler = compiler_for(
+                    compiled_store,
+                    adapter,
+                )
+                baseline = compiler.drain()
+                corrupt = structurally_corrupt_state(
+                    "gateway",
+                    2,
+                    adapter.read_state().semantic_fingerprint,
+                    corrupt_snapshot,
+                )
+                adapter.read_state = lambda value=corrupt: value
+
+                self.assertTrue(
+                    compiler.invalidate(
+                        "gateway",
+                        2,
+                        "{} binding corruption".format(label),
+                    )
+                )
+                failed = compiler.drain()
+
+                self.assertEqual(failed.status, CompileStatus.STALE)
+                self.assertTrue(failed.pending)
+                self.assertFalse(failed.published)
+                self.assertIs(
+                    failed.publication,
+                    baseline.publication,
+                )
+                self.assertEqual(compiled_store.writes, 1)
+                self.assertIn(
+                    "provider_read_failed",
+                    {issue.code for issue in failed.issues},
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

apps/predbat/tests/test_lattice_solis_fragment.py

@@ -398,7 +398,13 @@ def test_all_changes_invalidate_common_compiler_with_fresh_reads(self):
         self.assertTrue(offline.pending)
         self.assertTrue(adapter.remove())
         removed = compiler.drain()
-        self.assertEqual(removed.status, CompileStatus.STALE)
+        self.assertEqual(removed.status, CompileStatus.FRESH)
+        self.assertFalse(removed.pending)
+        self.assertEqual(removed.plan.topology["nodes"], ())
+        self.assertEqual(
+            dict(removed.publication.provider_generations),
+            {"solis": 5},
+        )
 
         self.assertEqual(
             tuple(item[:3] for item in invalidations),