diff --git a/docs/design-docs/working-memory-triage.md b/docs/design-docs/working-memory-triage.md index 00865888c..8615ff884 100644 --- a/docs/design-docs/working-memory-triage.md +++ b/docs/design-docs/working-memory-triage.md @@ -11,8 +11,8 @@ Findings from CodeRabbit review + bug reports. Tracking resolution before merge. ### Major -- [ ] **R2 — Bulletin fallback gate too aggressive** (`prompts/en/channel.md.j2:172`) - Condition `not working_memory and not knowledge_synthesis` hides bulletin when working memory exists but knowledge synthesis hasn't run yet. Should gate only on `not knowledge_synthesis`. +- [x] **R2 — Bulletin fallback gate too aggressive** (`prompts/en/channel.md.j2:172`) + Condition `not working_memory and not knowledge_synthesis` hides bulletin when working memory exists but knowledge synthesis hasn't run yet. **Fixed in PR #570:** fallback now depends on missing `knowledge_synthesis`, and prompt data preserves that original absence. - [ ] **R3 — Don't exclude participant-role facts yet** (`prompts/en/cortex_knowledge_synthesis.md.j2:21`) Exclusion of "The user is the CEO" drops participant context with nowhere else to live until Phase 6 ships. @@ -21,22 +21,22 @@ Findings from CodeRabbit review + bug reports. Tracking resolution before merge. `task` from user input persisted verbatim; could capture secrets/PII. Truncate and scrub. - [ ] **R5 — Dirty flag only bumps on merges** (`src/agent/cortex.rs:1958`) - Prunes and decays also change the memory set but don't trigger knowledge synthesis re-gen. Add `report.pruned > 0 || report.decayed > 0`. + Prunes and decays also change the memory set but don't trigger knowledge synthesis re-gen. Add `report.pruned > 0 || report.decayed > 0`. **Partial in PR #570:** prunes and merges now dirty synthesis; decay remains intentionally importance-only and needs a follow-up decision. - [ ] **R6 — Dirty-flag synthesis not mutex-guarded** (`src/agent/cortex.rs:2106`) - Can race with warmup synthesis path. Should acquire the same synthesis mutex. + Can race with warmup synthesis path. Should acquire the same synthesis mutex. **Still open:** PR #570 single-flights background refresh tasks, but lock parity with warmup still needs a focused verify/fix pass. -- [ ] **R7 — Intraday/daily synthesis blocks main cortex loop** (`src/agent/cortex.rs:2166`) - LLM calls awaited inline inside `tokio::select!`; events stop draining during synthesis. Spawn as background tasks. +- [x] **R7 — Intraday/daily synthesis blocks main cortex loop** (`src/agent/cortex.rs:2166`) + LLM calls awaited inline inside `tokio::select!`; events stop draining during synthesis. **Fixed in PR #570:** intraday and daily synthesis now run as background tasks with single-flight scheduling and failure backoff. -- [ ] **R8 — Empty sections treated as successful no-op** (`src/agent/cortex.rs:2558`) - Returns before tasks can contribute to synthesis; dirty flag never clears, causing infinite rescheduling. +- [x] **R8 — Empty sections treated as successful no-op** (`src/agent/cortex.rs:2558`) + Returns before tasks can contribute to synthesis; dirty flag never clears, causing infinite rescheduling. **Fixed in PR #570:** true empty input clears the target version, while gather failures fail the synthesis path and keep it retryable. - [ ] **R9 — Missing `default_max_turns(1)` + inline preambles** (`src/agent/cortex.rs:2579`) - Three cortex agent builders lack explicit max_turns; two have inline preamble strings instead of prompt files. + Three cortex agent builders lack explicit max_turns; two have inline preamble strings instead of prompt files. **Stacked in PR #571:** one-shot synthesis prompt hardening is kept out of PR #570 to keep the reliability diff focused. -- [ ] **R10 — Version snapshot after async work** (`src/agent/cortex.rs:2614`) - `knowledge_synthesis_last_version` read after LLM call; concurrent writes can advance the version past what was actually synthesized. Snapshot before. +- [x] **R10 — Version snapshot after async work** (`src/agent/cortex.rs:2614`) + `knowledge_synthesis_last_version` read after LLM call; concurrent writes can advance the version past what was actually synthesized. **Fixed in PR #570:** synthesis snapshots the target version before async work and only marks that version complete. - [x] **R11 — Unsynthesized yesterday events dropped** (`src/agent/cortex.rs:2916`) Raw events that didn't hit count/time trigger before midnight are lost from daily summary. Roll them into the summary. **Fixed:** daily summary now fetches all raw events, filters to the unsynthesized tail after the last intra-day synthesis, and includes them in the LLM input. diff --git a/flake.lock b/flake.lock index 65186f35e..994bbc795 100644 --- a/flake.lock +++ b/flake.lock @@ -2,11 +2,11 @@ "nodes": { "crane": { "locked": { - "lastModified": 1773857772, - "narHash": "sha256-5xsK26KRHf0WytBtsBnQYC/lTWDhQuT57HJ7SzuqZcM=", + "lastModified": 1776533550, + "narHash": "sha256-8mTHsQ8cB0jGlXE4WWKqpQFQPM/VotDnr2uzfrOGNKI=", "owner": "ipetkov", "repo": "crane", - "rev": "b556d7bbae5ff86e378451511873dfd07e4504cd", + "rev": "e24d86e91348e3d44014974fa24c9a22cfd663b5", "type": "github" }, "original": { @@ -35,11 +35,11 @@ }, "nixpkgs": { "locked": { - "lastModified": 1773628058, - "narHash": "sha256-hpXH0z3K9xv0fHaje136KY872VT2T5uwxtezlAskQgY=", + "lastModified": 1776329215, + "narHash": "sha256-a8BYi3mzoJ/AcJP8UldOx8emoPRLeWqALZWu4ZvjPXw=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "f8573b9c935cfaa162dd62cc9e75ae2db86f85df", + "rev": "b86751bc4085f48661017fa226dee99fab6c651b", "type": "github" }, "original": { diff --git a/prompts/en/channel.md.j2 b/prompts/en/channel.md.j2 index b2c84bab5..c7e83145f 100644 --- a/prompts/en/channel.md.j2 +++ b/prompts/en/channel.md.j2 @@ -198,7 +198,7 @@ When in doubt, skip. Being a lurker who speaks when it matters is better than be {{ knowledge_synthesis }} {%- endif %} -{%- if memory_bulletin and not working_memory and not knowledge_synthesis %} +{%- if memory_bulletin and not knowledge_synthesis %} ## Memory Context {{ memory_bulletin }} diff --git a/src/agent/cortex.rs b/src/agent/cortex.rs index 8081ac307..3de2e51d8 100644 --- a/src/agent/cortex.rs +++ b/src/agent/cortex.rs @@ -335,6 +335,159 @@ impl BulletinRefreshOutcome { } } +fn maybe_spawn_synthesis_task( + task: &mut Option>>, + backoff: &SynthesisTaskBackoff, + task_name: &'static str, + now: Instant, + spawn: impl FnOnce() -> tokio::task::JoinHandle>, +) -> bool { + if task.is_some() { + return false; + } + + if !backoff.can_spawn(now) { + tracing::debug!( + task = task_name, + failure_count = backoff.failure_count, + "cortex synthesis task scheduling skipped during backoff" + ); + return false; + } + + *task = Some(spawn()); + true +} + +fn spawn_intraday_synthesis_task( + deps: AgentDeps, + logger: CortexLogger, +) -> tokio::task::JoinHandle> { + tokio::spawn(async move { maybe_synthesize_intraday_batch(&deps, &logger).await }) +} + +fn spawn_daily_synthesis_task( + deps: AgentDeps, + logger: CortexLogger, +) -> tokio::task::JoinHandle> { + tokio::spawn(async move { maybe_synthesize_daily_summary(&deps, &logger).await }) +} + +fn mark_knowledge_synthesis_version_complete( + last_version: &std::sync::atomic::AtomicU64, + target_version: u64, +) { + last_version.store(target_version, std::sync::atomic::Ordering::Release); +} + +const SYNTHESIS_TASK_BACKOFF_INITIAL_SECS: u64 = 30; +const SYNTHESIS_TASK_BACKOFF_MAX_SECS: u64 = 5 * 60; + +#[derive(Debug, Clone)] +struct SynthesisTaskBackoff { + failure_count: u32, + next_allowed_instant: Instant, +} + +impl SynthesisTaskBackoff { + fn new(now: Instant) -> Self { + Self { + failure_count: 0, + next_allowed_instant: now, + } + } + + fn can_spawn(&self, now: Instant) -> bool { + now >= self.next_allowed_instant + } + + fn record_success(&mut self, now: Instant) { + self.failure_count = 0; + self.next_allowed_instant = now; + } + + fn record_failure(&mut self, now: Instant) { + self.failure_count = self.failure_count.saturating_add(1); + self.next_allowed_instant = now + synthesis_task_backoff_delay(self.failure_count); + } +} + +fn synthesis_task_backoff_delay(failure_count: u32) -> Duration { + let exponent = failure_count.saturating_sub(1).min(10); + let multiplier = 1_u64 << exponent; + let seconds = SYNTHESIS_TASK_BACKOFF_INITIAL_SECS + .saturating_mul(multiplier) + .min(SYNTHESIS_TASK_BACKOFF_MAX_SECS); + + Duration::from_secs(seconds) +} + +async fn collect_synthesis_task( + task: &mut Option>>, + task_name: &'static str, + backoff: &mut SynthesisTaskBackoff, + now: Instant, +) { + let Some(handle) = task.as_ref() else { + return; + }; + + if !handle.is_finished() { + return; + } + + let Some(handle) = task.take() else { + return; + }; + + match handle.await { + Ok(Ok(true)) => { + backoff.record_success(now); + tracing::debug!(task = task_name, "cortex synthesis task completed"); + } + Ok(Ok(false)) => { + backoff.record_success(now); + tracing::trace!(task = task_name, "cortex synthesis task skipped"); + } + Ok(Err(error)) => { + backoff.record_failure(now); + tracing::warn!( + %error, + task = task_name, + failure_count = backoff.failure_count, + "cortex synthesis task failed" + ); + } + Err(error) if error.is_cancelled() => { + backoff.record_failure(now); + tracing::debug!( + %error, + task = task_name, + failure_count = backoff.failure_count, + "cortex synthesis task cancelled" + ); + } + Err(error) if error.is_panic() => { + backoff.record_failure(now); + tracing::warn!( + %error, + task = task_name, + failure_count = backoff.failure_count, + "cortex synthesis task panicked" + ); + } + Err(error) => { + backoff.record_failure(now); + tracing::warn!( + %error, + task = task_name, + failure_count = backoff.failure_count, + "cortex synthesis task failed" + ); + } + } +} + const BRANCH_LATENCY_WINDOW_SIZE: usize = 32; #[derive(Debug, Clone)] @@ -1861,6 +2014,10 @@ async fn run_cortex_loop( let mut bulletin_refresh_circuit_open = false; let mut next_bulletin_refresh_allowed_at = Instant::now(); let mut last_maintenance = Instant::now(); + let mut intraday_synthesis_task: Option>> = None; + let mut daily_synthesis_task: Option>> = None; + let mut intraday_synthesis_backoff = SynthesisTaskBackoff::new(Instant::now()); + let mut daily_synthesis_backoff = SynthesisTaskBackoff::new(Instant::now()); loop { tokio::select! { @@ -1889,6 +2046,12 @@ async fn run_cortex_loop( if let Some(task) = refresh_task.take() { task.abort(); } + if let Some(task) = intraday_synthesis_task.take() { + task.abort(); + } + if let Some(task) = daily_synthesis_task.take() { + task.abort(); + } if let Some(task) = maintenance_task.take() { task.abort(); } @@ -1920,6 +2083,12 @@ async fn run_cortex_loop( if let Some(task) = refresh_task.take() { task.abort(); } + if let Some(task) = intraday_synthesis_task.take() { + task.abort(); + } + if let Some(task) = daily_synthesis_task.take() { + task.abort(); + } if let Some(task) = maintenance_task.take() { task.abort(); } @@ -1938,6 +2107,21 @@ async fn run_cortex_loop( let cortex_config = **cortex.deps.runtime_config.cortex.load(); let now = Instant::now(); + collect_synthesis_task( + &mut intraday_synthesis_task, + "intraday", + &mut intraday_synthesis_backoff, + now, + ) + .await; + collect_synthesis_task( + &mut daily_synthesis_task, + "daily", + &mut daily_synthesis_backoff, + now, + ) + .await; + if refresh_task .as_ref() .is_some_and(tokio::task::JoinHandle::is_finished) @@ -2026,8 +2210,9 @@ async fn run_cortex_loop( } maintenance_consecutive_failures = 0; maintenance_disabled_at = None; - // Merges change memory content — bump dirty flag. - if report.merged > 0 { + // Prunes and merges change memory content; decay is + // importance-only and does not dirty knowledge. + if report.pruned > 0 || report.merged > 0 { cortex.deps.runtime_config.bump_knowledge_synthesis_version(); } logger.log( @@ -2229,15 +2414,21 @@ async fn run_cortex_loop( last_maintenance = Instant::now(); } - // Working memory: intra-day synthesis (cheap SQL check, LLM only on threshold). - if let Err(error) = maybe_synthesize_intraday_batch(&cortex.deps, logger).await { - tracing::warn!(%error, "intra-day synthesis check failed"); - } + maybe_spawn_synthesis_task( + &mut intraday_synthesis_task, + &intraday_synthesis_backoff, + "intraday", + now, + || spawn_intraday_synthesis_task(cortex.deps.clone(), logger.clone()), + ); - // Working memory: daily summary for yesterday (idempotent, 1 LLM call/day max). - if let Err(error) = maybe_synthesize_daily_summary(&cortex.deps, logger).await { - tracing::warn!(%error, "daily summary check failed"); - } + maybe_spawn_synthesis_task( + &mut daily_synthesis_task, + &daily_synthesis_backoff, + "daily", + now, + || spawn_daily_synthesis_task(cortex.deps.clone(), logger.clone()), + ); // Working memory: prune old events (cheap SQL, runs every tick but deletes nothing most of the time). let wm_config = **cortex.deps.runtime_config.working_memory.load(); @@ -2616,6 +2807,18 @@ const KNOWLEDGE_SYNTHESIS_SECTIONS: &[BulletinSection] = &[ }, ]; +#[derive(Debug, Default)] +struct GatheredSections { + text: String, + failed_sections: usize, +} + +impl GatheredSections { + fn has_failures(&self) -> bool { + self.failed_sections > 0 + } +} + /// Generate a change-driven knowledge synthesis (Layer 5) and store it in RuntimeConfig. /// /// Uses the same programmatic gather + LLM synthesis pattern as the bulletin, @@ -2625,11 +2828,51 @@ const KNOWLEDGE_SYNTHESIS_SECTIONS: &[BulletinSection] = &[ pub async fn generate_knowledge_synthesis(deps: &AgentDeps, logger: &CortexLogger) -> bool { tracing::info!("cortex generating knowledge synthesis"); let started = Instant::now(); + let target_version = deps + .runtime_config + .knowledge_synthesis_version + .load(std::sync::atomic::Ordering::Acquire); - // Gather narrower sections (no identity, no events, no recent). - let raw_sections = gather_sections_from_list(deps, KNOWLEDGE_SYNTHESIS_SECTIONS).await; + let mut gathered_sections = gather_sections_from_list(deps, KNOWLEDGE_SYNTHESIS_SECTIONS).await; + let active_tasks_failed = match gather_active_tasks(deps).await { + Ok(tasks) => { + gathered_sections.text.push_str(&tasks); + false + } + Err(error) => { + tracing::warn!(%error, "failed to gather active tasks for knowledge synthesis"); + true + } + }; + let gather_failed = gathered_sections.has_failures() || active_tasks_failed; + let failed_memory_sections = gathered_sections.failed_sections; + let raw_sections = gathered_sections.text; let section_count = raw_sections.matches("### ").count(); + if gather_failed { + let duration_ms = started.elapsed().as_millis() as u64; + tracing::warn!( + failed_memory_sections, + active_tasks_failed, + duration_ms, + "knowledge synthesis input gather failed" + ); + update_warmup_status(deps, |status| { + status.last_error = Some("knowledge synthesis input gather failed".to_string()); + }); + logger.log( + "knowledge_synthesis_failed", + "Knowledge synthesis failed while gathering input", + Some(serde_json::json!({ + "duration_ms": duration_ms, + "failed_memory_sections": failed_memory_sections, + "active_tasks_failed": active_tasks_failed, + "target_version": target_version, + })), + ); + return false; + } + if raw_sections.is_empty() { tracing::info!("no memories found for knowledge synthesis"); deps.runtime_config @@ -2639,18 +2882,32 @@ pub async fn generate_knowledge_synthesis(deps: &AgentDeps, logger: &CortexLogge deps.runtime_config .memory_bulletin .store(Arc::new(String::new())); + mark_knowledge_synthesis_version_complete( + &deps.runtime_config.knowledge_synthesis_last_version, + target_version, + ); + update_warmup_status(deps, |status| { + status.last_refresh_unix_ms = Some(chrono::Utc::now().timestamp_millis()); + status.bulletin_age_secs = Some(0); + if status.state != crate::config::WarmupState::Warming { + status.state = crate::config::WarmupState::Warm; + status.last_error = None; + } + }); + logger.log( + "knowledge_synthesis_generated", + "Knowledge synthesis skipped: no memories or active tasks", + Some(serde_json::json!({ + "word_count": 0, + "sections": 0, + "duration_ms": started.elapsed().as_millis() as u64, + "target_version": target_version, + "skipped": true, + })), + ); return true; } - // Append active tasks (same as bulletin). - let raw_sections = match gather_active_tasks(deps).await { - Ok(tasks) => format!("{raw_sections}{tasks}"), - Err(error) => { - tracing::warn!(%error, "failed to gather active tasks for knowledge synthesis"); - raw_sections - } - }; - let cortex_config = **deps.runtime_config.cortex.load(); let prompt_engine = deps.runtime_config.prompts.load(); let synthesis_preamble = match prompt_engine.render_static("cortex_knowledge_synthesis") { @@ -2712,14 +2969,10 @@ pub async fn generate_knowledge_synthesis(deps: &AgentDeps, logger: &CortexLogge deps.runtime_config .memory_bulletin .store(Arc::new(synthesis)); - // Mark this version as synthesized. - let current = deps - .runtime_config - .knowledge_synthesis_version - .load(std::sync::atomic::Ordering::Relaxed); - deps.runtime_config - .knowledge_synthesis_last_version - .store(current, std::sync::atomic::Ordering::Relaxed); + mark_knowledge_synthesis_version_complete( + &deps.runtime_config.knowledge_synthesis_last_version, + target_version, + ); // Update warmup status. let refresh_ms = chrono::Utc::now().timestamp_millis(); update_warmup_status(deps, |status| { @@ -2766,8 +3019,11 @@ pub async fn generate_knowledge_synthesis(deps: &AgentDeps, logger: &CortexLogge /// /// Uses the same pattern as `gather_bulletin_sections` (empty-query metadata /// search) but accepts an arbitrary section list for narrower scoping. -async fn gather_sections_from_list(deps: &AgentDeps, sections: &[BulletinSection]) -> String { - let mut output = String::new(); +async fn gather_sections_from_list( + deps: &AgentDeps, + sections: &[BulletinSection], +) -> GatheredSections { + let mut gathered = GatheredSections::default(); for section in sections { let config = SearchConfig { @@ -2786,6 +3042,7 @@ async fn gather_sections_from_list(deps: &AgentDeps, sections: &[BulletinSection %error, "knowledge synthesis section query failed" ); + gathered.failed_sections += 1; continue; } }; @@ -2794,9 +3051,11 @@ async fn gather_sections_from_list(deps: &AgentDeps, sections: &[BulletinSection continue; } - output.push_str(&format!("### {}\n\n", section.label)); + gathered + .text + .push_str(&format!("### {}\n\n", section.label)); for result in &results { - output.push_str(&format!( + gathered.text.push_str(&format!( "- [{}] (importance: {:.1}) {}\n", result.memory.memory_type, result.memory.importance, @@ -2808,10 +3067,10 @@ async fn gather_sections_from_list(deps: &AgentDeps, sections: &[BulletinSection .unwrap_or(&result.memory.content), )); } - output.push('\n'); + gathered.text.push('\n'); } - output + gathered } /// Check if knowledge synthesis needs regeneration based on dirty flag and debounce. @@ -2819,11 +3078,11 @@ pub fn should_regenerate_knowledge_synthesis(deps: &AgentDeps) -> bool { let current_version = deps .runtime_config .knowledge_synthesis_version - .load(std::sync::atomic::Ordering::Relaxed); + .load(std::sync::atomic::Ordering::Acquire); let last_version = deps .runtime_config .knowledge_synthesis_last_version - .load(std::sync::atomic::Ordering::Relaxed); + .load(std::sync::atomic::Ordering::Acquire); if current_version == last_version { return false; @@ -2834,7 +3093,7 @@ pub fn should_regenerate_knowledge_synthesis(deps: &AgentDeps) -> bool { let last_change = deps .runtime_config .knowledge_synthesis_last_change - .load(std::sync::atomic::Ordering::Relaxed); + .load(std::sync::atomic::Ordering::Acquire); let now = chrono::Utc::now().timestamp(); let elapsed = now.saturating_sub(last_change) as u64; @@ -4291,13 +4550,15 @@ async fn fetch_memories_for_association( mod tests { use super::{ BULLETIN_REFRESH_CIRCUIT_OPEN_SECS, BULLETIN_REFRESH_CIRCUIT_OPEN_THRESHOLD, BranchTracker, - BulletinRefreshOutcome, CortexReceiverOutcome, HealthRuntimeState, + BulletinRefreshOutcome, CortexReceiverOutcome, GatheredSections, HealthRuntimeState, MAINTENANCE_TASK_CANCEL_GRACE_SECS, MaintenanceTimeoutAction, ReceiverClosedBehavior, - Signal, WorkerTracker, apply_cancelled_warmup_status, build_kill_targets, - claim_detached_completion, detached_timeout_transition, handle_cortex_receiver_result, - has_completed_initial_warmup, is_cancelled_control_result, is_terminal_control_result, - maintenance_task_timeout, maintenance_timeout_action, maybe_close_bulletin_refresh_circuit, - maybe_generate_bulletin_under_lock, parse_structured_success_flag, push_signal_into_buffer, + Signal, SynthesisTaskBackoff, WorkerTracker, apply_cancelled_warmup_status, + build_kill_targets, claim_detached_completion, collect_synthesis_task, + detached_timeout_transition, handle_cortex_receiver_result, has_completed_initial_warmup, + is_cancelled_control_result, is_terminal_control_result, maintenance_task_timeout, + maintenance_timeout_action, mark_knowledge_synthesis_version_complete, + maybe_close_bulletin_refresh_circuit, maybe_generate_bulletin_under_lock, + maybe_spawn_synthesis_task, parse_structured_success_flag, push_signal_into_buffer, record_bulletin_refresh_failure, should_execute_warmup, should_generate_bulletin_from_bulletin_loop, signal_from_event, summarize_signal_text, take_lagged_control_flag, @@ -4508,6 +4769,177 @@ mod tests { assert_eq!(calls.load(Ordering::SeqCst), 0); } + #[tokio::test] + async fn working_memory_synthesis_task_is_single_flight() { + let calls = Arc::new(AtomicUsize::new(0)); + let (release_tx, release_rx) = tokio::sync::oneshot::channel::<()>(); + let release_rx = Arc::new(tokio::sync::Mutex::new(Some(release_rx))); + let mut task: Option>> = None; + let now = Instant::now(); + let backoff = SynthesisTaskBackoff::new(now); + + let calls_for_first = Arc::clone(&calls); + let release_rx_for_first = Arc::clone(&release_rx); + assert!(maybe_spawn_synthesis_task( + &mut task, + &backoff, + "intraday", + now, + move || { + tokio::spawn(async move { + calls_for_first.fetch_add(1, Ordering::SeqCst); + let receiver = release_rx_for_first + .lock() + .await + .take() + .expect("release receiver should exist"); + receiver.await.expect("release oneshot dropped"); + Ok(true) + }) + } + )); + + let calls_for_second = Arc::clone(&calls); + assert!(!maybe_spawn_synthesis_task( + &mut task, + &backoff, + "intraday", + now, + move || { + tokio::spawn(async move { + calls_for_second.fetch_add(1, Ordering::SeqCst); + Ok(true) + }) + } + )); + + tokio::time::timeout(Duration::from_secs(2), async { + while calls.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("first synthesis task should start"); + assert_eq!(calls.load(Ordering::SeqCst), 1); + release_tx.send(()).expect("release should send"); + task.take() + .expect("task should exist") + .await + .expect("task should join") + .expect("task should succeed"); + } + + #[tokio::test] + async fn synthesis_task_failure_backs_off_before_respawn() { + let calls = Arc::new(AtomicUsize::new(0)); + let now = Instant::now(); + let mut backoff = SynthesisTaskBackoff::new(now); + let mut task: Option>> = None; + + let calls_for_first = Arc::clone(&calls); + assert!(maybe_spawn_synthesis_task( + &mut task, + &backoff, + "intraday", + now, + move || { + tokio::spawn(async move { + calls_for_first.fetch_add(1, Ordering::SeqCst); + Err(anyhow::anyhow!("backend unavailable")) + }) + } + )); + + tokio::time::timeout(Duration::from_secs(2), async { + while task.as_ref().is_some_and(|handle| !handle.is_finished()) { + tokio::task::yield_now().await; + } + }) + .await + .expect("failed synthesis task should finish"); + collect_synthesis_task(&mut task, "intraday", &mut backoff, now).await; + + assert!(task.is_none()); + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(backoff.failure_count, 1); + + let retry_at = backoff.next_allowed_instant; + let blocked_at = retry_at + .checked_sub(Duration::from_millis(1)) + .expect("retry instant should be after current instant"); + let calls_for_blocked = Arc::clone(&calls); + assert!(!maybe_spawn_synthesis_task( + &mut task, + &backoff, + "intraday", + blocked_at, + move || { + tokio::spawn(async move { + calls_for_blocked.fetch_add(1, Ordering::SeqCst); + Ok(true) + }) + } + )); + assert_eq!(calls.load(Ordering::SeqCst), 1); + + let calls_for_retry = Arc::clone(&calls); + assert!(maybe_spawn_synthesis_task( + &mut task, + &backoff, + "intraday", + retry_at, + move || { + tokio::spawn(async move { + calls_for_retry.fetch_add(1, Ordering::SeqCst); + Ok(true) + }) + } + )); + + tokio::time::timeout(Duration::from_secs(2), async { + while task.as_ref().is_some_and(|handle| !handle.is_finished()) { + tokio::task::yield_now().await; + } + }) + .await + .expect("retry synthesis task should finish"); + collect_synthesis_task(&mut task, "intraday", &mut backoff, retry_at).await; + + assert!(task.is_none()); + assert_eq!(backoff.failure_count, 0); + assert_eq!(backoff.next_allowed_instant, retry_at); + assert_eq!(calls.load(Ordering::SeqCst), 2); + } + + #[test] + fn knowledge_synthesis_completion_marks_target_version_not_current_version() { + let current_version = std::sync::atomic::AtomicU64::new(2); + let last_version = std::sync::atomic::AtomicU64::new(0); + let target_version = 1; + + mark_knowledge_synthesis_version_complete(&last_version, target_version); + + assert_eq!( + current_version.load(Ordering::Acquire), + 2, + "newer dirty version should still be pending" + ); + assert_eq!(last_version.load(Ordering::Acquire), target_version); + } + + #[test] + fn gathered_sections_fail_when_any_section_query_failed() { + let gathered = GatheredSections { + text: String::new(), + failed_sections: 1, + }; + + assert!( + gathered.has_failures(), + "failed section queries must keep synthesis retryable" + ); + } + #[test] fn summarize_signal_text_uses_first_non_empty_line() { let text = "\n\nfirst line\nsecond line"; diff --git a/src/api/projects.rs b/src/api/projects.rs index 03ae932d4..dbb43d221 100644 --- a/src/api/projects.rs +++ b/src/api/projects.rs @@ -1059,7 +1059,7 @@ pub(super) async fn disk_usage( }); } - entries.sort_by(|a, b| b.bytes.cmp(&a.bytes)); + entries.sort_by_key(|entry| std::cmp::Reverse(entry.bytes)); Ok(Json(DiskUsageResponse { total_bytes, diff --git a/src/api/usage.rs b/src/api/usage.rs index 68d9904b8..133dcffe0 100644 --- a/src/api/usage.rs +++ b/src/api/usage.rs @@ -393,7 +393,7 @@ pub(super) async fn get_usage( }; let mut by_model: Vec = all_by_model.into_values().collect(); - by_model.sort_by(|a, b| b.request_count.cmp(&a.request_count)); + by_model.sort_by_key(|model| std::cmp::Reverse(model.request_count)); let mut by_day: Vec = all_by_day.into_values().collect(); by_day.sort_by(|a, b| a.date.cmp(&b.date)); diff --git a/src/config/runtime.rs b/src/config/runtime.rs index 229b5a8bc..8df809491 100644 --- a/src/config/runtime.rs +++ b/src/config/runtime.rs @@ -195,12 +195,12 @@ impl RuntimeConfig { /// This bumps the dirty counter so the cortex regenerates knowledge synthesis. /// Do NOT call for importance-only changes (decay, access count). pub fn bump_knowledge_synthesis_version(&self) { - self.knowledge_synthesis_version - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); self.knowledge_synthesis_last_change.store( chrono::Utc::now().timestamp(), - std::sync::atomic::Ordering::Relaxed, + std::sync::atomic::Ordering::Release, ); + self.knowledge_synthesis_version + .fetch_add(1, std::sync::atomic::Ordering::AcqRel); } /// Compute the current dispatch-readiness signal. diff --git a/src/conversation/worker_transcript.rs b/src/conversation/worker_transcript.rs index 7a21669ac..b2fb7ebda 100644 --- a/src/conversation/worker_transcript.rs +++ b/src/conversation/worker_transcript.rs @@ -405,12 +405,10 @@ fn convert_history(history: &[rig::message::Message]) -> Vec { let mut parts = Vec::new(); for item in content.iter() { match item { - rig::message::AssistantContent::Text(text) => { - if !text.text.is_empty() { - parts.push(ActionContent::Text { - text: text.text.clone(), - }); - } + rig::message::AssistantContent::Text(text) if !text.text.is_empty() => { + parts.push(ActionContent::Text { + text: text.text.clone(), + }); } rig::message::AssistantContent::ToolCall(tool_call) => { let args_str = tool_call.function.arguments.to_string(); @@ -462,13 +460,13 @@ fn convert_history(history: &[rig::message::Message]) -> Vec { text: truncated, }); } - rig::message::UserContent::Text(text) => { + rig::message::UserContent::Text(text) + if !text.text.is_empty() && !text.text.starts_with("[System:") => + { // Skip compaction markers and system-injected messages - if !text.text.is_empty() && !text.text.starts_with("[System:") { - steps.push(TranscriptStep::UserText { - text: text.text.clone(), - }); - } + steps.push(TranscriptStep::UserText { + text: text.text.clone(), + }); } _ => {} } diff --git a/src/llm/model.rs b/src/llm/model.rs index 32c54b10b..deba032ce 100644 --- a/src/llm/model.rs +++ b/src/llm/model.rs @@ -2215,10 +2215,8 @@ fn stream_from_completion_response( for content in choice_items { match content { - AssistantContent::Text(text) => { - if !text.text.is_empty() { - yield Ok(RawStreamingChoice::Message(text.text)); - } + AssistantContent::Text(text) if !text.text.is_empty() => { + yield Ok(RawStreamingChoice::Message(text.text)); } AssistantContent::ToolCall(tool_call) => { yield Ok(RawStreamingChoice::ToolCall(RawStreamingToolCall { @@ -2286,10 +2284,8 @@ fn completion_choice_to_streaming_choices( for content in choice.iter() { match content { - AssistantContent::Text(text) => { - if !text.text.is_empty() { - events.push(RawStreamingChoice::Message(text.text.clone())); - } + AssistantContent::Text(text) if !text.text.is_empty() => { + events.push(RawStreamingChoice::Message(text.text.clone())); } AssistantContent::ToolCall(tool_call) => { events.push(RawStreamingChoice::ToolCall(RawStreamingToolCall { @@ -3353,13 +3349,11 @@ fn parse_openai_reasoning_fallback(message: &serde_json::Value) -> Option) { match value { - serde_json::Value::String(text) => { + serde_json::Value::String(text) if !text.is_empty() => { // Use is_empty() instead of trim().is_empty() to preserve whitespace-only // segments. Streaming providers (e.g. Kimi) sometimes send content chunks // that are just spaces; dropping those causes missing spaces in output. - if !text.is_empty() { - text_parts.push(text.to_string()); - } + text_parts.push(text.to_string()); } serde_json::Value::Array(items) => { for item in items { diff --git a/src/messaging/mattermost.rs b/src/messaging/mattermost.rs index 01b011478..b08049394 100644 --- a/src/messaging/mattermost.rs +++ b/src/messaging/mattermost.rs @@ -597,9 +597,12 @@ impl Messaging for MattermostAdapter { } // close match MattermostWsEvent } Some(Ok(WsMessage::Ping(data))) => { - if write.send(WsMessage::Pong(data)).await.is_err() { - tracing::warn!(adapter = %runtime_key, "failed to send pong"); - break; + match write.send(WsMessage::Pong(data.clone())).await { + Ok(()) => {} + Err(_) => { + tracing::warn!(adapter = %runtime_key, "failed to send pong"); + break; + } } } Some(Ok(WsMessage::Pong(_))) => {} diff --git a/src/prompts/engine.rs b/src/prompts/engine.rs index 691e14cee..9bb1e3f16 100644 --- a/src/prompts/engine.rs +++ b/src/prompts/engine.rs @@ -683,8 +683,6 @@ impl PromptEngine { participant_context: Option, direct_mode: bool, ) -> Result { - let knowledge_synthesis = knowledge_synthesis.or_else(|| memory_bulletin.clone()); - self.render( "channel", context! { @@ -823,5 +821,36 @@ mod tests { assert_eq!(prompt, "Base prompt"); } + + #[test] + fn renders_memory_context_when_knowledge_synthesis_is_absent() { + let engine = PromptEngine::new("en").expect("prompt engine should build"); + let prompt = engine + .render_channel_prompt_with_links( + None, + Some("Bulletin fallback".to_string()), + None, + None, + String::new(), + None, + None, + None, + None, + false, + None, + None, + None, + None, + None, + None, + None, + false, + ) + .expect("channel prompt should render"); + + assert!(prompt.contains("## Memory Context")); + assert!(prompt.contains("Bulletin fallback")); + assert!(!prompt.contains("## Knowledge Context")); + } } // to support multiple languages at compile time. diff --git a/src/secrets/scrub.rs b/src/secrets/scrub.rs index 3ffc354a3..7eb2dce3a 100644 --- a/src/secrets/scrub.rs +++ b/src/secrets/scrub.rs @@ -210,7 +210,7 @@ pub fn scrub_secrets(text: &str, tool_secrets: &[(String, String)]) -> String { // Sort by descending value length so longer secrets are replaced first. // This prevents partial replacement when one secret value is a prefix of another. let mut sorted: Vec<&(String, String)> = tool_secrets.iter().collect(); - sorted.sort_by(|a, b| b.1.len().cmp(&a.1.len())); + sorted.sort_by_key(|secret| std::cmp::Reverse(secret.1.len())); let mut result = text.to_string(); for (name, value) in sorted { if !value.is_empty() { diff --git a/tests/opencode_stream.rs b/tests/opencode_stream.rs index c0b4ae32e..bc11321d9 100644 --- a/tests/opencode_stream.rs +++ b/tests/opencode_stream.rs @@ -189,12 +189,12 @@ async fn stream_events_from_live_server() { saw_text = true; } } - SseEvent::SessionIdle { session_id: sid } => { - if sid == &session_id && saw_assistant { - saw_idle = true; - events.push(event); - break; - } + SseEvent::SessionIdle { session_id: sid } + if sid == &session_id && saw_assistant => + { + saw_idle = true; + events.push(event); + break; } _ => {} }