diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index 8d0e31f..565016b 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -3,10 +3,24 @@
 //! a [`SessionDigest`] of per-facet, prescriptive observations.
 //!
 //! Follows the `LlmEntityExtractor` pattern: a strict-JSON instruction with the
-//! schema in the prompt, and a **soft fallback** — any failure (transport,
-//! malformed JSON, empty) skips the session by returning an empty digest rather
-//! than aborting the run. Oversized sessions are windowed and digested in parts,
-//! and the observations are concatenated.
+//! schema in the prompt. Failure handling is deliberately **not** a silent
+//! soft-fallback (that was the data-loss bug this module fixes):
+//! - a provider/transport failure returns `Err` so the caller leaves the whole
+//!   session non-committable and retries it next run;
+//! - a truncated/unparseable response — the output-token cap cutting the JSON
+//!   array short — is first *recovered* in-process by re-splitting the window
+//!   into smaller pieces whose responses fit under the cap, and only a piece that
+//!   still won't parse at the minimum size is dropped-and-counted (surfaced in
+//!   the run report, never retried forever — a deterministically-bad window must
+//!   not starve the queue);
+//! - a cleanly-parsed empty response commits normally.
+//!
+//! Oversized sessions are windowed and digested part-by-part; each window is
+//! digested independently, so one bad window never discards the clean siblings
+//! that already digested — their observations are accumulated and kept.
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
 
 use anyhow::Result;
 use serde::Deserialize;
@@ -19,8 +33,75 @@ use crate::memory::store::safety::sanitize_text;
 /// Max characters of evidence sent in a single digest call. Larger sessions are
 /// split into windows and digested part-by-part.
 const WINDOW_CHARS: usize = 12_000;
-/// Output-token cap for a digest response (one small JSON object).
-const DIGEST_MAX_OUTPUT_TOKENS: u32 = 4_096;
+/// Output-token cap for a digest response.
+///
+/// A window holds up to [`WINDOW_CHARS`] of evidence, and a dense window of
+/// corrections/directives (e.g. Codex sessions) can distil into *many* facet
+/// observations, each carrying an `observation` string and a supporting `quote`.
+/// The response is a single JSON object, but its size scales with the observation
+/// count — not with the "one small object" a 4 K cap assumed. At 4 K the model
+/// ran out of output mid-array and emitted a well-formed prefix with no closing
+/// `]`, which `parse_digest` then rejected (`EOF while parsing a list`, observed
+/// at column ~15–19 K); B3 stops that from silently dropping the window, and this
+/// larger cap stops it from happening in the first place. 16 K comfortably covers
+/// the observed truncations while still bounding a runaway generation. Coupled to
+/// [`WINDOW_CHARS`]: raising the input window raises the observations a window can
+/// yield, so the two move together.
+const DIGEST_MAX_OUTPUT_TOKENS: u32 = 16_384;
+/// The smallest piece recovery will re-digest. A piece is split only while its
+/// half stays ≥ this (i.e. while the piece is ≥ ~2× this), so pieces of roughly
+/// this size *are* re-digested but are not split again: one that still won't parse
+/// at this size is treated as genuinely broken (not merely output-capped) and
+/// dropped-with-a-count. This floor is what guarantees recovery terminates.
+const MIN_WINDOW_CHARS: usize = 1_500;
+/// Max times recovery halves a truncated window before giving up on a sub-window.
+/// `12_000 → 6_000 → 3_000 → 1_500` reaches [`MIN_WINDOW_CHARS`], an ~8× cut in
+/// the per-call output that overran the cap — deep enough to recover real
+/// truncations, bounded so a deterministically-bad window can't loop.
+const MAX_RESPLIT_DEPTH: usize = 3;
+
+/// A shared, concurrency-safe ceiling on provider calls for one run.
+///
+/// `digest_window` consumes one permit per `chat_for_json`, so **windowing and
+/// truncation-recovery re-tries alike** count against the configured
+/// `max_llm_calls`: a recovery tree (up to ~15 calls per truncated window across
+/// many windows) can no longer overrun the run's call budget. Cheap to clone and
+/// share across the concurrently-digested sessions — it wraps an
+/// `Arc<AtomicUsize>` of the remaining calls.
+#[derive(Clone)]
+pub struct CallBudget(Arc<AtomicUsize>);
+
+impl CallBudget {
+    /// A budget of `max_calls` provider calls for the run.
+    pub fn new(max_calls: usize) -> Self {
+        Self(Arc::new(AtomicUsize::new(max_calls)))
+    }
+
+    /// A budget that never runs out — for callers/tests that don't gate calls.
+    pub fn unlimited() -> Self {
+        Self::new(usize::MAX)
+    }
+
+    /// Reserve one call, returning `false` once the budget is spent. Lock-free
+    /// and correct under the concurrent `buffered` digest stream.
+    fn try_acquire(&self) -> bool {
+        self.0
+            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1))
+            .is_ok()
+    }
+
+    /// Calls still available in this run.
+    pub fn remaining(&self) -> usize {
+        self.0.load(Ordering::SeqCst)
+    }
+}
+
+/// True when `err` (from [`digest_session`]) is the run's provider-call budget
+/// being hit mid-session — a clean checkpoint to resume next run, not a failure.
+pub fn is_budget_exhausted(err: &anyhow::Error) -> bool {
+    err.downcast_ref::<DigestError>()
+        .is_some_and(|e| matches!(e, DigestError::BudgetExhausted))
+}
 
 /// The strict-JSON system prompt: schema + extraction contract.
 fn system_prompt() -> String {
@@ -88,43 +169,181 @@ fn windows(session: &RawSession) -> Vec<String> {
     out
 }
 
-/// Digest one session into a [`SessionDigest`] via the chat provider.
+/// Outcome of digesting one session: the observations plus how many windows were
+/// **dropped** because they stayed unparseable even after truncation-recovery
+/// re-splitting.
+///
+/// A non-zero `windows_lost` is near-deterministic at temperature `0.0` (a
+/// heuristic — hosted providers aren't bit-exact), so retrying it in isolation
+/// only re-burns budget while starving newer sessions. The pipeline therefore
+/// commits such a session's cursor when the run produced observations elsewhere
+/// (a localized failure) but *withholds* it when the whole run yielded nothing
+/// (a systemic failure worth retrying) — see the pipeline's `systemic_digest_failure`
+/// handling. The count is surfaced in the run report so the drop is never silent.
+/// Only a *provider* failure (transport/budget/auth) is a non-committable `Err`
+/// from [`digest_session`] — that is transient and worth retrying the whole session.
+#[derive(Debug, Clone)]
+pub struct SessionOutcome {
+    /// The observations distilled from every digested (or recovered) window.
+    pub digest: SessionDigest,
+    /// Recovery leaf pieces dropped after re-splitting still could not parse them
+    /// (data intentionally skipped to keep the queue moving). Counts *pieces*, not
+    /// input windows: one fully-unparseable window contributes one per irreducible
+    /// leaf, so a window bottoming out at the private `MAX_RESPLIT_DEPTH` adds
+    /// `2^MAX_RESPLIT_DEPTH`.
+    pub windows_lost: usize,
+}
+
+/// Digest one session into a [`SessionOutcome`] via the chat provider.
 ///
-/// Distinguishes two failure modes so the pipeline can checkpoint correctly:
-/// - **Provider/transport failure** (a `chat_for_json` error — budget exhausted,
-///   401/403, transport) → returns `Err`. The caller must NOT commit the
-///   session's cursor, so the evidence is re-attempted on the next run.
-/// - **Model produced no usable output** (a valid call whose response wasn't
-///   parseable JSON, or yielded zero observations) → returns `Ok` with an empty
-///   digest. Re-running would reproduce it, so the cursor IS committed.
+/// Windows are digested independently and their observations accumulated, so a
+/// bad window never discards the clean siblings that already digested. Outcomes:
+/// - **Provider failure** (a `chat_for_json` error — budget exhausted, 401/403,
+///   transport) → returns `Err`. The caller must NOT commit the session's cursor,
+///   so the whole session is re-attempted next run. This is transient.
+/// - **Truncated/unparseable window** → recovered in-process by re-splitting (see
+///   the private `digest_window_recovering`); a piece that still won't parse at the
+///   minimum size is dropped and tallied in [`SessionOutcome::windows_lost`]. Such a
+///   session's cursor is committed only when the run produced observations
+///   elsewhere (a localized permanent failure — near-deterministic, so retrying is
+///   waste); if the *whole run* yielded nothing despite dropped windows the pipeline
+///   withholds the cursor and flags `systemic_digest_failure` for retry (never a
+///   silent skip). See the pipeline's deferred-commit handling.
+/// - **Genuinely empty digest** (`{"observations":[]}`) → `Ok` with an empty
+///   digest and `windows_lost = 0`. Re-running reproduces it, so the cursor commits.
+/// - **Call budget exhausted** mid-session → returns `Err` (classify with
+///   [`is_budget_exhausted`]); the session is left non-committable and resumes
+///   next run. A clean checkpoint, not a failure.
 pub async fn digest_session(
     provider: &dyn ChatProvider,
     session: &RawSession,
-) -> Result<SessionDigest> {
+    budget: &CallBudget,
+) -> Result<SessionOutcome> {
     if session.is_empty() {
-        return Ok(SessionDigest::empty(session.source.clone()));
+        return Ok(SessionOutcome {
+            digest: SessionDigest::empty(session.source.clone()),
+            windows_lost: 0,
+        });
     }
     let mut observations: Vec<DigestObservation> = Vec::new();
+    let mut windows_lost = 0usize;
     for window in windows(session) {
-        // A hard provider failure bubbles up (the whole session is retried next
-        // run); a soft parse failure yields an empty window and is tolerated.
-        let obs = digest_window(provider, session, &window).await?;
+        // Each window recovers from truncation on its own and never aborts its
+        // siblings; a hard provider failure or an exhausted call budget bubbles
+        // up as `Err` (the whole session is then retried next run, nothing
+        // committed).
+        let (obs, lost) = digest_window_recovering(provider, session, &window, budget)
+            .await
+            .map_err(anyhow::Error::new)?;
         observations.extend(obs);
+        windows_lost += lost;
     }
-    Ok(SessionDigest {
-        source: session.source.clone(),
-        observations,
+    Ok(SessionOutcome {
+        digest: SessionDigest {
+            source: session.source.clone(),
+            observations,
+        },
+        windows_lost,
     })
 }
 
-/// One window → observations. A `chat_for_json` failure bubbles up as `Err`
-/// (hard, non-committable); an unparseable-but-received response degrades to an
-/// empty window (`Ok(vec![])`) since retrying reproduces it.
+/// Why a window could not be digested into observations.
+#[derive(Debug, thiserror::Error)]
+enum DigestError {
+    /// The provider call itself failed (budget/auth/transport). The response was
+    /// never received — transient, so the caller retries the whole session.
+    #[error("digest provider call failed: {0:#}")]
+    Provider(#[source] anyhow::Error),
+    /// A response was received but could not be parsed — typically a JSON array
+    /// truncated at the output-token cap (a well-formed prefix with no closing
+    /// `]`). Recoverable by re-splitting the window into smaller pieces.
+    #[error("digest response unparseable (likely truncated at the output cap): {0:#}")]
+    Unparseable(#[source] anyhow::Error),
+    /// The run's provider-call budget (`max_llm_calls`) was spent before this
+    /// window could be digested. The session is left non-committable so it
+    /// resumes next run — a clean checkpoint, not a failure.
+    #[error("digest call budget exhausted")]
+    BudgetExhausted,
+}
+
+/// Digest one window, recovering from output-cap truncation by re-splitting.
+///
+/// The first attempt digests the whole window. If the response is unparseable —
+/// the signature of a JSON array cut off at the output-token cap — the window is
+/// halved and each half retried, which shrinks the per-call output below the cap
+/// and *recovers* the observations instead of discarding them. Halving is bounded
+/// by [`MAX_RESPLIT_DEPTH`] and [`MIN_WINDOW_CHARS`]; a sub-window still
+/// unparseable at the floor is genuinely broken (not merely truncated), so it is
+/// dropped and counted in the returned `lost` tally rather than retried forever —
+/// that is what stops a single deterministically-bad window from starving the
+/// queue on every run.
+///
+/// A provider/transport failure is *not* recovered here: it returns `Err` so the
+/// caller leaves the whole session non-committable and retries it next run.
+async fn digest_window_recovering(
+    provider: &dyn ChatProvider,
+    session: &RawSession,
+    window: &str,
+    budget: &CallBudget,
+) -> Result<(Vec<DigestObservation>, usize), DigestError> {
+    let mut observations = Vec::new();
+    let mut lost = 0usize;
+    // Work stack of (text, splits_remaining). Order does not matter — evidence
+    // ids are content-addressed, so folding is insensitive to window order.
+    let mut stack: Vec<(String, usize)> = vec![(window.to_string(), MAX_RESPLIT_DEPTH)];
+    while let Some((piece, splits_remaining)) = stack.pop() {
+        match digest_window(provider, session, &piece, budget).await {
+            Ok(obs) => observations.extend(obs),
+            // Transient / budget checkpoint — abort recovery and let the whole
+            // session retry next run (nothing committed).
+            Err(DigestError::Provider(e)) => return Err(DigestError::Provider(e)),
+            Err(DigestError::BudgetExhausted) => return Err(DigestError::BudgetExhausted),
+            Err(DigestError::Unparseable(e)) => {
+                let target = piece.chars().count() / 2;
+                let parts = if splits_remaining > 0 && target >= MIN_WINDOW_CHARS {
+                    split_window(&piece, target)
+                } else {
+                    Vec::new()
+                };
+                if parts.len() > 1 {
+                    for part in parts {
+                        stack.push((part, splits_remaining - 1));
+                    }
+                } else {
+                    // Irreducible and still unparseable: drop it, but loudly and
+                    // counted — a deterministic failure must make progress, not
+                    // hold the cursor and retry forever.
+                    log::warn!(
+                        "[persona] digest window unrecoverable after re-splitting for {} ({}); \
+                         dropping {} chars and committing the rest so the queue is not starved: {e:#}",
+                        session.source.kind.as_str(),
+                        session.source.session_id.as_deref().unwrap_or("?"),
+                        piece.chars().count(),
+                    );
+                    lost += 1;
+                }
+            }
+        }
+    }
+    Ok((observations, lost))
+}
+
+/// One window → observations.
+///
+/// Returns [`DigestError::Provider`] for a `chat_for_json` failure and
+/// [`DigestError::Unparseable`] for a truncated/unparseable response; a
+/// cleanly-parsed response with zero usable observations returns `Ok(vec![])`.
 async fn digest_window(
     provider: &dyn ChatProvider,
     session: &RawSession,
     window: &str,
-) -> Result<Vec<DigestObservation>> {
+    budget: &CallBudget,
+) -> Result<Vec<DigestObservation>, DigestError> {
+    // Reserve the call against the run budget *before* spending it, so windowing
+    // and every recovery re-try count toward `max_llm_calls`.
+    if !budget.try_acquire() {
+        return Err(DigestError::BudgetExhausted);
+    }
     let prompt = ChatPrompt {
         system: system_prompt(),
         user: user_prompt(session, window),
@@ -132,18 +351,19 @@ async fn digest_window(
         kind: "persona::digest",
         max_tokens: Some(DIGEST_MAX_OUTPUT_TOKENS),
     };
-    let raw = provider.chat_for_json(&prompt).await?;
-    let parsed: RawDigest = match parse_digest(&raw) {
-        Ok(p) => p,
-        Err(e) => {
-            log::warn!(
-                "[persona] digest parse failed for {} ({}): {e:#}",
-                session.source.kind.as_str(),
-                session.source.session_id.as_deref().unwrap_or("?")
-            );
-            return Ok(Vec::new());
-        }
-    };
+    let raw = provider
+        .chat_for_json(&prompt)
+        .await
+        .map_err(DigestError::Provider)?;
+    let parsed: RawDigest = parse_digest(&raw).map_err(|e| {
+        log::warn!(
+            "[persona] digest parse failed for {} ({}); attempting in-process \
+             re-split recovery before giving up: {e:#}",
+            session.source.kind.as_str(),
+            session.source.session_id.as_deref().unwrap_or("?")
+        );
+        DigestError::Unparseable(e)
+    })?;
     Ok(parsed
         .observations
         .into_iter()
@@ -151,6 +371,47 @@ async fn digest_window(
         .collect())
 }
 
+/// Split `text` into chunks of at most `target` **chars**, breaking on line
+/// boundaries; a single line longer than `target` is hard-split on char
+/// boundaries (UTF-8-safe). Used by truncation recovery to shrink a window whose
+/// digest overran the output-token cap. Char counts throughout (not byte lengths)
+/// so a multibyte-heavy window still halves to a genuinely smaller piece.
+fn split_window(text: &str, target: usize) -> Vec<String> {
+    let target = target.max(1);
+    let mut out = Vec::new();
+    let mut cur = String::new();
+    let mut cur_len = 0usize;
+    for line in text.split_inclusive('\n') {
+        let line_len = line.chars().count();
+        if cur_len > 0 && cur_len + line_len > target {
+            out.push(std::mem::take(&mut cur));
+            cur_len = 0;
+        }
+        if line_len > target {
+            // Oversized single line: hard-split on char boundaries.
+            let mut chunk = String::new();
+            let mut chunk_len = 0usize;
+            for ch in line.chars() {
+                if chunk_len >= target {
+                    out.push(std::mem::take(&mut chunk));
+                    chunk_len = 0;
+                }
+                chunk.push(ch);
+                chunk_len += 1;
+            }
+            cur.push_str(&chunk);
+            cur_len += chunk_len;
+        } else {
+            cur.push_str(line);
+            cur_len += line_len;
+        }
+    }
+    if !cur.trim().is_empty() {
+        out.push(cur);
+    }
+    out
+}
+
 /// Parse a digest response, tolerating models that wrap the JSON in prose or
 /// code fences by extracting the first `{...}` object.
 fn parse_digest(raw: &str) -> Result<RawDigest> {
diff --git a/src/memory/persona/distill_tests.rs b/src/memory/persona/distill_tests.rs
index 82541e9..bd2c1d1 100644
--- a/src/memory/persona/distill_tests.rs
+++ b/src/memory/persona/distill_tests.rs
@@ -22,6 +22,61 @@ impl ChatProvider for MockChat {
     }
 }
 
+/// A chat provider that truncates the JSON array (drops the closing `]`/`}`) when
+/// the prompt's evidence window is large, and returns a clean single-observation
+/// object once the window is small enough — modelling a real output-token cap
+/// that only trips on dense windows and clears once recovery re-splits them.
+struct SizeAwareChat {
+    /// Prompt-length threshold (chars) above which the response truncates.
+    threshold: usize,
+}
+
+#[async_trait]
+impl ChatProvider for SizeAwareChat {
+    fn name(&self) -> &str {
+        "size-aware"
+    }
+    async fn chat_for_json(&self, prompt: &ChatPrompt) -> anyhow::Result<String> {
+        if prompt.user.chars().count() > self.threshold {
+            // Well-formed prefix, no closing `]`/`}` — a hit output cap.
+            Ok(r#"{"observations":[
+                {"facet":"workflow","observation":"Commits small and often","quote":"commit","tier":"t2"}"#
+                .into())
+        } else {
+            Ok(r#"{"observations":[
+                {"facet":"stack","observation":"Uses Rust everywhere","quote":"cargo","tier":"t2"}
+            ]}"#
+            .into())
+        }
+    }
+}
+
+/// A chat provider that truncates the response iff the prompt contains `marker`,
+/// and returns a clean single-observation object otherwise — modelling a window
+/// that stays unparseable no matter how small it is re-split (the poison window).
+struct MarkerChat {
+    marker: &'static str,
+}
+
+#[async_trait]
+impl ChatProvider for MarkerChat {
+    fn name(&self) -> &str {
+        "marker"
+    }
+    async fn chat_for_json(&self, prompt: &ChatPrompt) -> anyhow::Result<String> {
+        if prompt.user.contains(self.marker) {
+            Ok(r#"{"observations":[
+                {"facet":"workflow","observation":"Commits small and often","quote":"commit","tier":"t2"}"#
+                .into())
+        } else {
+            Ok(r#"{"observations":[
+                {"facet":"stack","observation":"Uses Rust everywhere","quote":"cargo","tier":"t2"}
+            ]}"#
+            .into())
+        }
+    }
+}
+
 fn session_with(excerpts: &[(&str, EvidenceTier)]) -> RawSession {
     let src = EvidenceSource::new(PersonaSourceKind::ClaudeCode).with_scope("demo");
     let mut s = RawSession::new(src.clone());
@@ -37,6 +92,23 @@ fn session_with(excerpts: &[(&str, EvidenceTier)]) -> RawSession {
     s
 }
 
+/// A session whose evidence spans `total_chars` (one line), forcing a single
+/// large window that a size-aware provider will truncate until it is re-split.
+fn big_session(total_chars: usize) -> RawSession {
+    // Repeated ASCII with no PII patterns, so `sanitize_text` leaves it intact
+    // and the window length is predictable.
+    let filler = "abcdefghij".repeat(total_chars / 10 + 1);
+    session_with(&[(&filler[..total_chars], EvidenceTier::T2)])
+}
+
+/// Like [`big_session`] but `char_count` *multibyte* characters (byte length ≫
+/// char count), so the recovery splitter is exercised on char — not byte —
+/// boundaries.
+fn big_multibyte_session(char_count: usize) -> RawSession {
+    let filler: String = "日本語コード".chars().cycle().take(char_count).collect();
+    session_with(&[(filler.as_str(), EvidenceTier::T2)])
+}
+
 #[tokio::test]
 async fn parses_observations_from_json() {
     let body = r#"{"observations":[
@@ -47,11 +119,15 @@ async fn parses_observations_from_json() {
         body: Ok(body.into()),
     };
     let session = session_with(&[("commit small and often", EvidenceTier::T2)]);
-    let digest = digest_session(&provider, &session).await.unwrap();
-    assert_eq!(digest.observations.len(), 2);
-    assert_eq!(digest.observations[0].facet, PersonaFacet::Workflow);
-    assert_eq!(digest.observations[1].facet, PersonaFacet::CodingStyle);
-    assert_eq!(digest.observations[1].tier, EvidenceTier::T1);
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
+    assert_eq!(outcome.windows_lost, 0);
+    let obs = outcome.digest.observations;
+    assert_eq!(obs.len(), 2);
+    assert_eq!(obs[0].facet, PersonaFacet::Workflow);
+    assert_eq!(obs[1].facet, PersonaFacet::CodingStyle);
+    assert_eq!(obs[1].tier, EvidenceTier::T1);
 }
 
 #[tokio::test]
@@ -61,28 +137,168 @@ async fn tolerates_prose_wrapped_json() {
         body: Ok(body.into()),
     };
     let session = session_with(&[("cargo test", EvidenceTier::T2)]);
-    let digest = digest_session(&provider, &session).await.unwrap();
-    assert_eq!(digest.observations.len(), 1);
-    assert_eq!(digest.observations[0].facet, PersonaFacet::Stack);
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
+    assert_eq!(outcome.windows_lost, 0);
+    assert_eq!(outcome.digest.observations.len(), 1);
+    assert_eq!(outcome.digest.observations[0].facet, PersonaFacet::Stack);
 }
 
+/// A hard provider failure (transport/budget/auth) is transient: it surfaces as
+/// `Err` so the caller won't commit the cursor and retries the whole session.
 #[tokio::test]
-async fn soft_falls_back_on_error_and_bad_json() {
+async fn provider_failure_is_a_non_committable_error() {
     let session = session_with(&[("x", EvidenceTier::T2)]);
-
-    // A hard provider failure surfaces as Err (so the caller won't commit the
-    // cursor and the session is retried next run).
     let failing = MockChat {
         body: Err("402 requires more credits".into()),
     };
-    assert!(digest_session(&failing, &session).await.is_err());
+    assert!(digest_session(&failing, &session, &CallBudget::unlimited())
+        .await
+        .is_err());
+}
 
-    // A received-but-unparseable response is a soft failure: Ok + empty digest
-    // (re-running reproduces it, so the cursor may commit).
-    let garbage = MockChat {
-        body: Ok("not json at all".into()),
+/// A truncated window that a smaller call *can* parse must be **recovered**, not
+/// lost: recovery re-splits the window, the halves fit under the cap, and every
+/// observation is kept with nothing counted as lost. This is the data-loss the
+/// PR exists to fix, proven end-to-end through the re-split path.
+#[tokio::test]
+async fn truncated_window_is_recovered_by_resplitting() {
+    // One ~8 000-char window. The full window exceeds the threshold (truncates),
+    // but each half (~4 000) is under it and parses cleanly.
+    let session = big_session(8_000);
+    let provider = SizeAwareChat { threshold: 5_000 };
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
+    assert_eq!(
+        outcome.windows_lost, 0,
+        "a re-splittable truncation must recover, not drop"
+    );
+    assert!(
+        !outcome.digest.observations.is_empty(),
+        "recovered halves must yield observations"
+    );
+}
+
+/// Truncation recovery must draw from the run's `max_llm_calls` budget: a window
+/// that would re-split into several calls cannot exceed it. With a budget of one
+/// call, the first (truncated) attempt spends it and the re-split can't proceed,
+/// so the session is a non-committable `BudgetExhausted` checkpoint (not a silent
+/// partial). Guards against a recovery tree blowing the configured cap.
+#[tokio::test]
+async fn recovery_respects_the_call_budget() {
+    let session = big_session(8_000);
+    let provider = SizeAwareChat { threshold: 5_000 };
+    let budget = CallBudget::new(1);
+    let err = digest_session(&provider, &session, &budget)
+        .await
+        .expect_err("a one-call budget can't complete a re-splitting recovery");
+    assert!(
+        is_budget_exhausted(&err),
+        "budget exhaustion must be classifiable as a checkpoint, got: {err:#}"
+    );
+    assert_eq!(budget.remaining(), 0, "the one call was spent");
+}
+
+/// A provider that truncates *every* call, even the smallest sub-window, must not
+/// retry forever: recovery bottoms out at the minimum size, drops the window, and
+/// tallies it in `windows_lost` — returning `Ok` so the cursor can commit and the
+/// queue keeps moving (the permanent-starvation guard).
+#[tokio::test]
+async fn permanently_unparseable_window_terminates_and_is_counted() {
+    let truncated = r#"{"observations":[
+        {"facet":"workflow","observation":"Commits small and often","quote":"commit small","tier":"t2"},
+        {"facet":"coding_style","observation":"Insists on regression tests","quote":"add a test","tier":"t1"}"#;
+    let provider = MockChat {
+        body: Ok(truncated.into()),
+    };
+    let session = session_with(&[("x", EvidenceTier::T2)]);
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
+    assert_eq!(
+        outcome.windows_lost, 1,
+        "an irreducible truncation is dropped-and-counted, not looped forever"
+    );
+    assert!(outcome.digest.observations.is_empty());
+}
+
+/// The bounded re-splitting *descent*: a window sized to `MIN_WINDOW_CHARS <<
+/// MAX_RESPLIT_DEPTH` (12k) splits binarily the full 12k → 6k → 3k → 1.5k chain,
+/// digesting a complete recovery tree and dropping every leaf. Pinning the call
+/// count to the exact tree size — `sum(2^i for i in 0..=depth) = 2^(depth+1) - 1`
+/// — detects a fan-out regression, not merely a hang. (The test above covers the
+/// immediate below-floor drop; this exercises the `MAX_RESPLIT_DEPTH` path.)
+#[tokio::test]
+async fn large_permanently_unparseable_window_terminates_within_bounded_calls() {
+    let truncated = r#"{"observations":[
+        {"facet":"workflow","observation":"Commits small and often","quote":"commit small","tier":"t2"}"#;
+    let provider = MockChat {
+        body: Ok(truncated.into()),
+    };
+    let session = big_session(MIN_WINDOW_CHARS << MAX_RESPLIT_DEPTH); // 1500 * 8 = 12_000
+    let budget = CallBudget::new(1_000);
+    let outcome = digest_session(&provider, &session, &budget).await.unwrap();
+    assert!(outcome.digest.observations.is_empty());
+    // Leaves are the `2^depth` pieces at the floor (8 here), each dropped-and-counted.
+    assert_eq!(outcome.windows_lost, 1 << MAX_RESPLIT_DEPTH);
+    let calls = 1_000 - budget.remaining();
+    let expected_calls = (1usize << (MAX_RESPLIT_DEPTH + 1)) - 1; // 1+2+4+8 = 15
+    assert_eq!(
+        calls, expected_calls,
+        "recovery must spend exactly the bounded tree ({expected_calls}), spent {calls}"
+    );
+}
+
+/// Recovery re-splits on *character* boundaries, so a multibyte window that never
+/// parses walks the same descent without a byte-slice panic and drops the same
+/// leaf count as its ASCII twin.
+#[tokio::test]
+async fn recovery_splits_multibyte_windows_without_panicking() {
+    let truncated = r#"{"observations":[
+        {"facet":"workflow","observation":"Commits small and often","quote":"commit small","tier":"t2"}"#;
+    let provider = MockChat {
+        body: Ok(truncated.into()),
+    };
+    let session = big_multibyte_session(MIN_WINDOW_CHARS << MAX_RESPLIT_DEPTH);
+    let budget = CallBudget::new(1_000);
+    let outcome = digest_session(&provider, &session, &budget).await.unwrap();
+    assert!(outcome.digest.observations.is_empty());
+    assert_eq!(
+        outcome.windows_lost,
+        1 << MAX_RESPLIT_DEPTH,
+        "multibyte leaves drop-and-count exactly like ASCII, with no panic"
+    );
+}
+
+/// A multi-window session with one permanently-bad window keeps the observations
+/// from the clean window (no all-or-nothing abort) and counts only the bad one.
+#[tokio::test]
+async fn one_bad_window_does_not_discard_the_clean_siblings() {
+    // A near-full first line (~11 800 chars) flushes as its own clean window, then
+    // a short marker line (~300 chars, below the re-split floor) forms a second
+    // window that truncates and cannot be split further → dropped-and-counted.
+    let clean = "abcdefghij".repeat(1_200); // ~12 000 chars available
+    let poison = format!("POISONWINDOW {}", "z".repeat(280));
+    let session = session_with(&[
+        (&clean[..11_800], EvidenceTier::T2),
+        (poison.as_str(), EvidenceTier::T1),
+    ]);
+    let provider = MarkerChat {
+        marker: "POISONWINDOW",
     };
-    assert!(digest_session(&garbage, &session).await.unwrap().is_empty());
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
+    assert_eq!(
+        outcome.windows_lost, 1,
+        "only the irreducible poison window is lost"
+    );
+    assert!(
+        !outcome.digest.observations.is_empty(),
+        "the clean sibling window's observations are retained"
+    );
 }
 
 #[tokio::test]
@@ -91,10 +307,30 @@ async fn empty_session_yields_empty_digest() {
         body: Ok("{\"observations\":[]}".into()),
     };
     let session = RawSession::new(EvidenceSource::new(PersonaSourceKind::Codex));
-    assert!(digest_session(&provider, &session)
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
         .await
-        .unwrap()
-        .is_empty());
+        .unwrap();
+    assert!(outcome.digest.is_empty());
+    assert_eq!(outcome.windows_lost, 0);
+}
+
+/// A cleanly-parsed empty digest for a *non-empty* session is committable: it is
+/// `Ok` with no observations and nothing lost (the guard against over-correcting
+/// the truncation fix into a permanent retry of genuinely-empty windows).
+#[tokio::test]
+async fn clean_empty_observations_commit_without_loss() {
+    let provider = MockChat {
+        body: Ok("{\"observations\":[]}".into()),
+    };
+    let session = session_with(&[("just chatting, no rules here", EvidenceTier::T3)]);
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
+    assert!(outcome.digest.is_empty());
+    assert_eq!(
+        outcome.windows_lost, 0,
+        "a clean empty response is not a loss and must commit"
+    );
 }
 
 #[tokio::test]
@@ -109,7 +345,12 @@ async fn drops_unusable_observations() {
         body: Ok(body.into()),
     };
     let session = session_with(&[("db", EvidenceTier::T2)]);
-    let digest = digest_session(&provider, &session).await.unwrap();
-    assert_eq!(digest.observations.len(), 1);
-    assert_eq!(digest.observations[0].observation, "Prefers Postgres");
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
+    assert_eq!(outcome.digest.observations.len(), 1);
+    assert_eq!(
+        outcome.digest.observations[0].observation,
+        "Prefers Postgres"
+    );
 }
diff --git a/src/memory/persona/pipeline.rs b/src/memory/persona/pipeline.rs
index b65d29b..076b645 100644
--- a/src/memory/persona/pipeline.rs
+++ b/src/memory/persona/pipeline.rs
@@ -16,12 +16,12 @@ use serde::Serialize;
 
 use super::compile::{write_pack, PackInputs};
 use super::config::PersonaConfig;
-use super::distill::digest_session;
+use super::distill::{digest_session, is_budget_exhausted, CallBudget, SessionOutcome};
 use super::readers::{claude_code, codex, instruction, RawSession};
 use super::reduce::{fold_digest, fold_directives, seal_and_collect, FacetAsks, ReduceState};
 use super::state::PersonaStateStore;
 use super::state::{self, file_key, file_unchanged, record_file};
-use super::types::{PersonaFacet, SessionDigest};
+use super::types::PersonaFacet;
 use crate::memory::config::MemoryConfig;
 use crate::memory::score::extract::ChatProvider;
 use crate::memory::tree::Summariser;
@@ -62,46 +62,77 @@ pub struct RunReport {
     pub evidence_units: usize,
     /// Digest calls that produced at least one observation.
     pub digests: usize,
-    /// Sessions whose digest hit a hard provider failure (cursor NOT committed;
-    /// retried next run).
+    /// Sessions whose digest hit a hard **provider** failure — transport or auth
+    /// (cursor NOT committed; the whole session is retried next run). A spent call
+    /// budget is a clean checkpoint reported in [`Self::budget_hit`], not here, and
+    /// truncated/unparseable windows are recovered or counted in
+    /// [`Self::windows_lost`], not here.
     pub sessions_failed: usize,
+    /// Recovery leaf sub-windows dropped because their digest stayed unparseable
+    /// even after truncation-recovery re-splitting (one truncated 12k window can
+    /// contribute several, so this counts sub-windows, not top-level windows). A
+    /// session's cursor is committed on this path only when the run produced
+    /// observations elsewhere (near-deterministic at temperature 0.0); a run that
+    /// lost windows and yielded nothing withholds instead — see
+    /// [`Self::systemic_digest_failure`]. Surfaced so the drop is never silent.
+    pub windows_lost: usize,
     /// Observations distilled.
     pub observations: usize,
     /// Per-facet observation counts (facet wire-string → count).
     pub facet_counts: BTreeMap<String, usize>,
     /// True when a run budget stopped the run early (checkpointed).
     pub budget_hit: bool,
+    /// True when every digested session yielded zero observations *and* at least
+    /// one window was lost — the signature of a systemic provider failure (a
+    /// wrong/non-instruct model, refusal mode, a proxy returning prose). The
+    /// fully-lost sessions' cursors are then **withheld** so the backlog is
+    /// retried once the cause is fixed, rather than silently committed and skipped.
+    pub systemic_digest_failure: bool,
     /// Path of the compiled pack, if written.
     pub pack_path: Option<String>,
 }
 
-/// A run's budget accounting.
+/// A run's session-count budget. The provider-call ceiling (`max_llm_calls`) is
+/// enforced separately and precisely by [`CallBudget`] — per actual call, so
+/// multi-window sessions and truncation-recovery re-tries all count — rather than
+/// estimated here at one call per session.
 struct Budget {
     max_sessions: usize,
-    max_calls: usize,
     sessions: usize,
-    calls: usize,
 }
 
 impl Budget {
     fn from(cfg: &PersonaConfig) -> Self {
         Self {
             max_sessions: cfg.run_budget.max_sessions,
-            max_calls: cfg.run_budget.max_llm_calls as usize,
             sessions: 0,
-            calls: 0,
         }
     }
-    /// True if another digest would exceed the budget (stop cleanly).
+    /// True once the run has digested its full session allowance (stop cleanly).
     fn exhausted(&self) -> bool {
-        self.sessions >= self.max_sessions || self.calls >= self.max_calls
+        self.sessions >= self.max_sessions
     }
     fn charge(&mut self) {
         self.sessions += 1;
-        self.calls += 1;
     }
 }
 
+/// Digest-loop guards threaded through the ingest sources: the shared
+/// provider-call ceiling and the deferred cursor commits awaiting the run-level
+/// systemic-failure check in [`Pipeline::run`].
+struct DigestGuards {
+    call_budget: CallBudget,
+    /// Commits for fully-lost sessions (zero observations, ≥1 window dropped),
+    /// applied or withheld once the whole run's outcome is known.
+    deferred: Vec<(String, serde_json::Value)>,
+}
+
+/// Git-history ingestion (`ingest_git`), feature-gated and split into a sibling
+/// module to keep this file within the repo's size norm.
+#[cfg(feature = "git-diff")]
+#[path = "pipeline_git.rs"]
+mod pipeline_git;
+
 /// The pipeline binds the workspace, config, provider, summariser, and state
 /// store; `run` executes one pass.
 pub struct Pipeline<'a> {
@@ -123,6 +154,13 @@ impl Pipeline<'_> {
         let asks = self.persona.asks();
         let mut state = ReduceState::default();
         let mut budget = Budget::from(self.persona);
+        // Shared provider-call ceiling (so transcripts and git history draw from
+        // the same `max_llm_calls`) plus the deferred fully-lost cursor commits,
+        // resolved after all sources run.
+        let mut guards = DigestGuards {
+            call_budget: CallBudget::new(self.persona.run_budget.max_llm_calls as usize),
+            deferred: Vec::new(),
+        };
         let mut report = RunReport {
             mode: mode.as_str().to_string(),
             ..Default::default()
@@ -135,15 +173,67 @@ impl Pipeline<'_> {
         self.ingest_instructions(&mut state, &mut report).await?;
 
         // 2. Transcripts (Claude Code + Codex) — the digest map step.
-        self.ingest_transcripts(mode, &asks, &mut state, &mut budget, &mut report)
-            .await?;
+        self.ingest_transcripts(
+            mode,
+            &asks,
+            &mut state,
+            &mut budget,
+            &mut guards,
+            &mut report,
+        )
+        .await?;
 
         // 3. Git history (feature-gated).
         #[cfg(feature = "git-diff")]
-        self.ingest_git(mode, &asks, &mut state, &mut budget, &mut report)
-            .await?;
+        self.ingest_git(
+            mode,
+            &asks,
+            &mut state,
+            &mut budget,
+            &mut guards,
+            &mut report,
+        )
+        .await?;
 
-        report.budget_hit = budget.exhausted();
+        // Resolve the deferred (zero-observation, ≥1-window-lost) cursor commits.
+        // If the run produced observations anywhere, those sessions are localized
+        // permanent failures — commit them so the queue advances (no starvation).
+        // If the run yielded *nothing* despite dropping windows, treat it as a
+        // systemic provider failure: withhold every deferred commit so the backlog
+        // is retried once the cause is fixed, and flag it loudly.
+        if !guards.deferred.is_empty() {
+            if report.observations > 0 {
+                for (key, value) in &guards.deferred {
+                    // A single failed cursor write must not discard the whole run's
+                    // reduce output (the pack, written below): a cursor is a
+                    // fast-skip, not a correctness gate, so the session simply
+                    // re-digests next run. Log and carry on rather than `?`-abort.
+                    if let Err(e) = self.store.set(state::NAMESPACE, key, value).await {
+                        log::warn!(
+                            "[persona] deferred cursor commit failed for {key}; \
+                             it will be re-digested next run: {e:#}"
+                        );
+                    }
+                }
+            } else {
+                report.systemic_digest_failure = true;
+                log::error!(
+                    "[persona] {} session(s) digested to zero observations with {} window(s) \
+                     lost and no observations anywhere this run — likely a systemic provider \
+                     failure (wrong/non-instruct model, refusal, or a proxy returning prose). \
+                     Withholding {} cursor commit(s) so the backlog is retried, not skipped.",
+                    guards.deferred.len(),
+                    report.windows_lost,
+                    guards.deferred.len(),
+                );
+            }
+        }
+
+        // `budget_hit` is recorded precisely where a ceiling actually stops work:
+        // the selection loop when it drops a pending session, and the call-budget
+        // checkpoint on `BudgetExhausted`. It is deliberately NOT re-derived from
+        // `budget.exhausted()` here — that is true whenever the run merely filled
+        // its session allowance exactly, which drops nothing and is not a hit.
 
         // 4. Seal facet trees + compile the pack.
         let bodies = seal_and_collect(self.config, &asks, self.summariser).await?;
@@ -239,6 +329,7 @@ impl Pipeline<'_> {
         asks: &FacetAsks,
         state: &mut ReduceState,
         budget: &mut Budget,
+        guards: &mut DigestGuards,
         report: &mut RunReport,
     ) -> Result<()> {
         let mut files: Vec<(PathBuf, &'static str)> = Vec::new();
@@ -281,25 +372,28 @@ impl Pipeline<'_> {
                 .map(|v| (key, v));
             pending.push(Pending { session, commit });
         }
-        self.digest_and_fold(pending, asks, state, budget, report)
+        self.digest_and_fold(pending, asks, state, budget, guards, report)
             .await
     }
 
     /// Digest the `pending` sessions concurrently (the network-bound map step)
     /// and fold the results serially into the facet trees (SQLite writes must
     /// stay serial). Order is preserved (`buffered`, not `buffer_unordered`) so
-    /// trees fold oldest-first. Selection honours the shared run budget; each
-    /// selected session's cursor is committed only after it is folded, so a
-    /// budget-truncated tail is re-processed on the next run.
+    /// trees fold oldest-first. Selection honours the shared run budget; a
+    /// selected session's cursor is committed after it is folded, except a
+    /// fully-lost session (zero observations, ≥1 window dropped) whose commit is
+    /// pushed to `deferred` for the run-level systemic check in [`Self::run`].
     async fn digest_and_fold(
         &self,
         pending: Vec<Pending>,
         asks: &FacetAsks,
         state: &mut ReduceState,
         budget: &mut Budget,
+        guards: &mut DigestGuards,
         report: &mut RunReport,
     ) -> Result<()> {
-        // Select within budget up front.
+        // Select within the session-count budget up front; the provider-call
+        // budget is enforced per call inside `digest_session`.
         let mut selected: Vec<Pending> = Vec::new();
         for p in pending {
             if budget.exhausted() {
@@ -313,98 +407,66 @@ impl Pipeline<'_> {
             return Ok(());
         }
         let concurrency = self.persona.digest_concurrency.max(1);
-        let results: Vec<Result<SessionDigest>> = stream::iter(selected.iter())
-            .map(|p| digest_session(self.provider, &p.session))
+        let results: Vec<Result<SessionOutcome>> = stream::iter(selected.iter())
+            .map(|p| digest_session(self.provider, &p.session, &guards.call_budget))
             .buffered(concurrency)
             .collect()
             .await;
         for (p, result) in selected.iter().zip(results) {
-            let digest = match result {
-                Ok(d) => d,
+            let outcome = match result {
+                Ok(o) => o,
                 Err(e) => {
-                    // Hard provider failure: do NOT commit the cursor, so this
-                    // session is re-attempted on the next run.
-                    log::warn!("[persona] digest failed, cursor not committed: {e:#}");
-                    report.sessions_failed += 1;
+                    // Non-committable, cursor NOT committed so the session is
+                    // re-attempted next run. Two shapes reach here, neither a
+                    // silent drop: the run's call budget was spent mid-session (a
+                    // clean checkpoint), or a hard provider failure
+                    // (transport/auth). Truncated/unparseable windows never reach
+                    // here — they are recovered or dropped-and-counted inside
+                    // `digest_session`.
+                    if is_budget_exhausted(&e) {
+                        report.budget_hit = true;
+                    } else {
+                        log::warn!(
+                            "[persona] digest provider failure, cursor not committed: {e:#}"
+                        );
+                        report.sessions_failed += 1;
+                    }
                     continue;
                 }
             };
             report.sessions_processed += 1;
+            report.windows_lost += outcome.windows_lost;
+            let digest = outcome.digest;
+            let session_observations = digest.observations.len();
             if !digest.is_empty() {
                 report.digests += 1;
-                report.observations += digest.observations.len();
+                report.observations += session_observations;
                 fold_digest(self.config, &digest, asks, self.summariser, state).await?;
             }
-            // Commit the cursor/watermark now that the session is folded (a valid
-            // empty digest still commits — retrying would reproduce it).
-            if let Some((key, value)) = &p.commit {
-                self.store.set(state::NAMESPACE, key, value).await?;
+            let Some(commit) = &p.commit else { continue };
+            if session_observations == 0 && outcome.windows_lost > 0 {
+                // Yielded nothing but lost ≥1 window. In isolation this is a
+                // localized permanent-garbage window that should commit so the
+                // queue advances; run-wide it can instead be the symptom of a
+                // systemic provider failure. Defer the commit — `run` applies it
+                // only if the run produced observations somewhere, otherwise it
+                // withholds and flags rather than silently skipping the backlog.
+                guards.deferred.push(commit.clone());
+            } else if let Err(e) = self.store.set(state::NAMESPACE, &commit.0, &commit.1).await {
+                // Committed now that the session is folded: a recovered or
+                // genuinely-empty (zero-loss) session reproduces on re-run, so
+                // committing loses nothing. As on the deferred path, a failed
+                // write is logged and skipped rather than `?`-aborting the run and
+                // discarding the pack — the cursor is a fast-skip, so the session
+                // simply re-digests next run.
+                log::warn!(
+                    "[persona] cursor commit failed for {}; it will be re-digested next run: {e:#}",
+                    commit.0
+                );
             }
         }
         Ok(())
     }
-
-    #[cfg(feature = "git-diff")]
-    async fn ingest_git(
-        &self,
-        mode: RunMode,
-        asks: &FacetAsks,
-        state: &mut ReduceState,
-        budget: &mut Budget,
-        report: &mut RunReport,
-    ) -> Result<()> {
-        use super::readers::git_history::{self, GitReadConfig};
-
-        let git_cfg = GitReadConfig {
-            author_emails: self.persona.author_emails.clone(),
-            batch_size: self.persona.git.batch_size,
-            max_commits: self.persona.git.max_commits,
-            diff_sample_cap: self.persona.git.diff_sample_cap,
-            diff_size_cap_bytes: self.persona.git.diff_size_cap_bytes,
-            small_commit_max_files: self.persona.git.small_commit_max_files,
-        };
-        let author_hash = author_set_hash(&self.persona.author_emails);
-
-        // Read all qualifying repos into a pending list (serial, cheap), then
-        // digest concurrently + fold serially. A repo's HEAD watermark is
-        // attached to the LAST of its sessions, so it is only committed once the
-        // whole repo has been folded (a budget-truncated repo re-scans next run).
-        let mut pending: Vec<Pending> = Vec::new();
-        for repo in git_history::discover(&self.persona.project_roots) {
-            report.files_seen += 1;
-            let head = match git_head_sha(&repo) {
-                Some(h) => format!("{h}:{author_hash}"),
-                None => continue,
-            };
-            let key = state::git_key(&repo);
-            if mode == RunMode::Incremental
-                && state::watermark_unchanged(self.store, &key, &head).await?
-            {
-                report.sessions_skipped += 1;
-                continue;
-            }
-            let sessions = match git_history::read_repo(&repo, &git_cfg) {
-                Ok(s) => s,
-                Err(_) => continue,
-            };
-            for session in &sessions {
-                report.evidence_units += session.evidence.len();
-            }
-            if sessions.is_empty() {
-                // No author commits — watermark immediately (nothing to digest).
-                state::record_watermark(self.store, &key, &head).await?;
-                continue;
-            }
-            let head_value = serde_json::Value::String(head);
-            let last = sessions.len() - 1;
-            for (i, session) in sessions.into_iter().enumerate() {
-                let commit = (i == last).then(|| (key.clone(), head_value.clone()));
-                pending.push(Pending { session, commit });
-            }
-        }
-        self.digest_and_fold(pending, asks, state, budget, report)
-            .await
-    }
 }
 
 /// A read-but-not-yet-digested session plus an optional state commit (cursor or
@@ -429,29 +491,6 @@ fn file_mtime_ms(path: &Path) -> i64 {
     state::FileCursor::of(path).map(|c| c.mtime_ms).unwrap_or(0)
 }
 
-/// Stable short hash of the author-email set, so changing it forces a re-scan.
-#[cfg(feature = "git-diff")]
-fn author_set_hash(emails: &[String]) -> String {
-    use sha2::{Digest, Sha256};
-    let mut sorted: Vec<String> = emails.iter().map(|e| e.to_lowercase()).collect();
-    sorted.sort();
-    let mut h = Sha256::new();
-    h.update(sorted.join(",").as_bytes());
-    h.finalize()
-        .iter()
-        .take(4)
-        .map(|b| format!("{b:02x}"))
-        .collect()
-}
-
-/// Current HEAD sha of a repo, or `None` for an empty/broken repo.
-#[cfg(feature = "git-diff")]
-fn git_head_sha(repo: &Path) -> Option<String> {
-    let repo = git2::Repository::open(repo).ok()?;
-    let head = repo.head().ok()?;
-    head.target().map(|oid| oid.to_string())
-}
-
 #[cfg(test)]
 #[path = "pipeline_tests.rs"]
 mod tests;
diff --git a/src/memory/persona/pipeline_git.rs b/src/memory/persona/pipeline_git.rs
new file mode 100644
index 0000000..76d30c3
--- /dev/null
+++ b/src/memory/persona/pipeline_git.rs
@@ -0,0 +1,99 @@
+//! Git-history ingestion for the persona pipeline (doc 06 §6.7), feature-gated
+//! behind `git-diff`. Split out of `pipeline.rs` to keep that module within the
+//! repo's file-size norm; the digest/fold machinery it drives lives on
+//! [`Pipeline`] in the parent module.
+
+use std::path::Path;
+
+use anyhow::Result;
+
+use super::{Budget, DigestGuards, Pending, Pipeline, RunMode, RunReport};
+use crate::memory::persona::readers::git_history::{self, GitReadConfig};
+use crate::memory::persona::reduce::{FacetAsks, ReduceState};
+use crate::memory::persona::state;
+
+impl Pipeline<'_> {
+    /// Ingest each qualifying repo's author commits as digest sessions. A repo's
+    /// HEAD watermark is attached to the LAST of its sessions, so it commits only
+    /// once the whole repo has been folded (a budget-truncated repo re-scans next
+    /// run). `state` is the module (type namespace); the value binding is the
+    /// reduce accumulator.
+    pub(super) async fn ingest_git(
+        &self,
+        mode: RunMode,
+        asks: &FacetAsks,
+        state: &mut ReduceState,
+        budget: &mut Budget,
+        guards: &mut DigestGuards,
+        report: &mut RunReport,
+    ) -> Result<()> {
+        let git_cfg = GitReadConfig {
+            author_emails: self.persona.author_emails.clone(),
+            batch_size: self.persona.git.batch_size,
+            max_commits: self.persona.git.max_commits,
+            diff_sample_cap: self.persona.git.diff_sample_cap,
+            diff_size_cap_bytes: self.persona.git.diff_size_cap_bytes,
+            small_commit_max_files: self.persona.git.small_commit_max_files,
+        };
+        let author_hash = author_set_hash(&self.persona.author_emails);
+
+        // Read all qualifying repos into a pending list (serial, cheap), then
+        // digest concurrently + fold serially.
+        let mut pending: Vec<Pending> = Vec::new();
+        for repo in git_history::discover(&self.persona.project_roots) {
+            report.files_seen += 1;
+            let head = match git_head_sha(&repo) {
+                Some(h) => format!("{h}:{author_hash}"),
+                None => continue,
+            };
+            let key = state::git_key(&repo);
+            if mode == RunMode::Incremental
+                && state::watermark_unchanged(self.store, &key, &head).await?
+            {
+                report.sessions_skipped += 1;
+                continue;
+            }
+            let sessions = match git_history::read_repo(&repo, &git_cfg) {
+                Ok(s) => s,
+                Err(_) => continue,
+            };
+            for session in &sessions {
+                report.evidence_units += session.evidence.len();
+            }
+            if sessions.is_empty() {
+                // No author commits — watermark immediately (nothing to digest).
+                state::record_watermark(self.store, &key, &head).await?;
+                continue;
+            }
+            let head_value = serde_json::Value::String(head);
+            let last = sessions.len() - 1;
+            for (i, session) in sessions.into_iter().enumerate() {
+                let commit = (i == last).then(|| (key.clone(), head_value.clone()));
+                pending.push(Pending { session, commit });
+            }
+        }
+        self.digest_and_fold(pending, asks, state, budget, guards, report)
+            .await
+    }
+}
+
+/// Stable short hash of the author-email set, so changing it forces a re-scan.
+fn author_set_hash(emails: &[String]) -> String {
+    use sha2::{Digest, Sha256};
+    let mut sorted: Vec<String> = emails.iter().map(|e| e.to_lowercase()).collect();
+    sorted.sort();
+    let mut h = Sha256::new();
+    h.update(sorted.join(",").as_bytes());
+    h.finalize()
+        .iter()
+        .take(4)
+        .map(|b| format!("{b:02x}"))
+        .collect()
+}
+
+/// Current HEAD sha of a repo, or `None` for an empty/broken repo.
+fn git_head_sha(repo: &Path) -> Option<String> {
+    let repo = git2::Repository::open(repo).ok()?;
+    let head = repo.head().ok()?;
+    head.target().map(|oid| oid.to_string())
+}
diff --git a/src/memory/persona/pipeline_tests.rs b/src/memory/persona/pipeline_tests.rs
index f7c6a28..992d85b 100644
--- a/src/memory/persona/pipeline_tests.rs
+++ b/src/memory/persona/pipeline_tests.rs
@@ -39,6 +39,63 @@ impl ChatProvider for FailChat {
     }
 }
 
+/// A provider that returns a **truncated** observation array on every call — a
+/// well-formed prefix with the closing `]`/`}` missing, exactly what a hit
+/// output-token cap produces, and unrecoverable no matter how small the window is
+/// re-split. The transcript windows here are tiny (well under the re-split floor),
+/// so recovery drops-and-counts them rather than looping forever.
+struct TruncatedChat;
+#[async_trait]
+impl ChatProvider for TruncatedChat {
+    fn name(&self) -> &str {
+        "truncated"
+    }
+    async fn chat_for_json(&self, _p: &ChatPrompt) -> anyhow::Result<String> {
+        Ok(r#"{"observations":[
+            {"facet":"workflow","observation":"Commits small and often","quote":"commit","tier":"t2"},
+            {"facet":"communication","observation":"Terse and direct","quote":"do X","tier":"t2"}"#
+        .into())
+    }
+}
+
+/// A provider that truncates only when the prompt contains `POISON`, and returns
+/// a clean observation otherwise — so one transcript can be fully-lost while
+/// another in the same run digests normally (the localized-vs-systemic split).
+struct PoisonMarkerChat;
+#[async_trait]
+impl ChatProvider for PoisonMarkerChat {
+    fn name(&self) -> &str {
+        "poison-marker"
+    }
+    async fn chat_for_json(&self, p: &ChatPrompt) -> anyhow::Result<String> {
+        if p.user.contains("POISON") {
+            // Truncated array (unrecoverable at this tiny window size).
+            Ok(r#"{"observations":[
+                {"facet":"workflow","observation":"Commits small and often","quote":"c","tier":"t2"}"#
+                .into())
+        } else {
+            Ok(r#"{"observations":[
+                {"facet":"workflow","observation":"Commits small and often","quote":"c","tier":"t2"}
+            ]}"#
+            .into())
+        }
+    }
+}
+
+/// A provider that returns a cleanly-parsed **empty** observation set. A real,
+/// low-signal session (nothing worth distilling) looks exactly like this, and it
+/// must commit its cursor — re-running only reproduces the empty result.
+struct EmptyChat;
+#[async_trait]
+impl ChatProvider for EmptyChat {
+    fn name(&self) -> &str {
+        "empty"
+    }
+    async fn chat_for_json(&self, _p: &ChatPrompt) -> anyhow::Result<String> {
+        Ok(r#"{"observations":[]}"#.into())
+    }
+}
+
 fn user_turn(session: &str, ts: &str, text: &str) -> String {
     format!(
         r#"{{"type":"user","isSidechain":false,"cwd":"/work/demo","sessionId":"{session}","timestamp":"{ts}","message":{{"role":"user","content":"{text}"}}}}"#
@@ -147,6 +204,74 @@ async fn budget_cutoff_checkpoints() {
     assert!(report.pack_path.is_some());
 }
 
+#[tokio::test]
+async fn call_budget_exhaustion_checkpoints_and_resumes() {
+    // `max_llm_calls` is a hard ceiling on provider calls: with a budget of one
+    // call and two sessions (one window each, digested oldest-first), only the
+    // first commits; the second hits the spent budget, does NOT commit, and is
+    // re-digested on a later run once the budget refreshes.
+    let (ws, src, cfg, mut persona) = setup();
+    persona.run_budget.max_llm_calls = 1;
+    persona.digest_concurrency = 1; // deterministic oldest-first ordering
+    let summariser = ConcatSummariser::new();
+    let store = FileStateStore::open_in_workspace(ws.path()).unwrap();
+
+    let report = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &MockChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Backfill)
+    .await
+    .unwrap();
+    assert_eq!(
+        report.sessions_processed, 1,
+        "only one call's worth digested"
+    );
+    assert_eq!(
+        report.sessions_failed, 0,
+        "budget exhaustion is not a failure"
+    );
+    assert!(report.budget_hit, "the call budget stopped the run");
+
+    // Exactly one transcript cursor is committed; the budget-checkpointed one is not.
+    use crate::memory::persona::state::{file_key, PersonaStateStore, NAMESPACE};
+    let cc_root = src.path().join("claude/projects/-work-demo");
+    let mut committed = 0;
+    for name in ["a.jsonl", "b.jsonl"] {
+        let key = file_key("claude_code", &cc_root.join(name));
+        if PersonaStateStore::get(&store, NAMESPACE, &key)
+            .await
+            .unwrap()
+            .is_some()
+        {
+            committed += 1;
+        }
+    }
+    assert_eq!(
+        committed, 1,
+        "budget checkpoint commits exactly one session"
+    );
+
+    // A fresh run (new budget) digests the un-committed session.
+    let second = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &MockChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Incremental)
+    .await
+    .unwrap();
+    assert_eq!(
+        second.sessions_processed, 1,
+        "the budget-checkpointed session resumes"
+    );
+}
+
 #[tokio::test]
 async fn compile_only_reassembles_without_llm() {
     let (ws, _src, cfg, persona) = setup();
@@ -211,6 +336,229 @@ async fn hard_provider_failure_does_not_commit_cursor() {
     assert!(second.observations >= 2);
 }
 
+#[tokio::test]
+async fn systemic_truncation_withholds_commits_and_retries() {
+    // When *every* session digests to zero observations with windows lost — the
+    // signature of a systemic provider failure (wrong model, refusal, proxy prose)
+    // — the fully-lost cursors must be WITHHELD, not committed. Committing them
+    // would silently skip the whole backlog and require manual state deletion to
+    // recover once the cause is fixed. The run is flagged and a later run retries.
+    let (ws, src, cfg, mut persona) = setup();
+    persona.run_budget.max_llm_calls = 64; // decouple from the config default
+    let summariser = ConcatSummariser::new();
+    let store = FileStateStore::open_in_workspace(ws.path()).unwrap();
+
+    let report = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &TruncatedChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Backfill)
+    .await
+    .unwrap();
+    assert_eq!(report.sessions_processed, 2);
+    assert_eq!(
+        report.sessions_failed, 0,
+        "a truncation is not a provider failure"
+    );
+    assert_eq!(report.windows_lost, 2);
+    assert_eq!(report.observations, 0);
+    assert!(
+        report.systemic_digest_failure,
+        "zero observations run-wide with losses is a systemic failure"
+    );
+
+    // The transcript cursors are WITHHELD — nothing is silently skipped.
+    use crate::memory::persona::state::{file_key, PersonaStateStore, NAMESPACE};
+    let cc_root = src.path().join("claude/projects/-work-demo");
+    for name in ["a.jsonl", "b.jsonl"] {
+        let key = file_key("claude_code", &cc_root.join(name));
+        let stored = PersonaStateStore::get(&store, NAMESPACE, &key)
+            .await
+            .unwrap();
+        assert!(
+            stored.is_none(),
+            "cursor for {name} must be withheld under a systemic digest failure"
+        );
+    }
+
+    // Once the provider works, a later run re-digests the withheld backlog.
+    let second = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &MockChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Incremental)
+    .await
+    .unwrap();
+    assert_eq!(
+        second.sessions_processed, 2,
+        "the withheld sessions are retried, not skipped"
+    );
+    assert!(second.observations >= 2);
+    assert!(!second.systemic_digest_failure);
+}
+
+#[tokio::test]
+async fn fully_lost_session_commits_when_the_run_yields_observations() {
+    // A single fully-lost session amid a healthy run is a *localized* permanent
+    // failure, not systemic: because the run produced observations elsewhere, its
+    // deferred cursor IS committed so the queue advances (no starvation), and the
+    // run is not flagged systemic.
+    let ws = TempDir::new().unwrap();
+    let src = TempDir::new().unwrap();
+    let cc_root = src.path().join("claude/projects/-work-demo");
+    std::fs::create_dir_all(&cc_root).unwrap();
+    // good.jsonl (older) parses; poison.jsonl (newer) always truncates.
+    let mut good = std::fs::File::create(cc_root.join("good.jsonl")).unwrap();
+    writeln!(
+        good,
+        "{}",
+        user_turn("s1", "2026-07-01T10:00:00.000Z", "a normal working session")
+    )
+    .unwrap();
+    let mut poison = std::fs::File::create(cc_root.join("poison.jsonl")).unwrap();
+    writeln!(
+        poison,
+        "{}",
+        user_turn(
+            "s2",
+            "2026-07-02T10:00:00.000Z",
+            "POISON marks this one truncated"
+        )
+    )
+    .unwrap();
+
+    let cfg = MemoryConfig::new(ws.path());
+    let mut persona = PersonaConfig::with_home(src.path(), "me@example.com");
+    persona.claude_code_root = Some(src.path().join("claude/projects"));
+    persona.codex_root = None;
+    persona.project_roots = vec![];
+    persona.global_instruction_files = vec![];
+    persona.digest_concurrency = 1; // deterministic oldest-first
+    persona.run_budget.max_llm_calls = 64; // decouple from the config default
+
+    let summariser = ConcatSummariser::new();
+    let store = FileStateStore::open_in_workspace(ws.path()).unwrap();
+
+    let report = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &PoisonMarkerChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Backfill)
+    .await
+    .unwrap();
+    assert_eq!(report.sessions_processed, 2);
+    assert_eq!(
+        report.windows_lost, 1,
+        "only the poison session lost its window"
+    );
+    assert!(
+        report.observations >= 1,
+        "the good session yielded observations"
+    );
+    assert!(
+        !report.systemic_digest_failure,
+        "a run that produced observations is not systemic"
+    );
+
+    // Both cursors are committed — the good one directly, the poison one via the
+    // applied deferral — so a later run skips both.
+    use crate::memory::persona::state::{file_key, PersonaStateStore, NAMESPACE};
+    for name in ["good.jsonl", "poison.jsonl"] {
+        let key = file_key("claude_code", &cc_root.join(name));
+        assert!(
+            PersonaStateStore::get(&store, NAMESPACE, &key)
+                .await
+                .unwrap()
+                .is_some(),
+            "cursor for {name} must be committed in a healthy run"
+        );
+    }
+    let second = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &PoisonMarkerChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Incremental)
+    .await
+    .unwrap();
+    assert_eq!(
+        second.sessions_processed, 0,
+        "both sessions are cursor-skipped"
+    );
+}
+
+#[tokio::test]
+async fn clean_empty_digest_commits_cursor() {
+    // The counterpart invariant to the truncation fix: a cleanly-parsed empty
+    // digest is genuinely done, so its cursor MUST commit and a later run skips it.
+    // (Guards against over-correcting the truncation fix into a permanent retry of
+    // low-signal sessions.)
+    let (ws, src, cfg, mut persona) = setup();
+    persona.run_budget.max_llm_calls = 64; // decouple from the config default
+    let summariser = ConcatSummariser::new();
+    let store = FileStateStore::open_in_workspace(ws.path()).unwrap();
+
+    let report = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &EmptyChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Backfill)
+    .await
+    .unwrap();
+    assert_eq!(
+        report.sessions_processed, 2,
+        "both sessions digested cleanly"
+    );
+    assert_eq!(report.sessions_failed, 0);
+    assert_eq!(report.windows_lost, 0, "an empty digest is not a loss");
+    assert_eq!(report.observations, 0);
+
+    // Both cursors committed.
+    use crate::memory::persona::state::{file_key, PersonaStateStore, NAMESPACE};
+    let cc_root = src.path().join("claude/projects/-work-demo");
+    for name in ["a.jsonl", "b.jsonl"] {
+        let key = file_key("claude_code", &cc_root.join(name));
+        let stored = PersonaStateStore::get(&store, NAMESPACE, &key)
+            .await
+            .unwrap();
+        assert!(
+            stored.is_some(),
+            "cursor for {name} must be committed after a clean empty digest"
+        );
+    }
+
+    // Incremental re-run skips both.
+    let second = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &EmptyChat,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Incremental)
+    .await
+    .unwrap();
+    assert_eq!(
+        second.sessions_processed, 0,
+        "empty sessions are not re-digested"
+    );
+    assert!(second.sessions_skipped >= 2);
+}
+
 #[tokio::test]
 async fn removed_directive_drops_out_on_rerun() {
     // Editing an instruction file (removing a rule) must drop the stale rule
diff --git a/src/memory/persona/reduce_tests.rs b/src/memory/persona/reduce_tests.rs
index 00b1857..f04dcc1 100644
--- a/src/memory/persona/reduce_tests.rs
+++ b/src/memory/persona/reduce_tests.rs
@@ -8,7 +8,7 @@ use tempfile::TempDir;
 
 use crate::memory::config::MemoryConfig;
 use crate::memory::persona::compile::{compile_pack, PackInputs};
-use crate::memory::persona::distill::digest_session;
+use crate::memory::persona::distill::{digest_session, CallBudget};
 use crate::memory::persona::readers::RawSession;
 use crate::memory::persona::types::{
     EvidenceSource, EvidenceTier, PersonaEvidence, PersonaSourceKind,
@@ -65,7 +65,10 @@ async fn full_map_reduce_compile_offline() {
 
     // Two sessions from two different scopes → cross-project strength.
     for scope in ["projA", "projB"] {
-        let digest = digest_session(&provider, &session(scope)).await.unwrap();
+        let digest = digest_session(&provider, &session(scope), &CallBudget::unlimited())
+            .await
+            .unwrap()
+            .digest;
         fold_digest(&config, &digest, &asks, &summariser, &mut state)
             .await
             .unwrap();