diff --git a/src/memory/persona/distill.rs b/src/memory/persona/distill.rs index 8d0e31f..f37834a 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 { /// 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 = 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 { + 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