feat(memory_tree/jobs): JobOutcome::Defer + mark_deferred (#1256) (#1345)

src/openhuman/memory/tree/jobs/handlers/mod.rs

@@ -2,8 +2,12 @@
 //!
 //! Each handler parses its payload from `Job::payload_json`, performs its
 //! side effects (DB writes, LLM calls, follow-up enqueues), and returns
-//! `Ok(())` on success or an `anyhow::Error` on retryable failure.
-//! [`handle_job`] fans out to the handler matching the row's `kind`.
+//! `Ok(JobOutcome::Done)` on success or an `anyhow::Error` on retryable
+//! failure. A handler may also return `Ok(JobOutcome::Defer { … })` to
+//! re-queue the job with a wake-up time without burning the failure
+//! budget — useful for transient blockers like cloud rate limits or a
+//! warming-up model. [`handle_job`] fans out to the handler matching the
+//! row's `kind`.
 
 use anyhow::{Context, Result};
 
@@ -14,7 +18,7 @@ use crate::openhuman::memory::tree::content_store::{
 use crate::openhuman::memory::tree::jobs::store;
 use crate::openhuman::memory::tree::jobs::types::{
     AppendBufferPayload, AppendTarget, DigestDailyPayload, ExtractChunkPayload, FlushStalePayload,
-    Job, JobKind, NewJob, NodeRef, SealPayload, TopicRoutePayload,
+    Job, JobKind, JobOutcome, NewJob, NodeRef, SealPayload, TopicRoutePayload,
 };
 use crate::openhuman::memory::tree::score;
 use crate::openhuman::memory::tree::score::embed::{build_embedder_from_config, pack_checked};
@@ -28,7 +32,12 @@ use crate::openhuman::memory::tree::tree_source::{
 use crate::openhuman::memory::tree::tree_topic::curator;
 
 /// Dispatch a claimed job to the matching per-kind handler.
-pub async fn handle_job(config: &Config, job: &Job) -> Result<()> {
+///
+/// Existing handlers all return `Ok(JobOutcome::Done)` on success. The
+/// `Defer` outcome is wired through the worker but not yet emitted by any
+/// in-tree handler — consumers (cloud rate limiter, triage tiered
+/// fallback, embed warmup) land in follow-up issues.
+pub async fn handle_job(config: &Config, job: &Job) -> Result<JobOutcome> {
     match job.kind {
         JobKind::ExtractChunk => handle_extract(config, job).await,
         JobKind::AppendBuffer => handle_append_buffer(config, job).await,
@@ -39,7 +48,7 @@ pub async fn handle_job(config: &Config, job: &Job) -> Result<()> {
     }
 }
 
-async fn handle_extract(config: &Config, job: &Job) -> Result<()> {
+async fn handle_extract(config: &Config, job: &Job) -> Result<JobOutcome> {
     let payload: ExtractChunkPayload =
         serde_json::from_str(&job.payload_json).context("parse ExtractChunk payload")?;
     let Some(chunk) = chunk_store::get_chunk(config, &payload.chunk_id)? else {
@@ -47,7 +56,7 @@ async fn handle_extract(config: &Config, job: &Job) -> Result<()> {
             "[memory_tree::jobs] extract chunk missing chunk_id={}",
             payload.chunk_id
         );
-        return Ok(());
+        return Ok(JobOutcome::Done);
     };
 
     // Read the full body from disk (the `content` column in SQLite holds a
@@ -204,10 +213,10 @@ async fn handle_extract(config: &Config, job: &Job) -> Result<()> {
         super::worker::wake_workers();
     }
 
-    Ok(())
+    Ok(JobOutcome::Done)
 }
 
-async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> {
+async fn handle_append_buffer(config: &Config, job: &Job) -> Result<JobOutcome> {
     use crate::openhuman::memory::tree::tree_source::bucket_seal::should_seal;
     use crate::openhuman::memory::tree::tree_source::store as src_store;
 
@@ -221,7 +230,7 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> {
         NodeRef::Leaf { chunk_id } => {
             let Some(chunk) = chunk_store::get_chunk(config, chunk_id)? else {
                 log::warn!("[memory_tree::jobs] append_buffer chunk missing chunk_id={chunk_id}");
-                return Ok(());
+                return Ok(JobOutcome::Done);
             };
             let score_row = score_store::get_score(config, &chunk.id)?
                 .ok_or_else(|| anyhow::anyhow!("missing score row for chunk {}", chunk.id))?;
@@ -247,7 +256,7 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> {
                 log::warn!(
                     "[memory_tree::jobs] append_buffer summary missing summary_id={summary_id}"
                 );
-                return Ok(());
+                return Ok(JobOutcome::Done);
             };
             // Read the full body from disk — `summary.content` is a ≤500-char
             // preview after the MD-on-disk migration. The summariser receives
@@ -282,7 +291,7 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> {
         // topic_route and this append). Drop on the floor — the
         // topic_route was advisory and the source-tree path already
         // ran for this leaf.
-        return Ok(());
+        return Ok(JobOutcome::Done);
     };
 
     let is_source_target = matches!(payload.target, AppendTarget::Source { .. });
@@ -342,10 +351,10 @@ async fn handle_append_buffer(config: &Config, job: &Job) -> Result<()> {
     if did_enqueue_seal {
         super::worker::wake_workers();
     }
-    Ok(())
+    Ok(JobOutcome::Done)
 }
 
-async fn handle_seal(config: &Config, job: &Job) -> Result<()> {
+async fn handle_seal(config: &Config, job: &Job) -> Result<JobOutcome> {
     use crate::openhuman::memory::tree::tree_source::bucket_seal::{seal_one_level, should_seal};
     use crate::openhuman::memory::tree::tree_source::store as src_store;
     use crate::openhuman::memory::tree::tree_source::types::TreeKind;
@@ -357,7 +366,7 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> {
             "[memory_tree::jobs] seal tree missing tree_id={}",
             payload.tree_id
         );
-        return Ok(());
+        return Ok(JobOutcome::Done);
     };
 
     // Seal exactly one level. Parents only get sealed via a follow-up job
@@ -371,7 +380,7 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> {
             tree.id,
             payload.level
         );
-        return Ok(());
+        return Ok(JobOutcome::Done);
     }
     if !forced && !should_seal(&buf) {
         // Another job sealed this level out from under us (or the buffer
@@ -382,7 +391,7 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> {
             payload.level,
             buf.token_sum
         );
-        return Ok(());
+        return Ok(JobOutcome::Done);
     }
 
     // Pick the labeling strategy for this tree kind. Source trees mint
@@ -417,10 +426,10 @@ async fn handle_seal(config: &Config, job: &Job) -> Result<()> {
     }
 
     super::worker::wake_workers();
-    Ok(())
+    Ok(JobOutcome::Done)
 }
 
-async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> {
+async fn handle_topic_route(config: &Config, job: &Job) -> Result<JobOutcome> {
     let payload: TopicRoutePayload =
         serde_json::from_str(&job.payload_json).context("parse TopicRoute payload")?;
 
@@ -431,7 +440,7 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> {
         NodeRef::Leaf { chunk_id } => {
             if chunk_store::get_chunk(config, chunk_id)?.is_none() {
                 log::warn!("[memory_tree::jobs] topic_route chunk missing chunk_id={chunk_id}");
-                return Ok(());
+                return Ok(JobOutcome::Done);
             }
             chunk_id.clone()
         }
@@ -442,7 +451,7 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> {
                 log::warn!(
                     "[memory_tree::jobs] topic_route summary missing summary_id={summary_id}"
                 );
-                return Ok(());
+                return Ok(JobOutcome::Done);
             }
             summary_id.clone()
         }
@@ -451,7 +460,7 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> {
     let entity_ids = score_store::list_entity_ids_for_node(config, &node_id)?;
     if entity_ids.is_empty() {
         log::debug!("[memory_tree::jobs] topic_route no entities for node_id={node_id} — skipping");
-        return Ok(());
+        return Ok(JobOutcome::Done);
     }
 
     let summariser = build_summariser(config);
@@ -473,10 +482,10 @@ async fn handle_topic_route(config: &Config, job: &Job) -> Result<()> {
             }
         }
     }
-    Ok(())
+    Ok(JobOutcome::Done)
 }
 
-async fn handle_digest_daily(config: &Config, job: &Job) -> Result<()> {
+async fn handle_digest_daily(config: &Config, job: &Job) -> Result<JobOutcome> {
     let payload: DigestDailyPayload =
         serde_json::from_str(&job.payload_json).context("parse DigestDaily payload")?;
     let day = chrono::NaiveDate::parse_from_str(&payload.date_iso, "%Y-%m-%d")
@@ -491,10 +500,10 @@ async fn handle_digest_daily(config: &Config, job: &Job) -> Result<()> {
             log::debug!("[memory_tree::jobs] digest skipped existing_id={existing_id}");
         }
     }
-    Ok(())
+    Ok(JobOutcome::Done)
 }
 
-async fn handle_flush_stale(config: &Config, job: &Job) -> Result<()> {
+async fn handle_flush_stale(config: &Config, job: &Job) -> Result<JobOutcome> {
     let payload: FlushStalePayload =
         serde_json::from_str(&job.payload_json).context("parse FlushStale payload")?;
     let age_secs = payload
@@ -513,7 +522,7 @@ async fn handle_flush_stale(config: &Config, job: &Job) -> Result<()> {
             super::worker::wake_workers();
         }
     }
-    Ok(())
+    Ok(JobOutcome::Done)
 }
 
 #[cfg(test)]

src/openhuman/memory/tree/jobs/mod.rs

@@ -34,12 +34,12 @@ mod worker;
 
 pub use scheduler::{backfill_missing_digests, trigger_digest};
 pub use store::{
-    claim_next, count_by_status, count_total, enqueue, enqueue_tx, get_job, mark_done, mark_failed,
-    recover_stale_locks, DEFAULT_LOCK_DURATION_MS,
+    claim_next, count_by_status, count_total, enqueue, enqueue_tx, get_job, mark_deferred,
+    mark_done, mark_failed, recover_stale_locks, DEFAULT_LOCK_DURATION_MS,
 };
 pub use testing::drain_until_idle;
 pub use types::{
     AppendBufferPayload, AppendTarget, DigestDailyPayload, ExtractChunkPayload, FlushStalePayload,
-    Job, JobKind, JobStatus, NewJob, NodeRef, SealPayload, TopicRoutePayload,
+    Job, JobKind, JobOutcome, JobStatus, NewJob, NodeRef, SealPayload, TopicRoutePayload,
 };
 pub use worker::{start, wake_workers};

src/openhuman/memory/tree/jobs/store.rs

@@ -253,6 +253,57 @@ pub fn mark_failed(config: &Config, job: &Job, error: &str) -> Result<()> {
     })
 }
 
+/// Mark a claimed job as deferred: put it back to `ready` with
+/// `available_at_ms = until_ms` so [`claim_next`] will re-pick it once the
+/// wake-up time has passed. The handler ran successfully but chose not to
+/// make progress (cloud rate-limited, dependency unavailable, model
+/// warming up), so this path **does not** burn the failure budget — the
+/// `attempts` bump that [`claim_next`] applied at claim time is reverted.
+///
+/// `reason` is recorded in `last_error` for visibility and `started_at_ms`
+/// is cleared (mirroring the retry branch of [`mark_failed`]) so the next
+/// claim stamps a fresh start time.
+///
+/// Like [`mark_done`] / [`mark_failed`], the UPDATE is gated on the
+/// claim-token (`attempts` + `started_at_ms`) so a stale lessee's
+/// settlement is a silent no-op rather than clobbering an active lessee's
+/// row.
+pub fn mark_deferred(config: &Config, job: &Job, until_ms: i64, reason: &str) -> Result<()> {
+    let job_id = &job.id;
+    let claim_attempts = job.attempts as i64;
+    let pre_claim_attempts = claim_attempts.saturating_sub(1);
+    let claim_started_at = job.started_at_ms;
+    with_connection(config, |conn| {
+        let n = conn.execute(
+            "UPDATE mem_tree_jobs
+                SET status = 'ready',
+                    attempts = ?1,
+                    available_at_ms = ?2,
+                    locked_until_ms = NULL,
+                    started_at_ms = NULL,
+                    last_error = ?3
+              WHERE id = ?4
+                AND attempts = ?5
+                AND started_at_ms IS ?6",
+            params![
+                pre_claim_attempts,
+                until_ms,
+                reason,
+                job_id,
+                claim_attempts,
+                claim_started_at,
+            ],
+        )?;
+        if n == 0 {
+            log::warn!(
+                "[memory_tree::jobs] mark_deferred id={job_id} was a no-op \
+                 (stale lease: attempts={claim_attempts} started_at_ms={claim_started_at:?})"
+            );
+        }
+        Ok(())
+    })
+}
+
 /// Flip any `running` row whose `locked_until_ms` has expired back to
 /// `ready`. Called once at worker startup so a process crash mid-job
 /// doesn't leave work stranded. Returns the number of rows recovered.
@@ -627,4 +678,205 @@ mod tests {
         assert_eq!(count_by_status(&cfg, JobStatus::Done).unwrap(), 1);
         assert_eq!(count_by_status(&cfg, JobStatus::Ready).unwrap(), 2);
     }
+
+    /// Defer must NOT advance the failure-attempt counter. After a claim
+    /// (which bumps attempts to 1) and a deferral, the next claim should
+    /// see attempts==2 (just the second claim's bump) — proving the
+    /// transient deferral did not burn a slot from the row's failure
+    /// budget.
+    #[test]
+    fn mark_deferred_does_not_increment_attempts() {
+        let (_tmp, cfg) = test_config();
+        let payload = ExtractChunkPayload {
+            chunk_id: "c-defer-1".into(),
+        };
+        let nj = NewJob::extract_chunk(&payload).unwrap();
+        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");
+
+        let claim1 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        assert_eq!(claim1.id, id);
+        assert_eq!(claim1.attempts, 1, "first claim bumps attempts to 1");
+
+        // Defer with a wake-up time already in the past so the next
+        // claim_next is immediately eligible without sleeping.
+        mark_deferred(&cfg, &claim1, 0, "rate_limited").unwrap();
+
+        let row = get_job(&cfg, &id).unwrap().unwrap();
+        assert_eq!(row.status, JobStatus::Ready);
+        assert_eq!(
+            row.attempts, 0,
+            "Defer must revert the claim's attempts bump (back to pre-claim 0)"
+        );
+        assert_eq!(row.last_error.as_deref(), Some("rate_limited"));
+        assert_eq!(row.available_at_ms, 0);
+        assert!(row.locked_until_ms.is_none());
+        assert!(row.started_at_ms.is_none());
+
+        let claim2 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        assert_eq!(claim2.id, id);
+        assert_eq!(
+            claim2.attempts, 1,
+            "second claim bumps attempts to 1 again (Defer didn't count)"
+        );
+    }
+
+    /// A row deferred to a future `until_ms` must not be claimable until
+    /// the system clock crosses that threshold.
+    #[test]
+    fn deferred_row_not_claimable_until_until_ms() {
+        let (_tmp, cfg) = test_config();
+        let payload = ExtractChunkPayload {
+            chunk_id: "c-defer-2".into(),
+        };
+        let nj = NewJob::extract_chunk(&payload).unwrap();
+        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");
+
+        let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        let future_ms = Utc::now().timestamp_millis() + 60_000;
+        mark_deferred(&cfg, &claimed, future_ms, "warming_up").unwrap();
+
+        // Right now: not yet eligible.
+        let none = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap();
+        assert!(
+            none.is_none(),
+            "deferred row must not be claimable before until_ms"
+        );
+
+        // Force the row available again (proxy for "wall clock advanced
+        // past until_ms") and confirm claim_next picks it up.
+        with_connection(&cfg, |c| {
+            c.execute(
+                "UPDATE mem_tree_jobs SET available_at_ms = 0 WHERE id = ?1",
+                params![id],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+        let claim2 = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        assert_eq!(claim2.id, id);
+        assert_eq!(claim2.attempts, 1, "Defer left attempts at pre-claim 0");
+    }
+
+    /// Three rows with three different terminal verbs: Done, Failed (with
+    /// retry left, so it bounces back to ready with bumped attempts), and
+    /// Defer. After processing, each row's terminal state must reflect its
+    /// settlement verb. Critically, the Defer row keeps its pre-claim
+    /// `attempts` value while the failed row bumps.
+    #[test]
+    fn mixed_outcomes_stress() {
+        let (_tmp, cfg) = test_config();
+        let p_a = ExtractChunkPayload {
+            chunk_id: "c-mix-a".into(),
+        };
+        let p_b = ExtractChunkPayload {
+            chunk_id: "c-mix-b".into(),
+        };
+        let p_c = ExtractChunkPayload {
+            chunk_id: "c-mix-c".into(),
+        };
+        let id_a = enqueue(&cfg, &NewJob::extract_chunk(&p_a).unwrap())
+            .unwrap()
+            .unwrap();
+        let id_b = enqueue(&cfg, &NewJob::extract_chunk(&p_b).unwrap())
+            .unwrap()
+            .unwrap();
+        let id_c = enqueue(&cfg, &NewJob::extract_chunk(&p_c).unwrap())
+            .unwrap()
+            .unwrap();
+
+        // Claim the three rows in turn (FIFO within same kind/priority).
+        let claim_a = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        let claim_b = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        let claim_c = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        // Sanity: three distinct ids covering A/B/C.
+        let mut got: Vec<&str> = vec![&claim_a.id, &claim_b.id, &claim_c.id];
+        got.sort();
+        let mut want = vec![id_a.as_str(), id_b.as_str(), id_c.as_str()];
+        want.sort();
+        assert_eq!(got, want);
+
+        let until_ms = Utc::now().timestamp_millis() + 30_000;
+
+        // Settle: A=done, B=err (retry path), C=defer.
+        mark_done(&cfg, &claim_a).unwrap();
+        mark_failed(&cfg, &claim_b, "transient_error").unwrap();
+        mark_deferred(&cfg, &claim_c, until_ms, "rate_limited").unwrap();
+
+        // A: terminal done.
+        let row_a = get_job(&cfg, &id_a).unwrap().unwrap();
+        assert_eq!(row_a.status, JobStatus::Done);
+        assert!(row_a.completed_at_ms.is_some());
+
+        // B: retry path — back to ready with bumped attempts (1) and a
+        // future available_at_ms from exponential backoff.
+        let row_b = get_job(&cfg, &id_b).unwrap().unwrap();
+        assert_eq!(row_b.status, JobStatus::Ready);
+        assert_eq!(
+            row_b.attempts, 1,
+            "Err settlement keeps the claim's attempts bump"
+        );
+        assert!(row_b.available_at_ms > Utc::now().timestamp_millis());
+        assert_eq!(row_b.last_error.as_deref(), Some("transient_error"));
+
+        // C: deferred — back to ready with attempts reverted and
+        // available_at_ms == until_ms.
+        let row_c = get_job(&cfg, &id_c).unwrap().unwrap();
+        assert_eq!(row_c.status, JobStatus::Ready);
+        assert_eq!(
+            row_c.attempts, 0,
+            "Defer settlement reverts the claim's attempts bump"
+        );
+        assert_eq!(row_c.available_at_ms, until_ms);
+        assert_eq!(row_c.last_error.as_deref(), Some("rate_limited"));
+        assert!(row_c.locked_until_ms.is_none());
+        assert!(row_c.started_at_ms.is_none());
+    }
+
+    /// Stale-worker `mark_deferred` must be a no-op — same lease-token
+    /// gating as `mark_done` / `mark_failed`. After Worker A's claim
+    /// expires and Worker B re-claims, Worker A's stale Defer must not
+    /// clobber B's running row.
+    #[test]
+    fn mark_deferred_stale_lease_is_noop() {
+        let (_tmp, cfg) = test_config();
+        let payload = ExtractChunkPayload {
+            chunk_id: "c-defer-stale".into(),
+        };
+        let nj = NewJob::extract_chunk(&payload).unwrap();
+        let id = enqueue(&cfg, &nj).unwrap().expect("inserted");
+
+        // Worker A claims with already-expired lock.
+        let worker_a_job = claim_next(&cfg, -1).unwrap().unwrap();
+        assert_eq!(worker_a_job.attempts, 1);
+
+        // Lease expires; Worker B re-claims.
+        let recovered = recover_stale_locks(&cfg).unwrap();
+        assert_eq!(recovered, 1);
+        let worker_b_job = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+        assert_eq!(worker_b_job.attempts, 2);
+
+        // Worker A (stale) tries to defer using its old snapshot — must
+        // be a silent no-op.
+        let stale_until_ms = Utc::now().timestamp_millis() + 999_000;
+        mark_deferred(&cfg, &worker_a_job, stale_until_ms, "stale_defer").unwrap();
+
+        // Worker B's row must be untouched: still 'running' with
+        // attempts=2 and no stale_defer side effect.
+        let row = get_job(&cfg, &id).unwrap().unwrap();
+        assert_eq!(
+            row.status,
+            JobStatus::Running,
+            "stale mark_deferred must not clobber Worker B's running row"
+        );
+        assert_eq!(row.attempts, 2);
+        assert_ne!(
+            row.last_error.as_deref(),
+            Some("stale_defer"),
+            "stale defer reason must not be persisted"
+        );
+        assert_ne!(
+            row.available_at_ms, stale_until_ms,
+            "stale until_ms must not be persisted"
+        );
+    }
 }

src/openhuman/memory/tree/jobs/types.rs

@@ -65,6 +65,28 @@ impl JobKind {
     }
 }
 
+/// Outcome of a successful handler run. Workers translate this into a
+/// queue settlement: `Done` finalises the row, while `Defer` puts it back
+/// to `ready` with `available_at_ms = until_ms` and **does not** count
+/// toward the failure-attempt budget.
+///
+/// `Defer` exists so a handler that is transiently unable to make
+/// progress (cloud rate-limited, dependency unavailable, model warming
+/// up) can re-queue its job with a wake-up time without marking it
+/// failed. Handlers should still surface real errors via `Err(_)` — that
+/// path runs the existing exponential-backoff retry logic which **does**
+/// burn the failure budget.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum JobOutcome {
+    /// Handler ran to completion. Row is settled as `done`.
+    Done,
+    /// Handler chose not to make progress yet. Row is rescheduled to
+    /// `available_at_ms = until_ms` (UTC milliseconds) with `attempts`
+    /// reverted to its pre-claim value so the failure budget is not
+    /// touched. `reason` is recorded in `last_error` for visibility.
+    Defer { until_ms: i64, reason: String },
+}
+
 /// Lifecycle states persisted on `mem_tree_jobs.status`. Workers transition
 /// `ready → running → done|failed`. `Cancelled` is reserved for explicit
 /// admin actions (none surfaced yet).

src/openhuman/memory/tree/jobs/worker.rs

@@ -17,8 +17,10 @@ use tokio::sync::Notify;
 use crate::openhuman::config::Config;
 use crate::openhuman::memory::tree::jobs::handlers;
 use crate::openhuman::memory::tree::jobs::store::{
-    claim_next, mark_done, mark_failed, recover_stale_locks, DEFAULT_LOCK_DURATION_MS,
+    claim_next, mark_deferred, mark_done, mark_failed, recover_stale_locks,
+    DEFAULT_LOCK_DURATION_MS,
 };
+use crate::openhuman::memory::tree::jobs::types::JobOutcome;
 
 /// Number of concurrent job-worker tasks. Each worker claims one job
 /// at a time via `claim_next` (atomic UPDATE under SQLite WAL with
@@ -148,9 +150,29 @@ pub async fn run_once(config: &Config) -> Result<bool> {
     drop(llm_permit);
 
     match result {
-        Ok(()) => {
+        Ok(JobOutcome::Done) => {
+            log::debug!(
+                "[memory_tree::jobs] done id={} kind={}",
+                job.id,
+                job.kind.as_str()
+            );
             mark_done(config, &job)?;
         }
+        Ok(JobOutcome::Defer { until_ms, reason }) => {
+            // Defer is normal operation (transient blocker, e.g. rate
+            // limit) — log at info, not warn — and do NOT count this
+            // claim toward the failure-attempt budget. `mark_deferred`
+            // reverts the bump applied by `claim_next` so the row's
+            // attempts counter stays where it was before this claim.
+            log::info!(
+                "[memory_tree::jobs] deferred id={} kind={} until_ms={} reason={}",
+                job.id,
+                job.kind.as_str(),
+                until_ms,
+                reason
+            );
+            mark_deferred(config, &job, until_ms, &reason)?;
+        }
         Err(err) => {
             // Preserve the full anyhow cause chain in the persisted
             // last_error so a reader of mem_tree_jobs can see the root