diff --git a/desktop/electron/scripts/live-html-journey-business.mjs b/desktop/electron/scripts/live-html-journey-business.mjs
index 2def020967..90ecdab8d8 100644
--- a/desktop/electron/scripts/live-html-journey-business.mjs
+++ b/desktop/electron/scripts/live-html-journey-business.mjs
@@ -64,11 +64,22 @@ async function semanticControl(spec) {
   const element = candidates[0] || pool.find(item => name(item).includes(wanted))
   if (!element) throw new Error('SEMANTIC_CONTROL_MISSING')
   element.scrollIntoView({ block: 'center', inline: 'nearest', behavior: 'instant' })
-  let rect = element.getBoundingClientRect()
+  const nextFrameRect = () => new Promise((resolve, reject) => {
+    let frame
+    const unavailable = () => reject(new Error('SEMANTIC_CONTROL_UNAVAILABLE'))
+    const remaining = spec.deadline - Date.now()
+    if (remaining <= 0) { unavailable(); return }
+    const timer = setTimeout(() => { cancelAnimationFrame(frame); unavailable() }, remaining)
+    frame = requestAnimationFrame(() => {
+      clearTimeout(timer)
+      if (Date.now() >= spec.deadline) unavailable()
+      else resolve(element.getBoundingClientRect())
+    })
+  })
+  let rect = await nextFrameRect()
   let stable = false
   for (let attempt = 0; attempt < 8; attempt++) {
-    await new Promise(resolve => setTimeout(resolve, 16))
-    const next = element.getBoundingClientRect()
+    const next = await nextFrameRect()
     stable = ['x', 'y', 'width', 'height'].every(key => Math.abs(next[key] - rect[key]) <= 0.5)
     rect = next
     if (stable) break
@@ -101,12 +112,14 @@ export function createBusinessDriver(app, getPreviewId, capture) {
     const result = await app.evaluate(async ({ webContents }, request) => {
       const contents = webContents.fromId(request.id)
       if (!contents || contents.isDestroyed()) throw new Error('SELECTED_PREVIEW_DESTROYED')
-      let point
+      let point, lastSemanticError
       const until = Date.now() + 4000
       while (!point) {
-        try { point = await contents.executeJavaScript(`(${request.resolver})(${JSON.stringify({ ...request.spec, kind: request.kind })})`, true) }
+        try { point = await contents.executeJavaScript(`(${request.resolver})(${JSON.stringify({ ...request.spec, kind: request.kind, deadline: until })})`, true) }
         catch (error) {
-          if (!/SEMANTIC_CONTROL_(?:MISSING|UNAVAILABLE|COVERED)\b/.test(String(error)) || Date.now() >= until) throw error
+          if (!/SEMANTIC_CONTROL_(?:MISSING|UNAVAILABLE|COVERED)\b/.test(String(error))) throw error
+          if (Date.now() >= until) throw lastSemanticError || error
+          lastSemanticError = error
           await new Promise(resolve => setTimeout(resolve, 100))
         }
       }
diff --git a/desktop/electron/scripts/test-live-html-business.mjs b/desktop/electron/scripts/test-live-html-business.mjs
index 33c4f1d2d7..2eec9db2b1 100644
--- a/desktop/electron/scripts/test-live-html-business.mjs
+++ b/desktop/electron/scripts/test-live-html-business.mjs
@@ -1,6 +1,7 @@
 import assert from 'node:assert/strict'
 import { readFile } from 'node:fs/promises'
 import { after, before, test } from 'node:test'
+import { runInNewContext } from 'node:vm'
 import { chromium } from 'playwright'
 import { createBusinessDriver, verifyBusinessCase } from './live-html-journey-business.mjs'
 import { createSupplementalClient } from './live-html-supplemental-client.mjs'
@@ -49,6 +50,104 @@ test('moving modal close waits for actionability and receives a real pointer cli
   } finally { await f.close() }
 })
 
+function controlledFrameDriver({ renderFrames = true, covered = false } = {}) {
+  let now = 0, frame = 0, nextId = 0, pumping = false
+  const tasks = new Map(), inputs = [], deadlines = [], cancelledFrames = []
+  const pump = () => {
+    if (pumping || ![...tasks.values()].some(task => Number.isFinite(task.at))) return
+    pumping = true
+    setImmediate(() => {
+      pumping = false
+      const next = [...tasks.entries()].filter(([, task]) => Number.isFinite(task.at)).sort((a, b) => a[1].at - b[1].at)[0]
+      if (!next) return
+      const [id, task] = next
+      tasks.delete(id)
+      now = task.at
+      if (task.frame) frame++
+      task.callback(now)
+      pump()
+    })
+  }
+  const schedule = (callback, at, isFrame = false) => {
+    const id = ++nextId
+    tasks.set(id, { callback, at, frame: isFrame })
+    pump()
+    return id
+  }
+  const clock = {
+    Date: { now: () => now },
+    setTimeout(callback, milliseconds) { deadlines.push(now + milliseconds); return schedule(callback, now + milliseconds) },
+    clearTimeout(id) { tasks.delete(id) },
+    // At 30 fps, a 16 ms timer can run without a new rendering frame.
+    requestAnimationFrame(callback) { return schedule(callback, renderFrames ? now + 1000 / 30 : Infinity, true) },
+    cancelAnimationFrame(id) { cancelledFrames.push(id); tasks.delete(id) },
+  }
+  class Control {
+    tagName = 'BUTTON'
+    innerText = 'Close'
+    isConnected = true
+    getBoundingClientRect() {
+      const x = Math.max(100, 300 - frame * 100)
+      return { x, y: 100, width: 40, height: 20, left: x, right: x + 40, top: 100, bottom: 120 }
+    }
+    getAttribute() { return null }
+    hasAttribute() { return false }
+    scrollIntoView() {}
+    matches() { return false }
+    contains(other) { return other === this }
+  }
+  const button = new Control()
+  const renderer = {
+    ...clock, Element: Control,
+    document: { getElementById: () => null, querySelectorAll: () => [button], elementFromPoint: () => covered ? null : button },
+    getComputedStyle: () => ({ display: 'block', visibility: 'visible', opacity: '1' }),
+    innerWidth: 800, innerHeight: 700,
+  }
+  const contents = {
+    isDestroyed: () => false,
+    focus() {},
+    executeJavaScript: expression => runInNewContext(expression, renderer),
+    debugger: { isAttached: () => true, async sendCommand(command, args) { inputs.push({ command, ...args }) } },
+  }
+  const app = {
+    evaluate: (fn, request) => runInNewContext(`(${fn.toString()})(environment, request)`, {
+      ...clock, request, environment: { webContents: { fromId: () => contents } },
+    }),
+  }
+  return {
+    driver: createBusinessDriver(app, () => 1, async () => {}),
+    inputs, deadlines, cancelledFrames,
+    elapsed: () => now,
+    pending: () => tasks.size,
+  }
+}
+
+test('timer samples within one rendering frame do not make moving controls actionable', async () => {
+  const f = controlledFrameDriver()
+  const point = await f.driver.click('Close')
+  assert.equal(point.x, 120, 'click must use the position after movement stops between frames')
+  assert.equal(f.inputs.length, 3)
+  assert.ok(f.inputs.every(input => input.command === 'Input.dispatchMouseEvent' && input.x === 120))
+  assert.ok(f.deadlines.every(deadline => deadline === 4000), 'frame waits must share the action deadline')
+  assert.equal(f.pending(), 0, 'successful frame samples must clear their watchdogs')
+})
+
+test('a renderer that stops producing frames fails within the action budget without input', async () => {
+  const f = controlledFrameDriver({ renderFrames: false })
+  await assert.rejects(f.driver.click('Close'), /SEMANTIC_CONTROL_UNAVAILABLE/)
+  assert.equal(f.elapsed(), 4000)
+  assert.deepEqual(f.inputs, [])
+  assert.equal(f.cancelledFrames.length, 1)
+  assert.equal(f.pending(), 0, 'timed-out frame callbacks must be cancelled')
+})
+
+test('the action deadline preserves the last confirmed covered-control failure', async () => {
+  const f = controlledFrameDriver({ covered: true })
+  await assert.rejects(f.driver.click('Close'), /SEMANTIC_CONTROL_COVERED/)
+  assert.deepEqual(f.inputs, [])
+  assert.equal(f.pending(), 0)
+})
+
 for (const kind of ['disabled', 'covered']) {
   test(`${kind} control remains a failure without forced input`, async () => {
     const f = await fixture(`
diff --git a/src/opensquilla/gateway/rpc_cron.py b/src/opensquilla/gateway/rpc_cron.py
index c9b83057f0..9164e69a77 100644
--- a/src/opensquilla/gateway/rpc_cron.py
+++ b/src/opensquilla/gateway/rpc_cron.py
@@ -61,7 +61,6 @@
     DeliveryConfig,
     DeliveryMode,
     FailureDestination,
-    JobStatus,
     ReplyTargetSnapshot,
     ScheduleKind,
     SessionTarget,
@@ -771,21 +770,7 @@ async def _update_cron_job(
         patch["tz"] = tz_value if isinstance(tz_value, str) else ""
 
     if "enabled" in params:
-        # Resolve the enabled toggle but DO NOT early-return: fall through so
-        # sibling field updates in the same request (text/schedule/…) are
-        # applied too, and so a DISABLED/FAILED job (not just PAUSED) can be
-        # revived — those are exactly the states a user runs `--enabled` to fix.
-        job = await scheduler.get_job(job_id)
-        if params["enabled"]:
-            revivable = {
-                JobStatus.PAUSED.value,
-                JobStatus.DISABLED.value,
-                JobStatus.FAILED.value,
-            }
-            if job is not None and job.status.value in revivable:
-                await scheduler.resume_job(job_id)
-        else:
-            await scheduler.pause_job(job_id)
+        patch["enabled"] = bool(params["enabled"])
 
     current_job = await scheduler.get_job(job_id)
     if current_job is None:
diff --git a/src/opensquilla/scheduler/ops.py b/src/opensquilla/scheduler/ops.py
index 3a55b5da6f..be093aa7f4 100644
--- a/src/opensquilla/scheduler/ops.py
+++ b/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)
@@ -281,7 +292,13 @@ async def add(
             # CRON or EVERY with cron expression: scan forward
             job.next_run_at = _next_run(job, now)
 
-        return await self._store.create_or_get(job)
+        # A retry may arrive after the original execution time. Only validate a
+        # new row, after deduplication and with a fresh clock inside the lock.
+        validate_new = (
+            (lambda: _reject_past_at(cron_expr, self._now()))
+            if kind == ScheduleKind.AT else None
+        )
+        return await self._store.create_or_get(job, validate_new=validate_new)
 
     async def update(self, job_id: str, **patch) -> CronJob | None:
         """Apply a partial update to an existing job. Returns None if not found."""
@@ -302,7 +319,8 @@ async def update(self, job_id: str, **patch) -> CronJob | None:
         structured_kind = patch.pop("schedule_kind", None)
         structured_value = patch.pop("schedule_value", None)
         structured_tz = patch.pop("schedule_tz", None)
-        if structured_kind is not None and structured_value is not None:
+        schedule_updated = structured_kind is not None and structured_value is not None
+        if schedule_updated:
             kind, cron_expr = _validate_structured_schedule(structured_kind, structured_value)
             if structured_tz is not None:
                 raw_tz = (structured_tz or "").strip()
@@ -312,6 +330,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:
@@ -326,7 +345,7 @@ async def update(self, job_id: str, **patch) -> CronJob | None:
                 "pass schedule_kind + schedule_value instead"
             )
 
-        for field in ("name", "timeout_seconds", "enabled", "origin_session_key"):
+        for field in ("name", "timeout_seconds", "origin_session_key"):
             if field in patch:
                 setattr(job, field, patch.pop(field))
         if "tool_policy" in patch:
@@ -372,6 +391,19 @@ async def update(self, job_id: str, **patch) -> CronJob | None:
             job.origin_session_key,
         )
 
+        # Persist the enabled toggle with the validated schedule and payload.
+        # A rejected patch must never resume or pause the existing job.
+        if "enabled" in patch:
+            job.enabled = bool(patch.pop("enabled"))
+            if not job.enabled:
+                job.status = JobStatus.PAUSED
+            elif job.status in (JobStatus.PAUSED, JobStatus.DISABLED, JobStatus.FAILED):
+                job.status = JobStatus.PENDING
+                job.backoff_until = None
+                job.consecutive_errors = 0
+                if not schedule_updated and job.schedule_kind != ScheduleKind.AT:
+                    job.next_run_at = _next_run(job, now)
+
         job.updated_at = now
         await self._store.save(job)
         return job
diff --git a/src/opensquilla/scheduler/persistence.py b/src/opensquilla/scheduler/persistence.py
index 81bb4cdc36..ca714433ad 100644
--- a/src/opensquilla/scheduler/persistence.py
+++ b/src/opensquilla/scheduler/persistence.py
@@ -5,7 +5,7 @@
 import asyncio
 import json
 import uuid
-from collections.abc import AsyncIterator
+from collections.abc import AsyncIterator, Callable
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime
 
@@ -620,9 +620,13 @@ async def save(self, job: CronJob) -> None:
         await self._execute_save(job)
         await self._db().commit()
 
-    async def create_or_get(self, job: CronJob) -> CronJob:
-        """Atomically create an idempotent job or return the existing row."""
+    async def create_or_get(
+        self, job: CronJob, *, validate_new: Callable[[], None] | None = None,
+    ) -> CronJob:
+        """Return an existing row, or validate and create under the idempotency lock."""
         if not job.idempotency_key:
+            if validate_new is not None:
+                validate_new()
             await self.save(job)
             return job
 
@@ -631,6 +635,8 @@ async def create_or_get(self, job: CronJob) -> CronJob:
             if existing is not None:
                 existing.deduplicated = True
                 return existing
+            if validate_new is not None:
+                validate_new()
             try:
                 await self._execute_save(job)
                 await self._db().commit()
diff --git a/tests/test_gateway/test_rpc_cron_update_enabled.py b/tests/test_gateway/test_rpc_cron_update_enabled.py
index d5885f9650..fc2f6864d6 100644
--- a/tests/test_gateway/test_rpc_cron_update_enabled.py
+++ b/tests/test_gateway/test_rpc_cron_update_enabled.py
@@ -2,7 +2,11 @@
 
 from __future__ import annotations
 
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
+from unittest.mock import AsyncMock
+
+import pytest
 
 from opensquilla.gateway.rpc import RpcContext
 from opensquilla.gateway.rpc_cron import (
@@ -150,3 +154,101 @@ async def test_update_enabled_false_applies_sibling_fields(tmp_path: Path) -> No
         assert payload_text(after.payload, after.session_target) == "new prompt"
     finally:
         await store.close()
+
+
+@pytest.mark.parametrize(
+    ("status", "enabled"),
+    [
+        (JobStatus.PENDING, False),
+        (JobStatus.PAUSED, True),
+        (JobStatus.DISABLED, True),
+        (JobStatus.FAILED, True),
+    ],
+)
+@pytest.mark.parametrize(
+    ("patch", "error"),
+    [
+        ({"schedule": {"kind": "at", "at": "2035-01-01T19:59:59+08:00"}}, "in the past"),
+        ({"tz": "Invalid/Timezone"}, "Unknown timezone"),
+    ],
+    ids=["past-at", "invalid-timezone"],
+)
+async def test_update_invalid_patch_does_not_change_enabled_or_persist(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    status: JobStatus,
+    enabled: bool,
+    patch: dict,
+    error: str,
+) -> None:
+    now = datetime(2035, 1, 1, 12, tzinfo=UTC)
+    store = JobStore(str(tmp_path / "cron.db"))
+    await store.open()
+    engine = SchedulerEngine(store, clock=lambda: now)
+    try:
+        job = await engine.add_job(
+            name="original",
+            schedule_kind=ScheduleKind.AT,
+            schedule_value=(now + timedelta(hours=1)).isoformat(),
+            payload=make_agent_turn_payload("original reminder"),
+        )
+        job.status = status
+        job.enabled = not enabled
+        job.consecutive_errors = 2
+        job.backoff_until = now + timedelta(minutes=1)
+        await store.save(job)
+        before = await store.get(job.id)
+        save = AsyncMock(wraps=store.save)
+        monkeypatch.setattr(store, "save", save)
+
+        with pytest.raises(ValueError, match=error):
+            await _handle_cron_update(
+                {"id": job.id, "enabled": enabled, "name": "changed", **patch},
+                _ctx(engine),
+            )
+
+        assert await store.get(job.id) == before
+        save.assert_not_awaited()
+    finally:
+        await store.close()
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+async def test_update_enabled_and_schedule_persist_together_once(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, enabled: bool,
+) -> None:
+    now = datetime(2035, 1, 1, 12, tzinfo=UTC)
+    store = JobStore(str(tmp_path / "cron.db"))
+    await store.open()
+    engine = SchedulerEngine(store, clock=lambda: now)
+    try:
+        job = await engine.add_job(
+            name="original",
+            enabled=not enabled,
+            schedule_kind=ScheduleKind.AT,
+            schedule_value=(now + timedelta(hours=1)).isoformat(),
+            payload=make_agent_turn_payload("original reminder"),
+        )
+        save = AsyncMock(wraps=store.save)
+        monkeypatch.setattr(store, "save", save)
+        new_at = (now + timedelta(hours=2)).isoformat()
+
+        await _handle_cron_update(
+            {
+                "id": job.id,
+                "enabled": enabled,
+                "text": "updated reminder",
+                "schedule": {"kind": "at", "at": new_at},
+            },
+            _ctx(engine),
+        )
+
+        after = await store.get(job.id)
+        assert after is not None
+        assert after.status == (JobStatus.PENDING if enabled else JobStatus.PAUSED)
+        assert after.enabled is enabled
+        assert after.next_run_at == now + timedelta(hours=2)
+        assert payload_text(after.payload, after.session_target) == "updated reminder"
+        save.assert_awaited_once()
+    finally:
+        await store.close()
diff --git a/tests/test_scheduler/test_ops_strict_schedule.py b/tests/test_scheduler/test_ops_strict_schedule.py
index 4ac451e23e..2fb8504307 100644
--- a/tests/test_scheduler/test_ops_strict_schedule.py
+++ b/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,153 @@ 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,
+        )
+        before = await store.get(job.id)
+        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,
+            )
+        assert await store.get(job.id) == before
+    finally:
+        await store.close()
+
+
+async def test_ops_at_idempotent_retries_after_due_time_return_existing_job(tmp_path: Path) -> None:
+    store, _ = await _open_ops(tmp_path)
+    now = datetime(2035, 1, 1, 12, tzinfo=UTC)
+    ops = SchedulerOps(store, clock=lambda: now)
+    at = (now + timedelta(seconds=1)).isoformat()
+    try:
+        async def add_once(key: str = "same-request"):
+            return await ops.add(
+                name="once",
+                payload=make_agent_turn_payload("synthetic reminder"),
+                schedule_kind=ScheduleKind.AT,
+                schedule_value=at,
+                idempotency_key=key,
+            )
+
+        original = await add_once()
+        now += timedelta(seconds=2)
+        retries = await asyncio.gather(*(add_once() for _ in range(8)))
+
+        assert all(job.id == original.id and job.deduplicated for job in retries)
+        with pytest.raises(ValueError, match="in the past"):
+            await add_once("different-request")
+        assert len(await store.list_active()) == 1
+    finally:
+        await store.close()
+
+
+async def test_ops_at_checks_time_after_waiting_for_creation_lock(tmp_path: Path) -> None:
+    store, _ = await _open_ops(tmp_path)
+    now = datetime(2035, 1, 1, 12, tzinfo=UTC)
+    started = asyncio.Event()
+
+    def clock() -> datetime:
+        started.set()
+        return now
+
+    ops = SchedulerOps(store, clock=clock)
+    try:
+        async with store._idempotent_create_lock:
+            pending = asyncio.create_task(ops.add(
+                name="once",
+                payload=make_agent_turn_payload("synthetic reminder"),
+                schedule_kind=ScheduleKind.AT,
+                schedule_value=(now + timedelta(seconds=1)).isoformat(),
+                idempotency_key="new-request",
+            ))
+            await started.wait()
+            now += timedelta(seconds=2)
+        with pytest.raises(ValueError, match="in the past"):
+            await pending
+        assert await store.list_active() == []
+    finally:
+        await store.close()
+
+
+@pytest.mark.parametrize(
+    "at", ["2035-01-01T12:00:00Z", "2035-01-01T20:00:00+08:00"],
+)
+async def test_ops_at_accepts_the_current_instant_with_either_offset(
+    tmp_path: Path, at: str,
+) -> None:
+    store, _ = await _open_ops(tmp_path)
+    now = datetime(2035, 1, 1, 12, tzinfo=UTC)
+    ops = SchedulerOps(store, clock=lambda: now)
+    try:
+        job = await ops.add(
+            name="once",
+            payload=make_agent_turn_payload("synthetic reminder"),
+            schedule_kind=ScheduleKind.AT,
+            schedule_value=at,
+        )
+        assert job.next_run_at == now
+    finally:
+        await store.close()
+
+
+async def test_ops_overdue_at_metadata_update_preserves_schedule(tmp_path: Path) -> None:
+    store, _ = await _open_ops(tmp_path)
+    now = datetime(2035, 1, 1, 12, tzinfo=UTC)
+    ops = SchedulerOps(store, clock=lambda: now)
+    try:
+        job = await ops.add(
+            name="once",
+            payload=make_agent_turn_payload("synthetic reminder"),
+            schedule_kind=ScheduleKind.AT,
+            schedule_value="2035-01-01T21:00:00+08:00",
+        )
+        assert job.next_run_at == now + timedelta(hours=1)
+        now += timedelta(hours=2)
+
+        changed = await ops.update(job.id, name="renamed", enabled=False)
+
+        assert changed is not None
+        assert changed.name == "renamed"
+        assert changed.next_run_at == job.next_run_at
+        assert changed.enabled is False
+    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: