fix: make rollback manager proposal-only (#378)

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

docs/ARCHITECTURE.md

@@ -488,7 +488,7 @@ Notifications are dispatched through a shared `NotificationPublisher` that wraps
 | `drift_detector` | Detect strategy performance drift vs. backtest expectations |
 | `performance_monitor` | Real-time performance tracking (CAGR, Sharpe, drawdown) |
 | `ai_reviewer` | LLM-based strategy code review and performance commentary |
-| `rollback_manager` | Manage strategy version rollbacks with audit trail |
+| `rollback_manager` | Detect degradation and record no-order rollback proposals with an audit trail; platform-specific, owner-authorized rollback remains separate |
 | `shadow_validator` | Validate shadow-mode execution outcomes against live results |
 | `health_dashboard` | Aggregate health metrics across platforms |
 | `audit_log` | Immutable audit log for strategy changes |

src/quant_platform_kit/strategy_lifecycle/contracts.py

@@ -541,6 +541,7 @@ class UpdateStage(str, enum.Enum):
     DENIED = "denied"
     DEPLOYED = "deployed"
     RUNTIME_CONFIRMED = "runtime_confirmed"
+    ROLLBACK_PROPOSED = "rollback_proposed"
     ROLLED_BACK = "rolled_back"
 
 

src/quant_platform_kit/strategy_lifecycle/rollback_manager.py

@@ -1,9 +1,8 @@
-"""Rollback manager — monitors post-deployment performance and auto-rolls back on degradation."""
+"""Rollback monitor — records no-order rollback proposals on degradation."""
 
 from __future__ import annotations
 
 from collections.abc import Mapping
-from datetime import datetime, timezone
 from typing import Any
 
 import numpy as np
@@ -13,20 +12,20 @@
 from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore
 from quant_platform_kit.strategy_lifecycle.update_policy import UpdatePolicy
 
-
-def _now_iso() -> str:
-    return datetime.now(timezone.utc).isoformat()
-
-
 class RollbackManager:
-    """Monitors post-update performance and triggers rollback if needed.
+    """Monitors post-update performance and records rollback proposals.
+
+    This class has no deployment, runtime-target, broker, or order adapter.
+    A performance breach is therefore an auditable proposal, not an executed
+    rollback.  A platform-specific, owner-authorized control path must provide
+    any actual rollback separately.
 
     Usage::
 
         mgr = RollbackManager(store=store, policy=policy)
         decision = mgr.evaluate("global_etf_rotation", domain="us_equity")
         if decision["should_rollback"]:
-            mgr.rollback(...)
+            mgr.propose_rollback(...)
     """
 
     def __init__(
@@ -62,16 +61,25 @@ def evaluate(
             deployed_max_dd: Max drawdown at time of deployment.
 
         Returns:
-            Dict with "should_rollback", "reason", "live_sharpe", "live_max_dd".
+            Dict with a rollback recommendation and an explicit
+            ``rollback_execution_authorized=False`` boundary.
         """
         # Get latest live performance
         latest_snapshot = self._store.load_latest_snapshot(domain, strategy_profile)
         if latest_snapshot is None:
-            return {"should_rollback": False, "reason": "No live performance data available"}
+            return {
+                "should_rollback": False,
+                "reason": "No live performance data available",
+                "rollback_execution_authorized": False,
+            }
 
         ref_window = latest_snapshot.windows.get(126) or latest_snapshot.windows.get(252)
         if ref_window is None:
-            return {"should_rollback": False, "reason": "No window metrics available"}
+            return {
+                "should_rollback": False,
+                "reason": "No window metrics available",
+                "rollback_execution_authorized": False,
+            }
 
         live_sharpe = ref_window.sharpe_ratio
         live_max_dd = ref_window.max_drawdown
@@ -102,9 +110,10 @@ def evaluate(
             "reason": "; ".join(reasons) if reasons else "Performance within acceptable range",
             "live_sharpe": live_sharpe,
             "live_max_dd": live_max_dd,
+            "rollback_execution_authorized": False,
         }
 
-    def rollback(
+    def propose_rollback(
         self,
         strategy_profile: str,
         *,
@@ -113,27 +122,43 @@ def rollback(
         param_version_to: int,
         params_before: Mapping[str, Any],
         params_after: Mapping[str, Any],
-        reason: str = "Auto-rollback due to post-deployment performance degradation",
+        reason: str = "Rollback proposal due to post-deployment performance degradation",
     ) -> dict[str, Any]:
-        """Execute a rollback and record it in the audit log."""
+        """Record a rollback proposal without changing any external state."""
         entry = record_audit_entry(
             strategy_profile=strategy_profile,
             domain=domain,
-            stage=UpdateStage.ROLLED_BACK,
-            operator="auto_optimizer",
+            stage=UpdateStage.ROLLBACK_PROPOSED,
+            operator="rollback_monitor",
             param_version_from=param_version_from,
             param_version_to=param_version_to,
             params_before=params_before,
             params_after=params_after,
             reason=reason,
-            approval_source="auto",
+            approval_source="not_authorized",
+            store=self._store,
         )
 
         return {
-            "rolled_back": True,
+            "proposal_recorded": True,
+            "rolled_back": False,
+            "rollback_executed": False,
+            "execution_authorized": False,
+            "requires_owner_approval": True,
+            "stage": UpdateStage.ROLLBACK_PROPOSED.value,
             "strategy_profile": strategy_profile,
             "from_version": param_version_from,
             "to_version": param_version_to,
             "entry_id": entry.entry_id,
             "reason": reason,
         }
+
+    def rollback(self, *args: Any, **kwargs: Any) -> dict[str, Any]:
+        """Compatibility alias for :meth:`propose_rollback`.
+
+        Kept so callers do not fail at import time, but it never claims or
+        performs an external rollback.  Consumers must check
+        ``rollback_executed`` rather than treating an audit record as runtime
+        evidence.
+        """
+        return self.propose_rollback(*args, **kwargs)

tests/test_lifecycle_contracts.py

@@ -254,6 +254,7 @@ def test_update_stages(self) -> None:
         self.assertEqual(UpdateStage.PATCH_CREATED.value, "patch_created")
         self.assertEqual(UpdateStage.DEPLOYED.value, "deployed")
         self.assertEqual(UpdateStage.RUNTIME_CONFIRMED.value, "runtime_confirmed")
+        self.assertEqual(UpdateStage.ROLLBACK_PROPOSED.value, "rollback_proposed")
         self.assertEqual(UpdateStage.ROLLED_BACK.value, "rolled_back")
 
     def test_health_score(self) -> None:

tests/test_rollback_manager.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from quant_platform_kit.strategy_lifecycle.contracts import UpdateStage
+from quant_platform_kit.strategy_lifecycle.rollback_manager import RollbackManager
+
+
+def test_rollback_manager_records_only_an_owner_review_proposal() -> None:
+    store = object()
+    manager = RollbackManager(store=store, policy=object())
+
+    with patch(
+        "quant_platform_kit.strategy_lifecycle.rollback_manager.record_audit_entry",
+        return_value=SimpleNamespace(entry_id="proposal-123"),
+    ) as record_audit_entry:
+        result = manager.propose_rollback(
+            "soxl_soxx_trend_income",
+            domain="us_equity",
+            param_version_from=4,
+            param_version_to=3,
+            params_before={"risk_cap": 0.5},
+            params_after={"risk_cap": 0.3},
+        )
+
+    assert result["proposal_recorded"] is True
+    assert result["rolled_back"] is False
+    assert result["rollback_executed"] is False
+    assert result["execution_authorized"] is False
+    assert result["requires_owner_approval"] is True
+    assert result["stage"] == "rollback_proposed"
+    assert result["entry_id"] == "proposal-123"
+    assert record_audit_entry.call_args.kwargs["stage"] is UpdateStage.ROLLBACK_PROPOSED
+    assert record_audit_entry.call_args.kwargs["approval_source"] == "not_authorized"
+    assert record_audit_entry.call_args.kwargs["store"] is store
+
+
+def test_legacy_rollback_alias_preserves_the_no_execution_boundary() -> None:
+    manager = RollbackManager(store=object(), policy=object())
+
+    with patch.object(manager, "propose_rollback", return_value={"rollback_executed": False}) as propose:
+        result = manager.rollback("tqqq_growth_income", domain="us_equity")
+
+    assert result == {"rollback_executed": False}
+    propose.assert_called_once_with("tqqq_growth_income", domain="us_equity")