Fix manual cron session rotation

nerve/cron/service.py

@@ -240,6 +240,8 @@ def _is_overdue(job: CronJob, last_run: datetime, now: datetime) -> bool:
     async def _maybe_rotate_context(
         self, session_id: str, rotate_hours: int,
         rotate_at: str = "",
+        *,
+        force: bool = False,
     ) -> bool:
         """Check if a persistent cron session's context should be rotated.
 
@@ -252,28 +254,33 @@ async def _maybe_rotate_context(
         Returns True if rotation was performed.
         """
         session = await self.db.get_session(session_id)
-        if not session or not session.get("connected_at"):
-            return False
-
-        connected_at_str = session["connected_at"]
-        try:
-            ts = connected_at_str
-            if "T" not in ts:
-                ts = ts.replace(" ", "T")
-            if not ts.endswith(("Z", "+00:00")):
-                ts += "+00:00"
-            connected_at = datetime.fromisoformat(ts.replace("Z", "+00:00"))
-        except (ValueError, TypeError):
-            logger.warning(
-                "Invalid connected_at for %s: %s", session_id, connected_at_str,
-            )
+        if not session:
             return False
 
         now = datetime.now(timezone.utc)
-        should_rotate = False
-        reason = ""
+        should_rotate = force
+        reason = "manual" if force else ""
 
-        if rotate_at:
+        connected_at = None
+        connected_at_str = session.get("connected_at")
+        if connected_at_str:
+            try:
+                ts = connected_at_str
+                if "T" not in ts:
+                    ts = ts.replace(" ", "T")
+                if not ts.endswith(("Z", "+00:00")):
+                    ts += "+00:00"
+                connected_at = datetime.fromisoformat(ts.replace("Z", "+00:00"))
+            except (ValueError, TypeError):
+                logger.warning(
+                    "Invalid connected_at for %s: %s", session_id, connected_at_str,
+                )
+                if not force:
+                    return False
+        elif not force:
+            return False
+
+        if not should_rotate and rotate_at:
             # Time-of-day rotation: rotate if session started before today's
             # rotate_at and current time is past it.
             try:
@@ -291,7 +298,7 @@ async def _maybe_rotate_context(
             if now >= today_rotate_utc and connected_at < today_rotate_utc:
                 should_rotate = True
                 reason = f"rotate_at={rotate_at}"
-        elif rotate_hours > 0:
+        elif not should_rotate and rotate_hours > 0:
             age_hours = (now - connected_at).total_seconds() / 3600
             if age_hours >= rotate_hours:
                 should_rotate = True
@@ -631,8 +638,9 @@ async def rotate_session(self, job_id: str) -> dict:
             except (ValueError, TypeError):
                 pass
 
-        # Force rotation (rotate_hours=0 ensures any positive age passes)
-        rotated = await self._maybe_rotate_context(session_id, rotate_hours=0)
+        rotated = await self._maybe_rotate_context(
+            session_id, rotate_hours=0, force=True,
+        )
 
         logger.info(
             "Manual rotation for %s: rotated=%s age=%.1fh",

tests/test_cron.py

@@ -479,6 +479,30 @@ async def test_no_rotation_no_memorize(self, cron_service):
         assert rotated is False
         cron_service.engine.schedule_memorize.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    async def test_manual_rotation_forces_disabled_rotation_window(self, cron_service):
+        """Manual rotation clears context even when scheduled rotation is disabled."""
+        cron_service._jobs = [
+            _make_job(
+                id="pers", session_mode="persistent", context_rotate_hours=0,
+            ),
+        ]
+        cron_service.db.get_session = AsyncMock(return_value={
+            "connected_at": _hours_ago(1),
+            "sdk_session_id": "sdk-123",
+        })
+
+        result = await cron_service.rotate_session("pers")
+
+        assert result["rotated"] is True
+        assert result["session_age_hours"] is not None
+        cron_service.engine.schedule_memorize.assert_awaited_once_with(
+            "cron:pers",
+        )
+        cron_service.engine.sessions.mark_idle.assert_awaited_once_with(
+            "cron:pers", preserve_sdk_id=False,
+        )
+
 
 # ---------------------------------------------------------------------------
 # Run gates — service-level skip/run behaviour