diff --git a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs index 00751d63f6..bba3ff2e71 100644 --- a/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs +++ b/crates/tracedecay-code-index-runtime/tests/code_index_scheduler/serving.rs @@ -6889,6 +6889,11 @@ async fn graph_off_overflow_preserves_text_owner_progress_without_full_decode() .expect("remove restore witness to reproduce a cold preserved-profile mount"); let registry = CodeIndexSchedulerRegistryV1::with_background_reconcile_permits(1, 1); + let status_poll_admission = registry + .background_reconcile_admission() + .acquire_owned() + .await + .expect("hold text projection while establishing its first committed slice"); registry .mount_worktree_with_graph_policy( test_project_id(), @@ -6902,46 +6907,48 @@ async fn graph_off_overflow_preserves_text_owner_progress_without_full_decode() .mount_query_authority(fixture.path(), &scope, query_authority(privacy_domain)) .await .expect("mount query authority"); - let scheduler = { + let (scheduler, text_generation) = { let mounted = registry.mounted.lock().await; - Arc::clone( - &mounted - .get(&canonical_existing_identity(fixture.path()).expect("canonical root")) - .expect("mounted worktree") - .scheduler, + let worktree = mounted + .get(&canonical_existing_identity(fixture.path()).expect("canonical root")) + .expect("mounted worktree"); + ( + Arc::clone(&worktree.scheduler), + Arc::clone(&worktree.text_generation), ) }; - - let progress_deadline = std::time::Instant::now() + Duration::from_secs(10); - let (owner_epoch_before_overflow, progress_before_overflow) = loop { - let observed = { - let scheduler = scheduler - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - let progress = scheduler.build_progress_slot(); - let progress = progress - .read() - .unwrap_or_else(std::sync::PoisonError::into_inner); - (progress.owner_epoch, progress.snapshot()) - }; - if let (owner_epoch, Some(progress)) = observed - && progress.committed_pages > 0 - && progress.phase - != tracedecay_contracts::code_index_freshness::CodeIndexBuildPhaseV1::Ready - { - break (owner_epoch, progress); - } - assert!( - std::time::Instant::now() <= progress_deadline, - "text projection completed before exposing bounded live progress" + // Establish the real retained owner and advance one bounded page while + // background admission is held. Polling a running worker can miss every + // intermediate slice when a fast build finishes between observations. + let text = scheduler + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .servable_retained_text_generation() + .expect("restore retained text owner") + .expect("retained text owner is servable"); + assert!( + !text.advance_text_serving(1).expect("commit one text page"), + "the fixture must retain work after its first bounded page" + ); + *text_generation + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(text); + let (owner_epoch_before_overflow, progress_before_overflow) = { + let scheduler = scheduler + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let progress = scheduler.build_progress_slot(); + let progress = progress + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let snapshot = progress.snapshot().expect("committed text progress"); + assert!(snapshot.committed_pages > 0); + assert_ne!( + snapshot.phase, + tracedecay_contracts::code_index_freshness::CodeIndexBuildPhaseV1::Ready ); - tokio::time::sleep(Duration::from_millis(2)).await; + (progress.owner_epoch, snapshot) }; - let status_poll_admission = registry - .background_reconcile_admission() - .acquire_owned() - .await - .expect("pause text projection after a bounded committed slice"); let (last_reconciled_before_status_polls, original_staleness_threshold) = { let mut scheduler = scheduler .lock() diff --git a/crates/tracedecay-global-db/src/observation_projection.rs b/crates/tracedecay-global-db/src/observation_projection.rs index 61e0f40970..0d65142553 100644 --- a/crates/tracedecay-global-db/src/observation_projection.rs +++ b/crates/tracedecay-global-db/src/observation_projection.rs @@ -1,5 +1,6 @@ pub(crate) mod apply; mod codex_echo; +mod project_path_rearm; #[cfg(test)] mod rearm_tests; mod rebuild; @@ -19,6 +20,7 @@ pub(super) use apply::{ derive_projection_with_alias, retire_source_record, verify_workflow_effects, }; pub(crate) use codex_echo::converge_codex_response_echoes; +pub(crate) use project_path_rearm::rearm_project_path_collisions; pub use rebuild::{ converge_projection_predecessor, project_observation, project_queued_observations, rebuild_projection, diff --git a/crates/tracedecay-global-db/src/observation_projection/project_path_rearm.rs b/crates/tracedecay-global-db/src/observation_projection/project_path_rearm.rs new file mode 100644 index 0000000000..82b9677864 --- /dev/null +++ b/crates/tracedecay-global-db/src/observation_projection/project_path_rearm.rs @@ -0,0 +1,228 @@ +//! Session projection used to refuse a Codex observation whose session row +//! already stored a different concrete `project_path`, parking the queue row +//! on the terminal deadline with that collision as its last error. Projection +//! now keeps the stored path, so those rows are re-armed once per store and +//! project their messages. Collisions on any other field stay parked. + +use std::future::Future; +use std::pin::Pin; + +use tracedecay_domain::CanonicalObservationIdV1; +use tracedecay_runtime_core::db::engine::params; +use tracedecay_runtime_core::db::schema_version; +use tracedecay_runtime_core::db::{Database, DatabaseWriteTransaction}; +use tracedecay_session_temporal_store::record_canonical_observation_effect; +use tracedecay_store::{ + ObservationProjection, PROJECTION_TERMINAL_RETRY_MICROS, ProjectionSkipReason, + ProjectionStoreError, ProjectionStoreResult, SESSION_MESSAGE_PROJECTOR_VERSION, +}; + +use super::apply::verify_effect; +use super::rebuild::{REBUILD_PAGE_SIZE, reproject_project_path_collision}; +use super::state::{query_optional_i64, read_observation, storage}; +use crate::schema_contract::invariants::TEMPORAL_OBSERVATION_EFFECT_DELETE_GUARD_SQL; + +const OPERATION: &str = "rearm Codex project path collisions"; +const CONVERGENCE: &str = "codex-project-path-collision-rearm-v1"; + +/// Re-arms queue rows parked terminal only because of the `project_path` +/// collision, once per store, and returns how many it re-armed. +/// +/// The rowid cutoff excludes new work, which cannot emit the obsolete error. +/// Each committed window bounds both the scan and the update, including when +/// matching errors are sparse. An interrupted walk can revisit its committed +/// windows: their cleared deadlines no longer match, and the completion +/// marker is written only with the final window. +#[tracing::instrument( + name = "global_db.observation_projection.rearm_project_path_collisions", + level = "trace", + skip_all +)] +pub(crate) async fn rearm_project_path_collisions( + database: &Database, +) -> ProjectionStoreResult { + let last_rowid = query_optional_i64( + &database.read_connection(), + "SELECT MAX(rowid) FROM projection_queue", + params![], + ) + .await? + .unwrap_or(0); + let mut after_rowid = 0; + let mut rearmed = 0; + while let Some((batch_end, rows)) = + rearm_project_path_collision_batch(database, after_rowid, last_rowid, REBUILD_PAGE_SIZE) + .await? + { + after_rowid = batch_end; + rearmed += rows; + asupersync::runtime::yield_now().await; + } + Ok(rearmed) +} + +#[tracing::instrument( + name = "global_db.observation_projection.rearm_project_path_collisions.batch", + level = "trace", + skip_all +)] +pub(super) async fn rearm_project_path_collision_batch( + database: &Database, + after_rowid: i64, + last_rowid: i64, + batch_rows: i64, +) -> ProjectionStoreResult> { + let transaction = database + .begin_write_transaction(OPERATION) + .await + .map_err(|error| storage(OPERATION, error))?; + let converged = schema_version::read_version(&transaction, CONVERGENCE) + .await + .map_err(|error| storage(OPERATION, error))? + .is_some(); + if converged { + transaction + .commit() + .await + .map_err(|error| storage(OPERATION, error))?; + return Ok(None); + } + let batch_end = query_optional_i64( + &transaction, + "SELECT MAX(rowid) FROM ( + SELECT rowid FROM projection_queue + WHERE rowid > ?1 AND rowid <= ?2 + ORDER BY rowid LIMIT ?3 + )", + params![after_rowid, last_rowid, batch_rows], + ) + .await?; + let Some(batch_end) = batch_end else { + schema_version::record_version(&transaction, CONVERGENCE, 1) + .await + .map_err(|error| storage(OPERATION, error))?; + transaction + .commit() + .await + .map_err(|error| storage(OPERATION, error))?; + return Ok(None); + }; + let replay = completed_project_path_collisions(&transaction, after_rowid, batch_end).await?; + let rows = transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = 0 + WHERE rowid > ?1 AND rowid <= ?2 + AND next_retry_at_micros = ?3 + AND last_error LIKE 'projection session output collided at codex/%: project_path differs%'", + params![after_rowid, batch_end, PROJECTION_TERMINAL_RETRY_MICROS], + ) + .await + .map_err(|error| storage(OPERATION, error))?; + if !replay.is_empty() { + // beta.88 recorded an OutputCollision disposition, advanced the + // checkpoint, and wrote an empty temporal effect when it parked the + // row. Clearing the deadline alone would acknowledge that old skip as + // an ExactDuplicate. Replace only those verified empty derived + // effects; observations and sanitization receipts remain immutable. + // The guard is restored before the same transaction commits, and a + // failed repair rolls back the guard and every row together. + transaction + .execute_batch( + "DROP TRIGGER IF EXISTS session_temporal_observation_effects_immutable_delete_v1", + ) + .await + .map_err(|error| storage(OPERATION, error))?; + for observation_id in replay { + let repaired: Pin> + Send + '_>> = + Box::pin(repair_completed_collision(&transaction, &observation_id)); + repaired.await?; + } + transaction + .execute_batch(TEMPORAL_OBSERVATION_EFFECT_DELETE_GUARD_SQL) + .await + .map_err(|error| storage(OPERATION, error))?; + } + if batch_end == last_rowid { + schema_version::record_version(&transaction, CONVERGENCE, 1) + .await + .map_err(|error| storage(OPERATION, error))?; + } + transaction + .commit() + .await + .map_err(|error| storage(OPERATION, error))?; + Ok(Some((batch_end, rows))) +} + +async fn completed_project_path_collisions( + transaction: &DatabaseWriteTransaction<'_>, + after_rowid: i64, + batch_end: i64, +) -> ProjectionStoreResult> { + let mut rows = transaction + .query( + "SELECT queue.observation_id + FROM projection_queue AS queue + WHERE queue.rowid > ?1 AND queue.rowid <= ?2 + AND queue.next_retry_at_micros = ?3 + AND queue.last_error LIKE 'projection session output collided at codex/%: project_path differs%' + AND EXISTS ( + SELECT 1 FROM observation_projection_dispositions AS disposition + WHERE disposition.observation_id = queue.observation_id + AND disposition.projector_version = ?4 AND disposition.reason = ?5 + ) + ORDER BY queue.rowid", + params![ + after_rowid, + batch_end, + PROJECTION_TERMINAL_RETRY_MICROS, + SESSION_MESSAGE_PROJECTOR_VERSION, + ProjectionSkipReason::OutputCollision.as_str(), + ], + ) + .await + .map_err(|error| storage(OPERATION, error))?; + let mut observations = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| storage(OPERATION, error))? + { + let id = row + .get::(0) + .map_err(|error| storage(OPERATION, error))?; + observations + .push(CanonicalObservationIdV1::new(id).map_err(ProjectionStoreError::Contract)?); + } + Ok(observations) +} + +async fn repair_completed_collision( + transaction: &DatabaseWriteTransaction<'_>, + observation_id: &CanonicalObservationIdV1, +) -> ProjectionStoreResult<()> { + let Some((sequence, observation)) = read_observation(transaction, observation_id).await? else { + return Err(ProjectionStoreError::ObservationNotFound); + }; + let skipped = ObservationProjection::Skipped(ProjectionSkipReason::OutputCollision); + verify_effect(transaction, &observation, &skipped).await?; + record_canonical_observation_effect(transaction, sequence, &observation, &skipped).await?; + transaction + .execute( + "DELETE FROM observation_projection_dispositions + WHERE projector_version = ?1 AND observation_id = ?2 AND reason = ?3", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + observation_id.as_str(), + ProjectionSkipReason::OutputCollision.as_str(), + ], + ) + .await + .map_err(|error| storage(OPERATION, error))?; + Box::pin(reproject_project_path_collision( + transaction, + sequence, + &observation, + )) + .await +} diff --git a/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs b/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs index 04ac298504..b6e16e1849 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs @@ -7,8 +7,13 @@ use tracedecay_domain::{ SanitizationReceiptRefV1, SanitizationReceiptV1, SanitizerDispositionV1, SensitivityV1, }; use tracedecay_runtime_core::db::engine::params; +use tracedecay_runtime_core::db::schema_version; -use super::{RearmedProjectionRetries, rearm_queued_projection_retries}; +use super::project_path_rearm::rearm_project_path_collision_batch; +use super::state::query_optional_i64; +use super::{ + RearmedProjectionRetries, rearm_project_path_collisions, rearm_queued_projection_retries, +}; use crate::tests::harness::RegisteredGlobalDbHarness; const FAR_FUTURE_MICROS: i64 = 9_000_000_000_000_000; @@ -212,3 +217,179 @@ async fn daemon_admission_defers_rearm_to_background_convergence() { "background convergence re-arms the fresh mount's queue" ); } + +async fn parked_retry_deadlines(harness: &RegisteredGlobalDbHarness) -> Vec { + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT next_retry_at_micros FROM projection_queue ORDER BY observation_sequence", + (), + ) + .await + .unwrap(); + let mut deadlines = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + deadlines.push(row.get(0).unwrap()); + } + deadlines +} + +#[asupersync::test] +async fn project_path_collisions_rearm_once_and_genuine_collisions_stay_parked() { + let harness = RegisteredGlobalDbHarness::open("rearm-project-path-collisions").await; + seed_queue(&harness, 6, 0).await; + let errors = [ + "projection session output collided at codex/s-1: project_path differs", + "projection session output collided at codex/s-2: transcript_path differs", + "projection session output collided at codex/s-3: project_key differs", + "projection session output collided at codex/s-4: parent_session_id differs", + "projection session output collided at claude/s-5: project_path differs", + "projection session output collided at codex/s-6: project_path differs", + ]; + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + // Opening a fresh store already converges it. Model an older store whose + // parked rows predate the one-time project-path repair. + transaction + .execute( + "DELETE FROM tracedecay_schema_version + WHERE component = 'codex-project-path-collision-rearm-v1'", + (), + ) + .await + .unwrap(); + for (sequence, error) in (1_i64..).zip(errors) { + let deadline = if sequence == 6 { + FAR_FUTURE_MICROS + } else { + i64::MAX + }; + transaction + .execute( + "UPDATE projection_queue + SET next_retry_at_micros = ?2, last_error = ?3 + WHERE observation_sequence = ( + SELECT sequence FROM observations ORDER BY sequence LIMIT 1 OFFSET ?1)", + params![sequence - 1, deadline, error], + ) + .await + .unwrap(); + } + transaction.commit().await.unwrap(); + let parked = vec![ + i64::MAX, + i64::MAX, + i64::MAX, + i64::MAX, + i64::MAX, + FAR_FUTURE_MICROS, + ]; + assert_eq!(parked_retry_deadlines(&harness).await, parked); + + let database = harness.registered.runtime_database(); + assert_eq!(rearm_project_path_collisions(database).await.unwrap(), 1); + assert_eq!( + parked_retry_deadlines(&harness).await, + vec![0, i64::MAX, i64::MAX, i64::MAX, i64::MAX, FAR_FUTURE_MICROS], + "only a terminal Codex project_path collision is re-armed" + ); + + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?1, last_error = ?2 + WHERE next_retry_at_micros = 0", + params![i64::MAX, errors[0]], + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + assert_eq!( + rearm_project_path_collisions(database).await.unwrap(), + 0, + "the re-arm is recorded and never repeats" + ); + assert_eq!(parked_retry_deadlines(&harness).await, parked); +} + +#[asupersync::test] +async fn project_path_rearm_commits_bounded_windows_before_recording_completion() { + let harness = RegisteredGlobalDbHarness::open("rearm-project-path-windows").await; + seed_queue(&harness, 6, 0).await; + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "DELETE FROM tracedecay_schema_version + WHERE component = 'codex-project-path-collision-rearm-v1'", + (), + ) + .await + .unwrap(); + transaction + .execute( + "UPDATE projection_queue + SET next_retry_at_micros = ?1, + last_error = 'projection session output collided at codex/s: transcript_path differs'", + params![i64::MAX], + ) + .await + .unwrap(); + transaction + .execute( + "UPDATE projection_queue + SET last_error = 'projection session output collided at codex/s: project_path differs' + WHERE rowid = (SELECT MIN(rowid) FROM projection_queue) + OR rowid = (SELECT MAX(rowid) FROM projection_queue)", + (), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let database = harness.registered.runtime_database(); + let last_rowid = query_optional_i64( + &database.read_connection(), + "SELECT MAX(rowid) FROM projection_queue", + params![], + ) + .await + .unwrap() + .unwrap(); + let (_, rearmed) = rearm_project_path_collision_batch(database, 0, last_rowid, 2) + .await + .unwrap() + .unwrap(); + assert_eq!(rearmed, 1); + assert_eq!( + parked_retry_deadlines(&harness).await, + vec![0, i64::MAX, i64::MAX, i64::MAX, i64::MAX, i64::MAX], + "a window limits rows examined, not just matching collisions" + ); + assert_eq!( + schema_version::read_version( + &database.read_connection(), + "codex-project-path-collision-rearm-v1", + ) + .await + .unwrap(), + None, + "a committed partial repair must not claim completion" + ); + + // A new walker has no in-memory cursor from the interrupted one. It + // preserves that committed first window and repairs the remaining row. + assert_eq!(rearm_project_path_collisions(database).await.unwrap(), 1); + assert_eq!( + parked_retry_deadlines(&harness).await, + vec![0, i64::MAX, i64::MAX, i64::MAX, i64::MAX, 0] + ); + assert_eq!( + schema_version::read_version( + &database.read_connection(), + "codex-project-path-collision-rearm-v1", + ) + .await + .unwrap(), + Some(1) + ); + assert_eq!(rearm_project_path_collisions(database).await.unwrap(), 0); +} diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index 773df53880..e225c63ca3 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -33,7 +33,9 @@ use super::transition::{ MessageTransition, MessageTransitionState, WorkflowFactTarget, WorkflowFactTransition, message_transition, write_workflow_fact_transition, }; -use tracedecay_session_temporal_store::record_canonical_observation_effect; +use tracedecay_session_temporal_store::{ + record_canonical_observation_effect, request_session_temporal_reset, +}; pub(super) const REBUILD_PAGE_SIZE: i64 = 128; const REBUILD_MAX_STEPS_PER_INVOCATION: usize = 4; @@ -587,6 +589,117 @@ async fn project_observation_in_transaction( .map(|(outcome, _)| outcome) } +/// The one-time project-path repair has verified the empty temporal effect and +/// retired this observation's obsolete collision disposition. Its sequence +/// may already be behind the checkpoint, so write the recovered effect without +/// rewinding the frontier or replaying unrelated observations. The ordinary +/// collision guard still parks a real conflict uncovered by the repair. +pub(super) async fn reproject_project_path_collision( + transaction: &DatabaseWriteTransaction<'_>, + sequence: u64, + observation: &DurableObservationV1, +) -> ProjectionStoreResult<()> { + let mut effect = derive_projection_with_alias(transaction, observation).await?; + Box::pin(write_effect_converging_collisions( + transaction, + &CollisionGuardedWrite::Drain, + sequence, + observation, + &mut effect, + )) + .await?; + if effect.message().is_some() { + // The bounded migration transaction temporarily lifts only this + // delete guard. A still-colliding observation keeps its original + // zero-output effect, including its original recording timestamp. + transaction + .execute( + "DELETE FROM session_temporal_observation_effects + WHERE observation_id = ?1 AND output_count = 0", + params![observation.observation_id().as_str()], + ) + .await + .map_err(|error| storage("retire obsolete empty temporal effect", error))?; + } + record_canonical_observation_effect(transaction, sequence, observation, &effect).await?; + Box::pin(restage_recovered_collision( + transaction, + sequence, + observation, + )) + .await?; + // A temporal generation may already cover the old zero-output effect. + // Rebuilding that session makes the newly recovered message visible there + // as well as in the raw message read path. + request_session_temporal_reset(transaction, observation.observation_id().as_str()).await?; + if !queue_row_is_terminal_collision(transaction, observation.observation_id()).await? { + consume_projection_queue_item(transaction, observation.observation_id()).await?; + } + Ok(()) +} + +/// A rebuild suspended after this observation staged the obsolete skip. +/// Replace that one staged effect in the same transaction as the live repair, +/// so resuming the rebuild cannot remove the recovered message again. +async fn restage_recovered_collision( + transaction: &DatabaseWriteTransaction<'_>, + sequence: u64, + observation: &DurableObservationV1, +) -> ProjectionStoreResult<()> { + let Some(job) = read_optional_rebuild_job(transaction).await? else { + return Ok(()); + }; + if sequence_i64(sequence)? > job.staged_through { + return Ok(()); + } + let removed = transaction + .execute( + "DELETE FROM observation_projection_rebuild_dispositions + WHERE projector_version = ?1 AND generation = ?2 + AND observation_id = ?3 AND reason = ?4", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + job.generation.as_str(), + observation.observation_id().as_str(), + ProjectionSkipReason::OutputCollision.as_str(), + ], + ) + .await + .map_err(|error| storage("retire staged project path collision", error))?; + if removed != 1 { + return Err(ProjectionStoreError::ProvenanceCollision); + } + let mut effect = + derive_projection_for_rebuild(transaction, observation, &job.generation).await?; + Box::pin(write_effect_converging_collisions( + transaction, + &CollisionGuardedWrite::Stage { + generation: &job.generation, + }, + sequence, + observation, + &mut effect, + )) + .await?; + record_canonical_observation_effect(transaction, sequence, observation, &effect).await?; + transaction + .execute( + "UPDATE observation_projection_rebuilds + SET projected_rows = projected_rows + ?3, + skipped_observations = skipped_observations - 1 + ?4 + WHERE projector_version = ?1 AND generation = ?2", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + job.generation.as_str(), + usize_i64(effect.output_count())?, + i64::from(matches!(effect, ObservationProjection::Skipped(_))), + ], + ) + .await + .map_err(|error| storage("update repaired projection rebuild counts", error))?; + Ok(()) +} + /// Like [`project_observation_in_transaction`], also returning the /// observation's session identity so batched drains need no follow-up point /// read per projected item. @@ -973,8 +1086,9 @@ async fn activate_projection_rebuild_transaction( .map_err(|error| storage("requeue observations past rebuild frontier", error))?; transaction .execute( - "DELETE FROM projection_queue WHERE observation_sequence <= ?1", - params![job.frontier], + "DELETE FROM projection_queue + WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2", + params![job.frontier, PROJECTION_TERMINAL_RETRY_MICROS], ) .await .map_err(|error| storage("consume rebuilt projection queue", error))?; diff --git a/crates/tracedecay-global-db/src/observation_projection/state.rs b/crates/tracedecay-global-db/src/observation_projection/state.rs index 3eb75e74ab..29336c0809 100644 --- a/crates/tracedecay-global-db/src/observation_projection/state.rs +++ b/crates/tracedecay-global-db/src/observation_projection/state.rs @@ -310,7 +310,7 @@ async fn rearm_projection_retry_batch( Ok(Some((batch_end, rows))) } -async fn query_optional_i64( +pub(super) async fn query_optional_i64( conn: &impl QueryExecutor, sql: &str, params: impl IntoParams, diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants.rs b/crates/tracedecay-global-db/src/schema_contract/invariants.rs index fcbc2b4be0..446f788be7 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants.rs @@ -25,6 +25,7 @@ mod triggers; #[cfg(test)] pub(crate) use triggers::SOURCE_CURSOR_ADVANCE_DELETE_GUARD_SQL; +pub(crate) use triggers::TEMPORAL_OBSERVATION_EFFECT_DELETE_GUARD_SQL; use audit::{ AuditCheckpoint, AuditProgress, audit_checkpoint_is_plausible, ensure_audit_checkpoint_schema, diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs index ce199cef7c..97358fa9aa 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs @@ -3,8 +3,8 @@ use std::collections::BTreeSet; use tracedecay_domain::{ DurableObservationV1, ObservationOrderingDomainV1, ObservationSourceCursorV1, }; -use tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION; use tracedecay_store::observation::ObservationCoverageV1; +use tracedecay_store::{PROJECTION_TERMINAL_RETRY_MICROS, SESSION_MESSAGE_PROJECTOR_VERSION}; use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; use tracedecay_rusqlite_runtime::repository::observation_cursor_authority::PRUNE_SUPERSEDED_CURSOR_ADVANCES_SQL; @@ -150,8 +150,9 @@ pub(super) async fn repair_projection_frontier( .await .map_err(audit_read_error)?; conn.execute( - "DELETE FROM projection_queue WHERE observation_sequence <= ?1", - params![repaired_checkpoint], + "DELETE FROM projection_queue + WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2", + params![repaired_checkpoint, PROJECTION_TERMINAL_RETRY_MICROS], ) .await .map_err(audit_read_error)?; diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs index f81dff2357..d2a6a324c4 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs @@ -482,6 +482,12 @@ const SESSION_SUMMARY_AUTHORITY_IMMUTABILITY: &[Trigger] = &[ }, ]; +pub(crate) const TEMPORAL_OBSERVATION_EFFECT_DELETE_GUARD_SQL: &str = + "CREATE TRIGGER session_temporal_observation_effects_immutable_delete_v1 + BEFORE DELETE ON session_temporal_observation_effects BEGIN + SELECT RAISE(ABORT, 'session temporal observation effects are immutable'); + END"; + const SESSION_RECEIPT_IMMUTABILITY: &[Trigger] = &[ Trigger { name: "session_temporal_projection_receipts_insert_guard_v1", @@ -538,10 +544,7 @@ const SESSION_RECEIPT_IMMUTABILITY: &[Trigger] = &[ Trigger { name: "session_temporal_observation_effects_immutable_delete_v1", table: "session_temporal_observation_effects", - create_sql: "CREATE TRIGGER session_temporal_observation_effects_immutable_delete_v1 - BEFORE DELETE ON session_temporal_observation_effects BEGIN - SELECT RAISE(ABORT, 'session temporal observation effects are immutable'); - END", + create_sql: TEMPORAL_OBSERVATION_EFFECT_DELETE_GUARD_SQL, }, Trigger { name: "session_temporal_projection_receipts_immutable_update_v1", diff --git a/crates/tracedecay-global-db/src/schema_stages.rs b/crates/tracedecay-global-db/src/schema_stages.rs index 9f39094da3..54c5d1e4dc 100644 --- a/crates/tracedecay-global-db/src/schema_stages.rs +++ b/crates/tracedecay-global-db/src/schema_stages.rs @@ -1288,6 +1288,17 @@ async fn converge_registered_schema_on( rearmed.await.map_err(|error| { global_db_operation_message("rearm queued projection retries", error.durable_detail()) })?; + let parked: Pin< + Box> + Send + '_>, + > = Box::pin(observation_projection::rearm_project_path_collisions( + database, + )); + parked.await.map_err(|error| { + global_db_operation_message( + "rearm Codex project path collisions", + error.durable_detail(), + ) + })?; let echoes: Pin< Box> + Send + '_>, > = Box::pin(observation_projection::converge_codex_response_echoes( diff --git a/crates/tracedecay-global-db/src/session_project_rebind_tests.rs b/crates/tracedecay-global-db/src/session_project_rebind_tests.rs index ddfb44fe0a..6ce5b42fca 100644 --- a/crates/tracedecay-global-db/src/session_project_rebind_tests.rs +++ b/crates/tracedecay-global-db/src/session_project_rebind_tests.rs @@ -19,12 +19,14 @@ use tracedecay_runtime_core::db::engine::params; use tracedecay_sessions::runtime::shared::durable_project_path_key; use tracedecay_store::{ AnchoredObservationWrite, ObservationPersistOutcome, ObservationProjectionStore, - ObservationStore, ObservationWrite, + ObservationStore, ObservationWrite, ProjectionPersistOutcome, ProjectionSkipReason, + SESSION_MESSAGE_PROJECTOR_VERSION, }; use tracedecay_temporal_query::execution::ExecutionControl; use crate::tests::harness::{ - HostAdmissionScope, HostAdmissionTestRuntimeV1, reopen_project_sessions_database, + HostAdmissionScope, HostAdmissionTestRuntimeV1, RegisteredGlobalDbHarness, + bind_test_session_relation_graph, reopen_project_sessions_database, }; use crate::{GlobalDbObservationStore, RegisteredGlobalDb}; @@ -668,6 +670,429 @@ fn session_fact_with_transcript(cwd: &str, transcript_path: &str) -> CanonicalOb } } +/// beta.88 (950bf21f85) parked a session collision with a durable skip, an +/// advanced checkpoint, and a zero-output temporal effect. Exercise that full +/// state through a registered daemon remount, not just a queued retry deadline. +#[asupersync::test] +async fn registered_upgrade_recovers_parked_codex_messages_once() { + check_parked_codex_upgrade(false).await; +} + +#[asupersync::test] +async fn registered_upgrade_keeps_recovered_codex_messages_after_rebuild_resume() { + check_parked_codex_upgrade(true).await; +} + +async fn check_parked_codex_upgrade(resume_rebuild: bool) { + let paths = TempDir::new().unwrap(); + let old_cwd = paths.path().join("repo"); + let new_cwd = paths.path().join("worktree"); + std::fs::create_dir_all(&old_cwd).unwrap(); + std::fs::create_dir_all(&new_cwd).unwrap(); + let old_cwd = durable_project_path_key(&old_cwd.to_string_lossy()); + let new_cwd = new_cwd.to_string_lossy().into_owned(); + let old_transcript = paths + .path() + .join("old.jsonl") + .to_string_lossy() + .into_owned(); + let new_transcript = paths + .path() + .join("new.jsonl") + .to_string_lossy() + .into_owned(); + let harness = RegisteredGlobalDbHarness::open("recover-parked-codex-messages").await; + let store = harness.registered.observation_store(); + let sessions = [ + "session.recover-path", + "session.keep-transcript", + "session.keep-parent", + "session.path-masked-transcript", + ]; + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + for (index, session) in sessions.iter().enumerate() { + transaction + .execute( + "INSERT INTO sessions ( + provider, session_id, project_key, project_path, + transcript_path, parent_session_id + ) VALUES ('codex', ?1, 'user', ?2, ?3, ?4)", + params![ + *session, + old_cwd.as_str(), + (index != 2).then_some(old_transcript.as_str()), + (index == 2).then_some("session.parent-before"), + ], + ) + .await + .unwrap(); + } + transaction.commit().await.unwrap(); + let mut parked = Vec::new(); + for (index, session) in sessions.iter().enumerate() { + let record = format!("record.parked-{index}"); + let receipt_id = format!("receipt.parked-{index}"); + let mut write = observation( + "codex", + session, + ObservationScopeV1::Profile, + &record, + ORIGINAL_TEXT, + &receipt_id, + Some(if index == 2 { + session_fact(&new_cwd) + } else { + session_fact_with_transcript(&new_cwd, &new_transcript) + }), + ); + if index == 2 { + let mut payload = write.payload().clone(); + payload["relations"]["parent_session_id"] = json!("session.parent-after"); + write = DurableObservationV1::new( + write.identity().clone(), + receipt(&receipt_id, &payload), + write.retention_class().clone(), + payload, + ) + .unwrap(); + } + persist(&store, write.clone()).await; + assert!(matches!( + store + .project_observation(write.observation_id()) + .await + .unwrap(), + ProjectionPersistOutcome::Skipped { + reason: ProjectionSkipReason::OutputCollision, + .. + } + )); + parked.push(write); + } + let parked_frontier = projection_checkpoint(&harness.registered).await; + let later = observation( + "codex", + "session.after-parked", + ObservationScopeV1::Profile, + "record.after-parked", + CURRENT_TEXT, + "receipt.after-parked", + None, + ); + persist(&store, later.clone()).await; + drain_projection_queue(&store).await; + + // The production collision writer above installed every durable side + // effect of parking. Restore only the released project-path disagreement: + // the first row has no real transcript collision, while the last also + // carries one that the old path-first check masked. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "UPDATE sessions SET transcript_path = NULL + WHERE provider = 'codex' AND session_id = ?1", + params![sessions[0]], + ) + .await + .unwrap(); + for index in [0, 3] { + transaction + .execute( + "UPDATE projection_queue SET last_error = ?2 WHERE observation_id = ?1", + params![ + parked[index].observation_id().as_str(), + format!( + "projection session output collided at codex/{}: project_path differs", + sessions[index] + ), + ], + ) + .await + .unwrap(); + } + transaction + .execute( + "DELETE FROM tracedecay_schema_version + WHERE component = 'codex-project-path-collision-rearm-v1'", + (), + ) + .await + .unwrap(); + if resume_rebuild { + // A suspended rebuild already staged these zero-output skips before + // upgrade. Its frozen frontier predates the later healthy message. + transaction + .execute( + "INSERT INTO observation_projection_rebuilds ( + projector_version, generation, frontier_sequence, + aliases_staged_through, staged_through, projected_rows, + skipped_observations, state + ) VALUES (?1, 'generation.parked-codex', ?2, ?2, ?2, 0, ?3, 'ready')", + params![ + SESSION_MESSAGE_PROJECTOR_VERSION, + parked_frontier, + i64::try_from(parked.len()).unwrap(), + ], + ) + .await + .unwrap(); + transaction + .execute( + "INSERT INTO observation_projection_rebuild_dispositions ( + projector_version, generation, observation_id, receipt_id, reason + ) SELECT projector_version, 'generation.parked-codex', + observation_id, receipt_id, reason + FROM observation_projection_dispositions + WHERE projector_version = ?1", + params![SESSION_MESSAGE_PROJECTOR_VERSION], + ) + .await + .unwrap(); + } + transaction.commit().await.unwrap(); + let checkpoint_before = projection_checkpoint(&harness.registered).await; + for (index, write) in parked.iter().enumerate() { + assert_eq!( + parked_projection_state(&harness.registered, write.observation_id().as_str()).await, + (Some(i64::MAX), Some("output_collision".to_owned()), Some(0)), + ); + assert!( + harness + .registered + .get_session_message("codex", &format!("message.record.parked-{index}")) + .await + .unwrap() + .is_none() + ); + } + let genuine_attempts = [ + queue_attempts(&harness.registered, parked[1].observation_id().as_str()).await, + queue_attempts(&harness.registered, parked[2].observation_id().as_str()).await, + ]; + if !resume_rebuild { + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute_batch( + "CREATE TRIGGER interrupt_recovered_message BEFORE INSERT ON lcm_raw_messages + WHEN NEW.message_id = 'message.record.parked-0' + BEGIN SELECT RAISE(ABORT, 'interrupted project path recovery'); END", + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let error = crate::observation_projection::rearm_project_path_collisions( + harness.registered.runtime_database(), + ) + .await + .unwrap_err(); + assert!( + error + .durable_detail() + .contains("interrupted project path recovery") + ); + assert_eq!( + parked_projection_state(&harness.registered, parked[0].observation_id().as_str()).await, + (Some(i64::MAX), Some("output_collision".to_owned()), Some(0)), + "a failed writer rolls the deadline, disposition, and temporal effect back together", + ); + assert!( + crate::schema_contract::authority_invariant_triggers_intact( + &harness.registered.read_connection() + ) + .await + .unwrap(), + "rollback restores the delete guard before another writer runs" + ); + assert_eq!( + tracedecay_runtime_core::db::schema_version::read_version( + &harness.registered.read_connection(), + "codex-project-path-collision-rearm-v1", + ) + .await + .unwrap(), + None, + ); + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute_batch("DROP TRIGGER interrupt_recovered_message") + .await + .unwrap(); + transaction.commit().await.unwrap(); + } + drop(store); + + let (harness, convergence) = harness.restart_for_daemon().await; + bind_test_session_relation_graph(&harness.registered).unwrap(); + harness + .registered + .converge_schema(convergence) + .await + .unwrap(); + let store = harness.registered.observation_store(); + if resume_rebuild { + let rebuilt = store + .rebuild_projection(u64::try_from(parked_frontier).unwrap()) + .await + .unwrap(); + assert!(rebuilt.is_complete()); + assert_eq!(rebuilt.projected_rows(), 1); + assert_eq!(rebuilt.skipped_observations(), parked.len() - 1); + } + let pass = store + .project_queued_observations(16) + .await + .unwrap() + .unwrap(); + assert!(!pass.has_more, "one pass must settle the upgraded queue"); + let recovered = harness + .registered + .get_session_message("codex", "message.record.parked-0") + .await + .unwrap() + .expect("the formerly parked message is readable after upgrade"); + assert_eq!(recovered.text, ORIGINAL_TEXT); + assert_eq!( + session_project_binding(&harness.registered, "codex", sessions[0]).await, + ("user".to_owned(), old_cwd), + "recovering a message keeps the existing concrete session directory", + ); + assert_eq!( + parked_projection_state(&harness.registered, parked[0].observation_id().as_str()).await, + (None, None, Some(1)), + ); + assert_eq!( + projection_checkpoint(&harness.registered).await, + checkpoint_before + ); + for (index, observation) in parked.iter().enumerate().skip(1) { + assert_eq!( + parked_projection_state(&harness.registered, observation.observation_id().as_str()) + .await, + (Some(i64::MAX), Some("output_collision".to_owned()), Some(0)), + ); + assert!( + harness + .registered + .get_session_message("codex", &format!("message.record.parked-{index}")) + .await + .unwrap() + .is_none() + ); + } + assert_eq!( + [ + queue_attempts(&harness.registered, parked[1].observation_id().as_str()).await, + queue_attempts(&harness.registered, parked[2].observation_id().as_str()).await, + ], + genuine_attempts, + "unrelated transcript and lineage collisions retain their attempt history", + ); + assert!( + queue_last_error(&harness.registered, parked[3].observation_id().as_str()) + .await + .unwrap() + .contains("transcript_path") + ); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut resets = snapshot + .query( + "SELECT COUNT(*) FROM session_temporal_resets WHERE session_id = ?1", + params![sessions[0]], + ) + .await + .unwrap(); + assert_eq!( + resets.next().await.unwrap().unwrap().get::(0).unwrap(), + 1, + "temporal refresh must revisit the recovered zero-output frontier" + ); + drop(resets); + drop(snapshot); + assert!( + crate::schema_contract::authority_invariant_triggers_intact( + &harness.registered.read_connection() + ) + .await + .unwrap(), + "the migration must restore every immutability guard" + ); + assert_eq!( + harness + .registered + .get_session_message("codex", "message.record.after-parked") + .await + .unwrap() + .unwrap() + .text, + CURRENT_TEXT + ); + + drop(store); + let harness = harness.restart().await; + let store = harness.registered.observation_store(); + assert!(matches!( + store + .project_observation(parked[0].observation_id()) + .await + .unwrap(), + ProjectionPersistOutcome::ExactDuplicate(_) + )); + assert_eq!( + harness + .registered + .get_session_message("codex", "message.record.parked-0") + .await + .unwrap() + .unwrap() + .text, + ORIGINAL_TEXT + ); + for observation in parked.iter().skip(1) { + assert_eq!( + parked_projection_state(&harness.registered, observation.observation_id().as_str()) + .await, + (Some(i64::MAX), Some("output_collision".to_owned()), Some(0)), + ); + } + assert_eq!( + projection_checkpoint(&harness.registered).await, + checkpoint_before + ); +} + +async fn projection_checkpoint(database: &RegisteredGlobalDb) -> i64 { + let snapshot = database.read_snapshot().await.unwrap(); + let mut rows = snapshot.query( + "SELECT last_sequence FROM observation_projection_checkpoints WHERE projector_version = ?1", + params![SESSION_MESSAGE_PROJECTOR_VERSION], + ).await.unwrap(); + rows.next().await.unwrap().unwrap().get(0).unwrap() +} + +async fn parked_projection_state( + database: &RegisteredGlobalDb, + observation_id: &str, +) -> (Option, Option, Option) { + let snapshot = database.read_snapshot().await.unwrap(); + let mut rows = snapshot.query( + "SELECT queue.next_retry_at_micros, disposition.reason, effect.output_count + FROM observations AS observation + LEFT JOIN projection_queue AS queue ON queue.observation_id = observation.observation_id + LEFT JOIN observation_projection_dispositions AS disposition + ON disposition.observation_id = observation.observation_id AND disposition.projector_version = ?2 + LEFT JOIN session_temporal_observation_effects AS effect + ON effect.observation_id = observation.observation_id + WHERE observation.observation_id = ?1", + params![observation_id, SESSION_MESSAGE_PROJECTOR_VERSION], + ).await.unwrap(); + let row = rows.next().await.unwrap().unwrap(); + ( + row.get(0).unwrap(), + row.get(1).unwrap(), + row.get(2).unwrap(), + ) +} + async fn drain_projection_queue(store: &GlobalDbObservationStore) { while let Some(observation_id) = store.next_queued_observation().await.unwrap() { store diff --git a/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs b/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs index 8bbd0efbdb..609cd0ba20 100644 --- a/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs +++ b/crates/tracedecay/src/daemon/tests/rmcp_route/host_index_handoff.rs @@ -73,9 +73,30 @@ async fn committed_save_remains_noncurrent_until_its_broker_handoff() { }; // Complete-owner demand can queue a freshness pass after text readiness. // Settle that setup work before pinning the generation the save must retain. - let initial = ready_reading(&fixture, &project, "after initial graph admission").await; + ready_reading(&fixture, &project, "after initial graph admission").await; let replay = broker.begin_replay().await.unwrap(); assert_eq!(broker.pending_replay_count().await.unwrap(), 0); + // Every source-proof renewal takes the mounted scheduler mutex. Hold it + // across the save and durable admission so an independent source observer + // cannot publish the edit before the broker records its handoff debt. + let scheduler = registry.scheduler_handle(&project).await.unwrap(); + let (release_scheduler, scheduler_released) = tokio::sync::oneshot::channel::<()>(); + let (scheduler_acquired, scheduler_held) = tokio::sync::oneshot::channel(); + let held_scheduler = tokio::task::spawn_blocking(move || { + let _scheduler = scheduler + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + scheduler_acquired.send(()).unwrap(); + let _ = scheduler_released.blocking_recv(); + }); + scheduler_held.await.unwrap(); + // Graph demand may have advanced the initial owner. Read the transition's + // identity only after that demand and the scheduler hold are established. + let initial = registry + .dashboard_freshness_read(&project) + .await + .unwrap() + .unwrap(); std::fs::write( project.join("src/main.rs"), "fn main() { let saved = 42; }\n", @@ -97,6 +118,21 @@ async fn committed_save_remains_noncurrent_until_its_broker_handoff() { .unwrap() .unwrap(); assert_eq!(owed.latest_generation_id, initial.latest_generation_id); + assert_eq!( + owed.hook_hint_count, + Some(0), + "broker debt is not a scheduler hint" + ); + drop(release_scheduler); + held_scheduler.await.unwrap(); + // Pending broker debt fences currency, not immutable publication. Read + // again after releasing and joining the holder so these assertions cannot + // pass merely because a held scheduler makes verification unknown. + let owed = registry + .dashboard_freshness_read(&project) + .await + .unwrap() + .unwrap(); assert_ne!( owed.readiness(CodeIndexReadinessTargetV1::Ready), CodeIndexReadinessV1::Reached @@ -105,11 +141,6 @@ async fn committed_save_remains_noncurrent_until_its_broker_handoff() { owed.readiness(CodeIndexReadinessTargetV1::Fresh), CodeIndexReadinessV1::Reached ); - assert_eq!( - owed.hook_hint_count, - Some(0), - "broker debt is not a scheduler hint" - ); assert!(!registry.has_current_ready_decoded_for_root_scope(&project, &scope)); assert!( registry diff --git a/crates/tracedecay/tests/mcp_suite/changelog_behavior_test.rs b/crates/tracedecay/tests/mcp_suite/changelog_behavior_test.rs index b09684cbab..dabf087ad6 100644 --- a/crates/tracedecay/tests/mcp_suite/changelog_behavior_test.rs +++ b/crates/tracedecay/tests/mcp_suite/changelog_behavior_test.rs @@ -123,6 +123,8 @@ async fn changelog_rejects_missing_and_non_object_arguments() { "invalid arguments: tracedecay_changelog expects a JSON object" ) ); + + repo.harness.shutdown().await; } fn assert_changelog_refusal(response: &JsonRpcResponse, message: &str) { @@ -170,6 +172,8 @@ async fn changelog_unknown_ref_is_a_typed_git_error() { message.starts_with("cannot resolve 'no-such-changelog-ref':"), "unresolvable ref must name itself in the git error: {message}" ); + + repo.harness.shutdown().await; } #[tokio::test] @@ -228,6 +232,8 @@ symbols_modified: none symbols_removed: none " ); + + repo.harness.shutdown().await; } #[tokio::test] @@ -253,6 +259,8 @@ async fn changelog_deleted_subtree_lists_only_the_removed_file() { json_text(result), r#"{"changed_file_count":1,"changed_files":["crates/sub/keep.rs"],"from_ref":"HEAD~1","status":"partial","symbol_changes_coverage":{"reason":"exact_local_branch_required","retryable":false,"status":"unavailable"},"symbols_added":[],"symbols_modified":[],"symbols_removed":[],"to_ref":"HEAD"}"# ); + + repo.harness.shutdown().await; } #[tokio::test] @@ -295,6 +303,8 @@ async fn changelog_same_branch_tip_reports_no_changes() { .as_str() .unwrap_or_else(|| panic!("identical tips must name the head generation: {body}")); assert_eq!(base, head, "the same tip is one generation: {body}"); + + repo.harness.shutdown().await; } #[tokio::test] @@ -377,4 +387,6 @@ async fn changelog_between_local_branches_names_added_removed_and_modified_symbo })], "{body}" ); + + repo.harness.shutdown().await; } diff --git a/crates/tracedecay/tests/mcp_suite/context_lexical_anchor_eval_test.rs b/crates/tracedecay/tests/mcp_suite/context_lexical_anchor_eval_test.rs index c505fbab5d..eaacca8aae 100644 --- a/crates/tracedecay/tests/mcp_suite/context_lexical_anchor_eval_test.rs +++ b/crates/tracedecay/tests/mcp_suite/context_lexical_anchor_eval_test.rs @@ -263,6 +263,7 @@ async fn context_lexical_anchors_admit_every_exact_import_site() { "markdown must name the unmatched anchor:\n{markdown}" ); + drop(server); production.harness.shutdown().await; } @@ -473,6 +474,7 @@ async fn context_returns_a_rare_anchor_site_over_an_exact_word_neighborhood() { "markdown must name the dropped anchor sites:\n{markdown}" ); + drop(server); production.harness.shutdown().await; } @@ -527,5 +529,6 @@ async fn an_identifier_named_in_the_task_leads_plan_and_explore_context() { ); } + drop(server); production.harness.shutdown().await; } diff --git a/crates/tracedecay/tests/mcp_suite/context_related_order_test.rs b/crates/tracedecay/tests/mcp_suite/context_related_order_test.rs index f3ce8bb78a..9e401eff38 100644 --- a/crates/tracedecay/tests/mcp_suite/context_related_order_test.rs +++ b/crates/tracedecay/tests/mcp_suite/context_related_order_test.rs @@ -80,4 +80,7 @@ async fn context_ranks_related_symbols_before_the_cut_and_reports_the_omission() json!({"state": "ran", "budget": 3, "admitted": 3, "truncated": true}), "{payload}" ); + + drop(server); + production.shutdown().await; } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/info_census_cost_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/info_census_cost_test.rs index 325aa66f86..fe0c8f1a52 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/info_census_cost_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/info_census_cost_test.rs @@ -116,6 +116,7 @@ async fn observe(write: fn(&Path)) -> Vec<(Value, u64)> { } = call(&fixture, tool, arguments).await; observed.push((payload, catalog_symbols)); } + drop(server); fixture.harness.shutdown().await; observed } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/insert_at_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/insert_at_test.rs index c234c1fdf7..51effc4055 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/insert_at_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/insert_at_test.rs @@ -65,6 +65,7 @@ impl InsertProject { } async fn close(self) { + drop(self.server); close_production_source_edit_fixture(self.fixture).await; } } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/move_symbol_behavior_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/move_symbol_behavior_test.rs index 7b04b6f12f..780eb4de1c 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/move_symbol_behavior_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/move_symbol_behavior_test.rs @@ -8,7 +8,8 @@ use std::path::Path; use serde_json::{Value, json}; use crate::support::{ - handle_real_server_tool_call_raw, init_production_source_edit_project, test_temp_dir, + close_production_source_edit_fixture, handle_real_server_tool_call_raw, + init_production_source_edit_project, test_temp_dir, }; const LIB_RS: &str = "pub mod pricing;\npub mod orders;\n"; @@ -286,6 +287,9 @@ async fn dry_run_then_apply_moves_compute_grand_total() { fs::read_to_string(project.join("src/orders.rs")).unwrap(), ORDERS_RS ); + + drop(server); + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -412,4 +416,7 @@ async fn move_symbol_refuses_unsafe_or_stale_requests() { "stale expected_state: {stale_text}" ); assert_pricing_crate_unchanged(project); + + drop(server); + close_production_source_edit_fixture(fixture).await; } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/remote_status_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/remote_status_test.rs index b5b04dea18..d17280a782 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/remote_status_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/remote_status_test.rs @@ -120,6 +120,8 @@ async fn production_daemon_reports_unconfigured_remote_plane() { let json_result = daemon_status(&mounted, json!({"format": "json"})).await; assert_eq!(status_text(&json_result), UNCONFIGURED_JSON); + + mounted.harness.shutdown().await; } #[tokio::test] diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/source_edit_rollback_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/source_edit_rollback_test.rs index d8c02d9399..df06fa1031 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/source_edit_rollback_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/source_edit_rollback_test.rs @@ -3,8 +3,9 @@ //! preimages or refuses, and exact retries replay that receipt. use crate::support::{ - ProductionSourceEditFixture, TestTempDir, handle_real_server_tool_call_raw, - init_production_source_edit_project, test_temp_dir, warm_code_index_search, + ProductionSourceEditFixture, TestTempDir, close_production_source_edit_fixture, + handle_real_server_tool_call_raw, init_production_source_edit_project, test_temp_dir, + warm_code_index_search, }; use serde_json::{Value, json}; use std::fs; @@ -311,6 +312,8 @@ async fn source_edit_rollback_restores_move_preimages_and_replays_the_receipt() "config error: source edit rollback identity does not match the completed original effect", ); assert_original_sources(&moved.project); + + close_production_source_edit_fixture(moved.fixture).await; } #[tokio::test] @@ -395,6 +398,8 @@ async fn source_edit_rollback_keeps_foreign_bytes_and_replays_the_refusal() { read_project_file(&moved.project, "src/source.rs"), source_after_move ); + + close_production_source_edit_fixture(moved.fixture).await; } #[tokio::test] @@ -463,6 +468,8 @@ async fn source_edit_rollback_refuses_an_edit_without_retained_preimages() { read_project_file(&project, "src/main.rs"), "fn new_name() {}\n" ); + + close_production_source_edit_fixture(fixture).await; } /// A rollback the daemon refused reaches the host as a tool result carrying diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/str_replace_behavior_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/str_replace_behavior_test.rs index 8d74108015..44191dccfa 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/str_replace_behavior_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/str_replace_behavior_test.rs @@ -9,8 +9,9 @@ //! an apply must present, never as an expected result. use crate::support::{ - ProductionSourceEditFixture, TestTempDir, extract_first_json_content, - handle_real_server_tool_call, init_production_source_edit_project, test_temp_dir, + ProductionSourceEditFixture, TestTempDir, close_production_source_edit_fixture, + extract_first_json_content, handle_real_server_tool_call, init_production_source_edit_project, + test_temp_dir, }; use serde_json::{Value, json}; use std::fs; @@ -209,6 +210,8 @@ async fn str_replace_writes_the_unique_span_and_reports_the_completed_edit() { "source edit completed; detailed edit output was not retained", false, ); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -251,6 +254,8 @@ async fn str_replace_dry_run_previews_the_exact_diff_without_writing() { "source edit completed; detailed edit output was not retained", false, ); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -314,6 +319,8 @@ async fn str_replace_reports_a_missing_span_and_leaves_the_file() { "source edit completed; detailed edit output was not retained", false, ); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -347,6 +354,8 @@ async fn str_replace_refuses_an_ambiguous_span_and_leaves_the_file() { answer.payload ); assert!(answer.payload.get("diff").is_none(), "{}", answer.payload); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -371,6 +380,8 @@ async fn str_replace_apply_without_preview_state_is_refused() { ) ); assert_eq!(fs::read(&file).unwrap(), initial); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -421,6 +432,8 @@ async fn str_replace_refuses_a_stale_preview_and_keeps_concurrent_bytes() { "source edit failed before the effect", true, ); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -488,6 +501,8 @@ async fn str_replace_replay_does_not_apply_the_same_span_twice() { "source edit completed; detailed edit output was not retained", false, ); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -524,6 +539,8 @@ async fn str_replace_refuses_a_path_outside_the_worktree() { "source edit failed before the effect", true, ); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -562,6 +579,8 @@ async fn str_replace_deletes_a_unique_span_when_the_replacement_is_empty() { assert_eq!(answer.payload["replaced_span"], "REMOVE_ME\n"); assert_eq!(answer.payload["message"], "replacement successful"); assert_eq!(answer.payload["replayed"], false); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -598,6 +617,8 @@ async fn str_replace_preserves_crlf_bytes_around_the_span() { assert_client_outcome(&answer, true); assert_eq!(answer.payload["message"], "replacement successful"); assert_eq!(answer.payload["replaced_span"], "12"); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -627,6 +648,8 @@ async fn str_replace_identical_replacement_previews_no_changes() { assert_eq!(answer.payload["matched_str"], "12"); assert_eq!(answer.payload["new_str"], "12"); assert_eq!(answer.payload["replaced_span"], "12"); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -654,6 +677,8 @@ async fn str_replace_empty_old_str_reports_every_match() { assert_eq!(answer.payload["matched_str"], ""); assert_eq!(answer.payload["new_str"], "x"); assert!(answer.payload.get("diff").is_none(), "{}", answer.payload); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -676,6 +701,8 @@ async fn str_replace_missing_old_str_is_a_parameter_error() { crate::support::application_surface_refusal_error("missing required parameter: old_str") ); assert_eq!(fs::read(&file).unwrap(), initial); + + close_production_source_edit_fixture(fixture).await; } #[tokio::test] @@ -704,6 +731,8 @@ async fn str_replace_refuses_project_selectors() { Some("tracedecay_str_replace") ); assert_eq!(fs::read(&file).unwrap(), initial); + + close_production_source_edit_fixture(fixture).await; } /// Another process holds the store's writer lock past the admission deadline: @@ -775,4 +804,7 @@ async fn str_replace_behind_a_held_writer_lock_reports_the_lock_deadline_detail( "the text the host reads carries the same record" ); assert_eq!(fs::read(&file).unwrap(), initial); + + drop(server); + close_production_source_edit_fixture(fixture).await; } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_activate_definition_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_activate_definition_test.rs index b0d898ae5c..cb81ec65cb 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_activate_definition_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_activate_definition_test.rs @@ -598,4 +598,7 @@ async fn activate_definition_publishes_active_revision_three() { Some(&json!(effect_id)), "{still_active}" ); + + drop(server); + production.shutdown().await; } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_list_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_list_test.rs index de7104a383..e02669a158 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_list_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/workflow_list_test.rs @@ -184,4 +184,7 @@ async fn listed_definitions_are_the_registered_rows_and_unknown_fields_are_refus let (after_refusal_result, after_refusal) = call(&server, "tracedecay_workflow_list_definitions", json!({})).await; assert_list_success(&after_refusal_result, &after_refusal, &expected); + + drop(server); + production.shutdown().await; } diff --git a/crates/tracedecay/tests/mcp_suite/mcp_server_test/protocol_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_server_test/protocol_test.rs index 64eab3e931..8fb9b49166 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_server_test/protocol_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_server_test/protocol_test.rs @@ -260,6 +260,7 @@ async fn test_tools_list() { "content": "items" }) ); + drop(server); fixture.shutdown().await; } @@ -303,6 +304,7 @@ async fn test_tools_call_search() { .iter() .any(|c| c["text"].as_str().is_some_and(|t| t.contains("helper"))); assert!(has_helper, "search results should contain 'helper'"); + drop(server); fixture.harness.shutdown().await; } @@ -346,6 +348,7 @@ async fn test_tools_call_semantic_failure_sets_is_error() { .expect("tool result text"); let payload: Value = serde_json::from_str(text).expect("tool result JSON"); assert_eq!(payload["success"], false); + drop(server); fixture.harness.shutdown().await; } @@ -388,6 +391,7 @@ async fn test_tools_call_plain_text_failure_sets_is_error() { text.contains("## error") && text.contains("**kind:** git"), "expected rendered changelog git failure, got: {text}" ); + drop(server); fixture.harness.shutdown().await; } @@ -510,6 +514,7 @@ async fn cancellable_tool_call_is_answered_after_client_half_close() { extract_tool_text(&resp["result"]).contains("helper"), "expected search results, got: {resp}" ); + drop(server); fixture.harness.shutdown().await; } @@ -572,6 +577,7 @@ async fn cancellable_tool_call_is_dropped_on_full_peer_close() { "full peer close must drop the in-flight handler, got {:?}", transport.written ); + drop(server); fixture.harness.shutdown().await; } @@ -687,6 +693,7 @@ async fn cancellable_tool_call_is_cancelled_on_peer_read_failure() { "peer-loss cancellation must not write a tools/call response, got {:?}", transport.written ); + drop(server); fixture.harness.shutdown().await; } @@ -737,6 +744,7 @@ async fn cancellable_tool_call_fails_connection_on_peer_write_failure() { err.to_string().contains("peer write half gone"), "unexpected error: {err}" ); + drop(server); fixture.harness.shutdown().await; } @@ -1174,6 +1182,8 @@ async fn test_server_stats_include_response_handle_metrics() { assert_eq!(payload["fact"]["status"]["payload_access"], "deleted"); assert_eq!(payload["commit"]["fact_id"], fact_id); } + drop(cg); + drop(server); fixture.harness.shutdown().await; } #[tokio::test]