Upload sync_hf.py

scripts/sync_hf.py

@@ -2156,6 +2156,251 @@ class OpenClawFullSync:
         finally:
             conn.close()
 
+    # ── Durable session pruning (default: archived sessions are deleted) ──
+    #
+    # Deleting a session from the Control UI never sticks: the next boot's
+    # dataset restore copies the transcript JSONL back into the container and
+    # the .session-guard row-count comparison re-injects the deleted sqlite
+    # rows. The manifest records which sessions were requested so every boot
+    # re-applies the deletion AFTER the restore and BEFORE the guard copy.
+    # Defaults (no configuration needed):
+    #   (unset)                     → delete every archived session
+    #   PRUNE_SESSIONS=off          → never delete anything
+    #   PRUNE_SESSIONS=<key,id,...> → delete specific session keys / window ids
+
+    def _prune_manifest_path(self) -> Path:
+        return OPENCLAW_HOME / "agents" / ".session-prune-manifest.json"
+
+    def _load_prune_manifest(self):
+        try:
+            data = json.loads(
+                self._prune_manifest_path().read_text(encoding="utf-8"))
+        except Exception:
+            return []
+        entries = data.get("sessions") if isinstance(data, dict) else None
+        return [e for e in (entries or []) if isinstance(e, dict)]
+
+    def _save_prune_manifest(self, entries):
+        p = self._prune_manifest_path()
+        try:
+            p.parent.mkdir(parents=True, exist_ok=True)
+            p.write_text(json.dumps({"sessions": entries}, indent=1),
+                         encoding="utf-8")
+        except OSError as e:
+            print(f"[PRUNE] cannot write manifest: {str(e)[:140]}")
+
+    @staticmethod
+    def _looks_like_session_id(value: str) -> bool:
+        # A bare window id is a UUID; session keys may *contain* a UUID
+        # ('agent:main:recovered:<uuid>') but also have ':' — those are keys.
+        v = value.lower()
+        return (len(v) == 36 and v.count("-") == 4
+                and all(c in "0123456789abcdef-" for c in v))
+
+    def _archived_session_targets(self):
+        """[(session_key, session_id|None)] for nodes with archived_at set."""
+        out, seen = [], set()
+        for fp in self._agent_session_dbs():
+            if not fp.is_file():
+                continue
+            conn = sqlite3.connect(str(fp), timeout=10)
+            try:
+                conn.execute("PRAGMA busy_timeout=10000")
+                cur = conn.cursor()
+                try:
+                    rows = cur.execute(
+                        "SELECT w.session_key, w.session_id "
+                        "FROM session_windows w "
+                        "JOIN session_nodes n ON n.session_key = w.session_key "
+                        "WHERE n.archived_at IS NOT NULL").fetchall()
+                except sqlite3.DatabaseError:
+                    rows = []
+                for key, sid in rows:
+                    if (key, sid) not in seen:
+                        seen.add((key, sid))
+                        out.append((key, sid))
+                try:
+                    keys = [r[0] for r in cur.execute(
+                        "SELECT session_key FROM session_nodes "
+                        "WHERE archived_at IS NOT NULL")]
+                except sqlite3.DatabaseError:
+                    keys = []
+                for key in keys:
+                    if not any(k == key for k, _ in out):
+                        out.append((key, None))
+                        seen.add((key, None))
+            except sqlite3.Error:
+                pass
+            finally:
+                conn.close()
+        return out
+
+    def _prune_targets(self):
+        """(session_keys, window_ids) from manifest + PRUNE_SESSIONS env.
+
+        Unset → 'archived' (default-on). 'off' → nothing, ever.
+        """
+        spec = (os.environ.get("PRUNE_SESSIONS") or "").strip()
+        if spec.lower() in ("off", "0", "no", "false"):
+            return set(), set()
+        entries = self._load_prune_manifest()
+        seen = {(e.get("key") or "", e.get("id") or "") for e in entries}
+        if not spec:
+            spec = "archived"
+        added = []
+        for raw in spec.split(","):
+            w = raw.strip()
+            if not w:
+                continue
+            if w.lower() in ("archived", "all-archived"):
+                for key, sid in self._archived_session_targets():
+                    item = (key or "", sid or "")
+                    if item in seen or item == ("", ""):
+                        continue
+                    seen.add(item)
+                    added.append({"key": key, "id": sid,
+                                  "at": int(time.time() * 1000)})
+            elif self._looks_like_session_id(w):
+                if ("", w) not in seen:
+                    seen.add(("", w))
+                    added.append({"id": w, "at": int(time.time() * 1000)})
+            elif (w, "") not in seen:
+                seen.add((w, ""))
+                added.append({"key": w, "at": int(time.time() * 1000)})
+        if added:
+            entries.extend(added)
+            self._save_prune_manifest(entries)
+            print(f"[PRUNE] manifest seeded with {len(added)} "
+                  f"session target(s)")
+        keys = {e["key"] for e in entries if e.get("key")}
+        ids = {e["id"] for e in entries if e.get("id")}
+        return keys, ids
+
+    def _prune_db(self, fp: Path, keys, ids):
+        """Delete rows for the target sessions; returns (rows, window_ids)."""
+        if not fp.is_file():
+            return 0, set()
+        conn = sqlite3.connect(str(fp), timeout=10)
+        try:
+            conn.execute("PRAGMA busy_timeout=10000")
+            try:
+                conn.execute("PRAGMA foreign_keys=ON")
+            except sqlite3.DatabaseError:
+                pass
+            cur = conn.cursor()
+            try:
+                tables = [r[0] for r in cur.execute(
+                    "SELECT name FROM sqlite_master WHERE type='table' "
+                    "AND name NOT LIKE 'sqlite_%'")]
+            except sqlite3.DatabaseError:
+                return 0, set()
+            win_ids = set(ids)
+            if keys:
+                ph = ",".join("?" * len(keys))
+                try:
+                    for (sid,) in cur.execute(
+                            f"SELECT session_id FROM session_windows "
+                            f"WHERE session_key IN ({ph})", list(keys)):
+                        win_ids.add(sid)
+                except sqlite3.DatabaseError:
+                    pass
+            total = 0
+            for t in tables:
+                try:
+                    cols = [c[1] for c in
+                            cur.execute(f'PRAGMA table_info("{t}")')]
+                except sqlite3.DatabaseError:
+                    continue
+                if win_ids and "session_id" in cols:
+                    ph = ",".join("?" * len(win_ids))
+                    try:
+                        cur.execute(
+                            f'DELETE FROM "{t}" WHERE session_id IN ({ph})',
+                            sorted(win_ids))
+                        total += max(cur.rowcount, 0)
+                    except sqlite3.DatabaseError as e:
+                        print(f"[PRUNE] {fp.name}:{t} session_id delete "
+                              f"failed: {str(e)[:80]}")
+                # session_nodes has no session_id column but points at the
+                # window it owns — prune the node when only the id was given.
+                if win_ids and t == "session_nodes" and \
+                        "current_session_id" in cols:
+                    ph = ",".join("?" * len(win_ids))
+                    try:
+                        cur.execute(
+                            f'DELETE FROM session_nodes '
+                            f'WHERE current_session_id IN ({ph})',
+                            sorted(win_ids))
+                        total += max(cur.rowcount, 0)
+                    except sqlite3.DatabaseError as e:
+                        print(f"[PRUNE] {fp.name}:session_nodes id delete "
+                              f"failed: {str(e)[:80]}")
+                if keys and "session_key" in cols:
+                    ph = ",".join("?" * len(keys))
+                    try:
+                        cur.execute(
+                            f'DELETE FROM "{t}" WHERE session_key IN ({ph})',
+                            sorted(keys))
+                        total += max(cur.rowcount, 0)
+                    except sqlite3.DatabaseError as e:
+                        print(f"[PRUNE] {fp.name}:{t} session_key delete "
+                              f"failed: {str(e)[:80]}")
+            conn.commit()
+            try:
+                mode = cur.execute("PRAGMA journal_mode").fetchone()[0]
+                if str(mode).lower() == "wal":
+                    cur.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+            except sqlite3.DatabaseError:
+                pass
+            return total, win_ids
+        finally:
+            conn.close()
+
+    def _prune_session_files(self, ids) -> int:
+        """Remove transcript JSONL/trajectory files (and their import
+        archives) for the pruned windows — otherwise the next boot's restore
+        hands them straight back to the gateway, which re-imports the session.
+        """
+        if not ids or not (OPENCLAW_HOME / "agents").is_dir():
+            return 0
+        removed = 0
+        for dirpath, _dirs, files in os.walk(OPENCLAW_HOME / "agents"):
+            if os.path.basename(dirpath) not in (
+                    "sessions", "session-sqlite-import-archive"):
+                continue
+            for fn in files:
+                if ".sqlite" in fn or not any(i in fn for i in ids):
+                    continue
+                try:
+                    os.remove(os.path.join(dirpath, fn))
+                    removed += 1
+                except OSError:
+                    pass
+        return removed
+
+    def _prune_sessions(self):
+        keys, ids = self._prune_targets()
+        if not keys and not ids:
+            return 0
+        targets = list(self._agent_session_dbs())
+        targets += self._agent_session_dbs_from(OPENCLAW_HOME / ".session-guard")
+        state_db = OPENCLAW_HOME / "state" / "openclaw.sqlite"
+        if state_db.is_file():
+            targets.append(state_db)
+        total, found = 0, set()
+        for fp in targets:
+            try:
+                rows, fids = self._prune_db(fp, keys, ids)
+                total += rows
+                found |= fids
+            except Exception as e:
+                print(f"[PRUNE] {fp} failed: {str(e)[:140]}")
+        files = self._prune_session_files(found)
+        if total or files:
+            print(f"[PRUNE] deleted {total} row(s), {files} transcript file(s) "
+                  f"({len(keys)} session key(s), {len(found)} window id(s))")
+        return total
+
     def run_session_migration(self):
         """Migrate legacy JSON session stores → SQLite.
 
@@ -2760,7 +3005,15 @@ def main():
         #     the duplicate when the merged live DB has fewer rows.
         sync._dedupe_canonical_session_rows()
 
-        # 1b. Guard chat-history DBs BEFORE any step that can park/recreate
+        # 1b. Durable session deletions — archived sessions by default
+        #     (PRUNE_SESSIONS=off disables). Must run AFTER the restore
+        #     (which re-creates the transcript JSONL) and BEFORE the guard
+        #     copy, or the guard re-injects the deleted rows next boot.
+        pruned = sync._prune_sessions()
+        if pruned:
+            sync._session_db_report("after-prune")
+
+        # 1c. Guard chat-history DBs BEFORE any step that can park/recreate
         #     them (purge, doctor --fix, 'agents add', startup-gate healing).
         sync._guard_agent_session_dbs()