INNER CODE UNIT · Python

recover_interrupted

yashab-cyber/HackGpt · workbench/server.py:87

    def recover_interrupted(self):
        """Convert trustworthy stale running checkpoints into explicit interrupted reports."""
        recovered = 0
        with closing(sqlite3.connect(self.path, timeout=5)) as connection:
            rows = connection.execute(
                "SELECT id, content FROM active_runs ORDER BY started"
            ).fetchall()
            for run_id, raw in rows:
                try:
                    report = json.loads(raw)
                    if (
                        not isinstance(report, dict)
                        or report.get("id") != run_id
                        or report.get("status") != "running"
                    ):
                        raise ValueError("invalid active snapshot")
                    candidate = copy.deepcopy(report)
                    seal(candidate)

View source record →