fix(scheduler): reject illegal calendar timestamps instead of rolling them

loopx/control_plane/runtime_timestamp.ts

@@ -1,7 +1,13 @@
-/** Calendar-checked ISO codec; Date.parse alone silently rolls invalid dates. */
+/** Calendar-checked ISO codec; Date.parse alone silently rolls invalid dates.
+ * RFC 3339 §5.6 allows a lowercase "t" separator. Normalize only that
+ * separator before the existing calendar grammar. */
 export function parseIsoTimestamp(value: string): Date | null {
+  const source = value.trim().replace(
+    /^(\d{4}-\d{2}-\d{2})t(?=\d{2}:)/u,
+    "$1T",
+  );
   const match = /^(\d{4})-(\d{2})-(\d{2})(?:[T ](\d{2}):(\d{2})(?::(\d{2})(?:\.(\d+))?)?(Z|z|[+-]\d{2}(?::?\d{2})?)?)?$/u.exec(
-    value.trim(),
+    source,
   );
   if (match === null) return null;
   const [, yearText, monthText, dayText, hourText, minuteText, secondText, fraction, timezone] = match;
@@ -30,7 +36,7 @@ export function parseIsoTimestamp(value: string): Date | null {
     calendar.getUTCMilliseconds() !== millisecond
   ) return null;
   if (hourText === undefined) return calendar;
-  let text = value.trim().replace(" ", "T").replace(/z$/u, "Z");
+  let text = source.replace(" ", "T").replace(/z$/u, "Z");
   if (fraction !== undefined) text = text.replace(`.${fraction}`, `.${fraction.slice(0, 3)}`);
   if (timezone === undefined) text += "Z";
   else text = text.replace(/([+-]\d{2})$/u, "$1:00");

loopx/control_plane/scheduler/state_store.ts

@@ -17,6 +17,7 @@ import {
   requireNonEmptyString as requiredString,
   requireStringLiteral,
 } from "../runtime_decode.ts";
+import { parseIsoTimestamp } from "../runtime_timestamp.ts";
 
 export const SCHEDULER_STATE_SCHEMA_VERSION = "loopx_scheduler_state_v0";
 export const SCHEDULER_HOST_UPDATE_FAILURE_SCHEMA_VERSION =
@@ -216,14 +217,26 @@ export function normalizeSchedulerHostUpdateFailures(
   return normalized.slice(-HOST_UPDATE_FAILURE_CACHE_LIMIT);
 }
 
+/** Calendar-checked ISO instant. Blank is null; a missing timezone means UTC. */
 export function schedulerTimestampMilliseconds(value: unknown): number | null {
   const text = trimmed(value);
   if (!text) return null;
-  const timezoneAware = /(?:[zZ]|[+-]\d{2}(?::?\d{2})?)$/.test(text)
-    ? text
-    : `${text}Z`;
-  const parsed = Date.parse(timezoneAware);
-  return Number.isNaN(parsed) ? null : parsed;
+  const parsed = parseIsoTimestamp(text);
+  return parsed === null ? null : parsed.getTime();
+}
+
+/** Blank stays null. Present text that is not a calendar-checked ISO timestamp is rejected. */
+export function requireSchedulerTimestampMilliseconds(
+  value: unknown,
+  label: string,
+): number | null {
+  const text = trimmed(value);
+  if (!text) return null;
+  const parsed = schedulerTimestampMilliseconds(text);
+  if (parsed === null) {
+    throw new EffectRuntimeRequestError(`${label} must be an ISO timestamp`);
+  }
+  return parsed;
 }
 
 export function retainedSchedulerHostUpdateFailures(
@@ -232,7 +245,10 @@ export function retainedSchedulerHostUpdateFailures(
   observedHostRrule: unknown = null,
 ): JsonObject[] {
   const failures = normalizeSchedulerHostUpdateFailures(value);
-  const now = schedulerTimestampMilliseconds(referenceTime) ?? Date.now();
+  const now = requireSchedulerTimestampMilliseconds(
+    referenceTime,
+    "reference_time",
+  ) ?? Date.now();
   const cutoff = now - HOST_UPDATE_FAILURE_TTL_MS;
   const expectedHostRrule = normalizeSchedulerRrule(observedHostRrule);
   return failures.filter((failure) => {

loopx/control_plane/scheduler/state_transition_rules.ts

@@ -10,6 +10,7 @@ import {
   normalizeSchedulerHostUpdateFailures,
   normalizeSchedulerRrule,
   retainedSchedulerHostUpdateFailures,
+  requireSchedulerTimestampMilliseconds,
   rruleForMinutes,
   schedulerRruleIntervalMinutes,
   schedulerTimestampMilliseconds,
@@ -359,17 +360,40 @@ function schedulerProgressionIntervalElapsed(
   schedulerState: JsonObject,
   currentTime: string,
 ): boolean {
-  const updatedAt = schedulerTimestampMilliseconds(schedulerState.updated_at);
-  const current = schedulerTimestampMilliseconds(currentTime);
   const appliedInterval = schedulerRruleIntervalMinutes(
     schedulerState.last_applied_rrule,
   );
-  if (updatedAt === null || current === null || appliedInterval === null) {
-    return true;
-  }
+  // Missing interval or blank updated_at stays elapsed. A present illegal
+  // calendar is rejected instead of rolled or treated as elapsed.
+  if (appliedInterval === null) return true;
+  const updatedAt = requireSchedulerTimestampMilliseconds(
+    schedulerState.updated_at,
+    "updated_at",
+  );
+  if (updatedAt === null) return true;
+  const current = requireSchedulerTimestampMilliseconds(
+    currentTime,
+    "current_time",
+  );
+  if (current === null) return true;
   return current >= updatedAt + appliedInterval * 60_000;
 }
 
+// evaluateCadence reads the old clock only after identity matches and the
+// same-identity path is comparing the applied interval. An obsolete receipt
+// must not block identity_reset, an unacknowledged retry, or a held initial.
+function cadenceConsumesAppliedInterval(
+  statePresent: boolean,
+  identityMatches: boolean,
+  hasHostUpdateFailures: boolean,
+  currentCadenceAcknowledged: boolean,
+  advanceSameIdentity: boolean,
+): boolean {
+  if (!statePresent || !identityMatches) return false;
+  if (hasHostUpdateFailures && !currentCadenceAcknowledged) return false;
+  return advanceSameIdentity;
+}
+
 function monitorRruleAppliedWithinStaleTolerance(
   cadenceClass: string,
   lastAppliedRrule: string,
@@ -400,6 +424,7 @@ function evaluateBackoff(request: JsonObject): SchedulerBackoffTransitionResult
     "identity_signature",
   );
   const currentTime = requiredString(request.current_time, "current_time");
+  requireSchedulerTimestampMilliseconds(currentTime, "current_time");
   const cadenceClass = requiredString(request.cadence_class, "cadence_class");
   const staleToleranceMinutes = requiredInteger(
     request.stale_tolerance_minutes,
@@ -428,23 +453,33 @@ function evaluateBackoff(request: JsonObject): SchedulerBackoffTransitionResult
     : "";
   const currentCadenceAcknowledged = Boolean(appliedTargetRrule) &&
     normalizeSchedulerRrule(lastAppliedRrule) === appliedTargetRrule;
+  const statePresent = Object.keys(schedulerState).length > 0;
+  const identityMatches = statePresent &&
+    schedulerState.reset_token === resetToken &&
+    schedulerState.identity_signature === identitySignature;
+  const advanceSameIdentity = requiredBoolean(
+    request.advance_same_identity,
+    "advance_same_identity",
+  );
+  const hasHostUpdateFailures = allHostUpdateFailures.length > 0;
+  const appliedIntervalElapsed = cadenceConsumesAppliedInterval(
+    statePresent,
+    identityMatches,
+    hasHostUpdateFailures,
+    currentCadenceAcknowledged,
+    advanceSameIdentity,
+  )
+    ? schedulerProgressionIntervalElapsed(schedulerState, currentTime)
+    : false;
   const cadence = evaluateCadence({
     progression_size: progressionMinutes.length,
-    state_present: Object.keys(schedulerState).length > 0,
-    identity_matches: Object.keys(schedulerState).length > 0 &&
-      schedulerState.reset_token === resetToken &&
-      schedulerState.identity_signature === identitySignature,
+    state_present: statePresent,
+    identity_matches: identityMatches,
     applied_index: appliedIndex,
     current_cadence_acknowledged: currentCadenceAcknowledged,
-    advance_same_identity: requiredBoolean(
-      request.advance_same_identity,
-      "advance_same_identity",
-    ),
-    applied_interval_elapsed: schedulerProgressionIntervalElapsed(
-      schedulerState,
-      currentTime,
-    ),
-    has_host_update_failures: allHostUpdateFailures.length > 0,
+    advance_same_identity: advanceSameIdentity,
+    applied_interval_elapsed: appliedIntervalElapsed,
+    has_host_update_failures: hasHostUpdateFailures,
   });
   const currentIndex = cadence.current_index;
   const currentInterval = progressionMinutes[currentIndex];

tests/control_plane_ts/scheduler_state_store.test.ts

@@ -9,6 +9,7 @@ import {
   loadSchedulerState,
   normalizeSchedulerState,
   schedulerStatePath,
+  schedulerTimestampMilliseconds,
   SCHEDULER_STATE_OPERATION_REQUEST_SCHEMA,
   SCHEDULER_STATE_STORE_REQUEST_SCHEMA,
   writeSchedulerState,
@@ -160,6 +161,76 @@ test("failure retention accepts the ISO offsets persisted by Python", () => {
   assert.equal((result as unknown[]).length, 1);
 });
 
+function hostUpdateFailure(failedAt: string) {
+  return {
+    schema_version: "scheduler_host_update_failure_v0",
+    target_rrule: "FREQ=MINUTELY;INTERVAL=3",
+    observed_host_rrule: "FREQ=MINUTELY;INTERVAL=30",
+    failure_kind: "timeout",
+    failure_count: 1,
+    failed_at: failedAt,
+  };
+}
+
+test("scheduler timestamp codec rejects illegal calendars and keeps offset instants", () => {
+  assert.equal(schedulerTimestampMilliseconds("2026-02-31T00:00:00Z"), null);
+  assert.equal(schedulerTimestampMilliseconds("2026-02-29T00:00:00Z"), null);
+  assert.equal(schedulerTimestampMilliseconds("2024-02-29T00:00:00Z"), Date.UTC(2024, 1, 29));
+  assert.equal(schedulerTimestampMilliseconds("2026-11-31T10:04:00Z"), null);
+  assert.equal(schedulerTimestampMilliseconds("2026-01-32T12:10:00Z"), null);
+  assert.equal(schedulerTimestampMilliseconds("   "), null);
+  assert.equal(
+    schedulerTimestampMilliseconds("2026-08-25T10:00:00+08:00"),
+    Date.UTC(2026, 7, 25, 2, 0, 0),
+  );
+  assert.equal(
+    schedulerTimestampMilliseconds("2026-01-01T12:00:00"),
+    Date.UTC(2026, 0, 1, 12, 0, 0),
+  );
+  assert.equal(
+    schedulerTimestampMilliseconds("2026-01-01T12:00:00+0000"),
+    Date.UTC(2026, 0, 1, 12, 0, 0),
+  );
+  assert.equal(
+    schedulerTimestampMilliseconds("2026-12-04t01:15:00z"),
+    Date.UTC(2026, 11, 4, 1, 15, 0),
+  );
+  assert.equal(
+    schedulerTimestampMilliseconds("2026-12-04t01:15:00z"),
+    schedulerTimestampMilliseconds("2026-12-04T01:15:00Z"),
+  );
+  assert.equal(
+    schedulerTimestampMilliseconds("2026-08-25t10:00:00+08:00"),
+    schedulerTimestampMilliseconds("2026-08-25T10:00:00+08:00"),
+  );
+  assert.equal(schedulerTimestampMilliseconds("2026-11-31t10:04:00z"), null);
+  assert.equal(schedulerTimestampMilliseconds("2026-02-31t00:00:00z"), null);
+});
+
+test("failure retention does not roll illegal calendar timestamps into the TTL window", () => {
+  for (const [failedAt, referenceTime] of [
+    ["2026-02-31T00:00:00Z", "2026-03-03T00:00:00Z"],
+    ["2026-11-31T10:04:00Z", "2026-12-01T10:04:00Z"],
+  ] as const) {
+    const retained = evaluateSchedulerStateOperation({
+      schema_version: SCHEDULER_STATE_OPERATION_REQUEST_SCHEMA,
+      operation: "retain_failures",
+      value: [hostUpdateFailure(failedAt)],
+      reference_time: referenceTime,
+    }).value;
+    assert.deepEqual(retained, [], failedAt);
+  }
+  assert.throws(
+    () => evaluateSchedulerStateOperation({
+      schema_version: SCHEDULER_STATE_OPERATION_REQUEST_SCHEMA,
+      operation: "retain_failures",
+      value: [hostUpdateFailure("2026-03-03T00:00:00Z")],
+      reference_time: "2026-02-31T00:00:00Z",
+    }),
+    /reference_time must be an ISO timestamp/,
+  );
+});
+
 test("pure scheduler state operations do not mutate caller input", () => {
   const params = fixture.cases[7].params as Record<string, unknown>;
   const before = structuredClone(params);