From c97a40e0211a1bfe32a2968887f59c6f073ca73d Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Wed, 12 Aug 2026 18:33:23 +0530
Subject: [PATCH 01/10] fix(persona): retain observations when a digest is
 truncated

A dense window (e.g. a Codex session full of corrections/directives)
distils into many facet observations, and the 4 K output-token cap was
sized for "one small JSON object". The model ran out of output mid-array
and emitted a well-formed prefix with no closing `]` (observed
`EOF while parsing a list` at column ~15-19 K).

Two independent bugs made that lossy:

- `digest_window` collapsed any parse failure to `Ok(Vec::new())`, so a
  truncated response was indistinguishable from a genuinely-empty digest.
- `digest_and_fold` commits the window cursor for ANY `Ok` result, so the
  truncated window was marked done and its observations were dropped for
  good, never retried.

Fix:

- Raise `DIGEST_MAX_OUTPUT_TOKENS` 4_096 -> 16_384 so dense digests fit,
  and document the coupling to `WINDOW_CHARS`.
- Introduce `DigestError` (a distinct, retryable failure type) and return
  `Err` for an unparseable/truncated response instead of a fake-empty
  `Ok`. It flows through the caller's existing hard-failure arm, which
  already `continue`s without committing the cursor, so the window is
  re-attempted next run. A genuinely-empty digest still returns
  `Ok(vec![])` and commits, since re-running would only reproduce it.

Tests: flip the bad-JSON case (now a non-committable `Err`), add a
truncated-array case, and a pipeline test asserting a truncated window
leaves its cursor absent and is re-processed on the next run.

Closes #5510
---
 src/memory/persona/distill.rs        | 95 ++++++++++++++++++++--------
 src/memory/persona/distill_tests.rs  | 36 +++++++++--
 src/memory/persona/pipeline.rs       | 12 ++--
 src/memory/persona/pipeline_tests.rs | 79 +++++++++++++++++++++++
 4 files changed, 189 insertions(+), 33 deletions(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index 8d0e31f..728a12a 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -19,8 +19,21 @@ 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 strict-JSON system prompt: schema + extraction contract.
 fn system_prompt() -> String {
@@ -90,12 +103,16 @@ fn windows(session: &RawSession) -> Vec<String> {
 
 /// Digest one session into a [`SessionDigest`] 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
+/// Distinguishes two outcomes so the pipeline can checkpoint correctly:
+/// - **Non-committable failure** → returns `Err`. The caller must NOT commit the
+///   session's cursor, so the evidence is re-attempted on the next run. This
+///   covers both a provider/transport error (a `chat_for_json` error — budget
+///   exhausted, 401/403, transport) *and* a response we could not parse (most
+///   often a **truncated** JSON array — the model hit its output-token cap
+///   mid-list, so the observations it *did* find would be lost forever if we
+///   committed). See [`DigestError`].
+/// - **Genuinely empty digest** (a valid call whose response parsed to zero
+///   observations, e.g. `{"observations":[]}`) → returns `Ok` with an empty
 ///   digest. Re-running would reproduce it, so the cursor IS committed.
 pub async fn digest_session(
     provider: &dyn ChatProvider,
@@ -106,8 +123,9 @@ pub async fn digest_session(
     }
     let mut observations: Vec<DigestObservation> = Vec::new();
     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.
+        // A hard provider failure OR an unparseable/truncated window bubbles up
+        // as `Err` (the whole session is retried next run, nothing committed);
+        // only a cleanly-parsed empty window is tolerated as `Ok(vec![])`.
         let obs = digest_window(provider, session, &window).await?;
         observations.extend(obs);
     }
@@ -117,9 +135,35 @@ pub async fn digest_session(
     })
 }
 
-/// 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 committable observations.
+///
+/// Both variants are **retryable** — the caller must not commit the session's
+/// cursor for either, so the window is re-attempted on the next run. They are
+/// distinguished only for logging/telemetry clarity. A genuinely-empty result is
+/// *not* an error (it is `Ok(vec![])`); this type exists so a truncated or
+/// otherwise unparseable response can no longer masquerade as "empty" and be
+/// silently committed (the data-loss bug this fixes).
+#[derive(Debug, thiserror::Error)]
+enum DigestError {
+    /// The provider call itself failed (budget/auth/transport). The response was
+    /// never received.
+    #[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
+    /// `]`). Committing would drop the observations the model *did* produce.
+    #[error("digest response unparseable (likely truncated at the output cap): {0:#}")]
+    Unparseable(#[source] anyhow::Error),
+}
+
+/// One window → observations.
+///
+/// Returns `Err` for **both** a `chat_for_json` failure and an
+/// unparseable/truncated response — both are non-committable so the window is
+/// retried next run (a truncated array must NOT be treated as "empty and done",
+/// or the observations already generated are lost). A cleanly-parsed response
+/// with zero usable observations returns `Ok(vec![])`, which the caller commits
+/// because re-running reproduces it.
 async fn digest_window(
     provider: &dyn ChatProvider,
     session: &RawSession,
@@ -132,18 +176,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 {} ({}); NOT committing cursor so \
+             the window is retried next run: {e:#}",
+            session.source.kind.as_str(),
+            session.source.session_id.as_deref().unwrap_or("?")
+        );
+        DigestError::Unparseable(e)
+    })?;
     Ok(parsed
         .observations
         .into_iter()
diff --git a/src/memory/persona/distill_tests.rs b/src/memory/persona/distill_tests.rs
index 82541e9..39ee544 100644
--- a/src/memory/persona/distill_tests.rs
+++ b/src/memory/persona/distill_tests.rs
@@ -67,7 +67,7 @@ async fn tolerates_prose_wrapped_json() {
 }
 
 #[tokio::test]
-async fn soft_falls_back_on_error_and_bad_json() {
+async fn hard_and_unparseable_are_non_committable_errors() {
     let session = session_with(&[("x", EvidenceTier::T2)]);
 
     // A hard provider failure surfaces as Err (so the caller won't commit the
@@ -77,12 +77,40 @@ async fn soft_falls_back_on_error_and_bad_json() {
     };
     assert!(digest_session(&failing, &session).await.is_err());
 
-    // A received-but-unparseable response is a soft failure: Ok + empty digest
-    // (re-running reproduces it, so the cursor may commit).
+    // A received-but-unparseable response is ALSO an Err now (B3): it must NOT be
+    // treated as a committable empty digest, because "no JSON at all" is
+    // indistinguishable from a response that was cut off before any observation
+    // could be read. Committing it would mark the window done and drop it.
     let garbage = MockChat {
         body: Ok("not json at all".into()),
     };
-    assert!(digest_session(&garbage, &session).await.unwrap().is_empty());
+    assert!(digest_session(&garbage, &session).await.is_err());
+}
+
+/// A response truncated mid-array (a well-formed prefix with the closing `]`
+/// missing — exactly what a hit output-token cap produces) must surface as an
+/// Err, never as a committable empty digest. This is the data-loss case B3
+/// fixes: the model *did* produce observations, so silently dropping the window
+/// and committing its cursor loses them forever.
+#[tokio::test]
+async fn truncated_json_array_is_a_non_committable_error() {
+    // A well-formed prefix: two complete observation objects, but the array's
+    // closing `]` and the outer `}` never arrive (the model hit its output cap).
+    // Both parse attempts in `parse_digest` fail — the raw string, and the
+    // first-`{`..last-`}` slice, which still lacks the `]`/`}` — so this reports
+    // "EOF while parsing a list" rather than degrading to an empty digest.
+    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 result = digest_session(&provider, &session).await;
+    assert!(
+        result.is_err(),
+        "a truncated observation array must be a retryable Err, got: {result:?}"
+    );
 }
 
 #[tokio::test]
diff --git a/src/memory/persona/pipeline.rs b/src/memory/persona/pipeline.rs
index b65d29b..22d6aa8 100644
--- a/src/memory/persona/pipeline.rs
+++ b/src/memory/persona/pipeline.rs
@@ -322,8 +322,10 @@ impl Pipeline<'_> {
             let digest = match result {
                 Ok(d) => d,
                 Err(e) => {
-                    // Hard provider failure: do NOT commit the cursor, so this
-                    // session is re-attempted on the next run.
+                    // Non-committable failure — either a hard provider error or a
+                    // truncated/unparseable window (see `distill::DigestError`).
+                    // Do NOT commit the cursor, so this session is re-attempted on
+                    // the next run and its observations are not silently dropped.
                     log::warn!("[persona] digest failed, cursor not committed: {e:#}");
                     report.sessions_failed += 1;
                     continue;
@@ -335,8 +337,10 @@ impl Pipeline<'_> {
                 report.observations += digest.observations.len();
                 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).
+            // Commit the cursor/watermark now that the session is folded. Only a
+            // cleanly-digested session reaches here (a truncated/failed one took
+            // the `continue` above), so committing a *genuinely* empty digest is
+            // safe — re-running would reproduce it, not recover lost work.
             if let Some((key, value)) = &p.commit {
                 self.store.set(state::NAMESPACE, key, value).await?;
             }
diff --git a/src/memory/persona/pipeline_tests.rs b/src/memory/persona/pipeline_tests.rs
index f7c6a28..b30a5d7 100644
--- a/src/memory/persona/pipeline_tests.rs
+++ b/src/memory/persona/pipeline_tests.rs
@@ -39,6 +39,24 @@ impl ChatProvider for FailChat {
     }
 }
 
+/// A provider that returns a **truncated** observation array — a well-formed
+/// prefix with the closing `]`/`}` missing, exactly what a hit output-token cap
+/// produces. The response parses to neither valid JSON nor a recoverable
+/// `{...}` slice, so the digest is a non-committable failure (B3).
+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())
+    }
+}
+
 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}"}}}}"#
@@ -211,6 +229,67 @@ async fn hard_provider_failure_does_not_commit_cursor() {
     assert!(second.observations >= 2);
 }
 
+#[tokio::test]
+async fn truncated_digest_does_not_commit_cursor() {
+    // A window that truncates at the output-token cap must NOT checkpoint its
+    // transcript: the observations the model already produced would be lost if we
+    // marked the file done. Assert the file cursor is absent from the store after
+    // the run, and that a later working run re-processes the file.
+    let (ws, src, cfg, persona) = setup();
+    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, 0,
+        "truncated digests commit nothing"
+    );
+    assert_eq!(report.sessions_failed, 2, "both transcripts truncated");
+    assert_eq!(report.observations, 0);
+
+    // The transcript cursors must be absent — nothing was committed for them.
+    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 NOT be committed after a truncated digest, got: {stored:?}"
+        );
+    }
+
+    // A later working run re-digests both un-committed transcripts (evidence was
+    // retained, not silently dropped).
+    let good = MockChat;
+    let second = Pipeline {
+        config: &cfg,
+        persona: &persona,
+        provider: &good,
+        summariser: &summariser,
+        store: &store,
+    }
+    .run(RunMode::Incremental)
+    .await
+    .unwrap();
+    assert_eq!(
+        second.sessions_processed, 2,
+        "truncated sessions were retried on the next run"
+    );
+    assert!(second.observations >= 2);
+}
+
 #[tokio::test]
 async fn removed_directive_drops_out_on_rerun() {
     // Editing an instruction file (removing a rule) must drop the stale rule

From 5ffcfe08c9416c35b9a300f321a92e80214091d6 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Wed, 12 Aug 2026 20:24:06 +0530
Subject: [PATCH 02/10] docs(persona): code-span the private DigestError in
 digest_session docs

---
 src/memory/persona/distill.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index 728a12a..f37834a 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -110,7 +110,7 @@ fn windows(session: &RawSession) -> Vec<String> {
 ///   exhausted, 401/403, transport) *and* a response we could not parse (most
 ///   often a **truncated** JSON array — the model hit its output-token cap
 ///   mid-list, so the observations it *did* find would be lost forever if we
-///   committed). See [`DigestError`].
+///   committed). See the module-private `DigestError`.
 /// - **Genuinely empty digest** (a valid call whose response parsed to zero
 ///   observations, e.g. `{"observations":[]}`) → returns `Ok` with an empty
 ///   digest. Re-running would reproduce it, so the cursor IS committed.

From 30a392d4e89bc734bdb24d1dab1b384220c5e1c2 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 17:29:01 +0530
Subject: [PATCH 03/10] fix(persona): recover truncated windows and never
 starve the queue
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The truncation fix traded silent data-loss for two new failure modes a
reviewer flagged:

- A deterministically-unparseable window returned `Err` forever, so the
  caller never committed its cursor. With files selected oldest-first and a
  bounded per-pass budget, a run of such windows at the head of the queue
  consumed the whole budget every pass and starved every newer session,
  re-paying for each failed digest at temperature 0.0 where retrying can
  never help.
- `digest_session` aborted the whole session on the first bad window with
  `?`, discarding sibling windows that had already digested cleanly — a
  regression versus the pre-fix behaviour, which kept the good windows.

Digest each window independently and accumulate their observations, so one
bad window no longer discards its clean siblings. On a truncated/unparseable
window, recover in-process by re-splitting it into smaller pieces whose
responses fit under the output-token cap (bounded by MAX_RESPLIT_DEPTH /
MIN_WINDOW_CHARS); a piece still unparseable at the floor is dropped and
tallied in the new `RunReport::windows_lost` and its cursor committed, so a
genuinely-broken window makes progress instead of looping forever. Only a
provider/transport failure stays a non-committable `Err`.

`digest_session` now returns `SessionOutcome { digest, windows_lost }`.

Tests: re-split recovery keeps all observations with zero loss; an always-
truncating provider terminates and counts the loss instead of hanging; a
multi-window session retains the clean window when a sibling is poison; and
a cleanly-parsed empty digest still commits its cursor (the counterpart
invariant). Module/report docs updated to describe the recover-or-count
contract instead of the removed soft-fallback.
---
 src/memory/persona/distill.rs        | 229 ++++++++++++++++++++++-----
 src/memory/persona/distill_tests.rs  | 201 ++++++++++++++++++-----
 src/memory/persona/pipeline.rs       |  40 +++--
 src/memory/persona/pipeline_tests.rs | 129 ++++++++++++---
 src/memory/persona/reduce_tests.rs   |   5 +-
 5 files changed, 482 insertions(+), 122 deletions(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index f37834a..12f3003 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -3,10 +3,21 @@
 //! 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 anyhow::Result;
 use serde::Deserialize;
@@ -34,6 +45,16 @@ const WINDOW_CHARS: usize = 12_000;
 /// [`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;
+/// Smallest sub-window truncation-recovery ([`digest_window_recovering`]) will
+/// produce. Below this a response that still won't parse is treated as genuinely
+/// broken (not merely output-capped) and dropped-with-a-count rather than split
+/// further — the floor that 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;
 
 /// The strict-JSON system prompt: schema + extraction contract.
 fn system_prompt() -> String {
@@ -101,74 +122,153 @@ 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` still commits the session's cursor: the failure is
+/// deterministic (temperature `0.0`), so retrying would only re-burn budget
+/// without recovering anything, and holding the cursor would starve every newer
+/// session behind it. The count is surfaced in the run report so the drop is
+/// visible, 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 for.
+#[derive(Debug, Clone)]
+pub struct SessionOutcome {
+    /// The observations distilled from every digested (or recovered) window.
+    pub digest: SessionDigest,
+    /// Windows dropped after recovery could not parse them (data intentionally
+    /// skipped to keep the queue moving).
+    pub windows_lost: usize,
+}
+
+/// Digest one session into a [`SessionOutcome`] via the chat provider.
 ///
-/// Distinguishes two outcomes so the pipeline can checkpoint correctly:
-/// - **Non-committable failure** → returns `Err`. The caller must NOT commit the
-///   session's cursor, so the evidence is re-attempted on the next run. This
-///   covers both a provider/transport error (a `chat_for_json` error — budget
-///   exhausted, 401/403, transport) *and* a response we could not parse (most
-///   often a **truncated** JSON array — the model hit its output-token cap
-///   mid-list, so the observations it *did* find would be lost forever if we
-///   committed). See the module-private `DigestError`.
-/// - **Genuinely empty digest** (a valid call whose response parsed to zero
-///   observations, e.g. `{"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
+///   [`digest_window_recovering`]); a piece that still won't parse at the minimum
+///   size is dropped and tallied in [`SessionOutcome::windows_lost`]. The cursor
+///   is still committed — the failure is deterministic, so retrying is pure waste.
+/// - **Genuinely empty digest** (`{"observations":[]}`) → `Ok` with an empty
+///   digest and `windows_lost = 0`. Re-running reproduces it, so the cursor commits.
 pub async fn digest_session(
     provider: &dyn ChatProvider,
     session: &RawSession,
-) -> Result<SessionDigest> {
+) -> 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 OR an unparseable/truncated window bubbles up
-        // as `Err` (the whole session is retried next run, nothing committed);
-        // only a cleanly-parsed empty window is tolerated as `Ok(vec![])`.
-        let obs = digest_window(provider, session, &window).await?;
+        // Each window recovers from truncation on its own and never aborts its
+        // siblings; only a hard provider failure bubbles up as `Err` (the whole
+        // session is then retried next run, nothing committed).
+        let (obs, lost) = digest_window_recovering(provider, session, &window)
+            .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,
     })
 }
 
-/// Why a window could not be digested into committable observations.
-///
-/// Both variants are **retryable** — the caller must not commit the session's
-/// cursor for either, so the window is re-attempted on the next run. They are
-/// distinguished only for logging/telemetry clarity. A genuinely-empty result is
-/// *not* an error (it is `Ok(vec![])`); this type exists so a truncated or
-/// otherwise unparseable response can no longer masquerade as "empty" and be
-/// silently committed (the data-loss bug this fixes).
+/// 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.
+    /// 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
-    /// `]`). Committing would drop the observations the model *did* produce.
+    /// `]`). Recoverable by re-splitting the window into smaller pieces.
     #[error("digest response unparseable (likely truncated at the output cap): {0:#}")]
     Unparseable(#[source] anyhow::Error),
 }
 
+/// 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,
+) -> 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).await {
+            Ok(obs) => observations.extend(obs),
+            // Transient — abort recovery and let the whole session retry.
+            Err(DigestError::Provider(e)) => return Err(DigestError::Provider(e)),
+            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 `Err` for **both** a `chat_for_json` failure and an
-/// unparseable/truncated response — both are non-committable so the window is
-/// retried next run (a truncated array must NOT be treated as "empty and done",
-/// or the observations already generated are lost). A cleanly-parsed response
-/// with zero usable observations returns `Ok(vec![])`, which the caller commits
-/// because re-running reproduces it.
+/// 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>> {
+) -> Result<Vec<DigestObservation>, DigestError> {
     let prompt = ChatPrompt {
         system: system_prompt(),
         user: user_prompt(session, window),
@@ -182,8 +282,8 @@ async fn digest_window(
         .map_err(DigestError::Provider)?;
     let parsed: RawDigest = parse_digest(&raw).map_err(|e| {
         log::warn!(
-            "[persona] digest parse failed for {} ({}); NOT committing cursor so \
-             the window is retried next run: {e:#}",
+            "[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("?")
         );
@@ -196,6 +296,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 39ee544..17c2c73 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,15 @@ 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)])
+}
+
 #[tokio::test]
 async fn parses_observations_from_json() {
     let body = r#"{"observations":[
@@ -47,11 +111,13 @@ 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).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,44 +127,50 @@ 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).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 hard_and_unparseable_are_non_committable_errors() {
+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());
+}
 
-    // A received-but-unparseable response is ALSO an Err now (B3): it must NOT be
-    // treated as a committable empty digest, because "no JSON at all" is
-    // indistinguishable from a response that was cut off before any observation
-    // could be read. Committing it would mark the window done and drop it.
-    let garbage = MockChat {
-        body: Ok("not json at all".into()),
-    };
-    assert!(digest_session(&garbage, &session).await.is_err());
+/// 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).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"
+    );
 }
 
-/// A response truncated mid-array (a well-formed prefix with the closing `]`
-/// missing — exactly what a hit output-token cap produces) must surface as an
-/// Err, never as a committable empty digest. This is the data-loss case B3
-/// fixes: the model *did* produce observations, so silently dropping the window
-/// and committing its cursor loses them forever.
+/// 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 truncated_json_array_is_a_non_committable_error() {
-    // A well-formed prefix: two complete observation objects, but the array's
-    // closing `]` and the outer `}` never arrive (the model hit its output cap).
-    // Both parse attempts in `parse_digest` fail — the raw string, and the
-    // first-`{`..last-`}` slice, which still lacks the `]`/`}` — so this reports
-    // "EOF while parsing a list" rather than degrading to an empty digest.
+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"}"#;
@@ -106,10 +178,38 @@ async fn truncated_json_array_is_a_non_committable_error() {
         body: Ok(truncated.into()),
     };
     let session = session_with(&[("x", EvidenceTier::T2)]);
-    let result = digest_session(&provider, &session).await;
+    let outcome = digest_session(&provider, &session).await.unwrap();
+    assert_eq!(
+        outcome.windows_lost, 1,
+        "an irreducible truncation is dropped-and-counted, not looped forever"
+    );
+    assert!(outcome.digest.observations.is_empty());
+}
+
+/// 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",
+    };
+    let outcome = digest_session(&provider, &session).await.unwrap();
+    assert_eq!(
+        outcome.windows_lost, 1,
+        "only the irreducible poison window is lost"
+    );
     assert!(
-        result.is_err(),
-        "a truncated observation array must be a retryable Err, got: {result:?}"
+        !outcome.digest.observations.is_empty(),
+        "the clean sibling window's observations are retained"
     );
 }
 
@@ -119,10 +219,26 @@ async fn empty_session_yields_empty_digest() {
         body: Ok("{\"observations\":[]}".into()),
     };
     let session = RawSession::new(EvidenceSource::new(PersonaSourceKind::Codex));
-    assert!(digest_session(&provider, &session)
-        .await
-        .unwrap()
-        .is_empty());
+    let outcome = digest_session(&provider, &session).await.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).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]
@@ -137,7 +253,10 @@ 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).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 22d6aa8..4caefc6 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, 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,9 +62,16 @@ 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, budget,
+    /// or auth (cursor NOT committed; the whole session is retried next run).
+    /// Truncated/unparseable windows are recovered or counted in [`Self::windows_lost`],
+    /// not here.
     pub sessions_failed: usize,
+    /// Windows dropped because their digest stayed unparseable even after
+    /// truncation-recovery re-splitting. Their session's cursor IS committed (the
+    /// failure is deterministic at temperature 0.0), so this is data intentionally
+    /// skipped to keep the queue moving — surfaced so the drop is never silent.
+    pub windows_lost: usize,
     /// Observations distilled.
     pub observations: usize,
     /// Per-facet observation counts (facet wire-string → count).
@@ -313,34 +320,39 @@ impl Pipeline<'_> {
             return Ok(());
         }
         let concurrency = self.persona.digest_concurrency.max(1);
-        let results: Vec<Result<SessionDigest>> = stream::iter(selected.iter())
+        let results: Vec<Result<SessionOutcome>> = stream::iter(selected.iter())
             .map(|p| digest_session(self.provider, &p.session))
             .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) => {
-                    // Non-committable failure — either a hard provider error or a
-                    // truncated/unparseable window (see `distill::DigestError`).
+                    // Non-committable *provider* failure (transport/budget/auth).
                     // Do NOT commit the cursor, so this session is re-attempted on
                     // the next run and its observations are not silently dropped.
-                    log::warn!("[persona] digest failed, cursor not committed: {e:#}");
+                    // Truncated/unparseable windows never reach here — they are
+                    // recovered or dropped-and-counted inside `digest_session`.
+                    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;
             if !digest.is_empty() {
                 report.digests += 1;
                 report.observations += digest.observations.len();
                 fold_digest(self.config, &digest, asks, self.summariser, state).await?;
             }
-            // Commit the cursor/watermark now that the session is folded. Only a
-            // cleanly-digested session reaches here (a truncated/failed one took
-            // the `continue` above), so committing a *genuinely* empty digest is
-            // safe — re-running would reproduce it, not recover lost work.
+            // Commit the cursor/watermark now that the session is folded. A session
+            // reaches here once its windows are digested, recovered, or
+            // deterministically dropped-and-counted (`outcome.windows_lost`); only
+            // a provider failure took the `continue` above. Committing is safe —
+            // re-running a recovered or genuinely-empty session reproduces it, and
+            // a dropped window would only fail identically at temperature 0.0.
             if let Some((key, value)) = &p.commit {
                 self.store.set(state::NAMESPACE, key, value).await?;
             }
diff --git a/src/memory/persona/pipeline_tests.rs b/src/memory/persona/pipeline_tests.rs
index b30a5d7..06b0702 100644
--- a/src/memory/persona/pipeline_tests.rs
+++ b/src/memory/persona/pipeline_tests.rs
@@ -39,10 +39,11 @@ impl ChatProvider for FailChat {
     }
 }
 
-/// A provider that returns a **truncated** observation array — a well-formed
-/// prefix with the closing `]`/`}` missing, exactly what a hit output-token cap
-/// produces. The response parses to neither valid JSON nor a recoverable
-/// `{...}` slice, so the digest is a non-committable failure (B3).
+/// 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 {
@@ -57,6 +58,20 @@ impl ChatProvider for TruncatedChat {
     }
 }
 
+/// 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}"}}}}"#
@@ -230,11 +245,12 @@ async fn hard_provider_failure_does_not_commit_cursor() {
 }
 
 #[tokio::test]
-async fn truncated_digest_does_not_commit_cursor() {
-    // A window that truncates at the output-token cap must NOT checkpoint its
-    // transcript: the observations the model already produced would be lost if we
-    // marked the file done. Assert the file cursor is absent from the store after
-    // the run, and that a later working run re-processes the file.
+async fn unrecoverable_truncation_commits_and_counts_loss() {
+    // A window that stays truncated even after in-process re-splitting (the tiny
+    // transcript windows here are already below the re-split floor) must NOT hold
+    // its cursor forever — that would starve every newer session behind it. It is
+    // dropped-and-counted in `windows_lost` and the cursor commits, so a later run
+    // skips it rather than re-burning budget on a deterministically-bad window.
     let (ws, src, cfg, persona) = setup();
     let summariser = ConcatSummariser::new();
     let store = FileStateStore::open_in_workspace(ws.path()).unwrap();
@@ -250,13 +266,21 @@ async fn truncated_digest_does_not_commit_cursor() {
     .await
     .unwrap();
     assert_eq!(
-        report.sessions_processed, 0,
-        "truncated digests commit nothing"
+        report.sessions_processed, 2,
+        "both sessions are processed (recovery bottomed out, no provider error)"
+    );
+    assert_eq!(
+        report.sessions_failed, 0,
+        "a truncation is not a provider failure"
+    );
+    assert_eq!(
+        report.windows_lost, 2,
+        "both truncated windows are dropped-and-counted"
     );
-    assert_eq!(report.sessions_failed, 2, "both transcripts truncated");
     assert_eq!(report.observations, 0);
 
-    // The transcript cursors must be absent — nothing was committed for them.
+    // The transcript cursors ARE committed — the loss is deterministic, so the
+    // queue must move on instead of retrying forever.
     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"] {
@@ -265,18 +289,16 @@ async fn truncated_digest_does_not_commit_cursor() {
             .await
             .unwrap();
         assert!(
-            stored.is_none(),
-            "cursor for {name} must NOT be committed after a truncated digest, got: {stored:?}"
+            stored.is_some(),
+            "cursor for {name} must be committed after an unrecoverable truncation"
         );
     }
 
-    // A later working run re-digests both un-committed transcripts (evidence was
-    // retained, not silently dropped).
-    let good = MockChat;
+    // A later incremental run skips both — no starvation, no repeated paid calls.
     let second = Pipeline {
         config: &cfg,
         persona: &persona,
-        provider: &good,
+        provider: &TruncatedChat,
         summariser: &summariser,
         store: &store,
     }
@@ -284,10 +306,73 @@ async fn truncated_digest_does_not_commit_cursor() {
     .await
     .unwrap();
     assert_eq!(
-        second.sessions_processed, 2,
-        "truncated sessions were retried on the next run"
+        second.sessions_processed, 0,
+        "committed sessions are not re-digested"
     );
-    assert!(second.observations >= 2);
+    assert!(
+        second.sessions_skipped >= 2,
+        "both truncated sessions are cursor-skipped on the next run"
+    );
+}
+
+#[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, persona) = setup();
+    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]
diff --git a/src/memory/persona/reduce_tests.rs b/src/memory/persona/reduce_tests.rs
index 00b1857..d75c25c 100644
--- a/src/memory/persona/reduce_tests.rs
+++ b/src/memory/persona/reduce_tests.rs
@@ -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))
+            .await
+            .unwrap()
+            .digest;
         fold_digest(&config, &digest, &asks, &summariser, &mut state)
             .await
             .unwrap();

From a44ba2ca7ebf0344521c2184c6a98162d7dd4b71 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 17:53:24 +0530
Subject: [PATCH 04/10] docs(persona): drop private intra-doc link from
 digest_session

The public digest_session doc linked the private digest_window_recovering,
tripping rustdoc::private_intra_doc_links under the Rust SDK doc gate
(RUSTDOCFLAGS=-D warnings). Reference it as a plain code span instead.
---
 src/memory/persona/distill.rs | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index 12f3003..32ee471 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -150,8 +150,8 @@ pub struct SessionOutcome {
 ///   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
-///   [`digest_window_recovering`]); a piece that still won't parse at the minimum
-///   size is dropped and tallied in [`SessionOutcome::windows_lost`]. The cursor
+///   the private `digest_window_recovering`); a piece that still won't parse at the
+///   minimum size is dropped and tallied in [`SessionOutcome::windows_lost`]. The cursor
 ///   is still committed — the failure is deterministic, so retrying is pure waste.
 /// - **Genuinely empty digest** (`{"observations":[]}`) → `Ok` with an empty
 ///   digest and `windows_lost = 0`. Re-running reproduces it, so the cursor commits.

From affbacdc296c5383702c129dc51fe47d988ed110 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 18:07:36 +0530
Subject: [PATCH 05/10] fix(persona): enforce max_llm_calls across windows and
 recovery
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The run's provider-call budget was charged once per selected session, so it
undercounted multi-window sessions and — after the truncation-recovery change
— ignored recovery entirely: an unparseable window can fan out to ~15 provider
calls, and a session has many windows, so `max_llm_calls` could be overrun
substantially on a paid path.

Introduce a shared, concurrency-safe `CallBudget` (an `Arc<AtomicUsize>` of
remaining calls) consumed one permit per `chat_for_json`, so windowing and
every recovery re-try count against the ceiling. When the budget is spent
mid-session, `digest_window` returns `DigestError::BudgetExhausted`, recovery
stops, and the session is left non-committable — a clean checkpoint that
resumes next run, classified via `is_budget_exhausted` so the pipeline records
it as `budget_hit` rather than a failure. The per-session `Budget` now gates
only the session count; the call ceiling is enforced precisely per call.

Tests: recovery aborts as a budget checkpoint under a one-call budget
(`recovery_respects_the_call_budget`), and end-to-end a `max_llm_calls` of 1
commits exactly one of two sessions and resumes the other on a later run
(`call_budget_exhaustion_checkpoints_and_resumes`).
---
 src/memory/persona/distill.rs        | 75 ++++++++++++++++++++++++--
 src/memory/persona/distill_tests.rs  | 56 ++++++++++++++++----
 src/memory/persona/pipeline.rs       | 79 +++++++++++++++++++---------
 src/memory/persona/pipeline_tests.rs | 68 ++++++++++++++++++++++++
 src/memory/persona/reduce_tests.rs   |  4 +-
 5 files changed, 241 insertions(+), 41 deletions(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index 32ee471..f853026 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -19,6 +19,9 @@
 //! 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;
 
@@ -56,6 +59,49 @@ const MIN_WINDOW_CHARS: usize = 1_500;
 /// 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 {
     let facets = PersonaFacet::ALL
@@ -155,9 +201,13 @@ pub struct SessionOutcome {
 ///   is still committed — the failure is deterministic, so retrying is pure waste.
 /// - **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,
+    budget: &CallBudget,
 ) -> Result<SessionOutcome> {
     if session.is_empty() {
         return Ok(SessionOutcome {
@@ -169,9 +219,10 @@ pub async fn digest_session(
     let mut windows_lost = 0usize;
     for window in windows(session) {
         // Each window recovers from truncation on its own and never aborts its
-        // siblings; only a hard provider failure bubbles up as `Err` (the whole
-        // session is then retried next run, nothing committed).
-        let (obs, lost) = digest_window_recovering(provider, session, &window)
+        // 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);
@@ -198,6 +249,11 @@ enum DigestError {
     /// `]`). 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.
@@ -218,6 +274,7 @@ 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;
@@ -225,10 +282,12 @@ async fn digest_window_recovering(
     // 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).await {
+        match digest_window(provider, session, &piece, budget).await {
             Ok(obs) => observations.extend(obs),
-            // Transient — abort recovery and let the whole session retry.
+            // 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 {
@@ -268,7 +327,13 @@ async fn digest_window(
     provider: &dyn ChatProvider,
     session: &RawSession,
     window: &str,
+    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),
diff --git a/src/memory/persona/distill_tests.rs b/src/memory/persona/distill_tests.rs
index 17c2c73..3066f32 100644
--- a/src/memory/persona/distill_tests.rs
+++ b/src/memory/persona/distill_tests.rs
@@ -111,7 +111,9 @@ async fn parses_observations_from_json() {
         body: Ok(body.into()),
     };
     let session = session_with(&[("commit small and often", EvidenceTier::T2)]);
-    let outcome = digest_session(&provider, &session).await.unwrap();
+    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);
@@ -127,7 +129,9 @@ async fn tolerates_prose_wrapped_json() {
         body: Ok(body.into()),
     };
     let session = session_with(&[("cargo test", EvidenceTier::T2)]);
-    let outcome = digest_session(&provider, &session).await.unwrap();
+    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);
@@ -141,7 +145,9 @@ async fn provider_failure_is_a_non_committable_error() {
     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 truncated window that a smaller call *can* parse must be **recovered**, not
@@ -154,7 +160,9 @@ async fn truncated_window_is_recovered_by_resplitting() {
     // 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).await.unwrap();
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
     assert_eq!(
         outcome.windows_lost, 0,
         "a re-splittable truncation must recover, not drop"
@@ -165,6 +173,26 @@ async fn truncated_window_is_recovered_by_resplitting() {
     );
 }
 
+/// 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
@@ -178,7 +206,9 @@ async fn permanently_unparseable_window_terminates_and_is_counted() {
         body: Ok(truncated.into()),
     };
     let session = session_with(&[("x", EvidenceTier::T2)]);
-    let outcome = digest_session(&provider, &session).await.unwrap();
+    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"
@@ -202,7 +232,9 @@ async fn one_bad_window_does_not_discard_the_clean_siblings() {
     let provider = MarkerChat {
         marker: "POISONWINDOW",
     };
-    let outcome = digest_session(&provider, &session).await.unwrap();
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
     assert_eq!(
         outcome.windows_lost, 1,
         "only the irreducible poison window is lost"
@@ -219,7 +251,9 @@ async fn empty_session_yields_empty_digest() {
         body: Ok("{\"observations\":[]}".into()),
     };
     let session = RawSession::new(EvidenceSource::new(PersonaSourceKind::Codex));
-    let outcome = digest_session(&provider, &session).await.unwrap();
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
     assert!(outcome.digest.is_empty());
     assert_eq!(outcome.windows_lost, 0);
 }
@@ -233,7 +267,9 @@ async fn clean_empty_observations_commit_without_loss() {
         body: Ok("{\"observations\":[]}".into()),
     };
     let session = session_with(&[("just chatting, no rules here", EvidenceTier::T3)]);
-    let outcome = digest_session(&provider, &session).await.unwrap();
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
     assert!(outcome.digest.is_empty());
     assert_eq!(
         outcome.windows_lost, 0,
@@ -253,7 +289,9 @@ async fn drops_unusable_observations() {
         body: Ok(body.into()),
     };
     let session = session_with(&[("db", EvidenceTier::T2)]);
-    let outcome = digest_session(&provider, &session).await.unwrap();
+    let outcome = digest_session(&provider, &session, &CallBudget::unlimited())
+        .await
+        .unwrap();
     assert_eq!(outcome.digest.observations.len(), 1);
     assert_eq!(
         outcome.digest.observations[0].observation,
diff --git a/src/memory/persona/pipeline.rs b/src/memory/persona/pipeline.rs
index 4caefc6..dc5518a 100644
--- a/src/memory/persona/pipeline.rs
+++ b/src/memory/persona/pipeline.rs
@@ -16,7 +16,7 @@ use serde::Serialize;
 
 use super::compile::{write_pack, PackInputs};
 use super::config::PersonaConfig;
-use super::distill::{digest_session, SessionOutcome};
+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;
@@ -82,30 +82,28 @@ pub struct RunReport {
     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;
     }
 }
 
@@ -130,6 +128,9 @@ impl Pipeline<'_> {
         let asks = self.persona.asks();
         let mut state = ReduceState::default();
         let mut budget = Budget::from(self.persona);
+        // One provider-call budget shared across every digest source this run, so
+        // transcripts and git history draw from the same `max_llm_calls` ceiling.
+        let call_budget = CallBudget::new(self.persona.run_budget.max_llm_calls as usize);
         let mut report = RunReport {
             mode: mode.as_str().to_string(),
             ..Default::default()
@@ -142,15 +143,31 @@ 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,
+            &call_budget,
+            &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,
+            &call_budget,
+            &mut report,
+        )
+        .await?;
 
-        report.budget_hit = budget.exhausted();
+        // Either ceiling stopping the run counts as a budget hit — don't clobber a
+        // call-budget checkpoint recorded during digest with the session count.
+        report.budget_hit |= budget.exhausted();
 
         // 4. Seal facet trees + compile the pack.
         let bodies = seal_and_collect(self.config, &asks, self.summariser).await?;
@@ -246,6 +263,7 @@ impl Pipeline<'_> {
         asks: &FacetAsks,
         state: &mut ReduceState,
         budget: &mut Budget,
+        call_budget: &CallBudget,
         report: &mut RunReport,
     ) -> Result<()> {
         let mut files: Vec<(PathBuf, &'static str)> = Vec::new();
@@ -288,7 +306,7 @@ 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, call_budget, report)
             .await
     }
 
@@ -304,9 +322,11 @@ impl Pipeline<'_> {
         asks: &FacetAsks,
         state: &mut ReduceState,
         budget: &mut Budget,
+        call_budget: &CallBudget,
         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() {
@@ -321,7 +341,7 @@ impl Pipeline<'_> {
         }
         let concurrency = self.persona.digest_concurrency.max(1);
         let results: Vec<Result<SessionOutcome>> = stream::iter(selected.iter())
-            .map(|p| digest_session(self.provider, &p.session))
+            .map(|p| digest_session(self.provider, &p.session, call_budget))
             .buffered(concurrency)
             .collect()
             .await;
@@ -329,13 +349,21 @@ impl Pipeline<'_> {
             let outcome = match result {
                 Ok(o) => o,
                 Err(e) => {
-                    // Non-committable *provider* failure (transport/budget/auth).
-                    // Do NOT commit the cursor, so this session is re-attempted on
-                    // the next run and its observations are not silently dropped.
-                    // Truncated/unparseable windows never reach here — they are
-                    // recovered or dropped-and-counted inside `digest_session`.
-                    log::warn!("[persona] digest provider failure, 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;
                 }
             };
@@ -367,6 +395,7 @@ impl Pipeline<'_> {
         asks: &FacetAsks,
         state: &mut ReduceState,
         budget: &mut Budget,
+        call_budget: &CallBudget,
         report: &mut RunReport,
     ) -> Result<()> {
         use super::readers::git_history::{self, GitReadConfig};
@@ -418,7 +447,7 @@ impl Pipeline<'_> {
                 pending.push(Pending { session, commit });
             }
         }
-        self.digest_and_fold(pending, asks, state, budget, report)
+        self.digest_and_fold(pending, asks, state, budget, call_budget, report)
             .await
     }
 }
diff --git a/src/memory/persona/pipeline_tests.rs b/src/memory/persona/pipeline_tests.rs
index 06b0702..b29c86c 100644
--- a/src/memory/persona/pipeline_tests.rs
+++ b/src/memory/persona/pipeline_tests.rs
@@ -180,6 +180,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();
diff --git a/src/memory/persona/reduce_tests.rs b/src/memory/persona/reduce_tests.rs
index d75c25c..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,7 @@ 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))
+        let digest = digest_session(&provider, &session(scope), &CallBudget::unlimited())
             .await
             .unwrap()
             .digest;

From 305102737cc46b12e3727e40f18f2cfa5f1a0863 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 18:23:36 +0530
Subject: [PATCH 06/10] fix(persona): withhold cursors on a systemic zero-yield
 digest run
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Committing a fully-lost session (zero observations, ≥1 window dropped) is
right for a localized permanent-garbage window, but if the provider fails
systemically — wrong/non-instruct model, refusal mode, a proxy returning
prose — every session drops, every cursor commits with zero observations, and
later runs skip the whole backlog, needing manual state deletion to recover.
That is the original data-loss bug returning at run scale.

Defer the cursor commit for any fully-lost session and resolve it once the
run's outcome is known: if the run produced observations anywhere, the
fully-lost sessions are localized failures and their commits are applied (the
queue still advances, no starvation); if the run yielded nothing despite
dropped windows, withhold every deferred commit, set the new
`RunReport::systemic_digest_failure`, and log an error so the backlog is
retried once the cause is fixed rather than silently skipped.

The call-budget and deferred list travel together as `DigestGuards` through
the ingest sources. Docs corrected: `windows_lost` counts recovery leaf
sub-windows (a 12k window yields several), the effective re-split floor is
~2×MIN_WINDOW_CHARS, and temperature-0 determinism is a heuristic not a
guarantee.

Tests: `systemic_truncation_withholds_commits_and_retries` (all-lost run
withholds, flags, and retries next run) and
`fully_lost_session_commits_when_the_run_yields_observations` (a lone poison
session amid a healthy run still commits — no starvation).
---
 src/memory/persona/distill.rs        |  25 +++--
 src/memory/persona/pipeline.rs       | 112 ++++++++++++++-----
 src/memory/persona/pipeline_tests.rs | 161 +++++++++++++++++++++++----
 3 files changed, 236 insertions(+), 62 deletions(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index f853026..78e8727 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -48,10 +48,11 @@ const WINDOW_CHARS: usize = 12_000;
 /// [`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;
-/// Smallest sub-window truncation-recovery ([`digest_window_recovering`]) will
-/// produce. Below this a response that still won't parse is treated as genuinely
-/// broken (not merely output-capped) and dropped-with-a-count rather than split
-/// further — the floor that guarantees recovery terminates.
+/// Recovery stops splitting once the *half* it would produce falls below this, so
+/// the effective smallest window it will re-digest is ~2× this value (a ~3 000-char
+/// piece is the last one split). Below that a response that still won't parse is
+/// treated as genuinely broken (not merely output-capped) and dropped-with-a-count
+/// rather than split further — the floor that 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
@@ -172,13 +173,15 @@ fn windows(session: &RawSession) -> Vec<String> {
 /// **dropped** because they stayed unparseable even after truncation-recovery
 /// re-splitting.
 ///
-/// A non-zero `windows_lost` still commits the session's cursor: the failure is
-/// deterministic (temperature `0.0`), so retrying would only re-burn budget
-/// without recovering anything, and holding the cursor would starve every newer
-/// session behind it. The count is surfaced in the run report so the drop is
-/// visible, 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 for.
+/// 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.
diff --git a/src/memory/persona/pipeline.rs b/src/memory/persona/pipeline.rs
index dc5518a..dca213e 100644
--- a/src/memory/persona/pipeline.rs
+++ b/src/memory/persona/pipeline.rs
@@ -67,10 +67,13 @@ pub struct RunReport {
     /// Truncated/unparseable windows are recovered or counted in [`Self::windows_lost`],
     /// not here.
     pub sessions_failed: usize,
-    /// Windows dropped because their digest stayed unparseable even after
-    /// truncation-recovery re-splitting. Their session's cursor IS committed (the
-    /// failure is deterministic at temperature 0.0), so this is data intentionally
-    /// skipped to keep the queue moving — surfaced so the drop is never silent.
+    /// 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,
@@ -78,6 +81,12 @@ pub struct RunReport {
     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>,
 }
@@ -107,6 +116,16 @@ impl Budget {
     }
 }
 
+/// 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)>,
+}
+
 /// The pipeline binds the workspace, config, provider, summariser, and state
 /// store; `run` executes one pass.
 pub struct Pipeline<'a> {
@@ -128,9 +147,13 @@ impl Pipeline<'_> {
         let asks = self.persona.asks();
         let mut state = ReduceState::default();
         let mut budget = Budget::from(self.persona);
-        // One provider-call budget shared across every digest source this run, so
-        // transcripts and git history draw from the same `max_llm_calls` ceiling.
-        let call_budget = CallBudget::new(self.persona.run_budget.max_llm_calls as usize);
+        // 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()
@@ -148,7 +171,7 @@ impl Pipeline<'_> {
             &asks,
             &mut state,
             &mut budget,
-            &call_budget,
+            &mut guards,
             &mut report,
         )
         .await?;
@@ -160,11 +183,36 @@ impl Pipeline<'_> {
             &asks,
             &mut state,
             &mut budget,
-            &call_budget,
+            &mut guards,
             &mut report,
         )
         .await?;
 
+        // 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 {
+                    self.store.set(state::NAMESPACE, key, value).await?;
+                }
+            } 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(),
+                );
+            }
+        }
+
         // Either ceiling stopping the run counts as a budget hit — don't clobber a
         // call-budget checkpoint recorded during digest with the session count.
         report.budget_hit |= budget.exhausted();
@@ -263,7 +311,7 @@ impl Pipeline<'_> {
         asks: &FacetAsks,
         state: &mut ReduceState,
         budget: &mut Budget,
-        call_budget: &CallBudget,
+        guards: &mut DigestGuards,
         report: &mut RunReport,
     ) -> Result<()> {
         let mut files: Vec<(PathBuf, &'static str)> = Vec::new();
@@ -306,23 +354,24 @@ impl Pipeline<'_> {
                 .map(|v| (key, v));
             pending.push(Pending { session, commit });
         }
-        self.digest_and_fold(pending, asks, state, budget, call_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,
-        call_budget: &CallBudget,
+        guards: &mut DigestGuards,
         report: &mut RunReport,
     ) -> Result<()> {
         // Select within the session-count budget up front; the provider-call
@@ -341,7 +390,7 @@ impl Pipeline<'_> {
         }
         let concurrency = self.persona.digest_concurrency.max(1);
         let results: Vec<Result<SessionOutcome>> = stream::iter(selected.iter())
-            .map(|p| digest_session(self.provider, &p.session, call_budget))
+            .map(|p| digest_session(self.provider, &p.session, &guards.call_budget))
             .buffered(concurrency)
             .collect()
             .await;
@@ -370,19 +419,28 @@ impl Pipeline<'_> {
             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 session
-            // reaches here once its windows are digested, recovered, or
-            // deterministically dropped-and-counted (`outcome.windows_lost`); only
-            // a provider failure took the `continue` above. Committing is safe —
-            // re-running a recovered or genuinely-empty session reproduces it, and
-            // a dropped window would only fail identically at temperature 0.0.
-            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 {
+                // Committed now that the session is folded: a recovered or
+                // genuinely-empty (zero-loss) session reproduces on re-run, so
+                // committing loses nothing.
+                self.store
+                    .set(state::NAMESPACE, &commit.0, &commit.1)
+                    .await?;
             }
         }
         Ok(())
@@ -395,7 +453,7 @@ impl Pipeline<'_> {
         asks: &FacetAsks,
         state: &mut ReduceState,
         budget: &mut Budget,
-        call_budget: &CallBudget,
+        guards: &mut DigestGuards,
         report: &mut RunReport,
     ) -> Result<()> {
         use super::readers::git_history::{self, GitReadConfig};
@@ -447,7 +505,7 @@ impl Pipeline<'_> {
                 pending.push(Pending { session, commit });
             }
         }
-        self.digest_and_fold(pending, asks, state, budget, call_budget, report)
+        self.digest_and_fold(pending, asks, state, budget, guards, report)
             .await
     }
 }
diff --git a/src/memory/persona/pipeline_tests.rs b/src/memory/persona/pipeline_tests.rs
index b29c86c..36f966a 100644
--- a/src/memory/persona/pipeline_tests.rs
+++ b/src/memory/persona/pipeline_tests.rs
@@ -58,6 +58,30 @@ impl ChatProvider for TruncatedChat {
     }
 }
 
+/// 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.
@@ -313,12 +337,12 @@ async fn hard_provider_failure_does_not_commit_cursor() {
 }
 
 #[tokio::test]
-async fn unrecoverable_truncation_commits_and_counts_loss() {
-    // A window that stays truncated even after in-process re-splitting (the tiny
-    // transcript windows here are already below the re-split floor) must NOT hold
-    // its cursor forever — that would starve every newer session behind it. It is
-    // dropped-and-counted in `windows_lost` and the cursor commits, so a later run
-    // skips it rather than re-burning budget on a deterministically-bad window.
+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, persona) = setup();
     let summariser = ConcatSummariser::new();
     let store = FileStateStore::open_in_workspace(ws.path()).unwrap();
@@ -333,22 +357,19 @@ async fn unrecoverable_truncation_commits_and_counts_loss() {
     .run(RunMode::Backfill)
     .await
     .unwrap();
-    assert_eq!(
-        report.sessions_processed, 2,
-        "both sessions are processed (recovery bottomed out, no provider error)"
-    );
+    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,
-        "both truncated windows are dropped-and-counted"
-    );
+    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 committed — the loss is deterministic, so the
-    // queue must move on instead of retrying forever.
+    // 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"] {
@@ -357,16 +378,16 @@ async fn unrecoverable_truncation_commits_and_counts_loss() {
             .await
             .unwrap();
         assert!(
-            stored.is_some(),
-            "cursor for {name} must be committed after an unrecoverable truncation"
+            stored.is_none(),
+            "cursor for {name} must be withheld under a systemic digest failure"
         );
     }
 
-    // A later incremental run skips both — no starvation, no repeated paid calls.
+    // Once the provider works, a later run re-digests the withheld backlog.
     let second = Pipeline {
         config: &cfg,
         persona: &persona,
-        provider: &TruncatedChat,
+        provider: &MockChat,
         summariser: &summariser,
         store: &store,
     }
@@ -374,12 +395,104 @@ async fn unrecoverable_truncation_commits_and_counts_loss() {
     .await
     .unwrap();
     assert_eq!(
-        second.sessions_processed, 0,
-        "committed sessions are not re-digested"
+        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
+
+    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!(
-        second.sessions_skipped >= 2,
-        "both truncated sessions are cursor-skipped on the next run"
+        !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"
     );
 }
 

From 85cc1a3ee94b253c529ec08a46d01ce9526d9ad1 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 18:57:41 +0530
Subject: [PATCH 07/10] docs(persona): correct recovery-floor + commit
 contract, test the descent
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address CodeRabbit's re-review of the systemic-guard change:

- `MIN_WINDOW_CHARS` doc overstated the floor. Recovery re-digests pieces down
  to ~MIN_WINDOW_CHARS (1500) — a 3000-char piece splits into 1500-char pieces
  that each reach `digest_window`; a piece is only split while its half stays
  ≥ the floor, so ~1500 pieces are digested but not split again. Reworded.
- `digest_session`'s truncated-window outcome doc claimed the cursor is always
  committed on an irreducible drop. It's committed only when the run produced
  observations elsewhere; a run that yielded nothing despite dropped windows
  withholds the cursor and flags `systemic_digest_failure`. Doc now states this.
- Added `large_permanently_unparseable_window_terminates_within_bounded_calls`,
  which walks the full 12k→6k→3k→1.5k descent (the prior test only hit the
  immediate below-floor drop), asserting bounded termination and dropped leaves.
---
 src/memory/persona/distill.rs       | 18 +++++++++++-------
 src/memory/persona/distill_tests.rs | 26 ++++++++++++++++++++++++++
 2 files changed, 37 insertions(+), 7 deletions(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index 78e8727..bb9f134 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -48,11 +48,11 @@ const WINDOW_CHARS: usize = 12_000;
 /// [`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;
-/// Recovery stops splitting once the *half* it would produce falls below this, so
-/// the effective smallest window it will re-digest is ~2× this value (a ~3 000-char
-/// piece is the last one split). Below that a response that still won't parse is
-/// treated as genuinely broken (not merely output-capped) and dropped-with-a-count
-/// rather than split further — the floor that guarantees recovery terminates.
+/// 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
@@ -200,8 +200,12 @@ pub struct SessionOutcome {
 ///   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`]. The cursor
-///   is still committed — the failure is deterministic, so retrying is pure waste.
+///   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
diff --git a/src/memory/persona/distill_tests.rs b/src/memory/persona/distill_tests.rs
index 3066f32..530c581 100644
--- a/src/memory/persona/distill_tests.rs
+++ b/src/memory/persona/distill_tests.rs
@@ -216,6 +216,32 @@ async fn permanently_unparseable_window_terminates_and_is_counted() {
     assert!(outcome.digest.observations.is_empty());
 }
 
+/// The bounded re-splitting *descent*: a large window that never parses must walk
+/// the full 12k → 6k → 3k → 1.5k chain, drop every leaf piece, and terminate
+/// within a bounded number of calls — never loop. (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(12_000);
+    let budget = CallBudget::new(1_000);
+    let outcome = digest_session(&provider, &session, &budget).await.unwrap();
+    assert!(outcome.digest.observations.is_empty());
+    assert!(
+        outcome.windows_lost >= 1,
+        "the descent's leaf pieces are dropped-and-counted"
+    );
+    let calls = 1_000 - budget.remaining();
+    assert!(
+        (1..=60).contains(&calls),
+        "recovery must terminate within a bounded call count, spent {calls}"
+    );
+}
+
 /// 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]

From 9422747a49aeeb004fab259af0823d7cac1fc173 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 19:11:31 +0530
Subject: [PATCH 08/10] refactor(persona): fix budget_hit false-positive,
 harden commits, split file
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address CodeRabbit's re-review of the systemic-guard change:

- `budget_hit` was re-derived in `run` from `budget.exhausted()`, which is true
  whenever a run merely fills its session allowance exactly — dropping nothing.
  Removed: the selection loop already sets `budget_hit` precisely when it drops a
  pending session, and the call-budget path sets it on `BudgetExhausted`, so the
  flag now reflects only a real early stop.
- The deferred-commit loop used `?`, so one failed cursor write aborted the whole
  run and discarded the reduce output (the pack). It now logs and continues per
  key — a cursor is a fast-skip, not a correctness gate, so the session simply
  re-digests next run.
- Split the git-history ingestion (`ingest_git` + its helpers, feature-gated)
  into a sibling `pipeline_git.rs`, bringing `pipeline.rs` back under the repo's
  500-line limit (557 → 492).
- Pinned `max_llm_calls` in the two truncation pipeline tests so they isolate
  loss behavior and don't break if the default call ceiling is lowered later.

Tests: persona 65, persona+git-diff 69 (exercises the moved ingest_git),
all-features and persona matrix green; fmt / default clippy / doc gate clean.
---
 src/memory/persona/pipeline.rs       | 111 +++++----------------------
 src/memory/persona/pipeline_git.rs   |  99 ++++++++++++++++++++++++
 src/memory/persona/pipeline_tests.rs |   4 +-
 3 files changed, 123 insertions(+), 91 deletions(-)
 create mode 100644 src/memory/persona/pipeline_git.rs

diff --git a/src/memory/persona/pipeline.rs b/src/memory/persona/pipeline.rs
index dca213e..a3282bb 100644
--- a/src/memory/persona/pipeline.rs
+++ b/src/memory/persona/pipeline.rs
@@ -126,6 +126,12 @@ struct DigestGuards {
     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> {
@@ -197,7 +203,16 @@ impl Pipeline<'_> {
         if !guards.deferred.is_empty() {
             if report.observations > 0 {
                 for (key, value) in &guards.deferred {
-                    self.store.set(state::NAMESPACE, key, value).await?;
+                    // 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;
@@ -213,9 +228,11 @@ impl Pipeline<'_> {
             }
         }
 
-        // Either ceiling stopping the run counts as a budget hit — don't clobber a
-        // call-budget checkpoint recorded during digest with the session count.
-        report.budget_hit |= budget.exhausted();
+        // `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?;
@@ -445,69 +462,6 @@ impl Pipeline<'_> {
         }
         Ok(())
     }
-
-    #[cfg(feature = "git-diff")]
-    async fn ingest_git(
-        &self,
-        mode: RunMode,
-        asks: &FacetAsks,
-        state: &mut ReduceState,
-        budget: &mut Budget,
-        guards: &mut DigestGuards,
-        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, guards, report)
-            .await
-    }
 }
 
 /// A read-but-not-yet-digested session plus an optional state commit (cursor or
@@ -532,29 +486,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 36f966a..8470dc6 100644
--- a/src/memory/persona/pipeline_tests.rs
+++ b/src/memory/persona/pipeline_tests.rs
@@ -343,7 +343,8 @@ async fn systemic_truncation_withholds_commits_and_retries() {
     // — 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, persona) = setup();
+    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();
 
@@ -439,6 +440,7 @@ async fn fully_lost_session_commits_when_the_run_yields_observations() {
     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();

From 40579a4d9a1244dd533844aa37b9b7298202f5ab Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 19:32:52 +0530
Subject: [PATCH 09/10] test(persona): pin recovery-tree call count to the
 exact bound
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

CodeRabbit: the terminal-path test's (1..=60) window detected a hang but not a
fan-out regression at 16-60 calls. Size the window to MIN_WINDOW_CHARS <<
MAX_RESPLIT_DEPTH and assert the exact tree — calls == 2^(depth+1)-1 = 15, and
windows_lost == 2^depth = 8 — derived from the constants so it tracks them.
---
 src/memory/persona/distill_tests.rs | 23 ++++++++++++-----------
 1 file changed, 12 insertions(+), 11 deletions(-)

diff --git a/src/memory/persona/distill_tests.rs b/src/memory/persona/distill_tests.rs
index 530c581..b015a3c 100644
--- a/src/memory/persona/distill_tests.rs
+++ b/src/memory/persona/distill_tests.rs
@@ -216,9 +216,11 @@ async fn permanently_unparseable_window_terminates_and_is_counted() {
     assert!(outcome.digest.observations.is_empty());
 }
 
-/// The bounded re-splitting *descent*: a large window that never parses must walk
-/// the full 12k → 6k → 3k → 1.5k chain, drop every leaf piece, and terminate
-/// within a bounded number of calls — never loop. (The test above covers the
+/// 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() {
@@ -227,18 +229,17 @@ async fn large_permanently_unparseable_window_terminates_within_bounded_calls()
     let provider = MockChat {
         body: Ok(truncated.into()),
     };
-    let session = big_session(12_000);
+    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());
-    assert!(
-        outcome.windows_lost >= 1,
-        "the descent's leaf pieces are dropped-and-counted"
-    );
+    // 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();
-    assert!(
-        (1..=60).contains(&calls),
-        "recovery must terminate within a bounded call count, spent {calls}"
+    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}"
     );
 }
 

From a0489290d158e01165219d9c48b2dbbc8767d263 Mon Sep 17 00:00:00 2001
From: shanu <shanu@tinyhumans.ai>
Date: Tue, 18 Aug 2026 21:01:44 +0530
Subject: [PATCH 10/10] docs+test(persona): fix budget doc drift, align commit
 policy, multibyte test
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address CodeRabbit's re-review of the CallBudget/recovery changes:

- `SessionOutcome::windows_lost` doc now states it counts recovery leaf pieces,
  not input windows (a fully-lost window adds 2^MAX_RESPLIT_DEPTH).
- `RunReport::sessions_failed` doc no longer lists "budget" as a cause — a spent
  call budget is reported in `budget_hit`, matching the code that routes
  `is_budget_exhausted` there.
- The immediate cursor commit now logs-and-continues on a store write error,
  same as the deferred path, so one failed write can't `?`-abort the run and
  discard the pack.
- Pinned `max_llm_calls` in `clean_empty_digest_commits_cursor`.
- Added `recovery_splits_multibyte_windows_without_panicking`, proving the
  char-boundary re-split walks a multibyte window without a byte-slice panic and
  drops the same 2^depth leaves as its ASCII twin.

Deferred (replied on the PR): the git walk's blocking-in-async is pre-existing
and preserved unchanged by the module extraction; and pipeline_tests.rs at 603
lines is consistent with the repo's existing test files (store_settle_tests 628,
store_embed_tests 619) — the 500-line norm targets implementation files, and
pipeline.rs itself is 492.
---
 src/memory/persona/distill.rs        |  7 +++++--
 src/memory/persona/distill_tests.rs  | 29 ++++++++++++++++++++++++++++
 src/memory/persona/pipeline.rs       | 23 +++++++++++++---------
 src/memory/persona/pipeline_tests.rs |  3 ++-
 4 files changed, 50 insertions(+), 12 deletions(-)

diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs
index bb9f134..565016b 100644
--- a/src/memory/persona/distill.rs
+++ b/src/memory/persona/distill.rs
@@ -186,8 +186,11 @@ fn windows(session: &RawSession) -> Vec<String> {
 pub struct SessionOutcome {
     /// The observations distilled from every digested (or recovered) window.
     pub digest: SessionDigest,
-    /// Windows dropped after recovery could not parse them (data intentionally
-    /// skipped to keep the queue moving).
+    /// 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,
 }
 
diff --git a/src/memory/persona/distill_tests.rs b/src/memory/persona/distill_tests.rs
index b015a3c..bd2c1d1 100644
--- a/src/memory/persona/distill_tests.rs
+++ b/src/memory/persona/distill_tests.rs
@@ -101,6 +101,14 @@ fn big_session(total_chars: usize) -> RawSession {
     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":[
@@ -243,6 +251,27 @@ async fn large_permanently_unparseable_window_terminates_within_bounded_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]
diff --git a/src/memory/persona/pipeline.rs b/src/memory/persona/pipeline.rs
index a3282bb..076b645 100644
--- a/src/memory/persona/pipeline.rs
+++ b/src/memory/persona/pipeline.rs
@@ -62,10 +62,11 @@ 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 — transport, budget,
-    /// or auth (cursor NOT committed; the whole session is retried next run).
-    /// Truncated/unparseable windows are recovered or counted in [`Self::windows_lost`],
-    /// not here.
+    /// 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
@@ -451,13 +452,17 @@ impl Pipeline<'_> {
                 // only if the run produced observations somewhere, otherwise it
                 // withholds and flags rather than silently skipping the backlog.
                 guards.deferred.push(commit.clone());
-            } else {
+            } 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.
-                self.store
-                    .set(state::NAMESPACE, &commit.0, &commit.1)
-                    .await?;
+                // 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(())
diff --git a/src/memory/persona/pipeline_tests.rs b/src/memory/persona/pipeline_tests.rs
index 8470dc6..992d85b 100644
--- a/src/memory/persona/pipeline_tests.rs
+++ b/src/memory/persona/pipeline_tests.rs
@@ -504,7 +504,8 @@ async fn clean_empty_digest_commits_cursor() {
     // 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, persona) = setup();
+    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();