fix(retention): refuse a prune that would delete most of a table (#1644)

docs/memory/requirements/lifecycle-observability.md

@@ -22,8 +22,13 @@
   Purge runs the #816 `purge_agent_ownership` → `cascade_delete`
   primitive so all per-agent child rows are removed in one transaction;
   `KEEP`-policy tables (`schedule_executions`, `nevermined_payment_log`)
-  survive per their own retention discipline. Bounded by the shared
-  5000-row/cycle cap so a backlog drains gradually.
+  survive per their own retention discipline. Each purge additionally
+  removes the agent's Docker data volumes (#1581) and is therefore
+  **unrecoverable** — so the #1644 blast-radius guard floors this sweep
+  at **0**: any purge at all requires an explicit admin acknowledgement
+  before it runs. (`RETENTION_CHUNK_SIZE_PER_CYCLE` bounds each
+  *transaction*, not the call — there is no per-cycle row cap; see #1644.)
+
 - **Name reservation**: `is_agent_name_reserved()` is intentionally
   unfiltered — it sees soft-deleted rows so a soft-deleted name cannot
   be reused (and silently clobbered) before purge.
@@ -43,6 +48,49 @@
   deleted_at IS NOT NULL`. Migration
   `agent_ownership_soft_delete`.
 
+### RETENTION-GUARD-001: Retention prunes are blast-radius guarded
+- **Implements**: Issue #1644 (follow-up to #1638)
+- **Description**: before any window-driven destructive prune,
+  `services/retention_guard.py` counts the candidate set (bounded, so the
+  cost is O(threshold) not O(candidates)) and **refuses** the prune if it
+  exceeds the threshold, logging at ERROR and raising an `operator_queue`
+  alarm naming the setting, the window, the window's source
+  (`db-row`/`code-default`), and the counts. The prune proceeds only after
+  an admin acknowledges it. Covers all **7** window-driven prunes.
+- **Why**: #1638 fixed one *mechanism* (a retroactive default change).
+  It left every other route to a destructive window open — an unvalidated
+  `PUT /api/settings/ops/config`, a future default regression, a direct DB
+  write. The guard does not care how the bad window arrived.
+- **Acknowledgement**: `POST /api/settings/retention/acknowledge`
+  (admin **and** human-only) is **the gate**; the operator-queue item is an
+  alarm and authorizes nothing. An ack is **bound to the window in force**
+  (409 on mismatch — approving a prune at 30 days does not approve one at
+  1 day) and **single-use** (consumed once the prune runs, so the guard
+  re-arms and one approval can never authorize an unboundedly larger
+  future delete at the same window).
+- **Threshold**: a **fixed constant** (`retention_guard.MAX_ROWS_PER_SWEEP`,
+  1000) — deliberately NOT an operator setting. It was briefly configurable via
+  Settings; that was wrong twice over: nobody can reason about the right value
+  (it depends on per-cycle churn they cannot see — the panel needed a caption
+  explaining that *bigger is worse*, and a control that must explain which way is
+  safe is the wrong control), and a mutable constant read at action time gating a
+  destructive operation is #1638 one level up — raising it would silently disarm
+  the guard fleet-wide. Deleting the knob deleted its clamp, its endpoint, its
+  blocklist entry, and a whole fail-closed branch. Chosen against **steady state,
+  not table size**: only rows crossing the cutoff within one 5-min cycle are
+  candidates, so four digits means something changed. Lowering is always safe;
+  raising is a code change with a reviewer, not a text box. Surfaced read-only at
+  `GET /api/settings/retention` → `guard.max_rows`. Per-sweep floors: rows →
+  the constant, schedules → 100, agents → **0**.
+- **Fail-closed**: any error — the count throws, the ack lookup throws —
+  **refuses** the prune. A guard that fails open is worse than no guard
+  because it manufactures confidence. (There is no 'threshold unreadable'
+  path: the threshold is a constant, so that failure mode does not exist.)
+- **Expected behaviour**: a legitimate first-enable of retention on a
+  mature install *will* trip the guard once, and that is intended — the
+  guard cannot distinguish a large legitimate backlog from a mistyped
+  window, so it asks once and the operator acknowledges once.
+
 ### 33.2 Schedule Soft-Delete (#834 — Phase 1b)
 - **Implements**: Issue #834 Phase 1b (PR #839)
 - **Description**: `DELETE /api/agents/{name}/schedules/{id}` marks

src/backend/canary/snapshot.py

@@ -119,6 +119,12 @@
 #
 # The list intentionally errs on the side of catching more orphans rather
 # than fewer; false positives surface as L-03 violations operators triage.
+# #1644: mirrors services.retention_guard.ALARM_AGENT_NAME. Duplicated as a
+# literal rather than imported: `canary/` is a deterministic read-only library
+# that must not pull a service (which imports `database`) into its import graph.
+# `tests/unit/test_1644_retention_guard.py` asserts the two stay equal.
+_RETENTION_GUARD_AGENT = "_retention-guard"
+
 ORPHAN_SCAN_TABLES = [
     ("agent_sharing", "agent_name", None),
     ("agent_schedules", "agent_name", None),
@@ -135,7 +141,18 @@
     ("agent_tags", "agent_name", None),
     ("agent_shared_files", "agent_name", None),
     ("agent_public_links", "agent_name", None),
-    ("operator_queue", "agent_name", "status = 'pending'"),
+    # #1644: exclude the retention guard's reserved sentinel. L-03 hunts GHOST
+    # AGENTS — rows referencing an agent_name that a delete should have cascaded
+    # away. `_retention-guard` is not an agent and never was: it is a platform
+    # alarm host, deliberately un-createable (leading '_' is stripped by
+    # sanitize_agent_name), so it can never appear in agent_ownership and would
+    # otherwise report a permanent, un-fixable orphan. This narrows the predicate
+    # to what the invariant actually means; it does not weaken it.
+    (
+        "operator_queue",
+        "agent_name",
+        f"status = 'pending' AND agent_name != '{_RETENTION_GUARD_AGENT}'",
+    ),
     ("access_requests", "agent_name", "status = 'pending'"),
     # #918 — CASCADE table holding agent-published report payloads (can be
     # sensitive); watch for orphans referencing a deleted agent.

src/backend/database.py

@@ -1452,6 +1452,51 @@ def prune_execution_rows(self, retention_days: int, chunk_size: int = 500) -> in
         """Delete terminal schedule_executions rows older than retention_days (#772)."""
         return self._schedule_ops.prune_execution_rows(retention_days, chunk_size)
 
+    # --- #1644 blast-radius counts (bounded; share each prune's predicate) ---
+
+    def count_execution_log_candidates(self, retention_days: int, limit: int) -> int:
+        """Bounded count of what prune_execution_logs would null (#1644)."""
+        return self._schedule_ops.count_execution_log_candidates(retention_days, limit)
+
+    def count_execution_row_candidates(self, retention_days: int, limit: int) -> int:
+        """Bounded count of what prune_execution_rows would delete (#1644)."""
+        return self._schedule_ops.count_execution_row_candidates(retention_days, limit)
+
+    def count_soft_deleted_schedules_past_retention(
+        self, retention_days: int, limit: int
+    ) -> int:
+        """Bounded count of schedules the #834 sweep would purge (#1644)."""
+        return self._schedule_ops.count_soft_deleted_schedules_past_retention(
+            retention_days, limit
+        )
+
+    def count_health_check_candidates(self, days: int, limit: int) -> int:
+        """Bounded count of what cleanup_old_health_records would delete (#1644)."""
+        return self._monitoring_ops.count_health_check_candidates(days, limit)
+
+    def count_agent_reports_candidates(self, retention_days: int, limit: int) -> int:
+        """Bounded count of what prune_agent_reports would delete (#1644)."""
+        return self._report_ops.count_agent_reports_candidates(retention_days, limit)
+
+    def count_operator_queue_terminal_candidates(
+        self, retention_days: int, responded_retention_days: int, limit: int
+    ) -> int:
+        """Bounded count of what prune_terminal_items would delete (#1644)."""
+        return self._operator_queue_ops.count_terminal_candidates(
+            retention_days, responded_retention_days, limit
+        )
+
+    def count_soft_deleted_agents_past_retention(
+        self, retention_days: int, limit: int
+    ) -> int:
+        """Bounded count of agents the #834 sweep would hard-purge (#1644).
+
+        Each candidate is one agent whose data volumes are destroyed (#1581).
+        """
+        return self._agent_ops.count_soft_deleted_agents_past_retention(
+            retention_days, limit
+        )
+
     def get_running_executions_with_agent_info(self):
         """Get all running executions with schedule timeout info (Issue #129)."""
         return self._schedule_ops.get_running_executions_with_agent_info()

src/backend/db/agents.py

@@ -15,7 +15,7 @@
 from datetime import datetime
 from typing import Optional, List, Dict
 
-from sqlalchemy import select, insert, update, delete, func, or_
+from sqlalchemy import select, insert, update, delete, func, and_, or_
 from sqlalchemy.exc import IntegrityError
 
 from .engine import get_engine
@@ -40,6 +40,21 @@
 SYSTEM_AGENT_NAME = "trinity-system"
 
 
+def _soft_deleted_agents_predicate(cutoff: str):
+    """WHERE clause for the #834 Phase 1a soft-deleted agent hard-purge.
+
+    #1644: shared between `find_soft_deleted_agents_past_retention` (what the
+    sweep purges) and `count_soft_deleted_agents_past_retention` (what the guard
+    counts) so the two can never diverge. This purge chains into
+    `remove_agent_volumes` (#1581) — the count MUST describe exactly the set that
+    is about to become unrecoverable.
+    """
+    return and_(
+        agent_ownership.c.deleted_at.is_not(None),
+        agent_ownership.c.deleted_at < cutoff,
+    )
+
+
 class AgentOperations(
     SharingMixin,
     ResourcesMixin,
@@ -409,15 +424,37 @@ def find_soft_deleted_agents_past_retention(
         cutoff = iso_cutoff(hours=retention_days * 24)
         stmt = (
             select(agent_ownership.c.agent_name)
-            .where(
-                agent_ownership.c.deleted_at.is_not(None),
-                agent_ownership.c.deleted_at < cutoff,
-            )
+            .where(_soft_deleted_agents_predicate(cutoff))
             .limit(limit)
         )
         with get_engine().connect() as conn:
             return [row["agent_name"] for row in conn.execute(stmt).mappings()]
 
+    def count_soft_deleted_agents_past_retention(
+        self, retention_days: int, limit: int
+    ) -> int:
+        """#1644: how many agents `_sweep_soft_deleted_agents` would hard-purge.
+
+        Every candidate here is one agent whose Docker data volumes are destroyed
+        (#1581) — this count is not "rows", it is "unrecoverable losses". The guard
+        floors this sweep at 0 for that reason: any purge at all is worth one ack.
+        """
+        if retention_days <= 0 or limit <= 0:
+            return 0
+        from utils.helpers import iso_cutoff
+
+        cutoff = iso_cutoff(hours=retention_days * 24)
+        inner = (
+            select(agent_ownership.c.agent_name)
+            .where(_soft_deleted_agents_predicate(cutoff))
+            .limit(limit)
+            .subquery()
+        )
+        with get_engine().connect() as conn:
+            return int(
+                conn.execute(select(func.count()).select_from(inner)).scalar() or 0
+            )
+
     def recover_agent_ownership(self, agent_name: str) -> bool:
         """Recover a soft-deleted agent by clearing `deleted_at` (#834).
 

src/backend/db/monitoring.py

@@ -22,6 +22,15 @@
 from utils.helpers import iso_cutoff, utc_now_iso
 
 
+def _health_check_prune_predicate(cutoff: str):
+    """WHERE clause for the #772 health-check retention sweep.
+
+    #1644: shared with `count_health_check_candidates` so the guard's count and
+    the prune's delete can never describe different row sets.
+    """
+    return agent_health_checks.c.checked_at < cutoff
+
+
 class MonitoringOperations:
     """Database operations for agent health monitoring."""
 
@@ -249,6 +258,25 @@ def calculate_avg_latency(
         latencies = [h["latency_ms"] for h in history if h.get("latency_ms") is not None]
         return sum(latencies) / len(latencies) if latencies else None
 
+    def count_health_check_candidates(self, days: int, limit: int) -> int:
+        """#1644: how many rows `cleanup_old_records(days)` would DELETE.
+
+        Bounded at `limit` — the guard only asks "more than N?", never "how many?".
+        """
+        if days <= 0 or limit <= 0:
+            return 0
+        cutoff = iso_cutoff(hours=days * 24)
+        inner = (
+            select(agent_health_checks.c.id)
+            .where(_health_check_prune_predicate(cutoff))
+            .limit(limit)
+            .subquery()
+        )
+        with get_engine().connect() as conn:
+            return int(
+                conn.execute(select(func.count()).select_from(inner)).scalar() or 0
+            )
+
     def cleanup_old_records(self, days: int = 7, chunk_size: int = 1000) -> int:
         """
         Delete health check records older than specified days.
@@ -277,7 +305,7 @@ def cleanup_old_records(self, days: int = 7, chunk_size: int = 1000) -> int:
                     row["id"]
                     for row in conn.execute(
                         select(agent_health_checks.c.id)
-                        .where(agent_health_checks.c.checked_at < cutoff)
+                        .where(_health_check_prune_predicate(cutoff))
                         .limit(chunk_size)
                     ).mappings()
                 ]

src/backend/db/operator_queue.py

@@ -23,6 +23,34 @@
 from utils.helpers import utc_now_iso, iso_cutoff
 
 
+def _operator_queue_prune_predicate(
+    retention_days: int, responded_retention_days: int
+):
+    """WHERE clause for the #1142 terminal operator_queue retention sweep.
+
+    #1644: extracted so `prune_terminal_items` and `count_terminal_candidates`
+    share one definition. This predicate is the reason sharing is mandatory rather
+    than merely tidy — it derives a second cutoff (`resp_days`) internally, so any
+    hand-mirrored copy drifts the moment either window is edited.
+
+    `pending` rows are never matched (and so never deleted) — see the caller.
+    """
+    terminal_cutoff = iso_cutoff(hours=retention_days * 24)
+    # `responded` never uses a shorter window than the terminal one.
+    resp_days = max(responded_retention_days, retention_days)
+    responded_cutoff = iso_cutoff(hours=resp_days * 24)
+    return or_(
+        and_(
+            operator_queue.c.status.in_(("acknowledged", "cancelled", "expired")),
+            operator_queue.c.created_at < terminal_cutoff,
+        ),
+        and_(
+            operator_queue.c.status == "responded",
+            operator_queue.c.created_at < responded_cutoff,
+        ),
+    )
+
+
 class OperatorQueueOperations:
     """Database operations for the operator queue."""
 
@@ -379,26 +407,11 @@ def prune_terminal_items(
         """
         if retention_days <= 0 or limit <= 0:
             return 0
-        terminal_cutoff = iso_cutoff(hours=retention_days * 24)
-        # `responded` never uses a shorter window than the terminal one.
-        resp_days = max(responded_retention_days, retention_days)
-        responded_cutoff = iso_cutoff(hours=resp_days * 24)
 
         id_stmt = (
             select(operator_queue.c.id)
             .where(
-                or_(
-                    and_(
-                        operator_queue.c.status.in_(
-                            ("acknowledged", "cancelled", "expired")
-                        ),
-                        operator_queue.c.created_at < terminal_cutoff,
-                    ),
-                    and_(
-                        operator_queue.c.status == "responded",
-                        operator_queue.c.created_at < responded_cutoff,
-                    ),
-                )
+                _operator_queue_prune_predicate(retention_days, responded_retention_days)
             )
             .limit(limit)
         )
@@ -411,6 +424,34 @@ def prune_terminal_items(
             )
             return result.rowcount
 
+    def count_terminal_candidates(
+        self,
+        retention_days: int,
+        responded_retention_days: int,
+        limit: int,
+    ) -> int:
+        """#1644: how many rows `prune_terminal_items` would DELETE.
+
+        Shares the prune's predicate — which matters more here than anywhere else,
+        because that predicate derives a *second* cutoff internally
+        (``resp_days = max(responded_retention_days, retention_days)``). A
+        hand-mirrored count would drift the first time either knob is edited.
+        """
+        if retention_days <= 0 or limit <= 0:
+            return 0
+        inner = (
+            select(operator_queue.c.id)
+            .where(
+                _operator_queue_prune_predicate(retention_days, responded_retention_days)
+            )
+            .limit(limit)
+            .subquery()
+        )
+        with get_engine().connect() as conn:
+            return int(
+                conn.execute(select(func.count()).select_from(inner)).scalar() or 0
+            )
+
     def mark_acknowledged(self, item_id: str) -> bool:
         """Mark an item as acknowledged by the agent."""
         now = utc_now_iso()