Native takeover: refuse PHP schedules native cannot run yet

docs/rust-server-port.md

@@ -229,6 +229,31 @@ support is not evidence of multi-node safety.
 Newest checkpoints first. Each entry is exact to the source, tree and runs it
 names, including unsuccessful observations.
 
+October 21, takeover refuses schedules native cannot run:
+
+- **Found by.** A review of takeover against native's schedule support.
+  Native's scheduler fires only default-namespace schedules with one UTC
+  `P1D` or `PT2S` interval, the skip policy, no jitter and at most one run.
+  Once due, a PHP cron schedule taken over would have failed every
+  scheduler tick (native refuses its next occurrence), holding back the
+  schedules due after it. A named-namespace one would have silently stopped
+  firing.
+- **Source.** Every upgrade inspection now reports `schedule_inspection`: the
+  live schedules and, by reason, how many are outside native's subset. The
+  reasons are native's own schedule-creation refusals plus namespace, status,
+  memo/search attributes/labels, note, buffered actions and queue routing,
+  which native would not describe. Any such schedule makes the inspection
+  `unqualified`, and takeover refuses with
+  `upgrade_source_schedules_unsupported` before the fingerprint check or any
+  write. Deleted schedules do not count.
+- **Tests.** SQLite: a cron schedule and a named-namespace schedule each refuse
+  takeover and leave the file byte-identical; a runnable schedule and a
+  deleted cron schedule convert; inspection counts reasons without schedule
+  names or memo values. PostgreSQL 17, MySQL 8.0 and MariaDB 10.11 takeover
+  qualification each refuse the cron schedule with the source unchanged.
+- **Next.** Native schedule parity (cron, time zones, overlap policies,
+  jitter, quotas, metadata, namespaces) narrows this refusal.
+
 October 20, concurrent MySQL and MariaDB bootstrap:
 
 - **Found by.** A multi-node smoke run started three nodes together on an

docs/rust-server-upgrade-takeover.md

@@ -58,6 +58,16 @@ check exists, takeover refuses when `upgrade-payload-inspect` reports any
 external reference, and the operator's object manifest from the backup
 procedure remains the record for those bytes.
 
+Native schedules are still a bounded subset of PHP's: one UTC interval of a
+day or two seconds, the skip overlap policy, no jitter, at most one run, the
+default namespace, and no memo, search attributes, note, buffered actions or
+queue routing. Every inspection reports `schedule_inspection`: the number of
+live (not deleted) schedules and, by reason, how many fall outside that
+subset, without naming them. Such a schedule would stop firing or lose
+fields under native, so its inspection is `unqualified` and takeover refuses
+with `upgrade_source_schedules_unsupported` before any change. Deleted
+schedules keep only their audit history and do not block takeover.
+
 ## Exclusive transaction per backend
 
 The marker, the fingerprint check and the schema changes must not interleave

rust/src/runtime/mysql.rs

@@ -1388,6 +1388,10 @@ mod tests {
                 "INSERT INTO runtime_external_payloads(id,namespace,storage_uri,storage_uri_sha256,codec,sha256,size_bytes) VALUES ('p','default','s3://b/k','x','avro','y',1)",
                 "upgrade_source_external_payloads_unsupported",
             ),
+            (
+                "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s','nightly','default','{\"cron_expressions\":[\"0 2 * * *\"],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\"}','active','skip',0)",
+                "upgrade_source_schedules_unsupported",
+            ),
         ] {
             let db = Database::new().await;
             php_layout(&db, Some("php-2.9.7")).await;

rust/src/runtime/postgres.rs

@@ -921,6 +921,10 @@ mod tests {
                 "INSERT INTO runtime_external_payloads(id,namespace,storage_uri,storage_uri_sha256,codec,sha256,size_bytes) VALUES ('p','default','s3://b/k','x','avro','y',1)",
                 "upgrade_source_external_payloads_unsupported",
             ),
+            (
+                "INSERT INTO workflow_schedules(id,schedule_id,namespace,spec,action,status,overlap_policy,jitter_seconds) VALUES ('s','nightly','default','{\"cron_expressions\":[\"0 2 * * *\"],\"timezone\":\"UTC\"}','{\"workflow_type\":\"echo\",\"task_queue\":\"default\"}','active','skip',0)",
+                "upgrade_source_schedules_unsupported",
+            ),
         ] {
             let db = Database::new().await;
             let mut connection = php_297(&db).await;

rust/src/runtime/upgrade.rs

@@ -3,6 +3,7 @@ pub mod external_reference;
 pub mod fingerprint;
 pub mod takeover;
 mod payload_scan;
+mod schedule_scan;
 pub mod queue;
 pub mod queue_binding;
 mod queue_scan;
@@ -196,6 +197,7 @@ fn report(
     counts: BTreeMap<String, i64>,
     queue_inspection: Option<Value>,
     payload_inspection: Option<Value>,
+    schedule_inspection: Value,
 ) -> Result<Value> {
     let digest = Sha256::digest(serde_json::to_vec(catalog)?)
         .iter()
@@ -220,6 +222,10 @@ fn report(
         }
         report["payload_inspection"] = payloads;
     }
+    if schedule_inspection["unsupported"] != 0 {
+        report["status"] = json!("unqualified");
+    }
+    report["schedule_inspection"] = schedule_inspection;
     Ok(report)
 }
 
@@ -392,13 +398,15 @@ pub(super) async fn sqlite_inventory(
     } else {
         None
     };
+    let schedule_inspection = schedule_scan::inspect::<sqlx::Sqlite>(&mut *tx).await?;
     report(
         "sqlite",
         &catalog,
         layout,
         counts,
         queue_inspection,
         payload_inspection,
+        schedule_inspection,
     )
 }
 
@@ -661,6 +669,7 @@ pub(super) async fn mysql_inventory(
     } else {
         None
     };
+    let schedule_inspection = schedule_scan::inspect::<sqlx::MySql>(&mut *tx).await?;
     // MySQL's metadata is not an MVCC snapshot. Refuse a changed visible
     // catalog rather than joining counts to a different schema.
     if mysql_source_catalog(&mut *tx, fenced).await?.0 != catalog {
@@ -676,6 +685,7 @@ pub(super) async fn mysql_inventory(
         counts,
         queue_inspection,
         payload_inspection,
+        schedule_inspection,
     )
 }
 
@@ -742,13 +752,15 @@ pub(super) async fn postgres_inventory(
     } else {
         None
     };
+    let schedule_inspection = schedule_scan::inspect::<sqlx::Postgres>(&mut *tx).await?;
     report(
         "pgsql",
         &catalog,
         layout,
         counts,
         queue_inspection,
         payload_inspection,
+        schedule_inspection,
     )
 }
 

rust/src/runtime/upgrade/schedule_scan.rs

@@ -0,0 +1,190 @@
+//! Whether every live PHP schedule is one native can run and describe as PHP
+//! does. Native schedules are still a bounded subset (one UTC interval, the
+//! skip policy, no jitter, at most one run, the default namespace), so a
+//! source schedule outside it would stop firing, or lose fields, after
+//! takeover. The report counts them by reason and never contains schedule
+//! identities or values.
+use crate::runtime::{Result, RuntimeError, backend::Backend, schedules, store::now};
+use serde_json::{Value, json};
+use sqlx::{Executor, IntoArguments};
+use std::collections::BTreeMap;
+
+const SCHEDULES: &str = "SELECT schedule_id,namespace,status,spec,action,overlap_policy,jitter_seconds,max_runs,note,memo,search_attributes,visibility_labels,buffered_actions,connection,queue FROM workflow_schedules WHERE status<>'deleted' ORDER BY id";
+
+pub(super) async fn inspect<DB: Backend>(connection: &mut DB::Connection) -> Result<Value>
+where
+    for<'c> &'c mut DB::Connection: Executor<'c, Database = DB>,
+    DB::Arguments: IntoArguments<DB>,
+{
+    let rows = sqlx::query(sqlx::AssertSqlSafe(DB::statement(SCHEDULES)))
+        .fetch_all(&mut *connection)
+        .await?;
+    let mut reasons = BTreeMap::<&'static str, i64>::new();
+    for row in &rows {
+        let schedule = Schedule {
+            schedule_id: DB::string(row, "schedule_id")?,
+            namespace: DB::optional_string(row, "namespace")?,
+            status: DB::string(row, "status")?,
+            spec: DB::optional_document_row(row, "spec")?.unwrap_or(Value::Null),
+            action: DB::optional_document_row(row, "action")?.unwrap_or(Value::Null),
+            overlap_policy: DB::string(row, "overlap_policy")?,
+            jitter_seconds: DB::number(row, "jitter_seconds")?,
+            max_runs: DB::optional_number(row, "max_runs")?,
+            note: DB::optional_string(row, "note")?,
+            metadata: [
+                DB::optional_document_row(row, "memo")?,
+                DB::optional_document_row(row, "search_attributes")?,
+                DB::optional_document_row(row, "visibility_labels")?,
+            ],
+            buffered: DB::optional_document_row(row, "buffered_actions")?,
+            routing: [
+                DB::optional_string(row, "connection")?,
+                DB::optional_string(row, "queue")?,
+            ],
+        };
+        if let Some(reason) = schedule.unsupported() {
+            *reasons.entry(reason).or_default() += 1;
+        }
+    }
+    Ok(json!({
+        "schedules": rows.len(),
+        "unsupported": reasons.values().sum::<i64>(),
+        "reasons": reasons,
+    }))
+}
+
+struct Schedule {
+    schedule_id: String,
+    namespace: Option<String>,
+    status: String,
+    spec: Value,
+    action: Value,
+    overlap_policy: String,
+    jitter_seconds: i64,
+    max_runs: Option<i64>,
+    note: Option<String>,
+    metadata: [Option<Value>; 3],
+    buffered: Option<Value>,
+    routing: [Option<String>; 2],
+}
+
+/// A JSON column PHP leaves null when empty; an empty array or object is
+/// treated the same.
+fn present(value: &Option<Value>) -> bool {
+    match value {
+        None | Some(Value::Null) => false,
+        Some(Value::Array(items)) => !items.is_empty(),
+        Some(Value::Object(fields)) => !fields.is_empty(),
+        Some(_) => true,
+    }
+}
+
+impl Schedule {
+    /// The first reason native cannot take this schedule over, if any.
+    fn unsupported(&self) -> Option<&'static str> {
+        if self.namespace.as_deref() != Some("default") {
+            return Some("schedule_namespace_unsupported");
+        }
+        if !matches!(self.status.as_str(), "active" | "paused") {
+            return Some("schedule_status_unsupported");
+        }
+        if self.metadata.iter().any(present) {
+            return Some("schedule_metadata_unsupported");
+        }
+        if self.note.is_some() {
+            return Some("schedule_note_unsupported");
+        }
+        if present(&self.buffered) {
+            return Some("schedule_buffered_actions_unsupported");
+        }
+        if self.routing.iter().any(Option::is_some) {
+            return Some("schedule_queue_routing_unsupported");
+        }
+        // Exactly the definition native's own schedule creation accepts.
+        let body = json!({
+            "schedule_id": self.schedule_id, "spec": self.spec, "action": self.action,
+            "overlap_policy": self.overlap_policy, "jitter_seconds": self.jitter_seconds,
+            "max_runs": self.max_runs,
+        });
+        match schedules::definition(&body, now()) {
+            Ok(_) => None,
+            Err(RuntimeError::Refused { reason, .. }) => Some(reason),
+            Err(_) => Some("schedule_definition_unsupported"),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn runnable() -> Schedule {
+        Schedule {
+            schedule_id: "daily".into(),
+            namespace: Some("default".into()),
+            status: "active".into(),
+            spec: json!({"intervals":[{"every":"P1D"}],"timezone":"UTC"}),
+            action: json!({"workflow_type":"echo","task_queue":"default",
+                "input":{"codec":"avro","blob":"AA=="},
+                "execution_timeout_seconds":3600,"run_timeout_seconds":600}),
+            overlap_policy: "skip".into(),
+            jitter_seconds: 0,
+            max_runs: None,
+            note: None,
+            metadata: [None, Some(Value::Null), Some(json!([]))],
+            buffered: None,
+            routing: [None, None],
+        }
+    }
+
+    #[test]
+    fn only_schedules_native_runs_and_describes_unchanged_are_qualified() {
+        assert_eq!(runnable().unsupported(), None);
+        let cases: [(fn(&mut Schedule), &str); 10] = [
+            (
+                |s| s.namespace = Some("billing".into()),
+                "schedule_namespace_unsupported",
+            ),
+            (|s| s.namespace = None, "schedule_namespace_unsupported"),
+            (
+                |s| s.spec = json!({"cron_expressions":["0 9 * * *"],"timezone":"UTC"}),
+                "unsupported_request_field",
+            ),
+            (
+                |s| s.spec = json!({"intervals":[{"every":"PT1H"}],"timezone":"UTC"}),
+                "unsupported_schedule_interval",
+            ),
+            (|s| s.overlap_policy = "buffer_one".into(), "unsupported_schedule_policy"),
+            (|s| s.jitter_seconds = 30, "unsupported_schedule_policy"),
+            (|s| s.max_runs = Some(5), "unsupported_schedule_quota"),
+            (
+                |s| s.metadata[0] = Some(json!({"team":"ops"})),
+                "schedule_metadata_unsupported",
+            ),
+            (|s| s.note = Some("paused for audit".into()), "schedule_note_unsupported"),
+            (
+                |s| {
+                    s.action["execution_timeout_seconds"] = Value::Null;
+                },
+                "unsupported_schedule_action_timeout",
+            ),
+        ];
+        for (change, reason) in cases {
+            let mut schedule = runnable();
+            change(&mut schedule);
+            assert_eq!(schedule.unsupported(), Some(reason));
+        }
+        let mut buffered = runnable();
+        buffered.buffered = Some(json!([{"buffered_at":"2026-10-21T00:00:00+00:00"}]));
+        assert_eq!(
+            buffered.unsupported(),
+            Some("schedule_buffered_actions_unsupported")
+        );
+        let mut routed = runnable();
+        routed.routing[1] = Some("schedules".into());
+        assert_eq!(
+            routed.unsupported(),
+            Some("schedule_queue_routing_unsupported")
+        );
+    }
+}

rust/src/runtime/upgrade/takeover.rs

@@ -258,6 +258,13 @@ fn qualified(inventory: &Value) -> Result<&str> {
             return Err(refuse(StatusCode::CONFLICT, reason));
         }
     }
+    // A schedule native cannot run would stop firing after takeover.
+    if inventory["schedule_inspection"]["unsupported"] != 0 {
+        return Err(refuse(
+            StatusCode::CONFLICT,
+            "upgrade_source_schedules_unsupported",
+        ));
+    }
     if inventory["queue_inspection"]["status"] != "known_subset_bound" {
         return Err(refuse(
             StatusCode::CONFLICT,