Generic state merging, cover pullbacks

package/version

@@ -1 +1 @@
-0.1.384
+0.1.385

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "pyk"
-version = "0.1.384"
+version = "0.1.385"
 description = ""
 authors = [
     "Runtime Verification, Inc. <[email protected]>",

src/pyk/cterm.py

@@ -7,6 +7,7 @@
 
 from .kast.inner import KApply, KAtt, KInner, KRewrite, KVariable, Subst
 from .kast.manip import (
+    anti_unify,
     apply_existential_substitutions,
     count_vars,
     flatten_label,
@@ -137,6 +138,29 @@ def _ml_impl(antecedents: Iterable[KInner], consequents: Iterable[KInner]) -> KI
     def add_constraint(self, new_constraint: KInner) -> CTerm:
         return CTerm(self.config, [new_constraint] + list(self.constraints))
 
+    def anti_unify(self, other: CTerm) -> tuple[CTerm, CSubst, CSubst]:
+        new_config, _, _ = anti_unify(self.config, other.config)
+        common_constraints = [constraint for constraint in self.constraints if constraint in other.constraints]
+        new_constraints = []
+        fvs = free_vars(new_config)
+        len_fvs = 0
+        while len_fvs < len(fvs):
+            len_fvs = len(fvs)
+            for constraint in common_constraints:
+                if constraint not in new_constraints:
+                    constraint_fvs = free_vars(constraint)
+                    if any(fv in fvs for fv in constraint_fvs):
+                        new_constraints.append(constraint)
+                        fvs.extend(constraint_fvs)
+        new_cterm = CTerm(config=new_config, constraints=new_constraints)
+        self_csubst = new_cterm.match_with_constraint(self)
+        other_csubst = new_cterm.match_with_constraint(other)
+        if self_csubst is None or other_csubst is None:
+            raise ValueError(
+                f'Anti-unification failed to produce a more general state: {(new_cterm, (self, self_csubst), (other, other_csubst))}'
+            )
+        return (new_cterm, self_csubst, other_csubst)
+
 
 @dataclass(frozen=True, order=True)
 class CSubst:
@@ -173,6 +197,10 @@ def apply(self, cterm: CTerm) -> CTerm:
         _kast = self.subst(cterm.kast)
         return CTerm(_kast, [self.constraint])
 
+    @property
+    def ml_pred(self) -> KInner:
+        return mlAnd(flatten_label('#And', self.subst.ml_pred) + list(self.constraints))
+
 
 def remove_useless_constraints(cterm: CTerm, keep_vars: Iterable[str] = ()) -> CTerm:
     used_vars = free_vars(cterm.config) + list(keep_vars)

src/pyk/kast/manip.py

@@ -6,7 +6,7 @@
 
 from ..prelude.k import DOTS, GENERATED_TOP_CELL
 from ..prelude.kbool import FALSE, TRUE, andBool, impliesBool, notBool, orBool
-from ..prelude.ml import mlAnd, mlEqualsTrue, mlImplies, mlOr, mlTop
+from ..prelude.ml import mlAnd, mlEqualsTrue, mlOr
 from ..utils import find_common_items, hash_str
 from .inner import KApply, KRewrite, KSequence, KToken, KVariable, Subst, bottom_up, top_down, var_occurrences
 from .kast import EMPTY_ATT, KAtt, WithKAtt
@@ -597,43 +597,6 @@ def _rewrites_to_abstractions(_kast: KInner) -> KInner:
     return (abstracted_state, subst1, subst2)
 
 
-def anti_unify_with_constraints(
-    constrained_term_1: KInner,
-    constrained_term_2: KInner,
-    implications: bool = False,
-    constraint_disjunct: bool = False,
-    abstracted_disjunct: bool = False,
-) -> KInner:
-    def disjunction_from_substs(subst1: Subst, subst2: Subst) -> KInner:
-        if KToken('true', 'Bool') in [subst1.pred, subst2.pred]:
-            return mlTop()
-        return mlEqualsTrue(orBool([subst1.pred, subst2.pred]))
-
-    state1, constraint1 = split_config_and_constraints(constrained_term_1)
-    state2, constraint2 = split_config_and_constraints(constrained_term_2)
-    constraints1 = flatten_label('#And', constraint1)
-    constraints2 = flatten_label('#And', constraint2)
-    state, subst1, subst2 = anti_unify(state1, state2)
-
-    constraints = [c for c in constraints1 if c in constraints2]
-    constraint1 = mlAnd([c for c in constraints1 if c not in constraints])
-    constraint2 = mlAnd([c for c in constraints2 if c not in constraints])
-    implication1 = mlImplies(constraint1, subst1.ml_pred)
-    implication2 = mlImplies(constraint2, subst2.ml_pred)
-
-    if abstracted_disjunct:
-        constraints.append(disjunction_from_substs(subst1, subst2))
-
-    if implications:
-        constraints.append(implication1)
-        constraints.append(implication2)
-
-    if constraint_disjunct:
-        constraints.append(mlOr([constraint1, constraint2]))
-
-    return mlAnd([state] + constraints)
-
-
 def apply_existential_substitutions(constrained_term: KInner) -> KInner:
     state, constraint = split_config_and_constraints(constrained_term)
     constraints = flatten_label('#And', constraint)

src/pyk/kcfg/kcfg.py

@@ -673,6 +673,29 @@ def create_split(self, source_id: NodeIdLike, splits: Iterable[tuple[NodeIdLike,
         split = KCFG.Split(self.node(source_id), tuple((self.node(nid), csubst) for nid, csubst in splits))
         self._splits[source_id] = split
 
+    def create_merge(self, node_ids: Iterable[NodeIdLike]) -> NodeIdLike:
+        node_ids = list(node_ids)
+        if len(node_ids) < 2:
+            raise ValueError(f'Cannot create merge node with less than 2 input nodes: {node_ids}')
+        cterms = [self.node(nid).cterm for nid in node_ids]
+        new_cterm = cterms[0]
+        for cterm in cterms[1:]:
+            new_cterm, _, _ = new_cterm.anti_unify(cterm)
+        new_node = self.create_node(new_cterm)
+        return new_node.id
+
+    def pullback_covers(self, target_id: NodeIdLike) -> tuple[list[NodeIdLike], NodeIdLike] | None:
+        covers = self.covers(target_id=target_id)
+        source_ids: list[NodeIdLike] = [cover.source.id for cover in covers]
+        if len(source_ids) < 2:
+            return None
+        merge_id = self.create_merge(source_ids)
+        for cover in covers:
+            self.remove_cover(cover.source.id, target_id)
+            self.create_cover(cover.source.id, merge_id)
+        self.create_cover(merge_id, target_id)
+        return source_ids, merge_id
+
     def ndbranches(self, *, source_id: NodeIdLike | None = None, target_id: NodeIdLike | None = None) -> list[NDBranch]:
         source_id = self._resolve(source_id) if source_id is not None else None
         target_id = self._resolve(target_id) if target_id is not None else None

src/pyk/proof/reachability.py

@@ -275,6 +275,14 @@ def summary(self) -> CompositeSummary:
     def get_refutation_id(self, node_id: int) -> str:
         return f'{self.id}.node-infeasible-{node_id}'
 
+    def specialize_target_node(self) -> None:
+        pullback = self.kcfg.pullback_covers(self.target)
+        if pullback is None:
+            _LOGGER.warning(f'Could not make a cover pullback for target node: {self.target}')
+        else:
+            merge_id, source_ids = pullback
+            _LOGGER.info(f'Created specialized target subsumed node {self.id}: {pullback}')
+
     @staticmethod
     def read_proof_data(proof_dir: Path, id: str) -> APRProof:
         proof_subdir = proof_dir / id

src/tests/unit/kast/test_manip.py

@@ -5,9 +5,8 @@
 
 import pytest
 
-from pyk.kast.inner import KApply, KLabel, KRewrite, KSequence, KSort, KToken, KVariable, Subst
+from pyk.kast.inner import KApply, KLabel, KRewrite, KSequence, KSort, KVariable, Subst
 from pyk.kast.manip import (
-    anti_unify_with_constraints,
     bool_to_ml_pred,
     collapse_dots,
     minimize_term,
@@ -16,13 +15,12 @@
     remove_generated_cells,
     rename_generated_vars,
     simplify_bool,
-    split_config_and_constraints,
     split_config_from,
 )
 from pyk.prelude.k import DOTS, GENERATED_TOP_CELL
-from pyk.prelude.kbool import BOOL, FALSE, TRUE, andBool, notBool, orBool
+from pyk.prelude.kbool import BOOL, FALSE, TRUE, andBool, notBool
 from pyk.prelude.kint import INT, intToken
-from pyk.prelude.ml import mlAnd, mlEqualsTrue, mlTop
+from pyk.prelude.ml import mlEqualsTrue, mlTop
 
 from ..utils import a, b, c, f, k, x
 
@@ -341,94 +339,3 @@ def test_split_config_from(term: KInner, expected_config: KInner, expected_subst
     # Then
     assert actual_config == expected_config
     assert actual_subst == expected_subst
-
-
-def test_anti_unify_with_constraints() -> None:
-    cterm1 = mlAnd(
-        [
-            KApply(
-                '<generatedTop>',
-                [
-                    KApply('<k>', KToken('1', 'Int')),
-                    KApply('<generatedCounter>', KVariable('GENERATEDCOUNTER_CELL')),
-                ],
-            ),
-            mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')])),
-        ]
-    )
-    cterm2 = mlAnd(
-        [
-            KApply(
-                '<generatedTop>',
-                [
-                    KApply('<k>', KToken('2', 'Int')),
-                    KApply('<generatedCounter>', KVariable('GENERATEDCOUNTER_CELL')),
-                ],
-            ),
-            mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')])),
-        ]
-    )
-
-    anti_unifier = anti_unify_with_constraints(cterm1, cterm2, abstracted_disjunct=True)
-
-    config, constraints = split_config_and_constraints(anti_unifier)
-
-    assert type(config) is KApply
-    assert type(config.args[0]) is KApply
-    assert type(config.args[0].args[0]) is KVariable
-    var_name = config.args[0].args[0].name
-
-    expected = mlAnd(
-        [
-            mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')])),
-            mlEqualsTrue(
-                orBool(
-                    [
-                        KApply('_==K_', [KVariable(var_name), KToken('1', 'Int')]),
-                        KApply('_==K_', [KVariable(var_name), KToken('2', 'Int')]),
-                    ]
-                )
-            ),
-        ]
-    )
-
-    assert expected == constraints
-
-
-def test_anti_unify_with_constraints_subst_true() -> None:
-    cterm1 = mlAnd(
-        [
-            KApply(
-                '<generatedTop>',
-                [
-                    KApply('<k>', KToken('1', 'Int')),
-                    KApply('<generatedCounter>', KVariable('GENERATEDCOUNTER_CELL')),
-                ],
-            ),
-            mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')])),
-        ]
-    )
-    cterm2 = mlAnd(
-        [
-            KApply(
-                '<generatedTop>',
-                [
-                    KApply('<k>', KToken('1', 'Int')),
-                    KApply('<generatedCounter>', KVariable('GENERATEDCOUNTER_CELL')),
-                ],
-            ),
-            mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')])),
-        ]
-    )
-
-    anti_unifier = anti_unify_with_constraints(cterm1, cterm2, abstracted_disjunct=True)
-
-    config, constraints = split_config_and_constraints(anti_unifier)
-
-    expected = mlAnd(
-        [
-            mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')])),
-        ]
-    )
-
-    assert expected == constraints

src/tests/unit/test_cterm.py

@@ -5,13 +5,14 @@
 
 import pytest
 
-from pyk.cterm import CTerm, build_claim, build_rule
+from pyk.cterm import CSubst, CTerm, build_claim, build_rule
 from pyk.kast import KAtt
-from pyk.kast.inner import KApply, KLabel, KRewrite, KSequence, KVariable
+from pyk.kast.inner import KApply, KLabel, KRewrite, KSequence, KToken, KVariable, Subst
 from pyk.kast.outer import KClaim
 from pyk.prelude.k import GENERATED_TOP_CELL
-from pyk.prelude.kint import INT, intToken
+from pyk.prelude.kint import INT
 from pyk.prelude.ml import mlAnd, mlEqualsTrue
+from pyk.prelude.utils import token
 
 from .utils import a, b, c, f, g, h, k, x, y, z
 
@@ -106,7 +107,7 @@ def test_build_rule(lhs: KInner, rhs: KInner, keep_vars: list[str], expected: KI
 
 
 def constraint(v: KVariable) -> KInner:
-    return KApply('_<=Int_', intToken(0), v)
+    return KApply('_<=Int_', token(0), v)
 
 
 # (<k> V1 </k> #And { true #Equals 0 <=Int V2}) => <k> V2 </k>      expected: <k> _V1 => V2 </k> requires 0 <=Int V2
@@ -156,3 +157,88 @@ def test_build_claim(test_id: str, init: KInner, target: KInner, expected: KClai
 
     # Then
     assert actual == expected
+
+
+ANTI_UNIFY_TEST_DATA: list[tuple[str, CTerm, CTerm, tuple[CTerm, CSubst, CSubst] | None]] = [
+    (
+        'empty-subst',
+        CTerm(
+            KApply(
+                '<generatedTop>',
+                [
+                    KApply('<k>', KToken('1', 'Int')),
+                ],
+            ),
+            [],
+        ),
+        CTerm(
+            KApply(
+                '<generatedTop>',
+                [
+                    KApply('<k>', KToken('1', 'Int')),
+                ],
+            ),
+            [],
+        ),
+        (
+            CTerm(
+                KApply(
+                    '<generatedTop>',
+                    [
+                        KApply('<k>', [KToken('1', 'Int')]),
+                    ],
+                ),
+                [],
+            ),
+            CSubst(Subst({}), []),
+            CSubst(Subst({}), []),
+        ),
+    ),
+    (
+        'same-constraints',
+        CTerm(
+            KApply(
+                '<generatedTop>',
+                [
+                    KApply('<k>', KToken('1', 'Int')),
+                    KApply('<generatedCounter>', KVariable('GENERATEDCOUNTER_CELL')),
+                ],
+            ),
+            [mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')]))],
+        ),
+        CTerm(
+            KApply(
+                '<generatedTop>',
+                [
+                    KApply('<k>', KToken('2', 'Int')),
+                    KApply('<generatedCounter>', KVariable('GENERATEDCOUNTER_CELL')),
+                ],
+            ),
+            [mlEqualsTrue(KApply('_==K_', [KToken('1', 'Int'), KToken('1', 'Int')]))],
+        ),
+        (
+            CTerm(
+                KApply(
+                    '<generatedTop>',
+                    [
+                        KApply('<k>', [KVariable('V_93d8c352')]),
+                        KApply('<generatedCounter>', KVariable('GENERATEDCOUNTER_CELL')),
+                    ],
+                ),
+                [],
+            ),
+            CSubst(Subst({'V_93d8c352': token(1), 'GENERATEDCOUNTER_CELL': KVariable('GENERATEDCOUNTER_CELL')}), []),
+            CSubst(Subst({'V_93d8c352': token(2), 'GENERATEDCOUNTER_CELL': KVariable('GENERATEDCOUNTER_CELL')}), []),
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    'test_id,cterm1,cterm2,expected',
+    ANTI_UNIFY_TEST_DATA,
+    ids=[test_id for test_id, *_ in ANTI_UNIFY_TEST_DATA],
+)
+def test_anti_unify(test_id: str, cterm1: CTerm, cterm2: CTerm, expected: tuple[CTerm, CSubst, CSubst] | None) -> None:
+    actual = cterm1.anti_unify(cterm2)
+    assert actual == expected