fix(scheduler): reject past one-time 'at' cron schedules

src/opensquilla/scheduler/ops.py

@@ -32,6 +32,21 @@
 )
 
 
+def _reject_past_at(cron_expr: str, now: datetime) -> None:
+    """Reject a one-time ``at`` timestamp that is already in the past.
+
+    A past ``at`` fires on the next scheduler tick and the one-shot job is then
+    deleted, so the payload runs immediately with no future occurrence. That is
+    almost never what a caller scheduling a one-time reminder intends, so refuse
+    it at creation time instead of silently running it.
+    """
+    at_dt = parse_iso_at(cron_expr)
+    if at_dt < now:
+        raise ValueError(
+            f"schedule.at is in the past: {cron_expr}; one-time schedules must be in the future"
+        )
+
+
 def _validate_structured_schedule(
     kind: ScheduleKind | str,
     value: str,
@@ -211,11 +226,7 @@ async def add(
         # fall back to ISOLATED instead of failing creation. Headless cron
         # callers (no session context) get an isolated run rather than a hard
         # error.
-        if (
-            session_target == SessionTarget.CURRENT
-            and not session_key
-            and not origin_session_key
-        ):
+        if session_target == SessionTarget.CURRENT and not session_key and not origin_session_key:
             session_target = SessionTarget.ISOLATED
 
         origin_session_key = normalize_origin_session_key(session_target, origin_session_key)
@@ -270,6 +281,7 @@ async def add(
         )
 
         if kind == ScheduleKind.AT:
+            _reject_past_at(cron_expr, now)
             job.delete_after_run = True
             job.next_run_at = datetime.fromisoformat(cron_expr)
         elif kind == ScheduleKind.EVERY and cron_expr.isdigit():
@@ -312,6 +324,7 @@ async def update(self, job_id: str, **patch) -> CronJob | None:
             job.schedule_kind = kind
             job.cron_expr = cron_expr
             if kind == ScheduleKind.AT:
+                _reject_past_at(cron_expr, now)
                 job.anchor_at = None
                 job.next_run_at = datetime.fromisoformat(cron_expr)
             elif kind == ScheduleKind.EVERY:

tests/test_scheduler/test_ops_strict_schedule.py

@@ -132,9 +132,7 @@ async def test_cron_creator_authority_survives_persistence_without_widening_owne
         creator_is_owner or expected_persisted_host_execute
     )
     assert bool(envelope.metadata.get("cron_trusted_owner")) is creator_is_owner
-    assert bool(envelope.metadata.get("cron_trusted_host")) is (
-        expected_persisted_host_execute
-    )
+    assert bool(envelope.metadata.get("cron_trusted_host")) is (expected_persisted_host_execute)
     assert bool(envelope.metadata.get(PRINCIPAL_HOST_EXECUTE_METADATA_KEY)) is (
         expected_persisted_host_execute
     )
@@ -326,6 +324,51 @@ async def test_ops_add_at_rejects_naive_iso(tmp_path: Path) -> None:
         await store.close()
 
 
+async def test_ops_add_at_rejects_past_timestamp(tmp_path: Path) -> None:
+    """A one-time ``at`` in the past would fire immediately then delete itself;
+    reject it at creation time instead (issue #1516)."""
+    store, ops = await _open_ops(tmp_path)
+    try:
+        past = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+        with pytest.raises(ValueError, match="in the past"):
+            await ops.add(
+                name="stale",
+                handler_key="agent_run",
+                payload=make_agent_turn_payload("ping"),
+                session_target=SessionTarget.ISOLATED,
+                schedule_kind=ScheduleKind.AT,
+                schedule_value=past,
+            )
+        # Nothing should have been persisted.
+        assert await store.list_active() == []
+    finally:
+        await store.close()
+
+
+async def test_ops_update_at_rejects_past_timestamp(tmp_path: Path) -> None:
+    """Repointing a job to a past one-time ``at`` is rejected as well."""
+    store, ops = await _open_ops(tmp_path)
+    try:
+        future = (datetime.now(UTC) + timedelta(hours=1)).isoformat()
+        job = await ops.add(
+            name="once",
+            handler_key="agent_run",
+            payload=make_agent_turn_payload("ping"),
+            session_target=SessionTarget.ISOLATED,
+            schedule_kind=ScheduleKind.AT,
+            schedule_value=future,
+        )
+        past = (datetime.now(UTC) - timedelta(hours=1)).isoformat()
+        with pytest.raises(ValueError, match="in the past"):
+            await ops.update(
+                job.id,
+                schedule_kind=ScheduleKind.AT,
+                schedule_value=past,
+            )
+    finally:
+        await store.close()
+
+
 async def test_ops_add_every_rejects_zero_seconds(tmp_path: Path) -> None:
     store, ops = await _open_ops(tmp_path)
     try: