diff --git a/crates/tracedecay-privacy/src/detect.rs b/crates/tracedecay-privacy/src/detect.rs index 18ea9068c6..d5b8514a6e 100644 --- a/crates/tracedecay-privacy/src/detect.rs +++ b/crates/tracedecay-privacy/src/detect.rs @@ -409,9 +409,9 @@ pub enum DetectionError { #[error("privacy sanitizer quarantined an ambiguous structured document")] StructuredQuarantine, /// The document parsed, but key-anchored material cannot be redacted in - /// place: an object key carries credential material, or a key-proven - /// sensitive field cannot be located byte-exactly. Fail-closed quarantine, - /// not a receipt fault. + /// place: an object key carries credential material and the parser gave + /// no span to rewrite it at (YAML, the JSON value route), or the rewritten + /// document no longer parses. Fail-closed quarantine, not a receipt fault. #[error("privacy sanitizer quarantined credential-bearing keys")] CredentialKeyQuarantine, /// A parsed sensitive field could not be mapped back to one exact byte diff --git a/crates/tracedecay-privacy/src/structured_text.rs b/crates/tracedecay-privacy/src/structured_text.rs index 6d6230c3a4..60cb97d8e8 100644 --- a/crates/tracedecay-privacy/src/structured_text.rs +++ b/crates/tracedecay-privacy/src/structured_text.rs @@ -198,12 +198,15 @@ fn sanitize_structured_text_with( .map_err(detection_error_from_structured_sanitization)?; let mut quarantine_findings = Vec::new(); - let candidates = if parsed.fields.is_empty() { + let (candidates, keys) = if parsed.fields.is_empty() { tree_candidates(raw, &parsed, &policy, patterns, &mut quarantine_findings) } else { - line_candidates(raw, &parsed.fields, &policy, patterns) + ( + line_candidates(raw, &parsed.fields, &policy, patterns), + RedactedKeys::default(), + ) }; - if candidates.is_empty() && quarantine_findings.is_empty() { + if candidates.is_empty() && keys.redactions.is_empty() && quarantine_findings.is_empty() { let mut scanned = raw_only(raw, patterns); scanned.format = Some(parsed.format); return Ok(scanned); @@ -212,7 +215,7 @@ fn sanitize_structured_text_with( let ranks = ordinal_ranks(&candidates)?; let candidate_count = u32::try_from(candidates.len()).map_err(|_| DetectionError::ScanLimitExceeded)?; - let mut redactions: Vec<(Range, &'static str)> = candidates + let mut redactions: Vec<(Range, &str)> = candidates .iter() .flat_map(|candidate| { candidate @@ -220,6 +223,11 @@ fn sanitize_structured_text_with( .iter() .map(|span| (span.clone(), candidate.marker)) }) + .chain( + keys.redactions + .iter() + .map(|key| (key.span.clone(), key.placeholder.as_str())), + ) .collect(); redactions.sort_by(|(left, _), (right, _)| { left.start @@ -228,7 +236,7 @@ fn sanitize_structured_text_with( }); redactions.dedup_by(|(later, _), (earlier, _)| later.start < earlier.end); - let mut findings = Vec::new(); + let mut findings = keys.findings; let mut sanitized_text = String::with_capacity(raw.len()); let mut cursor = 0usize; for (span, marker) in redactions { @@ -281,19 +289,50 @@ fn sanitize_structured_text_with( ); } + // Text-located spans (YAML only) can cover more than the value, so that + // redaction keeps its format label only if it still parses as the input's + // format. Parser spans replace exactly the value token. + let text_located = candidates.iter().any(|candidate| candidate.text_located); + let reparsed = (text_located || !keys.redactions.is_empty()).then(|| { + let Ok(Some(reparsed)) = parse_structured_text(&sanitized_text) else { + return false; + }; + if reparsed.format != parsed.format { + return false; + } + if !keys.redactions.is_empty() { + // TOML's semantic table merges repeated path components and can + // report only one occurrence's span. Valid syntax alone cannot + // prove every escaped occurrence was replaced. Inspect all keys + // decoded by the canonical parser before allowing durable output. + TreeFieldCollector { + policy: &policy, + patterns, + fields: TreeFields::default(), + quarantine_findings: &mut quarantine_findings, + } + .quarantine_remaining_keys(&reparsed.value); + } + true + }); + // Some unlocated occurrences instead make the raw-scan rewrite invalid. + // They retain the same typed key quarantine. + if !keys.redactions.is_empty() && reparsed == Some(false) { + for key in &keys.redactions { + redact_text( + &mut key.key.clone(), + "$", + patterns, + &mut quarantine_findings, + SanitizationActionV1::Quarantined, + ); + } + } + let format = reparsed.unwrap_or(true).then_some(parsed.format); findings.sort(); findings.dedup(); quarantine_findings.sort(); quarantine_findings.dedup(); - // Text-located spans (YAML only) can cover more than the value, so that - // redaction keeps its format label only if it still parses as the input's - // format. Parser spans replace exactly the value token. - let format = (!candidates.iter().any(|candidate| candidate.text_located) - || matches!( - parse_structured_text(&sanitized_text), - Ok(Some(ref reparsed)) if reparsed.format == parsed.format - )) - .then_some(parsed.format); Ok(StructuredTextSanitizationV1 { format, sanitized_text, @@ -423,19 +462,20 @@ fn tree_candidates( policy: &ConfiguredSensitiveKeyPolicy<'_>, patterns: &CredentialPatternSet, quarantine_findings: &mut Vec, -) -> Vec { +) -> (Vec, RedactedKeys) { let Some(document) = document_node(raw, parsed) else { quarantine_findings.push(unlocatable_sensitive_field()); - return Vec::new(); + return (Vec::new(), RedactedKeys::default()); }; - let mut sensitive = Vec::new(); - collect_tree_fields( - document, + let mut collector = TreeFieldCollector { policy, patterns, - &mut sensitive, + fields: TreeFields::default(), quarantine_findings, - ); + }; + collector.document(document); + let TreeFields { sensitive, keys } = collector.fields; + let quarantine_findings = collector.quarantine_findings; let mut candidates = Vec::new(); for SensitiveScalar { @@ -478,7 +518,7 @@ fn tree_candidates( text_located, }); } - candidates + (candidates, keys) } fn unlocatable_sensitive_field() -> SanitizationFindingV1 { @@ -493,13 +533,22 @@ fn unlocatable_sensitive_field() -> SanitizationFindingV1 { /// One node of a parsed tree document. enum DocumentNode<'a> { - Object(Vec<(Cow<'a, str>, DocumentNode<'a>)>), + Object(Vec<(DocumentKey<'a>, DocumentNode<'a>)>), Array(Vec>), Scalar(ScalarValue), /// A boolean, null, or empty string: nothing to redact. Empty, } +/// An object key and, when the format's parser reports it, the byte range a +/// bare placeholder can replace while the document stays valid. +struct DocumentKey<'a> { + name: Cow<'a, str>, + /// `None` for formats parsed without spans (YAML): such a key cannot be + /// rewritten in place. + span: Option>, +} + /// Where a scalar's value sits in the raw text. enum ScalarValue { /// The byte range the format's own parser read, and the marker that keeps @@ -560,9 +609,17 @@ fn json_node(value: Json<'_>) -> DocumentNode<'_> { .properties .into_iter() .map(|property| { + // A string key's range includes both quotes; the + // placeholder replaces the body only. let key = match property.name { - ObjectPropName::String(name) => name.value, - ObjectPropName::Word(name) => Cow::Borrowed(name.value), + ObjectPropName::String(name) => DocumentKey { + span: Some(name.range.start + 1..name.range.end.saturating_sub(1)), + name: name.value, + }, + ObjectPropName::Word(name) => DocumentKey { + span: Some(name.range.start..name.range.end), + name: Cow::Borrowed(name.value), + }, }; (key, json_node(property.value)) }) @@ -597,7 +654,15 @@ fn toml_table_node<'a>(raw: &str, table: DeTable<'a>) -> DocumentNode<'a> { DocumentNode::Object( table .into_iter() - .map(|(key, value)| (key.into_inner(), toml_node(raw, value))) + .map(|(key, value)| { + // The span covers the whole key token, quotes included, and + // the placeholder is a valid bare key. + let key = DocumentKey { + span: Some(key.span()), + name: key.into_inner(), + }; + (key, toml_node(raw, value)) + }) .collect(), ) } @@ -642,7 +707,13 @@ fn decoded_node(value: &Value) -> DocumentNode<'_> { Value::Object(fields) => DocumentNode::Object( fields .iter() - .map(|(key, child)| (Cow::Borrowed(key.as_str()), decoded_node(child))) + .map(|(key, child)| { + let key = DocumentKey { + name: Cow::Borrowed(key.as_str()), + span: None, + }; + (key, decoded_node(child)) + }) .collect(), ), Value::Array(items) => DocumentNode::Array(items.iter().map(decoded_node).collect()), @@ -662,130 +733,200 @@ struct SensitiveScalar { decoded_value_matched: bool, } -fn collect_tree_fields( - node: DocumentNode<'_>, - policy: &ConfiguredSensitiveKeyPolicy<'_>, - patterns: &CredentialPatternSet, - sensitive: &mut Vec, - quarantine_findings: &mut Vec, -) { - match node { - DocumentNode::Object(fields) => { - for (key, child) in fields { - let mut key_evidence = key.to_string(); - redact_text( - &mut key_evidence, - "$", - patterns, - quarantine_findings, - SanitizationActionV1::Quarantined, - ); - match policy.classify(&NormalizedSensitiveKey::new(&key)) { - Some(origin) => collect_scalars(child, &key, origin, patterns, sensitive), - None => { - collect_detected_scalars( - child, - &key, - policy, - patterns, - sensitive, - quarantine_findings, - ); - } +/// An object key the detectors flagged, located byte-exactly in the raw text. +struct KeyRedaction { + span: Range, + placeholder: String, + /// The decoded key, kept only so a rewrite that breaks the document can be + /// reported as the key quarantine it falls back to. + key: String, +} + +/// Prefix of the placeholder that replaces a flagged object key. Like +/// [`REDACTED_STRUCTURED_FIELD`] it is a valid bare TOML key and needs no JSON +/// escaping, so the rewritten document still parses as its own format. +const REDACTED_KEY_PREFIX: &str = "TraceDecay-redacted-key-"; + +/// The flagged keys of one document and the detector findings that flagged +/// them. +#[derive(Default)] +struct RedactedKeys { + redactions: Vec, + findings: Vec, +} + +/// Everything a walk over a parsed tree document collects. +#[derive(Default)] +struct TreeFields { + sensitive: Vec, + keys: RedactedKeys, +} + +struct TreeFieldCollector<'a> { + policy: &'a ConfiguredSensitiveKeyPolicy<'a>, + patterns: &'a CredentialPatternSet, + fields: TreeFields, + quarantine_findings: &'a mut Vec, +} + +impl TreeFieldCollector<'_> { + fn document(&mut self, node: DocumentNode<'_>) { + match node { + DocumentNode::Object(fields) => self.object(fields, None), + DocumentNode::Array(items) => { + for item in items { + self.detected_scalars(item, "$"); } } + DocumentNode::Scalar(_) | DocumentNode::Empty => {} } - DocumentNode::Array(items) => { - for item in items { - collect_detected_scalars( - item, - "$", - policy, - patterns, - sensitive, - quarantine_findings, - ); - } - } - DocumentNode::Scalar(_) | DocumentNode::Empty => {} } -} -fn collect_scalars( - node: DocumentNode<'_>, - key: &str, - origin: SanitizationDetectorOriginV1, - patterns: &CredentialPatternSet, - sensitive: &mut Vec, -) { - match node { - DocumentNode::Scalar(value) => { - let decoded_value_matched = value - .decoded_value() - .is_some_and(|decoded| trips_a_detector(decoded, patterns)); - sensitive.push(SensitiveScalar { - key: key.to_owned(), - value, - origin, - decoded_value_matched, + fn object( + &mut self, + fields: Vec<(DocumentKey<'_>, DocumentNode<'_>)>, + sensitive_parent: Option<(&str, SanitizationDetectorOriginV1)>, + ) { + let evidence: Vec<_> = fields + .iter() + .map(|(key, _)| self.key_evidence(key)) + .collect(); + // Placeholders are numbered in document order, which a TOML table's + // sorted map does not preserve, and skip any sibling's literal name. + let mut flagged: Vec<_> = (0..fields.len()) + .filter(|&index| evidence[index].is_some()) + .collect(); + flagged.sort_by_key(|&index| fields[index].0.span.as_ref().map(|span| span.start)); + let mut placeholders = vec![None; fields.len()]; + let mut ordinal = 0usize; + for index in flagged { + placeholders[index] = Some(loop { + ordinal += 1; + let placeholder = format!("{REDACTED_KEY_PREFIX}{ordinal}"); + if !fields.iter().any(|(key, _)| key.name == placeholder) { + break placeholder; + } }); } - DocumentNode::Object(fields) => { - for (_, child) in fields { - collect_scalars(child, key, origin, patterns, sensitive); + for (((key, child), evidence), placeholder) in + fields.into_iter().zip(evidence).zip(placeholders) + { + let origin = self + .policy + .classify(&NormalizedSensitiveKey::new(&key.name)); + match (key.span, evidence, placeholder) { + (Some(span), Some(evidence), Some(placeholder)) => { + self.fields.keys.findings.extend(evidence); + self.scalars( + child, + &placeholder, + origin + .or(sensitive_parent.map(|(_, origin)| origin)) + .unwrap_or(SanitizationDetectorOriginV1::BuiltInDetectorKernel), + ); + self.fields.keys.redactions.push(KeyRedaction { + span, + placeholder, + key: key.name.into_owned(), + }); + } + (_, evidence, _) => { + self.quarantine_findings + .extend(evidence.into_iter().flatten()); + match origin + .map(|origin| (key.name.as_ref(), origin)) + .or(sensitive_parent) + { + Some((key, origin)) => self.scalars(child, key, origin), + None => self.detected_scalars(child, &key.name), + } + } } } - DocumentNode::Array(items) => { - for item in items { - collect_scalars(item, key, origin, patterns, sensitive); + } + + /// Detector findings for a key that trips one, recorded as a redaction + /// when the key can be rewritten in place and as a quarantine otherwise. + fn key_evidence(&self, key: &DocumentKey<'_>) -> Option> { + let action = if key.span.is_some() { + SanitizationActionV1::Redacted + } else { + SanitizationActionV1::Quarantined + }; + let mut probe = key.name.to_string(); + let mut findings = Vec::new(); + redact_text(&mut probe, "$", self.patterns, &mut findings, action).then_some(findings) + } + + fn quarantine_remaining_keys(&mut self, value: &Value) { + match value { + Value::Object(fields) => { + for (name, child) in fields { + let key = DocumentKey { + name: Cow::Borrowed(name), + span: None, + }; + if let Some(evidence) = self.key_evidence(&key) { + self.quarantine_findings.extend(evidence); + } + self.quarantine_remaining_keys(child); + } + } + Value::Array(items) => { + for item in items { + self.quarantine_remaining_keys(item); + } } + _ => {} } - DocumentNode::Empty => {} } -} -fn collect_detected_scalars( - node: DocumentNode<'_>, - key: &str, - policy: &ConfiguredSensitiveKeyPolicy<'_>, - patterns: &CredentialPatternSet, - sensitive: &mut Vec, - quarantine_findings: &mut Vec, -) { - match node { - DocumentNode::Scalar(value) => { - if value - .decoded_value() - .is_some_and(|decoded| trips_a_detector(decoded, patterns)) - { - sensitive.push(SensitiveScalar { + fn scalars(&mut self, node: DocumentNode<'_>, key: &str, origin: SanitizationDetectorOriginV1) { + match node { + DocumentNode::Scalar(value) => { + let decoded_value_matched = value + .decoded_value() + .is_some_and(|decoded| trips_a_detector(decoded, self.patterns)); + self.fields.sensitive.push(SensitiveScalar { key: key.to_owned(), value, - origin: SanitizationDetectorOriginV1::BuiltInDetectorKernel, - decoded_value_matched: true, + origin, + decoded_value_matched, }); } + DocumentNode::Object(fields) => self.object(fields, Some((key, origin))), + DocumentNode::Array(items) => { + for item in items { + self.scalars(item, key, origin); + } + } + DocumentNode::Empty => {} } - DocumentNode::Object(fields) => collect_tree_fields( - DocumentNode::Object(fields), - policy, - patterns, - sensitive, - quarantine_findings, - ), - DocumentNode::Array(items) => { - for item in items { - collect_detected_scalars( - item, - key, - policy, - patterns, - sensitive, - quarantine_findings, - ); + } + + fn detected_scalars(&mut self, node: DocumentNode<'_>, key: &str) { + match node { + DocumentNode::Scalar(value) => { + if value + .decoded_value() + .is_some_and(|decoded| trips_a_detector(decoded, self.patterns)) + { + self.fields.sensitive.push(SensitiveScalar { + key: key.to_owned(), + value, + origin: SanitizationDetectorOriginV1::BuiltInDetectorKernel, + decoded_value_matched: true, + }); + } + } + DocumentNode::Object(fields) => self.object(fields, None), + DocumentNode::Array(items) => { + for item in items { + self.detected_scalars(item, key); + } } + DocumentNode::Empty => {} } - DocumentNode::Empty => {} } } diff --git a/crates/tracedecay-privacy/src/structured_text_tests.rs b/crates/tracedecay-privacy/src/structured_text_tests.rs index f33337183d..cbb28610fc 100644 --- a/crates/tracedecay-privacy/src/structured_text_tests.rs +++ b/crates/tracedecay-privacy/src/structured_text_tests.rs @@ -26,6 +26,7 @@ use super::structured_text::{ }; mod code_shape; +mod key_redaction; /// Ordinary words: no credential prefix, no digits, shorter than the entropy /// floor, and containing no substring the credential patterns know. @@ -505,25 +506,6 @@ fn lcm_json_credential_bearing_keys_quarantine_instead_of_faulting_the_receipt() ); } -#[test] -fn lcm_non_json_credential_bearing_keys_are_key_quarantine_not_parse_ambiguity() { - // Same key-quarantine contract as the JSON container path, reached through - // the non-JSON structured-text route: a parsed TOML table whose *key* is - // itself credential material. (TOML rather than `key: value`, which the - // format probe reads as an HTTP header block whose line path never scans - // keys.) The refusal must name the key quarantine, not claim the document - // was ambiguous. It parsed fine. - let credential_key = ["sk", "-test-", "1234567890abcdef"].concat(); - let raw = format!("{credential_key} = \"ordinary-value\"\nregion = \"us-east\"\n"); - - let error = sanitize_lcm_payload_text(&raw).expect_err("key quarantine"); - assert_eq!(error, DetectionError::CredentialKeyQuarantine); - assert_eq!( - error.to_string(), - "privacy sanitizer quarantined credential-bearing keys" - ); -} - #[test] fn prose_opening_with_a_bracketed_annotation_line_is_not_toml_quarantine() { // LCM compaction emits scaffold messages such as @@ -594,21 +576,6 @@ fn sanitizer_verdicts_stay_distinguishable_from_sanitizer_faults() { } } -#[test] -fn code_source_credential_bearing_keys_are_key_quarantine_not_parse_ambiguity() { - let credential_key = ["sk", "-test-", "1234567890abcdef"].concat(); - let raw = format!("{credential_key} = \"ordinary-value\"\nregion = \"us-east\"\n"); - - let error = sanitize_code_source_bytes(raw.as_bytes(), CodeSourceShapeV1::StructuredData) - .map(|_| ()) - .expect_err("key quarantine"); - assert_eq!(error, DetectionError::CredentialKeyQuarantine); - assert_eq!( - error.to_string(), - "privacy sanitizer quarantined credential-bearing keys" - ); -} - #[test] fn code_source_parse_ambiguity_stays_a_structured_quarantine() { let raw = format!("vault_passphrase: {PLACEHOLDER}\n broken: [unclosed\n"); diff --git a/crates/tracedecay-privacy/src/structured_text_tests/key_redaction.rs b/crates/tracedecay-privacy/src/structured_text_tests/key_redaction.rs new file mode 100644 index 0000000000..f588779517 --- /dev/null +++ b/crates/tracedecay-privacy/src/structured_text_tests/key_redaction.rs @@ -0,0 +1,319 @@ +//! Object keys that trip a detector are rewritten in place, never exempted. +//! +//! Detection stays exactly as strict for keys as for values and prose: a key +//! that reads as a high-entropy token or a credential is replaced by a +//! placeholder unique within its object, and every scalar under it is +//! redacted. The document keeps its format, so the rest of it stays indexable. + +use serde_json::Value; +use tracedecay_domain::SanitizerDispositionV1; + +use super::{ + CodeSourceShapeV1, DetectionError, StructuredTextFormatV1, sanitize_code_source_bytes, + sanitize_lcm_payload_text, sanitize_structured_text, +}; + +/// Passphrase-shaped secrets that read like identifiers. No shape rule can +/// tell them apart from long type or field names, so they must stay redacted +/// in every position. +const PASSPHRASES: [&str; 3] = [ + "violin-pocket-jungle-quartz-island-778", + "RocketButtonStapleBridgeYellowOrange90", + "MySuperSecretPasswordForProduction2024Admin", +]; + +fn sanitize_data(raw: &str) -> String { + let sanitized = sanitize_code_source_bytes(raw.as_bytes(), CodeSourceShapeV1::StructuredData) + .unwrap_or_else(|error| panic!("structured data sanitizes in place ({error}): {raw}")); + assert_eq!( + sanitized.receipt().disposition(), + SanitizerDispositionV1::Redacted + ); + String::from_utf8(sanitized.into_parts().0).expect("sanitized text is UTF-8") +} + +fn assert_fixed_point(sanitized: &str) { + let replayed = sanitize_structured_text(sanitized).expect("replay scan runs"); + assert!(replayed.quarantine_findings().is_empty()); + assert_eq!( + replayed.sanitized_text(), + sanitized, + "sanitizing the sanitized payload must be a fixed point" + ); +} + +#[test] +fn json_secret_keys_become_placeholders_and_their_values_are_redacted() { + for secret in PASSPHRASES { + let raw = format!(r#"{{"api_keys": {{"{secret}": "admin-user"}}, "region": "us-east"}}"#); + + let text = sanitize_data(&raw); + + assert!(!text.contains(secret), "secret key survived: {text}"); + assert!( + !text.contains("admin-user"), + "value under it survived: {text}" + ); + let parsed: Value = serde_json::from_str(&text).expect("still valid JSON"); + assert_eq!(parsed["region"], "us-east"); + assert_eq!( + parsed["api_keys"]["TraceDecay-redacted-key-1"], + "TraceDecay-redacted-sensitive-field" + ); + assert_fixed_point(&text); + } +} + +#[test] +fn placeholders_are_unique_within_their_object() { + let [first, second, third] = PASSPHRASES; + let raw = format!( + "{{\n \"TraceDecay-redacted-key-1\": \"kept\",\n \"{first}\": 1,\n \ + \"{second}\": {{\"nested\": \"x\"}},\n \"inner\": {{\"{third}\": true}}\n}}\n" + ); + + let text = sanitize_data(&raw); + + for secret in PASSPHRASES { + assert!(!text.contains(secret), "secret key survived: {text}"); + } + let parsed: Value = serde_json::from_str(&text).expect("still valid JSON"); + let keys: Vec<&str> = parsed + .as_object() + .expect("object") + .keys() + .map(String::as_str) + .collect(); + assert_eq!( + keys, + [ + "TraceDecay-redacted-key-1", + "TraceDecay-redacted-key-2", + "TraceDecay-redacted-key-3", + "inner" + ], + "{text}" + ); + assert_eq!(parsed["TraceDecay-redacted-key-1"], "kept"); + assert_eq!( + parsed["TraceDecay-redacted-key-3"]["nested"], + "TraceDecay-redacted-sensitive-field" + ); + assert_eq!(parsed["inner"]["TraceDecay-redacted-key-1"], true); + assert_fixed_point(&text); +} + +#[test] +fn schema_definitions_with_long_type_names_stay_indexable() { + let raw = r##"{ + "$defs": { + "WorkDuplicateAdjudicationAppendOutcomeV1": {"type": "string", "enum": ["appended"]}, + "ProjectId": {"description": "Canonical project identity.", "type": "string"} + }, + "properties": {"outcome": {"$ref": "#/$defs/ProjectId"}} +} +"##; + + let text = sanitize_data(raw); + + assert!( + !text.contains("WorkDuplicateAdjudicationAppendOutcomeV1"), + "{text}" + ); + assert!( + text.contains( + r#""ProjectId": {"description": "Canonical project identity.", "type": "string"}"# + ), + "ordinary definitions are untouched: {text}" + ); + assert!(text.contains(r##""$ref": "#/$defs/ProjectId""##), "{text}"); + assert!( + text.contains(r#""TraceDecay-redacted-key-1": {"type""#), + "{text}" + ); + serde_json::from_str::(&text).expect("still valid JSON"); +} + +#[test] +fn benchmark_receipt_with_a_long_target_name_stays_indexable() { + let raw = r#"{"target_results": {"clone_peak_rss_added_at_most_1_gib_for_39582_symbols": true, "cold_index_completed": true}}"#; + + let text = sanitize_data(raw); + + assert_eq!( + text, + r#"{"target_results": {"TraceDecay-redacted-key-1": true, "cold_index_completed": true}}"# + ); +} + +#[test] +fn escaped_json_keys_are_replaced_byte_exactly() { + let raw = r#"{"Violin-pocket-jungle-quartz-island-778": "x", "ok": "y"}"#; + + let text = sanitize_data(raw); + + assert_eq!( + text, + r#"{"TraceDecay-redacted-key-1": "TraceDecay-redacted-sensitive-field", "ok": "y"}"# + ); +} + +#[test] +fn nested_escaped_json_keys_are_redacted_under_a_flagged_parent() { + let secret = PASSPHRASES[0]; + let raw = format!( + r#"{{"{secret}":{{"violin-pocket-ju\u006egle-quartz-island-778":true,"ordinary":"private-value"}}}}"# + ); + let original: Value = serde_json::from_str(&raw).expect("real JSON with an escaped nested key"); + assert_eq!(original[secret][secret], true); + + let text = sanitize_data(&raw); + let parsed: Value = serde_json::from_str(&text).expect("redaction keeps valid JSON"); + assert!( + !serde_json::to_string(&parsed).unwrap().contains(secret), + "{text}" + ); + assert!( + !text.contains("private-value"), + "forced value redaction survives: {text}" + ); + assert_fixed_point(&text); +} + +fn assert_repeated_toml_keys_are_redacted_or_withheld(raw: &str) { + let original: toml::Value = toml::from_str(raw).expect("valid repeated TOML path components"); + assert!(original.as_table().unwrap().contains_key(PASSPHRASES[0])); + let sanitized = + match sanitize_code_source_bytes(raw.as_bytes(), CodeSourceShapeV1::StructuredData) { + Ok(sanitized) => sanitized, + Err(DetectionError::CredentialKeyQuarantine) => { + return; + } + Err(error) => panic!("expected exact redaction or typed key quarantine: {error}"), + }; + assert_eq!( + sanitized.receipt().disposition(), + SanitizerDispositionV1::Redacted + ); + let text = String::from_utf8(sanitized.into_parts().0).unwrap(); + let parsed: toml::Value = toml::from_str(&text).expect("redacted TOML remains valid"); + assert!( + !toml::to_string(&parsed).unwrap().contains(PASSPHRASES[0]), + "{text}" + ); + assert_fixed_point(&text); +} + +#[test] +fn repeated_escaped_toml_dotted_keys_do_not_leave_decoded_secrets() { + let raw = concat!( + "\"violin-pocket-ju\\u006egle-quartz-island-778\".a=true\n", + "\"violin-pocket-ju\\u006egle-quartz-island-778\".b=true\n", + ); + assert_repeated_toml_keys_are_redacted_or_withheld(raw); + assert_eq!( + sanitize_lcm_payload_text(raw).map(|_| ()), + Err(DetectionError::CredentialKeyQuarantine) + ); +} + +#[test] +fn repeated_escaped_toml_quoted_headers_do_not_leave_decoded_secrets() { + let raw = concat!( + "[\"violin-pocket-ju\\u006egle-quartz-island-778\".a]\nx=true\n", + "[\"violin-pocket-ju\\u006egle-quartz-island-778\".b]\nx=true\n", + ); + assert_repeated_toml_keys_are_redacted_or_withheld(raw); + // LCM's existing JSON-intent guard withholds a leading `["` before the + // structured-source route gets to identify the quoted TOML header. + assert_eq!( + sanitize_lcm_payload_text(raw).map(|_| ()), + Err(DetectionError::StructuredQuarantine) + ); +} + +#[test] +fn toml_secret_keys_become_bare_placeholders() { + let [first, second, _] = PASSPHRASES; + let raw = format!( + "[users]\n{first} = \"admin-user\"\n\"{second}\" = {{ role = \"ops\" }}\nregion = \"us-east\"\n" + ); + + let scanned = sanitize_structured_text(&raw).expect("structured scan runs"); + assert_eq!(scanned.format(), Some(StructuredTextFormatV1::Toml)); + assert!(scanned.quarantine_findings().is_empty()); + let text = scanned.sanitized_text(); + + assert_eq!( + text, + "[users]\nTraceDecay-redacted-key-1 = \"TraceDecay-redacted-sensitive-field\"\n\ + TraceDecay-redacted-key-2 = { role = \"TraceDecay-redacted-sensitive-field\" }\n\ + region = \"us-east\"\n" + ); + assert_fixed_point(text); +} + +#[test] +fn credential_keys_redact_in_place_on_code_source_and_lcm_text_routes() { + let credential_key = ["sk", "-test-", "1234567890abcdef"].concat(); + let raw = format!("{credential_key} = \"ordinary-value\"\nregion = \"us-east\"\n"); + let expected = "TraceDecay-redacted-key-1 = \"TraceDecay-redacted-sensitive-field\"\nregion = \"us-east\"\n"; + + assert_eq!(sanitize_data(&raw), expected); + let sanitized = sanitize_lcm_payload_text(&raw).expect("key redacts in place"); + assert_eq!(sanitized.sanitized_text(), expected); +} + +/// YAML is parsed without key spans, so a key cannot be located byte-exactly +/// and the document stays withheld. +#[test] +fn yaml_secret_keys_still_quarantine() { + for secret in PASSPHRASES { + let raw = format!("users:\n {secret}: admin-user\nregion: us-east\n"); + + let error = sanitize_code_source_bytes(raw.as_bytes(), CodeSourceShapeV1::StructuredData) + .map(|_| ()) + .expect_err("unlocatable key quarantines"); + + assert_eq!(error, DetectionError::CredentialKeyQuarantine); + } +} + +/// A TOML table named twice has one parsed key but two occurrences. The +/// second is only raw-scanned, which breaks the header, so the redacted +/// document no longer parses and the file stays withheld. +#[test] +fn toml_secret_keys_that_cannot_be_rewritten_validly_still_quarantine() { + let secret = PASSPHRASES[0]; + let raw = format!("[{secret}]\nname = \"a\"\n\n[{secret}.child]\nname = \"b\"\n"); + + let error = sanitize_code_source_bytes(raw.as_bytes(), CodeSourceShapeV1::StructuredData) + .map(|_| ()) + .expect_err("unrewritable key quarantines"); + + assert_eq!(error, DetectionError::CredentialKeyQuarantine); +} + +#[test] +fn passphrases_in_values_and_free_text_stay_redacted() { + for secret in PASSPHRASES { + for raw in [ + format!("{{\"note\": \"{secret}\"}}"), + format!("{{\"url\": \"https://example.test/hooks/{secret}/run\"}}"), + ] { + let text = sanitize_data(&raw); + assert!(!text.contains(secret), "{text}"); + } + for raw in [ + format!("let label = \"{secret}\";\n"), + format!("Remember to use {secret} when you type the password.\n"), + format!("type Alias = {secret};\n"), + ] { + let sanitized = + sanitize_code_source_bytes(raw.as_bytes(), CodeSourceShapeV1::CodeOrProse) + .expect("prose sanitizes"); + let text = String::from_utf8(sanitized.into_parts().0).expect("UTF-8"); + assert!(!text.contains(secret), "{text}"); + } + } +} diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index 94243cdde5..e85db36ce8 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -26,6 +26,8 @@ use tracedecay_store::{ }; mod compression_ownership; +#[cfg(unix)] +mod summarizer_retries; fn session(provider: &str, session_id: &str) -> SessionRecord { SessionRecord { diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests/summarizer_retries.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests/summarizer_retries.rs new file mode 100644 index 0000000000..6539467359 --- /dev/null +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests/summarizer_retries.rs @@ -0,0 +1,231 @@ +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use tracedecay_global_db::RegisteredGlobalDbLeaseV1; +use tracedecay_lcm::summary_convergence::next_candidate; +use tracedecay_runtime_core::db::engine::params; +use tracedecay_runtime_core::test_executable::write_executable_script; + +use super::{ProjectSummarizerFixture, message, run_with_pinned_summarizer_binding, session}; +use crate::lcm_summary_convergence::{ + LcmSummaryConvergenceDisposition, run_summary_convergence_page, +}; +use crate::session_temporal_refresh_scheduler::registry::session_refresh_retry_delay; +use crate::session_temporal_refresh_scheduler::wake::SessionTemporalRefreshRetryClass; + +async fn seed_session(database: &RegisteredGlobalDbLeaseV1, session_id: &str) { + assert!( + database + .upsert_session(&session("cursor", session_id)) + .await + ); + for ordinal in 1..=8 { + let mut raw = message(session_id, ordinal); + raw.message_id = format!("{session_id}-{ordinal}"); + database + .lcm_ingest_raw_message(database.db_path().parent().unwrap(), &raw) + .await + .unwrap(); + } +} + +struct RetryState { + state: String, + failure_code: Option, + failure_count: i64, + next_attempt_at_ms: i64, +} + +async fn retry_state(database: &RegisteredGlobalDbLeaseV1, session_id: &str) -> RetryState { + let snapshot = database.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT state, failure_code, failure_count, next_attempt_at_ms + FROM lcm_summary_convergence_queue + WHERE provider = 'cursor' AND session_id = ?1", + params![session_id], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + RetryState { + state: row.get(0).unwrap(), + failure_code: row.get(1).unwrap(), + failure_count: row.get(2).unwrap(), + next_attempt_at_ms: row.get(3).unwrap(), + } +} + +#[test] +fn a_summarizer_timeout_is_counted_then_converges_under_the_same_binding() { + run_with_pinned_summarizer_binding(async { + let fixture = ProjectSummarizerFixture::open().await; + let database = fixture.db(); + let session_id = "timeout-then-recovery"; + seed_session(&database, session_id).await; + let executable = fixture.root.path().join("retrying-cursor-agent"); + let launches = fixture.root.path().join("summarizer-launches"); + write_executable_script( + &executable, + format!( + "#!/bin/sh\nif [ ! -e '{}' ]; then\n printf '%s\\n' first > '{}'\n exec /bin/sleep 60\nfi\nprintf '%s\\n' retry >> '{}'\nprintf '%s\\n' 'summary produced by retry'\n", + launches.display(), launches.display(), launches.display(), + ), + ) + .unwrap(); + fixture.pin_cursor_agent(&executable); + let binding = crate::lcm_summarization::summarizer_binding_identity(&database).unwrap(); + + // Model a session that already reached the bounded backoff plateau. + // The real provider run below must increment, rather than erase, its + // counted failures and start the next delay after that run settles. + let transaction = database.begin_write_transaction().await.unwrap(); + transaction + .execute( + "UPDATE lcm_summary_convergence_queue + SET state = 'retryable', failure_code = 'deadline_exceeded', + failure_count = 6, next_attempt_at_ms = 0, + attempted_raw_store_id = newest_raw_store_id + WHERE provider = 'cursor' AND session_id = ?1", + params![session_id], + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let before_run_ms = i64::try_from( + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis(), + ) + .unwrap(); + let failed = run_summary_convergence_page(database.clone(), 1) + .await + .unwrap(); + assert_eq!(failed.sessions.len(), 1); + assert_eq!(failed.sessions[0].summary_nodes_created, 0); + assert_eq!( + failed.sessions[0].disposition, + LcmSummaryConvergenceDisposition::Retryable { + reason: "deadline_exceeded".to_owned() + }, + ); + let retry = retry_state(&database, session_id).await; + assert_eq!(retry.state, "retryable"); + assert_eq!(retry.failure_code.as_deref(), Some("deadline_exceeded")); + assert_eq!(retry.failure_count, 7); + let delay = session_refresh_retry_delay(SessionTemporalRefreshRetryClass::Deadline, 7); + assert_eq!( + session_refresh_retry_delay(SessionTemporalRefreshRetryClass::Deadline, u32::MAX), + delay, + "repeated failures stay within the existing backoff bound", + ); + let minimum_run = Duration::from_secs( + *tracedecay_domain::configuration::LCM_SUMMARIZER_TIMEOUT_SECS_RANGE.start(), + ); + assert!( + retry.next_attempt_at_ms + >= before_run_ms + i64::try_from((minimum_run + delay).as_millis()).unwrap(), + "the persisted retry deadline must follow provider settlement, not page admission", + ); + assert!( + failed + .next_retry_delay + .is_some_and(|remaining| remaining <= delay) + ); + let snapshot = database.read_snapshot().await.unwrap(); + assert!( + next_candidate(&snapshot, retry.next_attempt_at_ms - 1) + .await + .unwrap() + .is_none() + ); + let due = next_candidate(&snapshot, retry.next_attempt_at_ms) + .await + .unwrap() + .unwrap(); + assert_eq!(due.session_id, session_id); + assert_eq!(due.failure_count, 7); + drop(snapshot); + let status = database + .lcm_status("cursor", Some(session_id)) + .await + .unwrap(); + assert_eq!(status.summary_node_count, 0); + assert_eq!(status.summary_convergence.retryable_session_count, 1); + assert_eq!(status.summary_convergence.unavailable_session_count, 0); + assert_eq!(std::fs::read_to_string(&launches).unwrap(), "first\n"); + + // Advance only this fixture's durable due time. Candidate selection, + // provider execution, and publication still use the production page; + // no wall-clock sleep or configuration change is needed for recovery. + let transaction = database.begin_write_transaction().await.unwrap(); + transaction + .execute( + "UPDATE lcm_summary_convergence_queue SET next_attempt_at_ms = 0 + WHERE provider = 'cursor' AND session_id = ?1", + params![session_id], + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let recovered = run_summary_convergence_page(database.clone(), 1) + .await + .unwrap(); + assert_eq!(recovered.parked_rows_requeued, 0); + assert_eq!(recovered.sessions.len(), 1); + assert!(recovered.sessions[0].summary_nodes_created > 0); + assert_eq!( + std::fs::read_to_string(&launches).unwrap(), + "first\nretry\n" + ); + assert_eq!( + crate::lcm_summarization::summarizer_binding_identity(&database).unwrap(), + binding + ); + let status = database + .lcm_status("cursor", Some(session_id)) + .await + .unwrap(); + assert!(status.summary_node_count > 0); + assert_eq!(status.summary_convergence.retryable_session_count, 0); + assert_eq!(status.summary_convergence.unavailable_session_count, 0); + assert!(status.lifecycle.current_frontier_store_id.is_some()); + }); +} + +#[test] +fn a_missing_configured_summarizer_remains_parked_without_retries() { + run_with_pinned_summarizer_binding(async { + let fixture = ProjectSummarizerFixture::open().await; + let database = fixture.db(); + let session_id = "missing-configured-summarizer"; + seed_session(&database, session_id).await; + fixture.pin_cursor_agent(&fixture.root.path().join("missing-cursor-agent")); + + let page = run_summary_convergence_page(database.clone(), 1) + .await + .unwrap(); + assert_eq!(page.sessions.len(), 1); + assert_eq!( + page.sessions[0].disposition, + LcmSummaryConvergenceDisposition::Pending { + reason: "cursor_agent_unavailable".to_owned(), + } + ); + assert_eq!(page.sessions[0].summary_nodes_created, 0); + let state = retry_state(&database, session_id).await; + assert_eq!(state.state, "unavailable"); + assert_eq!( + state.failure_code.as_deref(), + Some("cursor_agent_unavailable") + ); + assert_eq!(state.failure_count, 0); + assert_eq!(state.next_attempt_at_ms, 0); + assert!(page.next_retry_delay.is_none()); + let repeated = run_summary_convergence_page(database.clone(), 1) + .await + .unwrap(); + assert!(repeated.sessions.is_empty()); + assert!(repeated.next_retry_delay.is_none()); + }); +} diff --git a/crates/tracedecay-session-runtime/src/lcm_summarization/cursor_agent.rs b/crates/tracedecay-session-runtime/src/lcm_summarization/cursor_agent.rs index 3a98e421e9..bdf30df755 100644 --- a/crates/tracedecay-session-runtime/src/lcm_summarization/cursor_agent.rs +++ b/crates/tracedecay-session-runtime/src/lcm_summarization/cursor_agent.rs @@ -7,6 +7,7 @@ //! module never consults `PATH` or the process environment for either. use std::fmt::Write as _; +use std::io; use std::path::PathBuf; use std::process::{Command, Stdio}; use std::time::{Duration, Instant}; @@ -77,12 +78,14 @@ pub(super) fn summarize_with_cursor_agent( if Instant::now() >= deadline { let _ = child.kill(); let _ = child.wait(); - return Err(TraceDecayError::Config { - message: format!( + return Err(io::Error::new( + io::ErrorKind::TimedOut, + format!( "timed out waiting for `{}`", config.cursor_agent_bin.display() ), - }); + ) + .into()); } std::thread::sleep(Duration::from_millis(50)); } diff --git a/crates/tracedecay-session-runtime/src/lcm_summarization/provider_capabilities.rs b/crates/tracedecay-session-runtime/src/lcm_summarization/provider_capabilities.rs index a7491aa8e6..c9ea303037 100644 --- a/crates/tracedecay-session-runtime/src/lcm_summarization/provider_capabilities.rs +++ b/crates/tracedecay-session-runtime/src/lcm_summarization/provider_capabilities.rs @@ -7,11 +7,14 @@ //! The scan in the parent module stays provider-neutral: it reads rows, hands //! each one to the registered recognizers, and never names a provider. +use std::io::ErrorKind; use std::pin::Pin; use std::time::Duration; use serde_json::Value; +use tracedecay_contracts::automation::AgentTaskFailureClass; use tracedecay_domain::configuration::{LcmSummarizerExecutableV1, LcmSummarizerExecutablesV1}; +use tracedecay_domain::errors::TraceDecayError; use tracedecay_domain::{CanonicalObservationEnvelopeV1, CanonicalObservationFactV1}; use tracedecay_lcm::{LcmError, LcmSummaryRequest}; @@ -318,7 +321,12 @@ async fn cursor_agent_summary( }) .await .map_err(|_| SummaryResolutionError::Unavailable("cursor_agent_unavailable"))? - .map_err(|_| SummaryResolutionError::Unavailable("cursor_agent_unavailable"))?; + .map_err(|error| match error { + TraceDecayError::Io(error) if error.kind() == ErrorKind::TimedOut => { + LcmError::DeadlineExceeded.into() + } + _ => SummaryResolutionError::Unavailable("cursor_agent_unavailable"), + })?; Ok(AuthoritativeSummary { text, route: "cursor_agent".to_string(), @@ -364,7 +372,10 @@ async fn codex_app_server_summary( }) .await .map_err(|_| SummaryResolutionError::Unavailable("codex_app_server_unavailable"))? - .map_err(|_| SummaryResolutionError::Unavailable("codex_app_server_unavailable"))?; + .map_err(|error| match error.class { + AgentTaskFailureClass::Timeout => LcmError::DeadlineExceeded.into(), + _ => SummaryResolutionError::Unavailable("codex_app_server_unavailable"), + })?; Ok(AuthoritativeSummary { text: result.text, route: result.model.map_or_else( diff --git a/crates/tracedecay-session-runtime/src/lcm_summarization/summarizer_executable_tests.rs b/crates/tracedecay-session-runtime/src/lcm_summarization/summarizer_executable_tests.rs index 44101320a9..a27e1a606d 100644 --- a/crates/tracedecay-session-runtime/src/lcm_summarization/summarizer_executable_tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_summarization/summarizer_executable_tests.rs @@ -15,7 +15,7 @@ use tracedecay_domain::configuration::{LcmSummarizerExecutableV1, LcmSummarizerE use tracedecay_global_db::tests::harness::{ RegisteredGlobalDbHarness, RegisteredGlobalDbTestRuntime, }; -use tracedecay_lcm::{LcmSummaryRequest, LcmSummarySourceMessage, LcmSummarySourceRange}; +use tracedecay_lcm::{LcmError, LcmSummaryRequest, LcmSummarySourceMessage, LcmSummarySourceRange}; use tracedecay_runtime_core::test_executable::write_executable_script; use super::{ @@ -311,3 +311,90 @@ async fn configured_model_reaches_the_summarizer_in_a_private_workspace() { "the per-run summary workspace must be removed after the run" ); } + +#[tokio::test(flavor = "multi_thread")] +async fn configured_provider_timeouts_are_execution_failures() { + let root = tempfile::tempdir().unwrap(); + let project_root = root.path().join("project"); + let runtime = RegisteredGlobalDbTestRuntime::project( + root.path().join("profile"), + &project_root, + ProjectId::new("project.lcm-provider-timeout".to_owned()).unwrap(), + ) + .await + .unwrap(); + let database = runtime.project_database_arc().unwrap(); + let executable = root.path().join("unresponsive-summarizer"); + // Exec keeps the blocking fixture in the exact child the adapter reaps. + // A zero caller budget forces the deadline branch without a timing race. + write_executable_script(&executable, "#!/bin/sh\nexec /bin/sleep 60\n").unwrap(); + let configured = LcmSummarizerExecutableV1::configured(executable).unwrap(); + tracedecay_configuration::test_support::pin_lcm_summarizer_executables( + &database, + &project_root, + LcmSummarizerExecutablesV1 { + cursor_agent: configured.clone(), + codex: configured, + }, + ) + .unwrap(); + + for provider in ["cursor", "codex"] { + let error = generate_provider_summary( + &database, + provider, + &summary_request(provider), + Duration::ZERO, + ) + .await + .err() + .expect("the configured process cannot finish before its expired deadline"); + assert!( + matches!( + error, + SummaryResolutionError::Storage(LcmError::DeadlineExceeded) + ), + "{provider} must preserve a timeout as retryable execution failure" + ); + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn missing_configured_executables_remain_typed_unavailable() { + let root = tempfile::tempdir().unwrap(); + let project_root = root.path().join("project"); + let runtime = RegisteredGlobalDbTestRuntime::project( + root.path().join("profile"), + &project_root, + ProjectId::new("project.lcm-provider-missing".to_owned()).unwrap(), + ) + .await + .unwrap(); + let database = runtime.project_database_arc().unwrap(); + let missing = LcmSummarizerExecutableV1::configured(root.path().join("missing-host")).unwrap(); + tracedecay_configuration::test_support::pin_lcm_summarizer_executables( + &database, + &project_root, + LcmSummarizerExecutablesV1 { + cursor_agent: missing.clone(), + codex: missing, + }, + ) + .unwrap(); + + for (provider, reason) in [ + ("cursor", "cursor_agent_unavailable"), + ("codex", "codex_app_server_unavailable"), + ] { + let error = generate_provider_summary( + &database, + provider, + &summary_request(provider), + Duration::from_secs(5), + ) + .await + .err() + .expect("a missing executable cannot summarize"); + assert_eq!(unavailable_reason(error), reason); + } +} diff --git a/crates/tracedecay-session-runtime/src/lcm_summary_convergence.rs b/crates/tracedecay-session-runtime/src/lcm_summary_convergence.rs index 9a890fcba3..c297634147 100644 --- a/crates/tracedecay-session-runtime/src/lcm_summary_convergence.rs +++ b/crates/tracedecay-session-runtime/src/lcm_summary_convergence.rs @@ -125,7 +125,7 @@ pub(crate) async fn run_summary_convergence_page( }; let processed: Pin< Box> + Send + '_>, - > = Box::pin(process_candidate(&database, candidate, now_unix_ms)); + > = Box::pin(process_candidate(&database, candidate)); sessions.push(processed.await?); } let summary_retry: Pin, LcmError>> + Send + '_>> = @@ -140,6 +140,7 @@ pub(crate) async fn run_summary_convergence_page( (None, Some(relation)) => Some(relation), (None, None) => None, }; + let now_unix_ms = unix_millis()?; let next_retry_delay = next_retry_at .map(|retry_at| { u64::try_from(retry_at.saturating_sub(now_unix_ms).max(0)) @@ -336,7 +337,6 @@ async fn load_next_retry(database: &RegisteredGlobalDbLeaseV1) -> Result Result { let protection = match database .lcm_protect_session_raw_messages_page( @@ -349,7 +349,7 @@ async fn process_candidate( .await { Ok(page) => page, - Err(error) => return record_candidate_error(database, candidate, error, now_unix_ms).await, + Err(error) => return record_candidate_error(database, candidate, error).await, }; let Some(refreshed_candidate) = load_session_candidate(database, &candidate.provider, &candidate.session_id).await? @@ -398,7 +398,7 @@ async fn process_candidate( let (_, has_more) = match invalidation { Ok(page) => page, Err(error) => { - return record_candidate_error(database, candidate, error, now_unix_ms).await; + return record_candidate_error(database, candidate, error).await; } }; let Some(refreshed_candidate) = @@ -435,7 +435,7 @@ async fn process_candidate( { Ok(bounded) => bounded, Err(error) => { - return record_candidate_error(database, candidate, error, now_unix_ms).await; + return record_candidate_error(database, candidate, error).await; } }; let response = &bounded.response; @@ -445,10 +445,10 @@ async fn process_candidate( Ok(LcmCompressionOutcome::NeedsSummary(_)) => { let error = LcmError::Db("retained compression left an unresolved summary request".to_string()); - return record_candidate_error(database, candidate, error, now_unix_ms).await; + return record_candidate_error(database, candidate, error).await; } Err(error) => { - return record_candidate_error(database, candidate, error, now_unix_ms).await; + return record_candidate_error(database, candidate, error).await; } }; let (state, disposition, failure_code) = if let Some(reason) = deferred_reason { @@ -506,7 +506,6 @@ async fn record_candidate_error( database: &RegisteredGlobalDbLeaseV1, candidate: LcmSummaryConvergenceCandidate, error: LcmError, - now_unix_ms: i64, ) -> Result { if matches!( error, @@ -522,13 +521,16 @@ async fn record_candidate_error( _ => SessionTemporalRefreshRetryClass::Storage, }; let delay = session_refresh_retry_delay(class, failure_count); + // The failed provider run may have outlasted the backoff. Start the + // next attempt's delay when the failure settles, not when its page began. + let next_attempt_at_ms = unix_millis()?.saturating_add(duration_millis(delay)?); ( LcmSummaryConvergenceQueueState::Retryable, LcmSummaryConvergenceDisposition::Retryable { reason: reason.to_string(), }, failure_count, - now_unix_ms.saturating_add(duration_millis(delay)?), + next_attempt_at_ms, ) } else { (