feat: Subsumption

Author: lenianiva

pantograph/expr.py

@@ -158,6 +158,14 @@ def serial(self) -> dict:
             result["autoResume"] = self.auto_resume
         return result
 
+class Subsumption(Enum):
+    """
+    Subsumption result
+    """
+    NONE = 1
+    SUBSUMED = 2
+    CYCLE = 3
+
 @dataclass(frozen=True)
 class TacticHave:
     """

pantograph/server.py

@@ -17,6 +17,7 @@
     Goal,
     GoalState,
     Site,
+    Subsumption,
     Tactic,
     TacticHave,
     TacticLet,
@@ -342,6 +343,50 @@ async def goal_resume_async(self, state: GoalState, goals: list[Goal]) -> GoalSt
         return GoalState.parse(result, [], self.to_remove_goal_states)
     goal_resume = to_sync(goal_resume_async)
 
+    async def goal_subsume_async(
+            self,
+            state: GoalState,
+            goal: Goal,
+            candidates: list[Goal],
+            src_state: Optional[GoalState]=None
+    ) -> (Subsumption, Optional[GoalState], Optional[Goal]):
+        """
+        Detect subsumption by candidate goals
+
+        The candidate goals must all exist in `src_state`. If `src_state` is not
+        provided, they must exist in `state`. Returns a new goal state if a
+        subsumption does not lead to a cycle, and the subsumptor if there is any
+        subsumption happening.
+        """
+        args = {
+            "stateId": state.state_id,
+            "goal": goal.id,
+            "candidates": [g.id for g in candidates],
+        }
+        if src_state:
+            args["srcStateId"] = src_state.state_id
+        result = await self.run_async('goal.subsume', args)
+        if "error" in result:
+            raise ServerError(result)
+        nextState = None
+        if state_id := result.get("stateId"):
+            nextState = GoalState(
+                state_id=state_id,
+                goals=[g for g in state.goals if g.id != goal.id],
+                messages=[],
+                _sentinel=self.to_remove_goal_states,
+            )
+        sub = Subsumption[result["result"].upper()]
+        subsumptor = None
+        if subsumptor := result.get("subsumptor"):
+            gen = (g for g in candidates if g.id == subsumptor)
+            subsumptor = next(gen)
+            if subsumptor is None:
+                raise ServerError("Subsumptor should not be none")
+        return (sub, nextState, subsumptor)
+
+    goal_subsume = to_sync(goal_subsume_async)
+
     async def env_add_async(
             self, name: str, levels: list[str],
             t: Expr, v: Expr, is_theorem: bool = True):

pantograph/test_server.py

@@ -198,6 +198,22 @@ def test_dependent_mvars(self):
         self.assertEqual(state.goals[0].sibling_dep, {1})
         self.assertEqual(state.goals[1].sibling_dep, set())
 
+    def test_subsume(self):
+        server = Server()
+        state0 = server.goal_start("forall (p : Prop), p -> p")
+        state1 = server.goal_tactic(state0, "intro p")
+        state2 = server.goal_tactic(state1, "intro h")
+        state3 = server.goal_tactic(state2, "revert h")
+        src = state1.goals[0]
+        (sub, state, subsumptor) = server.goal_subsume(
+            state3,
+            state3.goals[0],
+            [state1.goals[0], state2.goals[0]],
+        )
+        self.assertEqual(sub, Subsumption.CYCLE)
+        self.assertEqual(state, None)
+        self.assertEqual(subsumptor, src)
+
     def test_env_add_inspect(self):
         server = Server()
         server.env_add(