fix(schedule): harden calendar transition handling

2026-08-08 05:45:49 +08:00
parent 925c0558a0
commit 7f5b301d5f
9 changed files with 133 additions and 9 deletions
@@ -66,7 +66,7 @@ The persistence coordinator supplies that acknowledgement only after its write p

 ### Live delivery lifecycle

-The Agent-scoped owner derives its active targets and latest recurring batch from the durable fold. Long targets use bounded timer segments, and every wake reads the wall clock again, so a rollback cannot fire early and a forward jump becomes overdue. A fixed-rate record treats its current `scheduledAt` as the earliest unaccepted point on the original sequence; integer division selects the latest due point directly. A Cron record treats its persisted target as a history-stable baseline, searches only for newer current matches, and persists the chosen occurrence and next target. Neither rule replays a missed backlog or shifts its authority to delivery time. Once one recurring record is overdue behind a closed gate, the owner arms that gate or an earlier one-shot instead of waking at intervening recurring targets. If a turn or another maintenance task already owns the Agent, `runMaintenance()` rejects the claim; the record stays active and one `whenIdle()` wait triggers a later retry. A rejected persistence preflight or contained framing/synchronous-enqueue failure also leaves the record active, but no private retry timer runs; later Agent activity reaching idle or a successful Schedule management preflight asks the owner to try again.
+The Agent-scoped owner derives its active targets and latest recurring batch from the durable fold. Long targets use bounded timer segments, and every wake reads the wall clock again, so a rollback cannot fire early and a forward jump becomes overdue. A fixed-rate record treats its current `scheduledAt` as the earliest unaccepted point on the original sequence; integer division selects the latest due point directly. A Cron record treats its persisted target as a history-stable baseline, searches only for newer current matches, and persists the chosen occurrence and next target. Neither rule replays a missed backlog or shifts its authority to delivery time. Once one recurring record is overdue behind a closed gate, the owner arms that gate or an earlier one-shot instead of waking at intervening recurring targets. If a turn or another maintenance task already owns the Agent, `runMaintenance()` rejects the claim; the record stays active and one `whenIdle()` wait triggers a later retry. A rejected persistence preflight, contained current-calendar resolution failure, or contained framing/synchronous-enqueue failure also leaves the record active, but no private retry timer runs; later Agent activity reaching idle or a successful Schedule management preflight asks the owner to try again.

 The accepted path first clears pending persistence and claims the true idle phase through `runMaintenance()`. Inside that task it refolds the exact Session suffix so a direct management mutation that won the claim race cannot be followed by a stale dispatch, then samples the decision clock once. A due one-shot bypasses the recurring gate and keeps the single fixed frame plus id-only dispatch. Otherwise the 300-second gate admits every overdue Every and Cron record in target/create order: the owner derives each latest occurrence, constructs the complete JSON batch before enqueue, synchronously queues one `followup()`, and appends an independent rule-specific dispatch per record. The gate's spacing directly limits every half-open 24-hour window to at most 288 recurring model turns; no second counter or quota exists. Waking input remains parked until maintenance settles, so the driver cannot claim the message before dispatch enters the log; only after the task releases the phase does the owner wait for the shared dispatch barrier. A framing or synchronous enqueue failure is contained and appends no dispatch. An append failure faults that owner because the message may already be queued. A later prompt-admission, request-checkpoint, or model failure cannot retract a dispatch.

@@ -66,7 +66,7 @@ persistence coordinator 只有在写路径完全停稳后才给出该确认。li

 ### Live 交付生命周期

-Agent-scoped owner 从持久 fold 派生活动目标与最近一次周期性 batch。超长目标使用有界 timer 分段，每次 wake 都重新读取墙钟，因此回拨不会提前触发，前跳则会形成 overdue。固定频率 record 将当前 `scheduledAt` 视为原始序列上最早尚未接受的点；整数除法会直接选出最近一次到期点。Cron record 将持久目标视为在 history 中保持稳定的 baseline，只搜索按当前规则求得且比 baseline 更新的 match，并持久化选定的 occurrence 与下一个目标。两种规则都不会回放错过期间积压的 occurrence，也不会把权威转移到交付时间。一旦有周期性 record 因门控关闭而处于 overdue，owner 就会将该门控或更早的一次性提醒设为唤醒点，而不再为其间的周期性目标安排唤醒。如果 agent 已被某个轮次或另一项 maintenance task 占用，`runMaintenance()` 会拒绝此次认领；record 保持活动，并由一个 `whenIdle()` wait 触发稍后的重试。被拒绝的 persistence preflight 或被收容的 framing／同步入队失败同样会让 record 保持活动，但不会运行私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 会要求 owner 再次尝试。
+Agent-scoped owner 从持久 fold 派生活动目标与最近一次周期性 batch。超长目标使用有界 timer 分段，每次 wake 都重新读取墙钟，因此回拨不会提前触发，前跳则会形成 overdue。固定频率 record 将当前 `scheduledAt` 视为原始序列上最早尚未接受的点；整数除法会直接选出最近一次到期点。Cron record 将持久目标视为在 history 中保持稳定的 baseline，只搜索按当前规则求得且比 baseline 更新的 match，并持久化选定的 occurrence 与下一个目标。两种规则都不会回放错过期间积压的 occurrence，也不会把权威转移到交付时间。一旦有周期性 record 因门控关闭而处于 overdue，owner 就会将该门控或更早的一次性提醒设为唤醒点，而不再为其间的周期性目标安排唤醒。如果 agent 已被某个轮次或另一项 maintenance task 占用，`runMaintenance()` 会拒绝此次认领；record 保持活动，并由一个 `whenIdle()` wait 触发稍后的重试。被拒绝的 persistence preflight、被收容的当前日历求值失败，或被收容的 framing／同步入队失败同样会让 record 保持活动，但不会运行私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 会要求 owner 再次尝试。

 获得准入的路径会先清空 pending persistence，并通过 `runMaintenance()` 认领真正的 idle phase。该任务会重新折叠确切的 Session 后缀，从而确保在认领竞态中胜出的直接管理变更之后不会跟随陈旧 dispatch；然后只采样一次 decision clock。到期的一次性提醒会绕过周期性门控，继续使用单条固定 reminder frame 和只含 id 的 dispatch。否则，300 秒门控会按目标／create 顺序接纳每条 overdue Every 与 Cron record：owner 为每条 record 派生最近一次到期的 occurrence，在入队前构造完整 JSON batch，同步排入一次 `followup()`，并为每条 record 追加与其规则对应的独立 dispatch。门控间隔直接将每个半开 24 小时窗口内由周期性提醒触发的模型轮次限制为至多 288 个；不存在第二个计数器或配额。触发唤醒的 input 会保持 parked，直到 maintenance 结束，因此 driver 无法在 dispatch 进入 log 前认领消息；只有该任务释放 phase 后，owner 才会等待共享 dispatch barrier。framing 或同步入队失败会被收容，且不会追加 dispatch。append 失败会使该 owner fault，因为消息可能已经入队。后续 prompt admission、request checkpoint 或模型失败都不能撤回 dispatch。

@@ -105,7 +105,7 @@ The reminder appends after existing history and preserves its reusable prefix. I
 ## Known Limitations and Deferred Work

 - **Session-local delivery only** — a reminder runs on time only while its original session is live; a cold session receives no external notification and processes an overdue record only after resume.
- **Activity-driven retry** — a rejected due preflight or contained framing/enqueue failure leaves the overdue record active but starts no private retry timer; the owner retries after later Agent activity reaches idle or a successful Schedule management preflight asks it to recompute.
+- **Activity-driven retry** — a rejected due preflight, contained current-calendar resolution failure, or contained framing/enqueue failure leaves the overdue record active but starts no private retry timer; the owner retries after later Agent activity reaches idle or a successful Schedule management preflight asks it to recompute.
 - **Restricted calendar language** — cron accepts only the documented numeric five-field subset with one unrestricted day field and an explicit IANA zone; it does not expose names, macros, seconds, years, Quartz operators, or user-selectable DST policy.
 - **Immutable Session zone** — a new Schedule Web Session captures one default browser zone and has no zone editor. Older headerless Sessions remain `unavailable`, and a mismatched or ambiguous request must name `time_zone` explicitly.
 - **Narrow crash duplicate window** — a crash after synchronous followup admission but before the dispatch checkpoint can repeat the reminder after recovery; the package does not claim model completion, user acknowledgement, or exactly-once external effects.
@@ -105,7 +105,7 @@ reminders_json: [{"schedule_id":<id>,"occurrence_at":<UTC RFC 3339>,"reminder_pr
 ## 已知限制与暂缓事项

 - **仅限会话本地交付**：提醒只有在原会话 live 时才能准时运行；cold 会话不会收到外部通知，只有恢复后才会处理 overdue 记录。
- **活动驱动的重试**：到期 preflight 被拒绝或 framing／入队失败被收容后，overdue 记录仍保持活动，但不会启动私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 要求 owner 重新计算后，owner 会重试。
+- **活动驱动的重试**：到期 preflight 被拒绝、当前日历求值失败被收容，或 framing／入队失败被收容后，overdue 记录仍保持活动，但不会启动私有重试 timer；后续 agent 活动进入 idle，或成功的 Schedule 管理 preflight 要求 owner 重新计算后，owner 会重试。
 - **受限的日历语言**：cron 只接受本文所述的数值五字段子集，其中一个日期字段必须不受限，并要求显式 IANA 时区；它不开放名称、macro、秒、年份、Quartz operator 或用户可选的 DST 策略。
 - **Session 时区不可变**：新的 Schedule Web Session 会记录一个默认浏览器时区，且没有时区编辑器。旧有的无 header Session 仍为 `unavailable`，不匹配或有歧义的请求必须显式指定 `time_zone`。
 - **存在狭窄的崩溃重复窗口**：同步 `followup` 获得准入后、dispatch 检查点完成前发生崩溃，可能使提醒在恢复后重复；此包不承诺模型完成、用户确认或外部副作用恰好一次。
@@ -689,6 +689,29 @@ function cronLocalFormatter(timeZone: string): Intl.DateTimeFormat {
  })
 }

+/** Skip a pre-standard-time sub-minute offset era without enumerating every Cron occurrence. */
+function cursorBeforeNextOffsetTransition(formatter: Intl.DateTimeFormat, epoch: number): number {
+  const initialOffset = localProjection(formatter, epoch).offset
+  let lower = epoch
+  let step = 366 * 86_400_000
+  let upper = epoch
+  while (upper < MAX_FOUR_DIGIT_YEAR_MS) {
+    upper = Math.min(MAX_FOUR_DIGIT_YEAR_MS, lower + step)
+    // IANA local-mean-time offsets do not return after a zone adopts standard time.
+    if (localProjection(formatter, upper).offset !== initialOffset) break
+    /* v8 ignore next 2 -- every supported IANA zone leaves local mean time before year 9999. */
+    if (upper === MAX_FOUR_DIGIT_YEAR_MS) return upper
+    lower = upper
+    step = Math.min(step * 2, MAX_FOUR_DIGIT_YEAR_MS - lower)
+  }
+  while (upper - lower > 1) {
+    const middle = lower + Math.floor((upper - lower) / 2)
+    if (localProjection(formatter, middle).offset === initialOffset) lower = middle
+    else upper = middle
+  }
+  return upper - 1
+}
+
 /** Whether local calendar fields satisfy one parsed rule. */
 function cronMatchesLocal(rule: ParsedCronRule, local: CalendarParts): boolean {
  const dayOfWeek = new Date(calendarEpoch(local)).getUTCDay()
@@ -727,6 +750,10 @@ function ownedLowYearCronInstant(
 ): number | undefined {
  const minYear = 1
  const maxYear = CRONER_LOW_YEAR_SEARCH_END
+  const formatter = cronLocalFormatter(timeZone)
+  const boundaryOffset = localProjection(formatter, boundary).offset
+  // IANA sub-minute local-mean-time offsets persist beyond this entire low-year bridge.
+  if (boundaryOffset % 60_000 !== 0) return undefined
  const utcYear = new Date(boundary).getUTCFullYear()
  const startYear = direction === 1
    ? Math.max(minYear, utcYear - 1)
@@ -769,6 +796,7 @@ function ownedLowYearCronInstant(
            /* v8 ignore next -- supported ICU data has no low-year transition gap to skip. */
            continue
          }
+          /* v8 ignore next -- a whole-minute low-year offset maps minute rules to whole-minute UTC. */
          if (candidate % 60_000 !== 0) continue
          if (direction === 1) {
            if (candidate > boundary) return candidate
@@ -812,6 +840,10 @@ function nextCronInstant(rule: ParsedCronRule, timeZone: string, after: number):
    }
    gapCorrections = 0
    if (epoch > MAX_FOUR_DIGIT_YEAR_MS) return undefined
+    if (epoch % 60_000 !== 0) {
+      cursor = cursorBeforeNextOffsetTransition(formatter, epoch)
+      continue
+    }
    if (isCanonicalCronCandidate(rule, formatter, timeZone, epoch)) return epoch
    cursor = epoch
  }
@@ -888,9 +920,6 @@ function validateLiveCronRule(record: CronScheduleRecord): {
  try {
    const rule = parseCronRule(record.cron)
    const timeZone = canonicalizeTimeZone(record.timeZone)
-    if (timeZone !== record.timeZone) {
-      throw new ScheduleLogError('live cron timeZone must use its current canonical IANA name')
-    }
    if (!rule.hasMatchingDate) {
      throw new ScheduleLogError('live cron rule must have a matching Gregorian date')
    }
@@ -904,6 +933,9 @@ function validateLiveCronRule(record: CronScheduleRecord): {
 function validateLiveCronRecord(record: CronScheduleRecord): void {
  const { rule, timeZone } = validateLiveCronRule(record)
  try {
+    if (timeZone !== record.timeZone) {
+      throw new ScheduleLogError('live cron timeZone must use its current canonical IANA name')
+    }
    const target = Date.parse(record.scheduledAt)
    if (nextCronInstant(rule, timeZone, target - 60_000) !== target) {
      throw new ScheduleLogError('live cron scheduledAt must match its rule in the current time-zone data')
@@ -235,6 +235,16 @@ export class ScheduleOwner {
    }
  }

+  /** Contain a current calendar-resolution failure without permanently faulting this owner. */
+  private decide(folded: FoldedSchedules, now: number): DueDecision | undefined {
+    try {
+      return dueDecision(folded, now)
+    } catch (error: unknown) {
+      this.ctx.logger.warn(`tool-schedule: calendar decision failed for agent "${this.agent.id}": ${renderThrown(error)}`)
+      return undefined
+    }
+  }
+
  /** Preflight, fold, arm, or dispatch the next one-shot or recurring batch. */
  private async driveOnce(): Promise<void> {
    this.clearTimer()
@@ -253,7 +263,8 @@ export class ScheduleOwner {
    const folded = this.readFolded()
    if (folded === undefined) return
    const wakeNow = Date.now()
-    const wakeDecision = dueDecision(folded, wakeNow)
+    const wakeDecision = this.decide(folded, wakeNow)
+    if (wakeDecision === undefined) return
    if (wakeDecision.kind === 'wait') {
      if (wakeDecision.target !== undefined) this.arm(wakeDecision.target, wakeNow)
      return
@@ -266,7 +277,8 @@ export class ScheduleOwner {
        const claimed = this.readFolded()
        if (claimed === undefined) return Promise.resolve(false)
        const decisionNow = Date.now()
-        const decision = dueDecision(claimed, decisionNow)
+        const decision = this.decide(claimed, decisionNow)
+        if (decision === undefined) return Promise.resolve(false)
        if (decision.kind === 'wait') {
          if (decision.target !== undefined) this.arm(decision.target, decisionNow)
          return Promise.resolve(false)
@@ -228,6 +228,14 @@ describe('Croner calendar adapter', () => {
      occurrenceAt: '2026-11-01T05:30:00.000Z',
      nextScheduledAt: '2026-11-02T06:30:00.000Z',
    })
+    expect(resolveCronOccurrence({
+      ...overlap,
+      cron: '0,30 1 * * *',
+      scheduledAt: '2026-10-31T05:30:00.000Z',
+    }, Date.parse('2026-11-01T07:00:00.000Z'))).toEqual({
+      occurrenceAt: '2026-11-01T05:30:00.000Z',
+      nextScheduledAt: '2026-11-02T06:00:00.000Z',
+    })
    expect(createCronScheduleRecord(
      ScheduleId('schedule-overlap-after-first'),
      'after first overlap instant',
@@ -237,6 +245,18 @@ describe('Croner calendar adapter', () => {
    ).scheduledAt).toBe('2026-11-02T06:30:00.000Z')
  })

+  it('skips a sub-minute local-mean-time era before iterating dense safe-year matches', () => {
+    const record = createCronScheduleRecord(
+      ScheduleId('schedule-sub-minute-offset'),
+      'standard-time handoff',
+      '*/5 * * * *',
+      'Europe/Amsterdam',
+      Date.parse('0100-01-01T00:00:00.000Z'),
+    )
+    expect(new Date(record.scheduledAt).getUTCFullYear()).toBeGreaterThan(109)
+    expect(Math.abs(Date.parse(record.scheduledAt) % 60_000)).toBe(0)
+  }, 1_000)
+
  it('selects the latest current match after a persisted baseline', () => {
    const record = createCronScheduleRecord(
      ScheduleId('schedule-latest'),
@@ -143,6 +143,28 @@ describe('Schedule package invariant', () => {
      }, 0)],
    })
    const fiber = await ctx.plugin(scheduleInvariant)
+    const alias = ctx.sessions.create(SessionId('schedule-historical-zone-alias'), {
+      seed: [event({
+        version: 1,
+        operation: 'create',
+        schedule: {
+          id: 'schedule-historical-zone-alias',
+          kind: 'cron',
+          prompt: 'historical zone alias',
+          cron: '0 9 * * *',
+          timeZone: 'US/Eastern',
+          scheduledAt: '2026-08-06T13:00:00.000Z',
+        },
+      }, 0)],
+    })
+    expect(() => alias.append('schedule/change', {
+      version: 1,
+      operation: 'dispatch',
+      id: ScheduleId('schedule-historical-zone-alias'),
+      occurrenceAt: '2026-08-07T13:00:00.000Z',
+      acceptedAt: '2026-08-07T14:00:00.000Z',
+      nextScheduledAt: '2026-08-08T13:00:00.000Z',
+    })).not.toThrow()
    const invalidLiveRules = [
      {
        id: 'schedule-historical-fast-cron',
@@ -4,6 +4,7 @@ import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
 import type { Agent, AgentCancelCause, InboxTarget } from '@deepseek-ai/dsh-agent'
 import type { UserMessage } from '@deepseek-ai/dsh-llm'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
+import { Cron } from 'croner'
 import {
  MIN_RECURRING_INTERVAL_SECONDS,
  ScheduleId,
@@ -168,6 +169,43 @@ afterEach(async () => {
 })

 describe('Schedule timer and admission runtime', () => {
+  it('contains calendar resolution failure without permanently faulting the owner', async () => {
+    const test = await harness()
+    const invalidId = ScheduleId('schedule-invalid-zone')
+    appendCron(test, invalidId, '0 0 * * *', Date.now() - 86_400_000)
+    const wakeFailure = vi.spyOn(Cron.prototype, 'previousRuns').mockImplementation(() => {
+      throw new Error('calendar unavailable')
+    })
+    const owner = ownerFor(test)
+    owner.start()
+    await settle()
+    expect(test.followed).toEqual([])
+    wakeFailure.mockRestore()
+
+    let restoreCalendarFailure: (() => void) | undefined
+    test.controls.onReserve = () => {
+      const calendarFailure = vi.spyOn(Cron.prototype, 'previousRuns').mockImplementation(() => {
+        throw new Error('calendar unavailable')
+      })
+      restoreCalendarFailure = () => { calendarFailure.mockRestore() }
+    }
+    owner.requestDrive()
+    await settle()
+    expect(test.followed).toEqual([])
+
+    restoreCalendarFailure?.()
+    test.controls.onReserve = undefined
+    test.agent.session.append('schedule/change', { version: 1, operation: 'delete', id: invalidId })
+    appendAfter(test, 'schedule-healthy-after', 1, Date.now() - 2_000)
+    owner.requestDrive()
+    await settle()
+    expect(test.followed).toHaveLength(1)
+    expect(test.agent.session.events.some(event =>
+      event.type === 'schedule/change'
+      && event.data.operation === 'dispatch'
+      && event.data.id === 'schedule-healthy-after')).toBe(true)
+  })
+
  it('segments waits beyond the Node timer limit and rechecks the wall clock', async () => {
    const test = await harness()
    const delaySeconds = Math.ceil((MAX_TIMER_DELAY_MS + 1_500) / 1_000)