fix(persona): retain observations when a digest is truncated

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 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.
 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()

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]

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?;
             }

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