Status bar support in proof loop (#4561)

Adds a couple things needed for supporting a proof status bar for
real-time updates in kevm/kontrol: Adds a callback parameter to the two
proof loops to get information about the proofs during execution of the
proof loop (without having to read from disk). Also adds a field to
`APRProof` which is the one-line summary which will be printed in the
status bar, along with a default implementation in `Proof`.

pyk/src/pyk/proof/proof.py

@@ -279,6 +279,10 @@ def lines(self) -> list[str]:
         subproofs_summaries = [subproof.summary for subproof in self.subproofs]
         return CompositeSummary([BaseSummary(self.id, self.status), *subproofs_summaries])
 
+    @property
+    def one_line_summary(self) -> str:
+        return self.status.name
+
     @abstractmethod
     def get_steps(self) -> Iterable[PS]:
         """Return all currently available steps associated with this Proof. Should not modify `self`."""
@@ -321,6 +325,7 @@ def parallel_advance_proof(
     max_iterations: int | None = None,
     fail_fast: bool = False,
     max_workers: int = 1,
+    callback: Callable[[P], None] = (lambda x: None),
 ) -> None:
     """Advance proof with multithreaded strategy.
 
@@ -342,6 +347,7 @@ def parallel_advance_proof(
         fail_fast: If the proof is failing after finishing a step,
           halt execution even if there are still available steps.
         max_workers: Maximum number of worker threads the pool can spawn.
+        callback: Callable to run in between each completed step, useful for getting real-time information about the proof.
     """
     pending: set[Future[Any]] = set()
     explored: set[PS] = set()
@@ -371,6 +377,7 @@ def submit_steps(_steps: Iterable[PS]) -> None:
                 for result in proof_results:
                     proof.commit(result)
                 proof.write_proof_data()
+                callback(proof)
                 iterations += 1
                 if max_iterations is not None and max_iterations <= iterations:
                     break
@@ -489,7 +496,13 @@ def init_proof(self, proof: P) -> None:
         """
         ...
 
-    def advance_proof(self, proof: P, max_iterations: int | None = None, fail_fast: bool = False) -> None:
+    def advance_proof(
+        self,
+        proof: P,
+        max_iterations: int | None = None,
+        fail_fast: bool = False,
+        callback: Callable[[P], None] = (lambda x: None),
+    ) -> None:
         """Advance a proof.
 
         Performs loop `Proof.get_steps()` -> `Prover.step_proof()` -> `Proof.commit()`.
@@ -499,6 +512,7 @@ def advance_proof(self, proof: P, max_iterations: int | None = None, fail_fast:
             max_iterations (optional): Maximum number of steps to take.
             fail_fast: If the proof is failing after finishing a step,
               halt execution even if there are still available steps.
+            callback: Callable to run in between each completed step, useful for getting real-time information about the proof.
         """
         iterations = 0
         _LOGGER.info(f'Initializing proof: {proof.id}')
@@ -520,5 +534,6 @@ def advance_proof(self, proof: P, max_iterations: int | None = None, fail_fast:
                 for result in results:
                     proof.commit(result)
                 proof.write_proof_data()
+                callback(proof)
         if proof.failed:
             proof.failure_info = self.failure_info(proof)

pyk/src/pyk/proof/reachability.py

@@ -502,6 +502,20 @@ def summary(self) -> CompositeSummary:
             ]
         )
 
+    @property
+    def one_line_summary(self) -> str:
+        nodes = len(self.kcfg.nodes)
+        pending = len(self.pending)
+        failing = len(self.failing)
+        branches = len(self.kcfg.ndbranches()) + len(self.kcfg.splits())
+        vacuous = len(self.kcfg.vacuous)
+        stuck = len(self.kcfg.stuck)
+        passed = len([cover for cover in self.kcfg.covers() if cover.target.id == self.target])
+        return (
+            super().one_line_summary
+            + f'/{nodes} nodes/{pending} pending/{passed} passed/{failing} failing/{branches} branches/{vacuous} vacuous/{stuck} stuck'
+        )
+
     def get_refutation_id(self, node_id: int) -> str:
         return f'{self.id}.node-infeasible-{node_id}'