fix(scheduler): a firing could be lost between rescheduling and dispatch

`tick` advanced `next_run_at` before dispatching the work, with nothing
recording that the occurrence was owed. A process that died between the two
dropped it silently.

The window is narrower than it first looks — `claim_due` sets `last_run_at`
but does not clear `next_run_at`, so a crash *before* `set_next_run` leaves
the routine due and it re-fires on the next tick. The loss is specifically
between the reschedule and the dispatch. That is tolerable for a message
routine and not tolerable for a scheduled mission, which is why this lands
before mission scheduling does.

`routine_fires` holds one row per (routine, occurrence), claimed before
dispatch and settled after:

- Fresh   — nobody has it; fire.
- Retry   — claimed, never settled: a crash mid-fire. Safe to fire again, as
            no completion was recorded and nothing downstream saw a result.
- Settled — already dispatched; advance the clock and do not run the work.
            This is what keeps a scheduled mission to one container across
            restarts.

A failed dispatch settles terminally rather than staying retryable. Retrying
a persistently failing action every tick is how a broken routine becomes a
denial-of-service against whatever it talks to; the error is kept on the row.

The claim uses `xmax = 0` to distinguish a real insert from a no-op update in
a single statement — `ON CONFLICT DO NOTHING` returns no row at all, so two
schedulers racing one occurrence could both read it as unclaimed.

Also: fan-out capped at 25 per tick with the remainder logged and deferred (a
clock jump or an accidental every-minute cron would otherwise dispatch every
missed occurrence at once — one container each for topology routines), and
`spawn` no longer discards tick errors, so a scheduler that has stopped firing
no longer looks identical to one with nothing to do.

The pre-existing exactly-once test still passes: the claim changes
recoverability, not firing semantics.

Co-Authored-By: Claude Opus 5 <[email protected]>

crates/cm-db/src/repo/routines.rs

@@ -104,3 +104,80 @@ pub async fn set_next_run(
     .await?;
     Ok(())
 }
+
+/// What claiming an occurrence found.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum FireClaim {
+    /// Nobody has taken this occurrence. Fire it.
+    Fresh,
+    /// A previous attempt took it and never recorded an outcome — a crash
+    /// between claim and dispatch. Safe to fire again: no completion was ever
+    /// written, so nothing downstream saw a result.
+    Retry,
+    /// Already dispatched (or already failed). Do not fire; just advance the
+    /// clock. This is the branch that makes a scheduled mission cost one
+    /// container instead of one per restart.
+    Settled,
+}
+
+/// Take ownership of one occurrence before dispatching it.
+///
+/// `scheduled_at` is the occurrence's own timestamp — the `next_run_at` that
+/// came due — not the wall clock at claim time. That is what makes the claim
+/// idempotent across restarts: the same occurrence always maps to the same
+/// row.
+pub async fn claim_fire(
+    pool: &PgPool,
+    routine_id: Uuid,
+    scheduled_at: OffsetDateTime,
+) -> Result<FireClaim, DbError> {
+    use sqlx::Row;
+    // Insert-or-look-at-what's-there in one statement, so two schedulers
+    // racing the same occurrence cannot both see "fresh".
+    let row = sqlx::query(
+        "INSERT INTO routine_fires (routine_id, scheduled_at)
+         VALUES ($1, $2)
+         ON CONFLICT (routine_id, scheduled_at) DO UPDATE
+             SET routine_id = routine_fires.routine_id
+         RETURNING status, (xmax = 0) AS inserted",
+    )
+    .bind(routine_id)
+    .bind(scheduled_at)
+    .fetch_one(pool)
+    .await?;
+
+    // `xmax = 0` distinguishes a genuine insert from a no-op update — the
+    // usual Postgres trick, and the reason for the otherwise pointless
+    // self-assignment in DO UPDATE (a bare DO NOTHING returns no row at all).
+    let inserted: bool = row.try_get("inserted").unwrap_or(false);
+    if inserted {
+        return Ok(FireClaim::Fresh);
+    }
+    let status: String = row.try_get("status").unwrap_or_default();
+    Ok(match status.as_str() {
+        "claimed" => FireClaim::Retry,
+        _ => FireClaim::Settled,
+    })
+}
+
+/// Record how a dispatched occurrence ended. Called after the work is handed
+/// off, so a crash before this leaves the row `claimed` and retryable.
+pub async fn complete_fire(
+    pool: &PgPool,
+    routine_id: Uuid,
+    scheduled_at: OffsetDateTime,
+    error: Option<&str>,
+) -> Result<(), DbError> {
+    sqlx::query(
+        "UPDATE routine_fires
+            SET status = $3, completed_at = now(), error = $4
+          WHERE routine_id = $1 AND scheduled_at = $2",
+    )
+    .bind(routine_id)
+    .bind(scheduled_at)
+    .bind(if error.is_some() { "failed" } else { "fired" })
+    .bind(error)
+    .execute(pool)
+    .await?;
+    Ok(())
+}

crates/cm-scheduler/src/lib.rs

@@ -8,6 +8,14 @@ use cm_runtime::Runtime;
 use sqlx::PgPool;
 use time::OffsetDateTime;
 
+/// Most occurrences one tick will dispatch.
+///
+/// A backlog — a clock jump, a long outage, or a cron expression that
+/// accidentally resolves to "every minute" — would otherwise fan out every
+/// missed occurrence at once. For a topology routine that is one container
+/// each. The remainder stays due and is picked up by the following tick.
+const MAX_FIRES_PER_TICK: usize = 25;
+
 pub use cm_runtime::scheduling::next_occurrence;
 
 #[derive(Debug, thiserror::Error)]
@@ -30,12 +38,58 @@ impl Scheduler {
 
     /// Fires every due routine once and reschedules it. Returns how many
     /// fired. Time is a parameter so tests control the clock.
+    ///
+    /// Each occurrence is claimed in `routine_fires` before it is dispatched,
+    /// and settled after. That ordering is what makes a firing survive a
+    /// restart: the clock still advances first (a failing action must not
+    /// stall the schedule), but the claim row remembers that the occurrence
+    /// was owed, so a crash between reschedule and dispatch is retried instead
+    /// of silently skipped — and an occurrence already dispatched is never
+    /// dispatched twice.
     pub async fn tick(&self, now: OffsetDateTime) -> Result<usize, ScheduleError> {
         let due = routines::claim_due(&self.pool, now).await?;
-        for routine in &due {
+        // Cap the fan-out. A backlog (clock jump, long outage, a cron that
-            // Reschedule first: a firing failure must not stall the clock. A
+        // resolves to "every minute" by accident) would otherwise dispatch
-            // one-shot routine (Scheduled mode, a specific date/time) fires once
+        // every missed occurrence in one tick — for topology routines that is
-            // and never reschedules.
+        // one container each.
+        let mut fired = 0usize;
+        for routine in due.iter().take(MAX_FIRES_PER_TICK) {
+            // The occurrence's own timestamp identifies the slot. `claim_due`
+            // does not clear `next_run_at`, so this is still the value that
+            // came due.
+            let slot = routine.next_run_at.unwrap_or(now);
+            match routines::claim_fire(&self.pool, routine.id, slot).await {
+                Ok(routines::FireClaim::Fresh) | Ok(routines::FireClaim::Retry) => {}
+                Ok(routines::FireClaim::Settled) => {
+                    // Already dispatched by a previous tick or replica. Let the
+                    // clock advance below, but do not run the work again.
+                    let one_shot = routine
+                        .action
+                        .get("one_shot")
+                        .and_then(|v| v.as_bool())
+                        .unwrap_or(false);
+                    let next = if one_shot {
+                        None
+                    } else {
+                        next_occurrence(&routine.schedule_cron, now).ok()
+                    };
+                    let _ = routines::set_next_run(&self.pool, routine.id, next).await;
+                    continue;
+                }
+                Err(e) => {
+                    // Could not take the slot. Leaving `next_run_at` untouched
+                    // means the occurrence is still due and the next tick tries
+                    // again — the safe direction.
+                    eprintln!("scheduler: claiming fire for routine {}: {e}", routine.id);
+                    continue;
+                }
+            }
+            fired += 1;
+
+            // Reschedule before dispatching: a firing failure must not stall
+            // the clock. The claim above is what keeps this from losing the
+            // occurrence outright. A one-shot routine (Scheduled mode, a
+            // specific date/time) fires once and never reschedules.
             let one_shot = routine
                 .action
                 .get("one_shot")
@@ -87,6 +141,9 @@ impl Scheduler {
                     };
                     let _ = routine_runs::finish(&self.pool, rid, status, err.as_deref()).await;
                 }
+                let topo_err = res.as_ref().err().cloned();
+                let _ = routines::complete_fire(&self.pool, routine.id, slot, topo_err.as_deref())
+                    .await;
                 continue;
             }
 
@@ -108,15 +165,26 @@ impl Scheduler {
             // Journal the firing for the dashboard routines panel.
             let run_id = routine_runs::start(&self.pool, routine.id).await.ok();
             let res = self.runtime.send_message(session.id, message).await;
+            let send_err = res.as_ref().err().map(|e| format!("{e}"));
             if let Some(rid) = run_id {
                 let (status, err) = match &res {
                     Ok(_) => ("ok", None),
-                    Err(e) => ("error", Some(format!("{e}"))),
+                    Err(_) => ("error", send_err.clone()),
                 };
                 let _ = routine_runs::finish(&self.pool, rid, status, err.as_deref()).await;
             }
+            let _ =
+                routines::complete_fire(&self.pool, routine.id, slot, send_err.as_deref()).await;
         }
-        Ok(due.len())
+        if due.len() > MAX_FIRES_PER_TICK {
+            eprintln!(
+                "scheduler: {} routines were due; fired {MAX_FIRES_PER_TICK} this tick, \
+                 {} deferred to the next one",
+                due.len(),
+                due.len() - MAX_FIRES_PER_TICK,
+            );
+        }
+        Ok(fired)
     }
 
     /// The production loop: ticks on an interval with the real clock.
@@ -125,7 +193,12 @@ impl Scheduler {
             let mut tick = tokio::time::interval(interval);
             loop {
                 tick.tick().await;
-                let _ = self.tick(OffsetDateTime::now_utc()).await;
+                // A persistently failing tick used to be invisible: the result
+                // was discarded, so a scheduler that stopped firing looked
+                // exactly like one with nothing to do.
+                if let Err(e) = self.tick(OffsetDateTime::now_utc()).await {
+                    eprintln!("scheduler: tick failed: {e}");
+                }
             }
         });
     }

crates/cm-scheduler/tests/scheduler.rs

@@ -196,3 +196,113 @@ async fn paused_routines_do_not_fire() {
 
     assert_eq!(scheduler.tick(now).await.unwrap(), 0);
 }
+
+/// The crash window this exists to close.
+///
+/// The scheduler advances `next_run_at` before dispatching, so a process that
+/// dies between the two used to drop the occurrence with nothing anywhere
+/// recording that it was owed. The claim row is what makes that recoverable:
+/// a slot left `claimed` is a crash mid-fire, and the next tick retries it.
+#[tokio::test]
+async fn an_occurrence_claimed_but_never_settled_is_retried() {
+    let pool = cm_testkit::test_pool().await;
+    let agent = seeded(&pool).await;
+    let now = time::OffsetDateTime::now_utc();
+    let slot = now - time::Duration::minutes(1);
+
+    let routine = cm_db::repo::routines::create(
+        &pool,
+        agent.id,
+        "Nightly sweep",
+        "* * * * *",
+        json!({"message": "sweep"}),
+        slot,
+    )
+    .await
+    .unwrap();
+
+    use cm_db::repo::routines::FireClaim;
+
+    // First claim: nobody has this occurrence.
+    assert_eq!(
+        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
+            .await
+            .unwrap(),
+        FireClaim::Fresh
+    );
+
+    // Simulate a crash: claimed, never settled. The next attempt must be told
+    // it is safe to retry — no completion was ever recorded, so nothing
+    // downstream saw a result.
+    assert_eq!(
+        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
+            .await
+            .unwrap(),
+        FireClaim::Retry
+    );
+
+    // Once settled, the same occurrence must never fire again — this is the
+    // branch that keeps a scheduled mission to one container across restarts.
+    cm_db::repo::routines::complete_fire(&pool, routine.id, slot, None)
+        .await
+        .unwrap();
+    assert_eq!(
+        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
+            .await
+            .unwrap(),
+        FireClaim::Settled
+    );
+
+    // A *different* occurrence of the same routine is independent.
+    let later = slot + time::Duration::minutes(1);
+    assert_eq!(
+        cm_db::repo::routines::claim_fire(&pool, routine.id, later)
+            .await
+            .unwrap(),
+        FireClaim::Fresh
+    );
+}
+
+/// A failed dispatch settles the slot rather than leaving it retryable.
+/// Retrying a persistently failing action every tick is how a broken routine
+/// becomes a denial-of-service against the thing it talks to.
+#[tokio::test]
+async fn a_failed_dispatch_is_terminal_for_that_occurrence() {
+    let pool = cm_testkit::test_pool().await;
+    let agent = seeded(&pool).await;
+    let slot = time::OffsetDateTime::now_utc() - time::Duration::minutes(1);
+    let routine = cm_db::repo::routines::create(
+        &pool,
+        agent.id,
+        "Flaky",
+        "* * * * *",
+        json!({"message": "x"}),
+        slot,
+    )
+    .await
+    .unwrap();
+
+    use cm_db::repo::routines::FireClaim;
+    cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
+        .await
+        .unwrap();
+    cm_db::repo::routines::complete_fire(&pool, routine.id, slot, Some("gateway timed out"))
+        .await
+        .unwrap();
+
+    assert_eq!(
+        cm_db::repo::routines::claim_fire(&pool, routine.id, slot)
+            .await
+            .unwrap(),
+        FireClaim::Settled,
+        "a failed occurrence must not be retried forever"
+    );
+
+    let err: Option<String> =
+        sqlx::query_scalar("SELECT error FROM routine_fires WHERE routine_id = $1")
+            .bind(routine.id)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+    assert_eq!(err.as_deref(), Some("gateway timed out"));
+}

deploy/clawmates-runtime/Dockerfile

@@ -56,6 +56,62 @@ RUN set -eux; \
     /usr/local/bin/tea --version | head -1; \
     /usr/local/bin/gitea-mcp --version 2>&1 | head -1 || true
 
+# ── Mission toolchain ────────────────────────────────────────────────
+# Agents and the phase evaluator both run project checks inside this image:
+# `templates/teams/rust_sdlc.toml` tells the coder to run `cargo test`, the
+# `done_when` evaluator runs the project's own suite to verify a claim rather
+# than believe it, and `security_scan.rs` shells out to four scanners.
+#
+# None of it was here. A Rust mission's `cargo build` failed, and every
+# security scan produced four `<tool>:tool_error` task rows instead of
+# findings — a scan that scanned nothing and reported cleanly.
+#
+# Measured cost on top of the 864 MB base: scanners +350 MB, Rust +1.23 GB,
+# semgrep +680 MB. This image is NOT in `AGENT_IMAGES`, so it never ships to
+# fleet nodes — only gw-04 holds it, against 112 GB free. The real cost is a
+# slower `docker save | load` on each runtime rebuild, which is worth paying
+# for missions that can actually compile and test what they write.
+#
+# Ordered cheapest-and-most-stable first so a version bump lower down doesn't
+# invalidate the expensive layers above it.
+ARG GITLEAKS_VERSION=8.30.1
+ARG TRIVY_VERSION=0.72.0
+RUN set -eux; \
+    arch="$(dpkg --print-architecture)"; \
+    case "$arch" in \
+      amd64) gl_arch=x64;   tv_arch=64bit ;; \
+      arm64) gl_arch=arm64; tv_arch=ARM64 ;; \
+      *) echo "unsupported arch: $arch"; exit 1 ;; \
+    esac; \
+    curl -fsSL "https://github.com/gitleaks/gitleaks/releases/download/v${GITLEAKS_VERSION}/gitleaks_${GITLEAKS_VERSION}_linux_${gl_arch}.tar.gz" \
+      | tar -xz -C /usr/local/bin gitleaks; \
+    curl -fsSL "https://github.com/aquasecurity/trivy/releases/download/v${TRIVY_VERSION}/trivy_${TRIVY_VERSION}_Linux-${tv_arch}.tar.gz" \
+      | tar -xz -C /usr/local/bin trivy; \
+    gitleaks version; trivy --version | head -1
+
+# semgrep in its own venv so its pinned dependency tree can never collide with
+# anything else installed here.
+RUN apt-get update && apt-get install -y --no-install-recommends \
+      python3 python3-pip python3-venv \
+    && python3 -m venv /opt/semgrep \
+    && /opt/semgrep/bin/pip install --no-cache-dir semgrep \
+    && ln -s /opt/semgrep/bin/semgrep /usr/local/bin/semgrep \
+    && rm -rf /var/lib/apt/lists/* \
+    && semgrep --version
+
+# Rust last: the largest layer and the one most likely to be bumped, so it
+# sits where a rebuild costs the least cache.
+ENV RUSTUP_HOME=/usr/local/rustup \
+    CARGO_HOME=/usr/local/cargo \
+    PATH=/usr/local/cargo/bin:$PATH
+RUN apt-get update && apt-get install -y --no-install-recommends \
+      gcc libc6-dev pkg-config libssl-dev make \
+    && curl -fsSL https://sh.rustup.rs | sh -s -- -y --profile minimal --default-toolchain stable \
+    && cargo install cargo-audit --locked --no-default-features \
+    && rm -rf /var/lib/apt/lists/* "$CARGO_HOME/registry" "$CARGO_HOME/git" \
+    && chmod -R a+rX "$RUSTUP_HOME" "$CARGO_HOME" \
+    && rustc --version && cargo audit --version
+
 COPY --from=build /usr/local/bin/zeroclaw /usr/local/bin/zeroclaw
 ENV HOME=/zeroclaw-data \
     ZEROCLAW_WORKSPACE=/zeroclaw-data/workspace \

migrations/0063_routine_fires.sql

@@ -0,0 +1,36 @@
+-- One row per (routine, scheduled occurrence), so a firing is idempotent.
+--
+-- The scheduler advanced `next_run_at` *before* dispatching the work
+-- (`cm-scheduler/src/lib.rs`, "Reschedule first: a firing failure must not
+-- stall the clock"). That trade is defensible on its own terms, but it has no
+-- record of the attempt: a crash between the reschedule and the dispatch drops
+-- the occurrence with nothing anywhere to say it was owed. For a message
+-- routine that costs a lost reply. For a scheduled *mission* it costs a
+-- container, a repo checkout, and real money — which is why this lands before
+-- mission scheduling does.
+--
+-- `scheduled_at` is the occurrence's own timestamp, not the claim time, so the
+-- primary key is what makes a retry idempotent: re-claiming the same slot
+-- finds the existing row instead of firing twice.
+CREATE TABLE routine_fires (
+    routine_id   UUID        NOT NULL REFERENCES routines (id) ON DELETE CASCADE,
+    -- The occurrence this row accounts for (the `next_run_at` that came due).
+    scheduled_at TIMESTAMPTZ NOT NULL,
+    claimed_at   TIMESTAMPTZ NOT NULL DEFAULT now(),
+    completed_at TIMESTAMPTZ,
+    -- `claimed`  — taken, dispatch not yet known to have finished. A row stuck
+    --              here is a crash mid-fire and is safe to retry.
+    -- `fired`    — dispatch completed; never fire this slot again.
+    -- `failed`   — dispatch returned an error. Terminal: the clock has already
+    --              moved on, and silently retrying a failing action every tick
+    --              is how a broken routine becomes a denial-of-service.
+    status       TEXT        NOT NULL DEFAULT 'claimed'
+                             CHECK (status IN ('claimed', 'fired', 'failed')),
+    error        TEXT,
+    PRIMARY KEY (routine_id, scheduled_at)
+);
+
+-- The reaper's query: rows still `claimed` past a grace period are crashes.
+CREATE INDEX routine_fires_stuck_idx
+    ON routine_fires (status, claimed_at)
+    WHERE status = 'claimed';