diff --git a/crates/pa-core/src/session/manager/queries.rs b/crates/pa-core/src/session/manager/queries.rs index ebb02ea28..9f227b12c 100644 --- a/crates/pa-core/src/session/manager/queries.rs +++ b/crates/pa-core/src/session/manager/queries.rs @@ -171,6 +171,40 @@ impl SessionManager { branch } + /// Whether the active branch still grows through `prefix_leaf` — the + /// leaf a compaction prepared against — with no compaction row between + /// the two. The compaction commit's structural-prefix check: a plain + /// append chains onto the leaf behind `prefix_leaf` and stays on the + /// path (it joins the retained tail behind the committed summary); + /// a branch switch or another compaction rebuilds the path and + /// invalidates the prepared cut. + pub(crate) fn compaction_prefix_intact(&self, prefix_leaf: Option<&str>) -> bool { + let Some(prefix_leaf) = prefix_leaf else { + return self.leaf_id.is_none(); + }; + let mut visited = std::collections::HashSet::new(); + let mut id = self.leaf_id.as_deref(); + while let Some(current) = id { + // A corrupt file can hold a parent cycle; the check must not hang. + if !visited.insert(current) { + return false; + } + // The prefix counts only as an entry of the active branch: a + // dangling parent id that left the tree is a different branch. + let Some(&index) = self.by_id.get(current) else { + return false; + }; + if current == prefix_leaf { + return true; + } + if matches!(self.file_entries[index], FileEntry::Compaction { .. }) { + return false; + } + id = self.file_entries[index].parent_id(); + } + false + } + #[must_use] pub fn active_goal_state(&self) -> Option { if let Some(window) = &self.window { diff --git a/crates/pa-core/src/session_engine/compact_session.rs b/crates/pa-core/src/session_engine/compact_session.rs index 7c1918893..d2e1ed00d 100644 --- a/crates/pa-core/src/session_engine/compact_session.rs +++ b/crates/pa-core/src/session_engine/compact_session.rs @@ -106,8 +106,7 @@ pub struct CompactRun { pub result: CompactionResult, pub entry: pa_types::session::CompactionEntry, /// The whole compaction's wall duration (the run's - /// `compaction_duration_ms`; measured here once, centrally, for every - /// arm). + /// `compaction_duration_ms`). pub duration_ms: u64, /// The post-compaction `ipython_state` notice, when a kernel was running: /// the row is already durable and in the live context; surfaces broadcast it. @@ -122,26 +121,39 @@ pub enum CompactOutcome { Skipped(&'static str), } -/// Run compaction over the session: summarize the pre-cut prefix, persist the -/// entry, and return the rebuilt post-compaction context messages. -/// -/// # Errors -/// -/// Returns an error when the preparation or the summarizer call fails, -/// or the entry cannot be persisted; a skip is a normal `Ok` outcome. -pub async fn execute_compaction( - session: &mut SessionManager, - options: CompactOptions<'_>, -) -> anyhow::Result { - let started_at = std::time::Instant::now(); +pub(crate) struct CompactionAttempt { + /// The session leaf at prepare: the commit is valid while the active + /// branch still grows through it with no compaction between. + prefix_leaf: Option, + cut: CutPointResult, + previous_summary: Option, + recent_state_anchor: Option, + history: Vec, + turn_prefix_messages: Vec, + tokens_before: u64, + details: CompactionDetails, + first_kept_entry: String, + semantic_compaction: Option, +} + +pub(crate) struct PreparedCompaction { + pub(crate) result: CompactionResult, + pub(crate) entry: pa_types::session::CompactionEntry, +} + +/// PREPARE under the caller's session lock: resolve the cut, begin the +/// ledger guard, and record the leaf the commit's structural check walks +/// back to. +pub(crate) fn prepare_attempt( + session: &SessionManager, + options: &CompactOptions<'_>, +) -> Result { let entries = session.retained_entries().to_vec(); - let preparation = match prepare_compaction(&entries, options.settings.keep_recent_tokens) { - Ok(preparation) => preparation, - Err(skip) => return Ok(CompactOutcome::Skipped(skip.user_message())), - }; + let prefix_leaf = session.get_leaf_id().map(str::to_string); + let preparation = prepare_compaction(&entries, options.settings.keep_recent_tokens)?; // TS `beginCompaction` after the preparation resolved: a skipped // compaction makes no events (the begin comes after the skip throw). - let mut semantic_compaction = options + let semantic_compaction = options .semantic_edges .as_ref() .map(|recorder| recorder.begin_compaction(options.abort)); @@ -193,7 +205,25 @@ pub async fn execute_compaction( file_op_messages.extend(turn_prefix_messages.iter().cloned()); let details: CompactionDetails = details_for(&file_op_messages, &entries, prev_compaction_index); + Ok(CompactionAttempt { + prefix_leaf, + cut, + previous_summary, + recent_state_anchor, + history, + turn_prefix_messages, + tokens_before, + details, + first_kept_entry, + semantic_compaction, + }) +} +/// SUMMARIZE: run the summarizer calls with no session lock held. +pub(crate) async fn summarize_attempt( + attempt: &CompactionAttempt, + options: &CompactOptions<'_>, +) -> anyhow::Result { // A run aborted before the summarizer request never starts one. pa_agent::abort::throw_if_aborted_signal(options.abort)?; @@ -205,11 +235,11 @@ pub async fn execute_compaction( let (model, api_key, summary_headers) = match options.auxiliary { Some(context) => { let required = estimate_summary_request_tokens( - &history, - &turn_prefix_messages, - cut.is_split_turn, - previous_summary.as_deref(), - recent_state_anchor.as_deref(), + &attempt.history, + &attempt.turn_prefix_messages, + attempt.cut.is_split_turn, + attempt.previous_summary.as_deref(), + attempt.recent_state_anchor.as_deref(), options.custom_instructions, options.settings.reserve_tokens, ); @@ -253,7 +283,10 @@ pub async fn execute_compaction( let history_call = async { // The stand-in applies only inside the split arm; a cut without // a turn prefix makes the history call below. - if cut.is_split_turn && !turn_prefix_messages.is_empty() && history.is_empty() { + if attempt.cut.is_split_turn + && !attempt.turn_prefix_messages.is_empty() + && attempt.history.is_empty() + { // The literal stand-in rides the live sink too, exactly // like the committed summary. if let Some(sink) = options.summary_delta.as_ref() { @@ -269,17 +302,17 @@ pub async fn execute_compaction( }); } let request = build_summarization_request( - &history, + &attempt.history, options.custom_instructions, - previous_summary.as_deref(), - recent_state_anchor.as_deref(), + attempt.previous_summary.as_deref(), + attempt.recent_state_anchor.as_deref(), options.settings.reserve_tokens, ); // Each summary wire call carries its own request id under the // compaction's guard (TS `summaryCall`): the id's headers merge // over the routed model's. super::semantic_edges::summary_slice_call( - semantic_compaction.as_ref(), + attempt.semantic_compaction.as_ref(), summary_headers.clone(), |headers| { complete_summary_call( @@ -296,12 +329,12 @@ pub async fn execute_compaction( .await }; let turn_prefix_call = async { - if !cut.is_split_turn || turn_prefix_messages.is_empty() { + if !attempt.cut.is_split_turn || attempt.turn_prefix_messages.is_empty() { return Ok::, anyhow::Error>(None); } - let request = build_turn_prefix_request(&turn_prefix_messages); + let request = build_turn_prefix_request(&attempt.turn_prefix_messages); let slice = super::semantic_edges::summary_slice_call( - semantic_compaction.as_ref(), + attempt.semantic_compaction.as_ref(), summary_headers.clone(), |headers| { complete_summary_call( @@ -355,8 +388,8 @@ pub async fn execute_compaction( None => String::new(), }; remainder.push_str(&file_ops_block( - &details.read_files, - &details.modified_files, + &attempt.details.read_files, + &attempt.details.modified_files, )); if !remainder.is_empty() { sink(&remainder); @@ -369,8 +402,8 @@ pub async fn execute_compaction( None => history_slice.summary.clone(), }; summary.push_str(&file_ops_block( - &details.read_files, - &details.modified_files, + &attempt.details.read_files, + &attempt.details.modified_files, )); let mut slices = vec![history_slice]; if let Some(prefix) = turn_prefix_slice { @@ -378,8 +411,8 @@ pub async fn execute_compaction( } let result = CompactionResult { summary, - first_kept_entry_id: first_kept_entry.clone(), - tokens_before, + first_kept_entry_id: attempt.first_kept_entry.clone(), + tokens_before: attempt.tokens_before, usage: summed_usage(&slices), }; // The digest snapshot and its fingerprint attach at the commit and never flow @@ -399,35 +432,55 @@ pub async fn execute_compaction( ); let entry = compaction_entry_for( &result, - &details, + &attempt.details, options.custom_instructions, harness_digest, harness_state_fingerprint, ); + Ok(PreparedCompaction { result, entry }) +} + +/// COMMIT under the session lock: an aborted run never commits; the +/// prepared prefix still decides the commit's validity; the ledger settles +/// and the compaction row persists behind the mid-window tail. +/// `Ok(false)` is a structural conflict — the caller re-prepares. +/// +/// # Errors +/// +/// The abort marker, or the durable append's I/O error (main's +/// `append_entry` then pops only the compaction row; every acked row is +/// already durable). +pub(crate) fn commit_attempt( + session: &mut SessionManager, + attempt: &mut CompactionAttempt, + prepared: &PreparedCompaction, + abort: Option<&pa_agent::abort::AbortSignal>, +) -> anyhow::Result { + pa_agent::abort::throw_if_aborted_signal(abort)?; + if !session.compaction_prefix_intact(attempt.prefix_leaf.as_deref()) { + return Ok(false); + } // Ledger before effect (TS: `compactionRecorded` and the terminal // event land before `appendCompaction`): a failed persist still // leaves a completed compaction on the ledger, exactly like TS. - if let Some(compaction) = semantic_compaction.as_mut() { + if let Some(compaction) = attempt.semantic_compaction.as_mut() { compaction.commit(); } // TS `appendCompaction` persists the full record: `details`, // `fromHook`, `customInstructions`, `usage`, and the `harnessDigest` // snapshot ride on the durable row alongside the summary, boundary, - // and token count. - session.append_compaction(entry.clone())?; + // and token count. The row's parent is the CURRENT leaf, so the + // mid-window tail sits between `first_kept_entry` and the compaction + // row and rides the retained tail. + session.append_compaction(prepared.entry.clone())?; super::compaction_trace::trace( "compact.entry_appended", &serde_json::json!({ - "firstKeptEntryId": first_kept_entry, + "firstKeptEntryId": attempt.first_kept_entry, "persisted": session.is_persisted(), }), ); - Ok(CompactOutcome::Ran(Box::new(CompactRun { - result, - entry, - duration_ms: started_at.elapsed().as_millis() as u64, - ipython_state: None, - }))) + Ok(true) } /// Rebuild the live agent context after compaction. Keep session-only roles diff --git a/crates/pa-core/src/session_engine/compact_session/tests.rs b/crates/pa-core/src/session_engine/compact_session/tests.rs index 7298c8fe6..0b1c6def5 100644 --- a/crates/pa-core/src/session_engine/compact_session/tests.rs +++ b/crates/pa-core/src/session_engine/compact_session/tests.rs @@ -78,3 +78,25 @@ mod prepare_compaction; mod second_compaction; mod skip_guards; mod split_turn; + +async fn execute_compaction( + session: &mut SessionManager, + options: CompactOptions<'_>, +) -> anyhow::Result { + let started_at = std::time::Instant::now(); + let mut attempt = match prepare_attempt(session, &options) { + Ok(attempt) => attempt, + Err(skip) => return Ok(CompactOutcome::Skipped(skip.user_message())), + }; + let prepared = summarize_attempt(&attempt, &options).await?; + assert!( + commit_attempt(session, &mut attempt, &prepared, options.abort)?, + "no concurrent writer in a unit test" + ); + Ok(CompactOutcome::Ran(Box::new(CompactRun { + result: prepared.result, + entry: prepared.entry, + duration_ms: started_at.elapsed().as_millis() as u64, + ipython_state: None, + }))) +} diff --git a/crates/pa-core/src/session_engine/compaction_arms.rs b/crates/pa-core/src/session_engine/compaction_arms.rs index f439ceec7..0493c2cac 100644 --- a/crates/pa-core/src/session_engine/compaction_arms.rs +++ b/crates/pa-core/src/session_engine/compaction_arms.rs @@ -164,39 +164,41 @@ impl AgentSession { api_key: Option, abort: Option<&pa_agent::abort::AbortSignal>, ) -> anyhow::Result { - // The digest inputs come from the live (pre-compaction) context; - // harness state reads fresh from disk when the snapshot renders. compaction_trace::trace( "compact.enter", &serde_json::json!({ "customInstructions": custom_instructions.is_some(), }), ); - let digest_inputs = self.harness_digest_inputs().await; - compaction_trace::trace("compact.digest_captured", &serde_json::Value::Null); - let mut outcome = { - let mut session = self.session.lock().await; - let summary_delta = self - .compaction_summary_sink - .lock() - .expect("compaction summary sink lock") - .clone(); - crate::session_engine::compact_session::execute_compaction( - &mut session, - crate::session_engine::compact_session::CompactOptions { - model: model.clone(), - api_key, - custom_instructions, - settings: self.compaction_settings(), - abort, - harness_digest: digest_inputs, - auxiliary: self.auxiliary_model.as_ref(), - summary_delta, - semantic_edges: self.semantic_edges(), - }, - ) - .await? + let started_at = std::time::Instant::now(); + let summary_delta = self + .compaction_summary_sink + .lock() + .expect("compaction summary sink lock") + .clone(); + let options = crate::session_engine::compact_session::CompactOptions { + model: model.clone(), + api_key, + custom_instructions, + settings: self.compaction_settings(), + abort, + // Captured per attempt inside `compaction_attempts`: a + // conflict retry summarizes a new branch, and inputs captured + // here would rank the abandoned branch's terms. + harness_digest: None, + auxiliary: self.auxiliary_model.as_ref(), + summary_delta, + semantic_edges: self.semantic_edges(), }; + // The flight spans the attempts, the rebuilt-context replace, and + // the kernel-state notice: every path here REPLACES the live loop + // context, and the other replacers — `refine_with_refiner`'s + // outcome push and `record_compaction_outcome`'s failure row — + // take the same flight. A row pushed onto the context must not + // interleave with the replace: it would duplicate in the rebuilt + // view or vanish under it while staying durable either way. + let _flight = self.compaction_flight.lock().await; + let mut outcome = self.compaction_attempts(&options, started_at).await?; if matches!(outcome, CompactOutcome::Skipped(_)) { compaction_trace::trace("compact.skipped", &serde_json::Value::Null); return Ok(outcome); @@ -235,6 +237,75 @@ impl AgentSession { Ok(outcome) } + /// The PREPARE -> SUMMARIZE -> COMMIT loop: prepare under the session + /// lock, summarize with no lock held (mid-window appends chain onto + /// the leaf and ride the retained tail), then commit under the lock — + /// a structural conflict retries from PREPARE, an aborted run stops. + async fn compaction_attempts( + &self, + options: &crate::session_engine::compact_session::CompactOptions<'_>, + started_at: std::time::Instant, + ) -> anyhow::Result { + /// The conflict-retry bound: a branch that keeps changing under + /// the compaction fails instead of re-summarizing forever. + const MAX_COMPACTION_ATTEMPTS: usize = 3; + for _ in 0..MAX_COMPACTION_ATTEMPTS { + let mut attempt = { + let session = self.session.lock().await; + match crate::session_engine::compact_session::prepare_attempt(&session, options) { + Ok(attempt) => attempt, + Err(skip) => return Ok(CompactOutcome::Skipped(skip.user_message())), + } + }; + // The digest ranks the branch THIS attempt summarizes: a + // conflict retry prepares a new tree, and inputs captured + // once before the loop would rank the abandoned branch's + // terms. The capture takes the session lock itself, so it + // runs off the prepare hold. + let digest_inputs = self.harness_digest_inputs().await; + compaction_trace::trace("compact.digest_captured", &serde_json::Value::Null); + let attempt_options = crate::session_engine::compact_session::CompactOptions { + model: options.model.clone(), + api_key: options.api_key.clone(), + custom_instructions: options.custom_instructions, + settings: options.settings, + abort: options.abort, + harness_digest: digest_inputs, + auxiliary: options.auxiliary, + summary_delta: options.summary_delta.clone(), + semantic_edges: options.semantic_edges.clone(), + }; + let prepared = crate::session_engine::compact_session::summarize_attempt( + &attempt, + &attempt_options, + ) + .await?; + let committed = { + let mut session = self.session.lock().await; + crate::session_engine::compact_session::commit_attempt( + &mut session, + &mut attempt, + &prepared, + options.abort, + )? + }; + if committed { + return Ok(CompactOutcome::Ran(Box::new( + crate::session_engine::compact_session::CompactRun { + result: prepared.result, + entry: prepared.entry, + duration_ms: started_at.elapsed().as_millis() as u64, + ipython_state: None, + }, + ))); + } + pa_agent::abort::throw_if_aborted_signal(options.abort)?; + } + Err(anyhow::anyhow!( + "compaction retried {MAX_COMPACTION_ATTEMPTS} times while the branch kept changing; try again" + )) + } + /// Record an unsuccessful compaction outcome: append the durable /// `compaction_outcome` row and push it onto the live loop context. The /// row is a user-facing disclosure, never model context (`convert_to_llm` drops it). @@ -249,6 +320,11 @@ impl AgentSession { outcome: crate::session_engine::messages::CompactionOutcomeKind, content: &str, ) -> anyhow::Result { + // The same flight as the other live-context replacers: the push + // below read-modify-writes the live context, and a replace + // interleaving it would roll the pushed row — or the replace — + // back. + let _flight = self.compaction_flight.lock().await; let row = crate::session_engine::messages::create_compaction_outcome_message( content, reason, outcome, ); @@ -331,6 +407,12 @@ impl AgentSession { refine_call: crate::refinement::executor::RefinerFn, global_harness_dir: std::path::PathBuf, ) -> anyhow::Result { + // The flight serializes every live-context replacer against a + // compaction's rebuild: this round's outcome rows persist and then + // push onto the context the compaction replaces, so the push must + // not interleave with a replace (duplicating in the rebuilt view + // or vanishing under it). + let _flight = self.compaction_flight.lock().await; // The transcript's consumed artifacts are extracted under this first // lock straight from the retained rows: no second clone of the rows. let parts = self.session.lock().await.refine_transcript_parts(); diff --git a/crates/pa-core/src/session_engine/compaction_unblocked_tests.rs b/crates/pa-core/src/session_engine/compaction_unblocked_tests.rs new file mode 100644 index 000000000..0b574ab80 --- /dev/null +++ b/crates/pa-core/src/session_engine/compaction_unblocked_tests.rs @@ -0,0 +1,322 @@ +use std::sync::Arc; + +use pa_agent::agent::{Agent, AgentOptions}; +use pa_types::ai::UserContent; +use pa_types::session::AgentMessage as SessionAgentMessage; + +use super::compact_session::CompactOutcome; +use super::{AgentSession, FileEntry, SessionManager}; + +fn user_turn(text: &str) -> SessionAgentMessage { + SessionAgentMessage::User(pa_types::ai::UserMessage { + content: UserContent::Text(text.to_string()), + timestamp: 0, + rest: pa_types::JsonMap::default(), + }) +} + +fn faux_reply(text: &str) -> pa_types::ai::AssistantMessage { + pa_ai::faux::faux_assistant_text_message( + text, + pa_ai::faux::FauxAssistantMessageOptions::default(), + ) +} + +fn assistant_turn(text: &str) -> SessionAgentMessage { + SessionAgentMessage::Assistant(faux_reply(text)) +} + +fn follow_up(text: &str) -> pa_ai::faux::FauxResponseStep { + pa_ai::faux::FauxResponseStep::Message(faux_reply(text)) +} + +/// A held summarizer: the first `holds` responses block until released +/// (each call signals its start, then waits for one release); the +/// follow-ups answer immediately. +struct HeldSummarizer { + registration: pa_ai::faux::FauxProviderRegistration, + started: tokio::sync::mpsc::UnboundedReceiver<()>, + release: std::sync::mpsc::Sender<()>, + model: pa_types::ai::Model, +} + +fn held_summarizer(holds: usize, follow_ups: Vec) -> HeldSummarizer { + let (started, started_rx) = tokio::sync::mpsc::unbounded_channel(); + let (release, release_rx) = std::sync::mpsc::channel(); + let release_rx = Arc::new(std::sync::Mutex::new(release_rx)); + // The registration must be unique so concurrent tests never share a queue. + let api = format!("faux-held-{:?}", std::time::SystemTime::now()); + let registration = + pa_ai::faux::register_faux_provider(pa_ai::faux::RegisterFauxProviderOptions { + api: Some(api), + provider: Some("faux-held".to_string()), + models: Some(vec![pa_ai::faux::FauxModelDefinition { + id: "held-m".to_string(), + ..Default::default() + }]), + ..Default::default() + }); + let held = { + let release_rx = Arc::clone(&release_rx); + move |_: &pa_types::ai::Context, + _: Option<&pa_ai::types::StreamOptions>, + _: u64, + _: &pa_types::ai::Model| { + let _ = started.send(()); + let _ = release_rx.lock().expect("release lock").recv(); + Ok(faux_reply("## Goal\nsummarized goal")) + } + }; + let mut responses = vec![pa_ai::faux::FauxResponseStep::Factory(Arc::new(held)); holds]; + responses.extend(follow_ups); + registration.set_responses(responses); + let model = registration.get_model(); + HeldSummarizer { + registration, + started: started_rx, + release, + model, + } +} + +/// A session with `turns` exchanges whose compaction cuts at +/// `keep_recent_tokens` 12. +async fn held_engine(persisted: bool, turns: usize) -> (Arc, tempfile::TempDir) { + let tmp = tempfile::tempdir().unwrap(); + let mut manager = if persisted { + SessionManager::persisted(tmp.path(), tmp.path()) + } else { + SessionManager::in_memory(tmp.path()) + }; + for i in 0..turns { + manager + .append_message(user_turn(&format!("turn {i} with a few words"))) + .and_then(|_| manager.append_message(assistant_turn(&format!("reply {i}")))) + .unwrap(); + } + let agent = Arc::new(Agent::new(AgentOptions::default())); + let engine = AgentSession::new(agent, manager, Vec::new()).await.unwrap(); + engine.set_compaction_settings(crate::session_engine::compaction::CompactionSettings { + keep_recent_tokens: 12, + ..Default::default() + }); + (Arc::new(engine), tmp) +} + +async fn wait_for_held_summarizer(summarizer: &mut HeldSummarizer) { + let window = std::time::Duration::from_secs(3); + let Ok(started) = tokio::time::timeout(window, summarizer.started.recv()).await else { + panic!("the summarizer request never started in the window"); + }; + started.expect("the summarizer request started"); +} + +fn release(summarizer: &HeldSummarizer) { + summarizer + .release + .send(()) + .expect("the summarizer released"); +} + +/// The session file's rows, parsed straight off disk. +fn file_entries(path: &std::path::Path) -> Vec { + crate::session::parse_session_entries(&std::fs::read_to_string(path).expect("the file reads")) +} + +fn spawn_compact( + engine: &Arc, + model: &pa_types::ai::Model, +) -> tokio::task::JoinHandle> { + let engine = Arc::clone(engine); + let model = model.clone(); + tokio::spawn(async move { engine.compact(None, &model, None, None).await }) +} + +/// A mid-window write is durable in the session file right away, visible +/// to `retained_entries()` reads (what `register_spawn` walks), and the +/// committed context survives a file reopen. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn mid_summarize_writes_are_durable_now_and_ride_the_retained_tail() { + let (engine, tmp) = held_engine(true, 3).await; + let session_file = engine + .shared_persistence() + .lock() + .await + .get_session_file() + .expect("the session file") + .to_path_buf(); + let mut summarizer = held_summarizer(1, Vec::new()); + let compact = spawn_compact(&engine, &summarizer.model); + wait_for_held_summarizer(&mut summarizer).await; + + let assistant_id = { + let persistence = engine.shared_persistence(); + let mut session = persistence.lock().await; + let (id, _) = session.append_message_retained(assistant_turn("mid-window reply")); + let last = session + .retained_entries() + .iter() + .rev() + .find_map(|entry| match entry { + FileEntry::Message { + message: SessionAgentMessage::Assistant(_), + base, + } => base.id.clone(), + _ => None, + }); + assert_eq!( + last, + Some(id.clone()), + "the mid-window row is visible in retained_entries()" + ); + id + }; + assert!( + file_entries(&session_file) + .iter() + .any(|entry| entry.id() == Some(assistant_id.as_str())), + "the mid-window row is durable in the session file during the window" + ); + + release(&summarizer); + compact + .await + .expect("the compact task joined") + .expect("the compaction ran"); + let rebuilt_live = { + let session = engine.shared_persistence(); + let session = session.lock().await; + crate::session_engine::compact_session::rebuilt_context_after_compaction(&session) + }; + let reopened = SessionManager::open(tmp.path(), tmp.path(), &session_file); + assert_eq!( + reopened.active_context().messages, + rebuilt_live, + "the reopened file rebuilds the same context" + ); + summarizer.registration.unregister(); +} + +/// A branch move mid-window is a structural conflict: the compaction +/// retries from PREPARE and commits against the moved-to tree. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_branch_move_mid_summarize_conflicts_and_commits_against_the_new_tree() { + let (engine, _tmp) = held_engine(false, 6).await; + let mut summarizer = held_summarizer(1, vec![follow_up("## Goal\nretry summary")]); + let compact = spawn_compact(&engine, &summarizer.model); + wait_for_held_summarizer(&mut summarizer).await; + + let entries = engine.entries().await; + let branch_entries = entries[..entries.len() - 4].to_vec(); + let moved_to_leaf = branch_entries + .last() + .and_then(FileEntry::id) + .expect("the moved-to leaf") + .to_string(); + engine + .rebuild_branch_context(branch_entries) + .await + .expect("the branch context rebuilt"); + + release(&summarizer); + compact + .await + .expect("the compact task joined") + .expect("the retried compaction ran"); + assert_eq!( + summarizer.registration.call_count(), + 2, + "the branch move conflicted and retried the summarizer" + ); + let entries = engine.entries().await; + let compaction = entries + .iter() + .rev() + .find(|entry| matches!(entry, FileEntry::Compaction { .. })) + .expect("the compaction row"); + assert_eq!( + compaction.parent_id(), + Some(moved_to_leaf.as_str()), + "the compaction committed against the moved-to tree" + ); + summarizer.registration.unregister(); +} + +/// Overlapping compactions serialize: the second waits for the +/// in-flight run instead of summarizing the same prefix again. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_second_compaction_waits_for_the_in_flight_one() { + let (engine, _tmp) = held_engine(false, 3).await; + // The follow-up response would let a racing second summarizer finish. + let mut summarizer = held_summarizer(1, vec![follow_up("## Goal\nracing summary")]); + let first = spawn_compact(&engine, &summarizer.model); + wait_for_held_summarizer(&mut summarizer).await; + let second = spawn_compact(&engine, &summarizer.model); + + release(&summarizer); + first + .await + .expect("the first compact joined") + .expect("the first compaction ran"); + second + .await + .expect("the second compact joined") + .expect("the second compaction resolved"); + assert_eq!( + summarizer.registration.call_count(), + 1, + "only the in-flight compaction summarized" + ); + let compactions = engine + .entries() + .await + .iter() + .filter(|entry| matches!(entry, FileEntry::Compaction { .. })) + .count(); + assert_eq!(compactions, 1, "exactly one committed compaction"); + summarizer.registration.unregister(); +} + +/// A branch rebuild during every attempt: the compaction fails with the +/// retry-cap error after three summarizer calls, never committing. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn repeated_branch_conflicts_fail_the_compaction_after_three_attempts() { + let (engine, _tmp) = held_engine(false, 6).await; + let mut summarizer = held_summarizer(3, Vec::new()); + let compact = spawn_compact(&engine, &summarizer.model); + // Two disjoint branch halves to alternate between: every rebuild is + // a structural conflict and every re-prepare stays compactable. + let full = engine.entries().await; + let trees = [ + full[..full.len() / 2].to_vec(), + full[full.len() / 2..].to_vec(), + ]; + for tree in 0..3 { + wait_for_held_summarizer(&mut summarizer).await; + engine + .rebuild_branch_context(trees[tree % 2].clone()) + .await + .expect("the branch context rebuilt"); + release(&summarizer); + } + let error = compact + .await + .expect("the compact task joined") + .expect_err("the capped compaction failed"); + assert_eq!( + error.to_string(), + "compaction retried 3 times while the branch kept changing; try again" + ); + assert_eq!( + summarizer.registration.call_count(), + 3, + "each attempt summarized once" + ); + let committed = engine + .entries() + .await + .iter() + .any(|entry| matches!(entry, FileEntry::Compaction { .. })); + assert!(!committed, "the capped compaction never committed"); + summarizer.registration.unregister(); +} diff --git a/crates/pa-core/src/session_engine/mod.rs b/crates/pa-core/src/session_engine/mod.rs index 85c5c7885..e1c1e2b61 100644 --- a/crates/pa-core/src/session_engine/mod.rs +++ b/crates/pa-core/src/session_engine/mod.rs @@ -158,6 +158,10 @@ pub struct AgentSession { /// daemon can install it after the build; `None` (the default) /// keeps the one-shot summarizer completion. compaction_summary_sink: std::sync::Mutex>, + /// One compaction at a time: a second `compact` waits for the + /// in-flight run and prepares against its result; reads and writes + /// keep using the session lock meanwhile. + compaction_flight: tokio::sync::Mutex<()>, /// The session's semantic-edge recorder (TS /// `AgentSession._semanticEdges`): `None` in sessions the engine /// built without a semantic identity (verification harnesses @@ -244,6 +248,7 @@ impl AgentSession { semantic_edges: std::sync::Mutex::new(None), side_question_stream_fn: std::sync::Mutex::new(None), agent_dir: None, + compaction_flight: tokio::sync::Mutex::new(()), }; this.ensure_harness_digest_context().await?; Ok(this) @@ -480,3 +485,6 @@ mod slash_session_tests; #[cfg(test)] mod compaction_outcome_tests; + +#[cfg(test)] +mod compaction_unblocked_tests; diff --git a/crates/pa-core/src/session_engine/semantic_edges.rs b/crates/pa-core/src/session_engine/semantic_edges.rs index 2838b6a18..67a374ae1 100644 --- a/crates/pa-core/src/session_engine/semantic_edges.rs +++ b/crates/pa-core/src/session_engine/semantic_edges.rs @@ -30,7 +30,7 @@ use std::sync::{Mutex, MutexGuard}; use serde::{Deserialize, Serialize}; -pub(crate) use compaction::summary_slice_call; +pub(crate) use compaction::{summary_slice_call, SemanticCompaction}; pub(crate) use stream::wrap_stream_fn; /// The outbound request-id header (TS `MODEL_REQUEST_ID_HEADER`). diff --git a/crates/pa-daemon/src/auto_compaction.rs b/crates/pa-daemon/src/auto_compaction.rs index 4a215594e..622d9db94 100644 --- a/crates/pa-daemon/src/auto_compaction.rs +++ b/crates/pa-daemon/src/auto_compaction.rs @@ -82,12 +82,13 @@ impl AgentSessionEngine { .expect("auto compaction abort lock") = Some(std::sync::Arc::clone(&controller)); } let api_key = self.resolve_request_api_key(&model); + // The lock covers the clone only; the summarizer call below must not ride it. + let engine = self.session.blocking_lock().clone(); + let Some(engine) = engine else { + self.clear_auto_compaction_abort(&controller); + return AutoCompactionRun::NotDue; + }; let outcome = { - let guard = self.session.blocking_lock(); - let Some(engine) = guard.as_deref() else { - self.clear_auto_compaction_abort(&controller); - return AutoCompactionRun::NotDue; - }; // The abort race drops the summarizer request in flight; the // signal also lands the pre-commit check inside the compaction. let compact = async { diff --git a/crates/pa-daemon/src/overflow_compaction.rs b/crates/pa-daemon/src/overflow_compaction.rs index 56fa62cca..8a386bf7b 100644 --- a/crates/pa-daemon/src/overflow_compaction.rs +++ b/crates/pa-daemon/src/overflow_compaction.rs @@ -274,12 +274,13 @@ impl AgentSessionEngine { .expect("auto compaction abort lock") = Some(std::sync::Arc::clone(&controller)); } let api_key = self.resolve_request_api_key(&model); + // The lock covers the clone only; the summarizer call below must not ride it. + let engine = self.session.blocking_lock().clone(); + let Some(engine) = engine else { + self.clear_auto_compaction_abort(&controller); + return OverflowAttempt::None; + }; let outcome = { - let guard = self.session.blocking_lock(); - let Some(engine) = guard.as_deref() else { - self.clear_auto_compaction_abort(&controller); - return OverflowAttempt::None; - }; let compact = async { engine .session