diff --git a/crates/tracedecay-domain/src/lib.rs b/crates/tracedecay-domain/src/lib.rs index 3c73c73636..be90aade95 100644 --- a/crates/tracedecay-domain/src/lib.rs +++ b/crates/tracedecay-domain/src/lib.rs @@ -279,8 +279,9 @@ pub use observation::{ ProviderUsageCursorV1, ProviderUsageModelV1, ProviderUsageObservationV1, ProviderUsageReadV1, ProviderUsageScopeV1, SanitizationReceiptV1, SanitizerDispositionV1, SensitivityV1, classify_observation_collision, cline_native_source_successor_id, - cline_task_native_observation_id, is_canonical_payload_revision_replay, - prove_cline_native_source_transition, tool_result_output_cut_state, tool_result_visible_text, + cline_task_native_observation_id, codex_rollout_siblings, is_canonical_payload_revision_replay, + is_codex_rollout_source, prove_cline_native_source_transition, tool_result_output_cut_state, + tool_result_visible_text, }; pub use remote::{ CredentialRevocationReceiptV1, CredentialRotationReceiptV1, CurrentRemoteAuthorityStateV1, diff --git a/crates/tracedecay-domain/src/observation.rs b/crates/tracedecay-domain/src/observation.rs index 3412285e80..b2ab609d4e 100644 --- a/crates/tracedecay-domain/src/observation.rs +++ b/crates/tracedecay-domain/src/observation.rs @@ -356,6 +356,26 @@ pub fn cline_native_source_successor_id( CanonicalObservationIdV1::derive(&material).map(Some) } +/// Two observations of one Codex native record read from two live rollout +/// files of one thread: the thread rollout under the session's source and a +/// `thread/revert` rollout under its own. Codex keeps both files, so both +/// observations stay offered and share the one message the record projects. +pub fn codex_rollout_siblings(left: &DurableObservationV1, right: &DurableObservationV1) -> bool { + let (left_source, right_source) = (left.source(), right.source()); + is_codex_rollout_source(left_source) + && is_codex_rollout_source(right_source) + && left_source.session_id() == right_source.session_id() + && left_source != right_source + && left.scope() == right.scope() + && left.identity().native_record_id().is_some() + && left.identity().native_record_id() == right.identity().native_record_id() +} + +/// A keyed Codex source feeds one rollout file of a thread. +pub fn is_codex_rollout_source(source: &ObservationSourceIdentityV1) -> bool { + source.provider().as_str() == "codex" && source.explicit_source_key().is_some() +} + /// Permits only the evidenced combined-file to native-file source transition. /// Authored facts, native identity, scope and sanitization remain exact; a UI /// record may lose only the API-array length formerly added to its position. diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 57990c86a4..5bc6062563 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/crates/tracedecay-global-db/src/observation_adapter.rs @@ -12,8 +12,9 @@ use tracedecay_domain::{ ObservationSourceIdentityV1, PayloadDigestV1, PayloadReferenceV1, ProjectionGenerationId, RetrievalAnchorId, RetrievalAnchorRecord, SanitizationReceiptV1, canonical_json_bytes, canonical_json_bytes_and_sha256, canonical_sha256, classify_observation_collision, - cline_native_source_successor_id, cline_task_native_observation_id, - is_canonical_payload_revision_replay, prove_cline_native_source_transition, sha256_hex_suffix, + cline_native_source_successor_id, cline_task_native_observation_id, codex_rollout_siblings, + is_canonical_payload_revision_replay, is_codex_rollout_source, + prove_cline_native_source_transition, sha256_hex_suffix, }; use tracedecay_store::observation::{ CursorAdvanceOutcome, ObservationCoverageReason, ObservationCursorAdvance, @@ -574,7 +575,9 @@ impl GlobalDbObservationStore { Box::new(write), )); } - if !preflight.accepts_pending_cline_alias(&write, &existing.anchor_id)? { + if !preflight.accepts_pending_cline_alias(&write, &existing.anchor_id)? + && !preflight.accepts_codex_rollout_sibling_alias(&write, &existing.anchor_id) + { return Err(ObservationStoreError::RetrievalAnchorAliasCollision { alias: Box::new(alias.clone()), existing_anchor_id: Box::new(existing.anchor_id), @@ -742,6 +745,7 @@ struct ObservationPreflightSnapshot { admission_refusals: HashMap<(String, String), PayloadDigestV1>, stored_observations: HashMap, retrieval_aliases: HashMap<(String, String, String), RetrievalAnchorId>, + alias_owners: HashMap, cline_supersessions: HashMap, source_cursors: HashMap<(ObservationSourceIdentityV1, ObservationScopeV1), ObservationSourceCursorV1>, @@ -774,6 +778,23 @@ impl ObservationPreflightSnapshot { self.stored_observations.get(observation_id.as_str()) } + fn accepts_codex_rollout_sibling_alias( + &self, + write: &AnchoredObservationWrite, + existing_anchor: &RetrievalAnchorId, + ) -> bool { + let Some(owner) = self + .alias_owners + .get(existing_anchor) + .and_then(|owner_id| self.stored_observation(owner_id)) + else { + return false; + }; + owner.retrieval_anchor_id() == existing_anchor + && anchors_share_access(owner.retrieval_anchor(), write.retrieval_anchor()) + && codex_rollout_siblings(owner.observation(), write.observation()) + } + fn accepts_pending_cline_alias( &self, write: &AnchoredObservationWrite, @@ -805,19 +826,8 @@ impl ObservationPreflightSnapshot { return Ok(false); }; let prior_anchor = predecessor.retrieval_anchor(); - let next_anchor = write.retrieval_anchor(); - let prior_auth = prior_anchor.authorization(); - let next_auth = next_anchor.authorization(); Ok(prior_anchor.anchor_id() == existing_anchor - && prior_anchor.owner() == next_anchor.owner() - && prior_anchor.aliases() == next_anchor.aliases() - && prior_anchor.payload_access() == next_anchor.payload_access() - && prior_anchor.retention_class() == next_anchor.retention_class() - && prior_anchor.durability() == next_anchor.durability() - && prior_auth.resolved_scope_id == next_auth.resolved_scope_id - && prior_auth.privacy_domain_id == next_auth.privacy_domain_id - && prior_auth.access_policy_digest == next_auth.access_policy_digest - && prior_auth.capability_id == next_auth.capability_id + && anchors_share_access(prior_anchor, write.retrieval_anchor()) && prove_cline_native_source_transition(predecessor.observation(), write.observation()) .is_some()) } @@ -943,6 +953,7 @@ async fn load_observation_preflight( let admission_refusals = read_admission_refusals_from_snapshot(&snapshot, writes, OPERATION).await?; let mut seen_observation_ids = HashSet::new(); + let mut refused_observation_ids = HashSet::new(); let mut observation_ids = Vec::new(); for write in writes { let observation = write.observation(); @@ -951,9 +962,9 @@ async fn load_observation_preflight( observation_id.to_owned(), observation.payload_reference().digest().as_str().to_owned(), ); - if !admission_refusals.contains_key(&refusal_key) - && seen_observation_ids.insert(observation_id.to_owned()) - { + if admission_refusals.contains_key(&refusal_key) { + refused_observation_ids.insert(observation_id.to_owned()); + } else if seen_observation_ids.insert(observation_id.to_owned()) { observation_ids.push(observation.observation_id().clone()); } for counterpart in [ @@ -972,10 +983,32 @@ async fn load_observation_preflight( } } } + let (retrieval_aliases, alias_owners) = + read_retrieval_aliases_from_snapshot(&snapshot, writes, OPERATION).await?; + let mut sibling_candidate_anchors = HashSet::new(); + for write in writes { + let observation = write.observation(); + if !is_codex_rollout_source(observation.source()) { + continue; + } + for alias in write.retrieval_anchor().aliases() { + if let Some(anchor_id) = + retrieval_aliases.get(&retrieval_alias_key(observation.scope(), alias)?) + { + sibling_candidate_anchors.insert(anchor_id.clone()); + } + } + } + for (anchor_id, owner_id) in &alias_owners { + if sibling_candidate_anchors.contains(anchor_id) + && !refused_observation_ids.contains(owner_id.as_str()) + && seen_observation_ids.insert(owner_id.as_str().to_owned()) + { + observation_ids.push(owner_id.clone()); + } + } let stored_observations = read_stored_observations_from_snapshot(&snapshot, &observation_ids, OPERATION).await?; - let retrieval_aliases = - read_retrieval_aliases_from_snapshot(&snapshot, writes, OPERATION).await?; let source_cursors = read_source_cursors_from_snapshot(&snapshot, writes, OPERATION).await?; let cline_supersessions = read_cline_supersessions_from_snapshot(&snapshot, &stored_observations, writes, OPERATION) @@ -988,6 +1021,7 @@ async fn load_observation_preflight( admission_refusals, stored_observations, retrieval_aliases, + alias_owners, cline_supersessions, source_cursors, }) @@ -1190,6 +1224,20 @@ async fn read_source_cursors_from_snapshot( Ok(cursors) } +fn anchors_share_access(prior: &RetrievalAnchorRecord, next: &RetrievalAnchorRecord) -> bool { + let prior_auth = prior.authorization(); + let next_auth = next.authorization(); + prior.owner() == next.owner() + && prior.aliases() == next.aliases() + && prior.payload_access() == next.payload_access() + && prior.retention_class() == next.retention_class() + && prior.durability() == next.durability() + && prior_auth.resolved_scope_id == next_auth.resolved_scope_id + && prior_auth.privacy_domain_id == next_auth.privacy_domain_id + && prior_auth.access_policy_digest == next_auth.access_policy_digest + && prior_auth.capability_id == next_auth.capability_id +} + fn retrieval_alias_key( scope: &ObservationScopeV1, alias: &tracedecay_domain::NativeAlias, @@ -1208,7 +1256,10 @@ async fn read_retrieval_aliases_from_snapshot( snapshot: &DatabaseEngineReadSnapshot, writes: &[AnchoredObservationWrite], operation: &'static str, -) -> ObservationStoreResult> { +) -> ObservationStoreResult<( + HashMap<(String, String, String), RetrievalAnchorId>, + HashMap, +)> { let mut requested = HashSet::new(); for write in writes { for alias in write.retrieval_anchor().aliases() { @@ -1216,7 +1267,7 @@ async fn read_retrieval_aliases_from_snapshot( } } if requested.is_empty() { - return Ok(HashMap::new()); + return Ok((HashMap::new(), HashMap::new())); } let requested = requested .into_iter() @@ -1234,8 +1285,10 @@ async fn read_retrieval_aliases_from_snapshot( let mut rows = snapshot .query( "SELECT alias.owner_json, alias.alias_kind, alias.locator_digest, - alias.anchor_id + alias.anchor_id, binding.observation_id FROM retrieval_anchor_aliases AS alias + LEFT JOIN observation_retrieval_anchors AS binding + ON binding.anchor_id = alias.anchor_id JOIN json_each(?1) AS requested ON alias.owner_json = json_extract(requested.value, '$.scope_json') @@ -1248,6 +1301,7 @@ async fn read_retrieval_aliases_from_snapshot( .await .map_err(|error| runtime_storage_error(operation, error))?; let mut aliases = HashMap::new(); + let mut owners = HashMap::new(); while let Some(row) = rows .next() .await @@ -1266,9 +1320,19 @@ async fn read_retrieval_aliases_from_snapshot( .map_err(|error| runtime_storage_error(operation, error))?, ) .map_err(|error| runtime_storage_error(operation, error))?; + if let Some(owner_id) = row + .get::>(4) + .map_err(|error| runtime_storage_error(operation, error))? + { + owners.insert( + anchor_id.clone(), + CanonicalObservationIdV1::new(owner_id) + .map_err(|error| runtime_storage_error(operation, error))?, + ); + } aliases.insert(key, anchor_id); } - Ok(aliases) + Ok((aliases, owners)) } fn observation_batch_row_projection() -> String { diff --git a/crates/tracedecay-global-db/src/observation_batch_tests.rs b/crates/tracedecay-global-db/src/observation_batch_tests.rs index 1f04efa226..5898e9ff89 100644 --- a/crates/tracedecay-global-db/src/observation_batch_tests.rs +++ b/crates/tracedecay-global-db/src/observation_batch_tests.rs @@ -214,6 +214,14 @@ fn sequential_observation( fn anchored_write( observation: DurableObservationV1, expected_cursor: Option, +) -> AnchoredObservationWrite { + anchored_write_under(observation, expected_cursor, "observation-batch-test") +} + +fn anchored_write_under( + observation: DurableObservationV1, + expected_cursor: Option, + authority: &str, ) -> AnchoredObservationWrite { let next_cursor = ObservationSourceCursorV1::for_ordering( observation.source().clone(), @@ -228,7 +236,7 @@ fn anchored_write( ProjectionGenerationId::new("projection.observation-batch.v1").unwrap(); let authorization = tracedecay_store::build_observation_resolution_authorization_v1( write.observation(), - "observation-batch-test", + authority, ) .unwrap(); let anchor = tracedecay_store::build_observation_retrieval_anchor( @@ -627,6 +635,208 @@ async fn intra_batch_alias_collision_settles_like_sequential_writes() { assert_eq!(effect.retained, vec![true, false]); } +#[derive(Clone, Copy)] +struct SharedRecordSource<'a> { + provider: &'a str, + source_key: Option<&'a str>, + retention_class: &'a str, + authority: &'a str, +} + +const THREAD_ROLLOUT: SharedRecordSource<'static> = SharedRecordSource { + provider: "codex", + source_key: Some("rollout.thread"), + retention_class: "retention.observation-batch", + authority: "observation-batch-test", +}; + +const REVERT_ROLLOUT: SharedRecordSource<'static> = SharedRecordSource { + source_key: Some("rollout.revert"), + ..THREAD_ROLLOUT +}; + +impl SharedRecordSource<'_> { + fn write(self) -> AnchoredObservationWrite { + let session_id = SessionId::new("session.observation-batch.shared-record").unwrap(); + let provider = ProviderId::new(self.provider).unwrap(); + let source = match self.source_key { + Some(key) => ObservationSourceIdentityV1::for_provider_source( + provider.clone(), + session_id.clone(), + SessionId::new(key).unwrap(), + ), + None => ObservationSourceIdentityV1::for_provider(provider.clone(), session_id.clone()), + } + .unwrap(); + let range = ObservationSourceRangeV1::new(0, 1).unwrap(); + let record = ObservationId::new("record.batch.shared").unwrap(); + let relations = CanonicalObservationRelationsV1::new(session_id) + .with_message_id(ObservationId::new("message.batch.shared").unwrap()); + let envelope = CanonicalObservationEnvelopeV1::new( + provider, + "message", + record.clone(), + relations, + vec![CanonicalObservationFactV1::Message { + role: CanonicalMessageRoleV1::Assistant, + content: json!({"text": "shared record"}), + model: None, + timestamp: Some(1_750_000_000), + }], + CanonicalObservationEvidenceV1::new(ObservationOrderingDomainV1::FileBytes, range), + ) + .unwrap(); + let payload = serde_json::to_value(envelope).unwrap(); + let identity = ObservationIdentityMaterialV1::for_native_record( + source, + ObservationScopeV1::Profile, + ObservationSourceGenerationV1::new(1).unwrap(), + range, + ObservationOrderingDomainV1::FileBytes, + record, + ) + .unwrap(); + let receipt_id = format!( + "receipt.batch.shared.{}.{}", + self.provider, + self.source_key.unwrap_or("session") + ); + let observation = DurableObservationV1::new( + identity, + fixture_receipt(&receipt_id, &payload), + RetentionClass::new(self.retention_class).unwrap(), + payload, + ) + .unwrap(); + anchored_write_under(observation, None, self.authority) + } +} + +async fn assert_shared_alias_is_refused( + first: SharedRecordSource<'_>, + second: SharedRecordSource<'_>, +) { + let (batch, effect) = + assert_batch_settles_like_sequential_writes(vec![first.write(), second.write()]).await; + + assert!( + matches!( + batch, + Err(ObservationStoreError::RetrievalAnchorAliasCollision { .. }) + ), + "expected an alias collision, got {batch:?}" + ); + assert_eq!(effect.retained, vec![true, false]); +} + +#[asupersync::test] +async fn codex_revert_rollout_shares_the_thread_rollout_alias() { + let (batch, effect) = assert_batch_settles_like_sequential_writes(vec![ + THREAD_ROLLOUT.write(), + REVERT_ROLLOUT.write(), + ]) + .await; + + let outcomes = batch.expect("both rollouts of one thread commit"); + assert!( + outcomes + .iter() + .all(|outcome| matches!(outcome.outcome(), ObservationPersistOutcome::Committed(_))) + ); + assert_eq!(effect.retained, vec![true, true]); +} + +#[asupersync::test] +async fn keyed_sources_of_another_provider_do_not_share_an_alias() { + let thread = SharedRecordSource { + provider: BATCH_PROVIDER, + ..THREAD_ROLLOUT + }; + let revert = SharedRecordSource { + provider: BATCH_PROVIDER, + ..REVERT_ROLLOUT + }; + + assert_shared_alias_is_refused(thread, revert).await; +} + +#[asupersync::test] +async fn unkeyed_codex_source_does_not_share_an_alias() { + let unkeyed = SharedRecordSource { + source_key: None, + ..THREAD_ROLLOUT + }; + + assert_shared_alias_is_refused(THREAD_ROLLOUT, unkeyed).await; + assert_shared_alias_is_refused(unkeyed, REVERT_ROLLOUT).await; +} + +#[asupersync::test] +async fn codex_rollouts_with_different_retention_do_not_share_an_alias() { + let revert = SharedRecordSource { + retention_class: "retention.observation-batch.other", + ..REVERT_ROLLOUT + }; + + assert_shared_alias_is_refused(THREAD_ROLLOUT, revert).await; +} + +#[asupersync::test] +async fn codex_rollouts_under_different_authorities_do_not_share_an_alias() { + let revert = SharedRecordSource { + authority: "observation-batch-other", + ..REVERT_ROLLOUT + }; + let thread_auth = THREAD_ROLLOUT + .write() + .retrieval_anchor() + .authorization() + .clone(); + let revert_auth = revert.write().retrieval_anchor().authorization().clone(); + assert_ne!(thread_auth.resolved_scope_id, revert_auth.resolved_scope_id); + assert_ne!( + thread_auth.access_policy_digest, + revert_auth.access_policy_digest + ); + + assert_shared_alias_is_refused(THREAD_ROLLOUT, revert).await; +} + +#[asupersync::test] +async fn non_codex_alias_collision_stays_typed_over_a_corrupt_alias_owner() { + let (_tmp, runtime, store) = open_profile_store().await; + let owner = SharedRecordSource { + provider: BATCH_PROVIDER, + source_key: None, + ..THREAD_ROLLOUT + } + .write(); + let claimant = SharedRecordSource { + provider: BATCH_PROVIDER, + ..REVERT_ROLLOUT + } + .write(); + assert!(matches!( + store.persist_observation(owner.clone()).await.unwrap(), + ObservationPersistOutcome::Committed(_) + )); + crate::observation_collision_tests::corrupt_stored_observation_row( + &runtime, + owner.observation().observation_id().as_str(), + ) + .await; + + let error = store.persist_observation(claimant).await.unwrap_err(); + + assert!( + matches!( + error, + ObservationStoreError::RetrievalAnchorAliasCollision { .. } + ), + "a non-codex write must not decode the alias owner, got {error:?}" + ); +} + #[asupersync::test] async fn persist_observations_dispatches_one_runtime_command_independent_of_batch_size() { let mut expected_snapshot_probes = None; diff --git a/crates/tracedecay-global-db/src/observation_collision_tests.rs b/crates/tracedecay-global-db/src/observation_collision_tests.rs index eecd1bcf8d..b248d6e83e 100644 --- a/crates/tracedecay-global-db/src/observation_collision_tests.rs +++ b/crates/tracedecay-global-db/src/observation_collision_tests.rs @@ -205,7 +205,7 @@ async fn overwrite_stored_observation_row( /// Original stored-row authority bytes captured before the tripwire arms, so /// restart-bearing tests can restore the row before remount, mount-time /// invariant convergence legitimately decodes committed observation rows. -struct StoredRowBytes { +pub(super) struct StoredRowBytes { payload_digest: String, observation_json: String, committed_cursor_json: String, @@ -223,7 +223,7 @@ struct StoredRowBytes { /// contract into a behavioral proof: if a regression reintroduces stored-row /// decode, identity re-derivation, or payload re-hashing on the fast path, /// the corrupted bytes make it fail loudly instead of passing silently. -async fn corrupt_stored_observation_row( +pub(super) async fn corrupt_stored_observation_row( runtime: &HostAdmissionTestRuntimeV1, observation_id: &str, ) -> StoredRowBytes { diff --git a/crates/tracedecay-global-db/src/observation_projection/state.rs b/crates/tracedecay-global-db/src/observation_projection/state.rs index 720fe3b53a..3eb75e74ab 100644 --- a/crates/tracedecay-global-db/src/observation_projection/state.rs +++ b/crates/tracedecay-global-db/src/observation_projection/state.rs @@ -720,6 +720,7 @@ pub(super) fn same_projection_lineage( ) -> bool { (candidate.source() == owner.source() && candidate.scope() == owner.scope()) || tracedecay_domain::prove_cline_native_source_transition(owner, candidate).is_some() + || tracedecay_domain::codex_rollout_siblings(owner, candidate) } pub(super) async fn verify_output_state( diff --git a/crates/tracedecay-global-db/src/observation_projection/transition.rs b/crates/tracedecay-global-db/src/observation_projection/transition.rs index 6e1189a3ab..2758b76628 100644 --- a/crates/tracedecay-global-db/src/observation_projection/transition.rs +++ b/crates/tracedecay-global-db/src/observation_projection/transition.rs @@ -66,7 +66,8 @@ impl MessageTransitionState { latest_observation, observation, ) - .is_none(), + .is_none() + && !tracedecay_domain::codex_rollout_siblings(latest_observation, observation), projector_owned, } } @@ -358,6 +359,73 @@ mod tests { )); } + fn codex_rollout_observation(source_key: &str) -> DurableObservationV1 { + use tracedecay_domain::{ + ComponentVersion, ObservationId, ObservationIdentityMaterialV1, + ObservationOrderingDomainV1, ObservationScopeV1, ObservationSourceGenerationV1, + ObservationSourceIdentityV1, ObservationSourceRangeV1, PayloadReferenceV1, ProviderId, + RetentionClass, SanitizationReceiptId, SanitizationReceiptRefV1, SanitizationReceiptV1, + SanitizerDispositionV1, SensitivityV1, SessionId, + }; + + let source = ObservationSourceIdentityV1::for_provider_source( + ProviderId::new("codex").unwrap(), + SessionId::new("codex-thread").unwrap(), + SessionId::new(source_key).unwrap(), + ) + .unwrap(); + let identity = ObservationIdentityMaterialV1::for_native_record( + source, + ObservationScopeV1::Profile, + ObservationSourceGenerationV1::new(7).unwrap(), + ObservationSourceRangeV1::new(0, 1).unwrap(), + ObservationOrderingDomainV1::SnapshotOrder, + ObservationId::new("record-1").unwrap(), + ) + .unwrap(); + let payload = serde_json::json!({"text": "shared"}); + let receipt = SanitizationReceiptV1::new( + SanitizationReceiptRefV1::new( + SanitizationReceiptId::new(format!("receipt-{source_key}")).unwrap(), + ComponentVersion::new("sanitizer.transition-test.v1").unwrap(), + ) + .unwrap(), + SanitizerDispositionV1::Accepted, + SensitivityV1::NonSensitive, + Some(PayloadReferenceV1::for_payload(&payload).unwrap()), + ) + .unwrap(); + DurableObservationV1::new( + identity, + receipt, + RetentionClass::new("retention.transition-test").unwrap(), + payload, + ) + .unwrap() + } + + #[test] + fn codex_rollout_siblings_supersede_within_one_generation() { + let thread = codex_rollout_observation("sha256:thread"); + let revert = codex_rollout_observation("sha256:revert"); + let expected = message("expected"); + + let sibling = MessageTransitionState::new(&revert, &thread, 1, true); + assert!(sibling.same_lineage); + assert!(!sibling.same_generation); + assert_eq!( + classify_message_transition(2, &expected, Some(false), Some(sibling)).unwrap(), + MessageTransition::Supersede + ); + + let same_source = MessageTransitionState::new(&revert, &revert, 1, true); + assert!(same_source.same_generation); + assert!(matches!( + classify_message_transition(2, &expected, Some(false), Some(same_source)), + Err(ProjectionStoreError::OutputCollision { .. }) + )); + } + #[test] fn rollover_only_supersedes_exclusively_owned_rows() { let expected = message("expected"); diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs b/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs index 086eed7356..8320b5b4a6 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/observation/authority.rs @@ -9,7 +9,7 @@ use tracedecay_domain::{ AnchorSourceGeneration, DurableObservationV1, EvidenceAvailabilityV1, FactOwnerV1, GenerationBoundRepositoryProvenanceV1, ObservationSourceCursorV1, RepositoryProvenanceV1, RetrievalAnchorRecord, RetrievalAnchorRecordParts, RetrievalAnchorTarget, - prove_cline_native_source_transition, + codex_rollout_siblings, prove_cline_native_source_transition, }; use tracedecay_store::{ AnchorDispositionReasonClassV1, AnchorDispositionStateV1, AnchoredObservationWrite, @@ -180,7 +180,7 @@ fn verify_retrieval_anchor( owned_aliases += 1; } else if !matches!( stored_anchor_id.as_deref(), - Some(current) if cline_alias_transition_is_valid(connection, anchor, current)? + Some(current) if shared_alias_is_valid(connection, anchor, current)? ) { return Err(invalid("retrieval anchor alias collision")); } @@ -197,10 +197,7 @@ fn verify_retrieval_anchor( Ok(()) } -// Alias promotion belongs to the projector transaction: a newly captured -// successor may wait behind the still-readable predecessor. Once promoted, -// immutable historical anchor replay requires the exact supersession receipt. -fn cline_alias_transition_is_valid( +fn shared_alias_is_valid( connection: &rusqlite::Connection, anchor: &RetrievalAnchorRecord, current_anchor_id: &str, @@ -258,6 +255,9 @@ fn cline_alias_transition_is_valid( super::decode_stored_observation(connection, anchor_json.as_str())?; let current_observation: DurableObservationV1 = super::decode_stored_observation(connection, current_json.as_str())?; + if codex_rollout_siblings(¤t_observation, &candidate_observation) { + return Ok(true); + } if prove_cline_native_source_transition(¤t_observation, &candidate_observation).is_some() { // Pending successor: leave current alias and predecessor availability diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs index e32f0c0515..cae50b9f87 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex.rs @@ -1080,7 +1080,7 @@ impl CodexRetainedScan { current: None, }, directories: Vec::new(), - epoch: CodexCorpusEpoch::initial(), + epoch: CodexCorpusEpoch::under_rollout_source_reading(), complete: true, skipped_oversized_entries: 0, files_considered: 0, @@ -1825,6 +1825,9 @@ impl CodexSource { } } +/// Bump when a rollout file's observation source mapping changes; every +/// completion claim recorded under the old reading then misses once. +const CODEX_ROLLOUT_SOURCE_READING: &[u8] = b"tracedecay.codex.rollout-source-reading.v2\0"; const CODEX_FRONTIER_SWEEPING_V2: u64 = 2; const CODEX_FRONTIER_COMPLETE_V2: u64 = 3; @@ -1950,6 +1953,12 @@ impl CodexCorpusEpoch { self.high == 0 && self.low == 0 && self.files == 0 } + fn under_rollout_source_reading() -> Self { + let mut epoch = Self::initial(); + epoch.fold(Sha256::digest(CODEX_ROLLOUT_SOURCE_READING).into()); + epoch + } + const fn from_parse_offset(offset: ParseOffset) -> Self { Self { high: offset.byte_offset, @@ -1966,13 +1975,17 @@ impl CodexCorpusEpoch { } } - fn observe(&mut self, digest: [u8; 32]) -> TranscriptIngestResult<()> { + fn fold(&mut self, digest: [u8; 32]) { let mut high = [0u8; 8]; let mut low = [0u8; 8]; high.copy_from_slice(&digest[..8]); low.copy_from_slice(&digest[8..16]); self.high ^= u64::from_be_bytes(high); self.low ^= u64::from_be_bytes(low); + } + + fn observe(&mut self, digest: [u8; 32]) -> TranscriptIngestResult<()> { + self.fold(digest); self.files = self.files.checked_add(1).ok_or( TranscriptIngestError::InvalidCodexDiscoveryFrontier { detail: "Codex corpus file count overflowed", diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs index ad1203baee..bcbe3fc504 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs @@ -4,6 +4,7 @@ use std::collections::{HashMap, VecDeque}; use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex, MutexGuard, OnceLock, PoisonError, Weak}; +use sha2::{Digest as _, Sha256}; use tokio::sync::Notify; use tokio::sync::futures::OwnedNotified; @@ -40,6 +41,8 @@ use tracedecay_runtime_core::resident_memory::ProcessSharedMemoryReservationV1; mod meta_cache_tests; #[cfg(test)] mod retired_source_tests; +#[cfg(test)] +mod rollout_source_tests; const CODEX_OBSERVATION_RETENTION: &str = "retention.provider-observation"; pub const CODEX_HOOK_MAX_NEW_BYTES: u64 = crate::runtime::source::MAX_JSONL_RECORD_BYTES as u64; @@ -403,7 +406,7 @@ impl CodexObservationAdmission<'_> { &'b self, session_cwd: &'b Path, record_cwd: &'b Path, - transcript_path: &'b Path, + transcript_path: Option<&'b Path>, model: Option<&'b str>, ) -> CodexObservationContext<'b> { let project_path = match self { @@ -413,7 +416,7 @@ impl CodexObservationAdmission<'_> { CodexObservationContext { project_path: Some(project_path), location_path: Some(record_cwd), - transcript_path: Some(transcript_path), + transcript_path, model, } } @@ -432,12 +435,11 @@ struct CodexAdmissionContext<'a> { admission: &'a dyn HostAdmission, meta: &'a super::meta::CodexMeta, native_thread_id: Option<&'a str>, + rollout: CodexRolloutFile<'a>, cancellation: &'a ObservationCancellation, } fn replay_identity(session_id: &str, domain: &[u8]) -> String { - use sha2::{Digest as _, Sha256}; - let mut hasher = Sha256::new(); hasher.update(domain); hasher.update([0]); @@ -458,6 +460,55 @@ pub fn codex_observation_source_v2( )?) } +/// Codex `thread/revert` keeps the thread id and writes a new rollout named +/// `rollout--_.jsonl` and leaves the original +/// `rollout--.jsonl` in place. +#[derive(Clone, Copy)] +enum CodexRolloutFile<'a> { + Thread, + Revert { rollout_id: &'a str }, +} + +impl<'a> CodexRolloutFile<'a> { + fn of(rollout: &'a Path, session_id: &str) -> Self { + rollout + .file_name() + .and_then(|name| name.to_str()) + .and_then(|name| name.strip_prefix("rollout-")) + .and_then(|name| name.strip_suffix(".jsonl")) + .filter(|stem| stem.get(19..20) == Some("-")) + .and_then(|stem| stem.get(20..)) + .and_then(|ids| ids.strip_prefix(session_id)) + .and_then(|ids| ids.strip_prefix('_')) + .filter(|rollout_id| !rollout_id.is_empty() && *rollout_id != session_id) + .map_or(Self::Thread, |rollout_id| Self::Revert { rollout_id }) + } + + fn source(self, session_id: &str) -> TranscriptIngestResult { + let Self::Revert { rollout_id } = self else { + return codex_observation_source_v2(session_id); + }; + let mut hasher = Sha256::new(); + hasher.update(b"tracedecay.codex.observation-source.revert-rollout.v1"); + hasher.update([0]); + hasher.update(session_id.as_bytes()); + hasher.update([0]); + hasher.update(rollout_id.as_bytes()); + Ok(ObservationSourceIdentityV1::for_provider_source( + ProviderId::new(PROVIDER)?, + SessionId::new(session_id.to_string())?, + SessionId::new(format!( + "sha256:{}", + encode_lowercase_hex(&hasher.finalize()) + ))?, + )?) + } + + fn session_transcript(self, rollout: &Path) -> Option<&Path> { + matches!(self, Self::Thread).then_some(rollout) + } +} + async fn shared_session_meta_with_provenance( path: &Path, cancellation: &ObservationCancellation, @@ -572,13 +623,15 @@ async fn try_admit_codex_jsonl_observations( if !admission_scope.accepts_session(&meta.session_id) { return Ok(CodexJsonlAdmissionProgress::default()); } - let canonical_source = codex_observation_source_v2(&meta.session_id)?; + let rollout = CodexRolloutFile::of(path, &meta.session_id); + let canonical_source = rollout.source(&meta.session_id)?; let context = CodexAdmissionContext { path, scope: &admission_scope, admission, meta: &meta, native_thread_id: native_thread_id.as_deref(), + rollout, cancellation, }; let scope = admission_scope.scope(); @@ -602,6 +655,7 @@ async fn admit_codex_jsonl_page( admission, meta, native_thread_id, + rollout, cancellation, } = context; let scope = admission_scope.scope(); @@ -693,7 +747,7 @@ async fn admit_codex_jsonl_page( admission_scope.observation_context( meta.cwd.as_path(), state.context.cwd.as_path(), - path, + rollout.session_transcript(path), state.context.model.as_deref(), ), )?; diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/rollout_source_tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/rollout_source_tests.rs new file mode 100644 index 0000000000..615c35e9ad --- /dev/null +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/rollout_source_tests.rs @@ -0,0 +1,49 @@ +use super::*; + +const SESSION_ID: &str = "codex-pinned-session"; +const SHIPPED_V2_SOURCE: &str = r#"{"provider":"codex","session_id":"codex-pinned-session","source_key":"sha256:1a03b304c22f4282bb1ce885d69e3c8f2c0e53d1a81320ca4b8df677234d216b"}"#; + +fn source(rollout: &str) -> ObservationSourceIdentityV1 { + CodexRolloutFile::of(Path::new(rollout), SESSION_ID) + .source(SESSION_ID) + .unwrap() +} + +#[test] +fn single_rollout_sessions_keep_the_shipped_v2_source_bytes() { + for rollout in [ + "/home/.codex/sessions/2026/01/01/rollout-2026-01-01T00-00-00-codex-pinned-session.jsonl", + "/home/.codex/archived_sessions/rollout-2026-01-01T00-00-00-codex-pinned-session.jsonl", + "/tmp/rollout.jsonl", + "/tmp/rollout-2026-01-01T00-00-00-another-thread_codex-pinned-session.jsonl", + "/tmp/rollout-2026-01-01T00-00-00-codex-pinned-session_codex-pinned-session.jsonl", + "/tmp/rollout-2026-01-01T00-00-00-codex-pinned-session_.jsonl", + ] { + assert_eq!( + serde_json::to_string(&source(rollout)).unwrap(), + SHIPPED_V2_SOURCE, + "{rollout}" + ); + } + assert_eq!( + serde_json::to_string(&codex_observation_source_v2(SESSION_ID).unwrap()).unwrap(), + SHIPPED_V2_SOURCE + ); +} + +#[test] +fn each_revert_rollout_of_a_thread_is_its_own_source() { + let original = source("/s/rollout-2026-01-01T00-00-00-codex-pinned-session.jsonl"); + let first_revert = + source("/s/rollout-2026-01-02T00-00-00-codex-pinned-session_rollout-a.jsonl"); + let second_revert = + source("/s/rollout-2026-01-03T00-00-00-codex-pinned-session_rollout-b.jsonl"); + let archived_first_revert = + source("/a/rollout-2026-01-02T00-00-00-codex-pinned-session_rollout-a.jsonl"); + + assert_ne!(first_revert, original); + assert_ne!(second_revert, original); + assert_ne!(first_revert, second_revert); + assert_eq!(first_revert, archived_first_revert); + assert_eq!(first_revert.session_id().as_str(), SESSION_ID); +} diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs index e00702e4ef..b453fedc65 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/tests.rs @@ -2809,6 +2809,59 @@ mod recent_first_discovery_tests { )); } + #[test] + fn a_completion_claim_from_the_shipped_reading_offers_every_rollout_again() { + let temp = TempDir::new().unwrap(); + let home = temp.path(); + let sorted = |mut paths: Vec| { + paths.sort(); + paths + }; + let rollouts = sorted(vec![ + write_dated_rollout(home, ("2026", "01", "01"), "2026-01-01T00-00-00-thread"), + write_dated_rollout( + home, + ("2026", "01", "02"), + "2026-01-02T00-00-00-thread_revert", + ), + ]); + let source = CodexSource::with_home(home); + let bounds = TranscriptDiscoveryBounds::from_discovered_units(8); + + let first = source + .discover_transcript_paths_with_frontier(bounds, CodexDiscoveryFrontier::initial()) + .unwrap(); + assert_eq!(sorted(first.report.paths.clone()), rollouts); + assert!(first.next_frontier.is_complete()); + + let mut shipped = CodexCorpusEpoch::initial(); + for file in &first.selected_sources { + shipped.observe(file.identity).unwrap(); + } + let upgraded = source + .discover_transcript_paths_with_frontier( + bounds, + CodexDiscoveryFrontier::complete(shipped), + ) + .unwrap(); + assert_eq!( + sorted(upgraded.report.paths), + rollouts, + "the shipped binary's claim covers nothing this binary reads" + ); + assert!(!upgraded.next_frontier.is_complete()); + + let unchanged = source + .discover_transcript_paths_with_frontier(bounds, first.next_frontier) + .unwrap(); + assert!( + unchanged.report.paths.is_empty(), + "this binary's own claim still covers the unchanged corpus: {:?}", + unchanged.report.paths + ); + assert!(unchanged.next_frontier.is_complete()); + } + /// The durable epoch stores both digest halves directly. This catches a /// regression back to bit packing, masking, saturation, or clamping. #[test] diff --git a/crates/tracedecay/tests/transcript_ingest_suite/codex_rollout_sources.rs b/crates/tracedecay/tests/transcript_ingest_suite/codex_rollout_sources.rs new file mode 100644 index 0000000000..31932cc891 --- /dev/null +++ b/crates/tracedecay/tests/transcript_ingest_suite/codex_rollout_sources.rs @@ -0,0 +1,356 @@ +use std::path::{Path, PathBuf}; + +use tempfile::TempDir; +use tracedecay_sessions::runtime::SessionProvider; + +use crate::codex::write_jsonl; +use crate::restart_atomicity::{ + ProjectSessionTestRuntime, ingest_global_sources_for_provider, open_project_session_db, +}; +use crate::support::setup; + +const THREAD: &str = "codex-revert-thread"; +const REVERT_ROLLOUT: &str = "codex-revert-rollout"; +const LEDGER_MIGRATION: &str = "The ledger migration is ready."; +const LEDGER_BACKFILL: &str = "The ledger backfill is ready."; +const INVOICE_RETRY: &str = "The invoice export retry is ready."; +const INVOICE_ROLLBACK: &str = "The invoice export rollback is ready."; +const SHARED_QUESTION: &str = "The shared ledger question is ready."; + +fn session_meta(day: &str, cwd: &Path) -> serde_json::Value { + serde_json::json!({ + "timestamp": format!("2026-01-{day}T00:00:00.000Z"), + "type": "session_meta", + "payload": {"id": THREAD, "cwd": cwd.to_string_lossy()} + }) +} + +fn agent_message(day: &str, second: usize, reply: &str) -> serde_json::Value { + serde_json::json!({ + "timestamp": format!("2026-01-{day}T00:00:{second:02}.000Z"), + "type": "event_msg", + "payload": {"type": "agent_message", "message": reply} + }) +} + +fn rollout_path(home: &Path, day: &str, file_ids: &str) -> PathBuf { + let dir = home.join(format!(".codex/sessions/2026/01/{day}")); + std::fs::create_dir_all(&dir).unwrap(); + dir.join(format!("rollout-2026-01-{day}T00-00-00-{file_ids}.jsonl")) +} + +fn revert_rollout_path(home: &Path, day: &str) -> PathBuf { + rollout_path(home, day, &format!("{THREAD}_{REVERT_ROLLOUT}")) +} + +fn rollout(home: &Path, day: &str, file_ids: &str, cwd: &Path, replies: &[&str]) -> PathBuf { + let mut lines = vec![session_meta(day, cwd)]; + lines.extend( + replies + .iter() + .enumerate() + .map(|(index, reply)| agent_message(day, index + 1, reply)), + ); + let path = rollout_path(home, day, file_ids); + write_jsonl(&path, &lines); + path +} + +fn thread_rollout(home: &Path, cwd: &Path, replies: &[&str]) -> PathBuf { + rollout(home, "01", THREAD, cwd, replies) +} + +fn revert_rollout(home: &Path, cwd: &Path, replies: &[&str]) -> PathBuf { + rollout( + home, + "02", + &format!("{THREAD}_{REVERT_ROLLOUT}"), + cwd, + replies, + ) +} + +async fn replies(db: &ProjectSessionTestRuntime) -> Vec { + let mut texts: Vec = db + .search_session_messages("codex", None, "ready", 20) + .await + .into_iter() + .filter(|hit| hit.message.session_id == THREAD) + .map(|hit| hit.message.text) + .collect(); + texts.sort(); + texts +} + +fn sorted(replies: &[&str]) -> Vec { + let mut texts: Vec = replies.iter().map(|reply| (*reply).to_owned()).collect(); + texts.sort(); + texts +} + +async fn ingest(home: &Path, db: &ProjectSessionTestRuntime, project: &Path) { + ingest_global_sources_for_provider(home, db, project, Some(SessionProvider::Codex)).await; +} + +async fn revert_rollout_the_shipped_v2_binary_read( + home: &Path, + db: &ProjectSessionTestRuntime, + project: &Path, + day: &str, + replies: &[&str], +) -> PathBuf { + let staged = rollout(home, day, THREAD, project, replies); + ingest(home, db, project).await; + let revert = revert_rollout_path(home, day); + std::fs::rename(&staged, &revert).unwrap(); + revert +} + +async fn assert_upgrade_leaves( + home: &Path, + db: &ProjectSessionTestRuntime, + project: &Path, + before: &[&str], + expected: &[&str], +) { + assert_eq!(replies(db).await, sorted(before), "before the upgrade"); + for pass in 1..=2 { + ingest(home, db, project).await; + let after = replies(db).await; + for reply in sorted(before) { + assert!( + after.contains(&reply), + "pass {pass} lost {reply:?} that was searchable before the upgrade: {after:?}" + ); + } + assert_eq!(after, sorted(expected), "pass {pass}"); + } +} + +#[tokio::test] +async fn reverted_codex_thread_projects_both_live_rollouts() { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + let thread = thread_rollout(&home, &project, &[LEDGER_MIGRATION, LEDGER_BACKFILL]); + revert_rollout(&home, &project, &[INVOICE_RETRY, INVOICE_ROLLBACK]); + + let db = open_project_session_db(&project).await.unwrap(); + for _ in 0..2 { + ingest(&home, &db, &project).await; + assert_eq!( + replies(&db).await, + sorted(&[ + LEDGER_MIGRATION, + LEDGER_BACKFILL, + INVOICE_RETRY, + INVOICE_ROLLBACK + ]), + "each rollout of the thread keeps its own messages on every pass" + ); + } + let session = db + .get_session("codex", THREAD) + .await + .expect("codex session should be stored"); + crate::support::assert_path_text_eq( + session + .transcript_path + .as_deref() + .expect("session transcript path"), + &thread, + ); +} + +#[tokio::test] +async fn rewriting_a_revert_rollout_supersedes_only_its_own_messages() { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + thread_rollout(&home, &project, &[LEDGER_MIGRATION]); + revert_rollout(&home, &project, &[INVOICE_RETRY]); + let db = open_project_session_db(&project).await.unwrap(); + ingest(&home, &db, &project).await; + + revert_rollout(&home, &project, &[INVOICE_ROLLBACK]); + ingest(&home, &db, &project).await; + + assert_eq!( + replies(&db).await, + sorted(&[LEDGER_MIGRATION, INVOICE_ROLLBACK]), + "the rewrite retires the revert rollout's old reply and leaves the thread rollout alone" + ); +} + +#[tokio::test] +async fn rewriting_the_thread_rollout_supersedes_only_its_own_messages() { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + thread_rollout(&home, &project, &[LEDGER_MIGRATION]); + revert_rollout(&home, &project, &[INVOICE_RETRY]); + let db = open_project_session_db(&project).await.unwrap(); + ingest(&home, &db, &project).await; + + thread_rollout(&home, &project, &[LEDGER_BACKFILL]); + ingest(&home, &db, &project).await; + + assert_eq!( + replies(&db).await, + sorted(&[LEDGER_BACKFILL, INVOICE_RETRY]), + "the rewrite retires the thread rollout's old reply and leaves the revert rollout alone" + ); +} + +async fn assert_a_shared_record_outlives_a_rewrite_of_either( + thread_lines: &[serde_json::Value], + revert_lines: &[serde_json::Value], +) { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + let thread = rollout_path(&home, "01", THREAD); + let revert = revert_rollout_path(&home, "02"); + let with_meta = |day: &str, lines: &[serde_json::Value]| { + let mut all = vec![session_meta(day, &project)]; + all.extend_from_slice(lines); + all + }; + write_jsonl(&thread, &with_meta("01", thread_lines)); + write_jsonl(&revert, &with_meta("02", revert_lines)); + let db = open_project_session_db(&project).await.unwrap(); + ingest(&home, &db, &project).await; + assert_eq!( + replies(&db).await, + sorted(&[SHARED_QUESTION, LEDGER_MIGRATION, INVOICE_RETRY]) + ); + + write_jsonl( + &revert, + &with_meta("02", &[agent_message("02", 2, INVOICE_ROLLBACK)]), + ); + ingest(&home, &db, &project).await; + assert_eq!( + replies(&db).await, + sorted(&[SHARED_QUESTION, LEDGER_MIGRATION, INVOICE_ROLLBACK]), + "the thread rollout still holds the shared record" + ); + + write_jsonl( + &thread, + &with_meta("01", &[agent_message("01", 2, LEDGER_BACKFILL)]), + ); + ingest(&home, &db, &project).await; + assert_eq!( + replies(&db).await, + sorted(&[LEDGER_BACKFILL, INVOICE_ROLLBACK]), + "no rollout holds the shared record any more" + ); +} + +#[tokio::test] +async fn a_record_both_rollouts_hold_outlives_a_rewrite_of_either() { + let shared = agent_message("01", 1, SHARED_QUESTION); + assert_a_shared_record_outlives_a_rewrite_of_either( + &[shared.clone(), agent_message("01", 2, LEDGER_MIGRATION)], + &[shared, agent_message("02", 2, INVOICE_RETRY)], + ) + .await; +} + +#[tokio::test] +async fn a_record_both_rollouts_hold_at_different_offsets_outlives_a_rewrite_of_either() { + let shared = agent_message("01", 1, SHARED_QUESTION); + assert_a_shared_record_outlives_a_rewrite_of_either( + &[shared.clone(), agent_message("01", 2, LEDGER_MIGRATION)], + &[agent_message("02", 2, INVOICE_RETRY), shared], + ) + .await; +} + +#[tokio::test] +async fn upgrade_keeps_both_rollouts_the_shipped_v2_binary_imported_together() { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + thread_rollout(&home, &project, &[LEDGER_MIGRATION, LEDGER_BACKFILL]); + let db = open_project_session_db(&project).await.unwrap(); + revert_rollout_the_shipped_v2_binary_read( + &home, + &db, + &project, + "02", + &[INVOICE_RETRY, INVOICE_ROLLBACK], + ) + .await; + + assert_upgrade_leaves( + &home, + &db, + &project, + &[LEDGER_MIGRATION, LEDGER_BACKFILL], + &[ + LEDGER_MIGRATION, + LEDGER_BACKFILL, + INVOICE_RETRY, + INVOICE_ROLLBACK, + ], + ) + .await; +} + +#[tokio::test] +async fn upgrade_revives_a_revert_rollout_the_shipped_v2_binary_retired() { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + thread_rollout(&home, &project, &[LEDGER_MIGRATION, LEDGER_BACKFILL]); + let db = open_project_session_db(&project).await.unwrap(); + ingest(&home, &db, &project).await; + assert_eq!( + replies(&db).await, + sorted(&[LEDGER_MIGRATION, LEDGER_BACKFILL]) + ); + revert_rollout_the_shipped_v2_binary_read( + &home, + &db, + &project, + "02", + &[INVOICE_RETRY, INVOICE_ROLLBACK], + ) + .await; + + assert_upgrade_leaves( + &home, + &db, + &project, + &[], + &[INVOICE_RETRY, INVOICE_ROLLBACK], + ) + .await; +} + +#[tokio::test] +async fn upgrade_keeps_a_revert_rollout_the_shipped_v2_binary_read_last() { + let tmp = TempDir::new().unwrap(); + let (home, project) = setup(&tmp); + rollout( + &home, + "02", + THREAD, + &project, + &[LEDGER_MIGRATION, LEDGER_BACKFILL], + ); + let db = open_project_session_db(&project).await.unwrap(); + revert_rollout_the_shipped_v2_binary_read( + &home, + &db, + &project, + "01", + &[INVOICE_RETRY, INVOICE_ROLLBACK], + ) + .await; + + assert_upgrade_leaves( + &home, + &db, + &project, + &[INVOICE_RETRY, INVOICE_ROLLBACK], + &[INVOICE_RETRY, INVOICE_ROLLBACK], + ) + .await; +} diff --git a/crates/tracedecay/tests/transcript_ingest_suite/main.rs b/crates/tracedecay/tests/transcript_ingest_suite/main.rs index fbe15891e7..6727eb379b 100644 --- a/crates/tracedecay/tests/transcript_ingest_suite/main.rs +++ b/crates/tracedecay/tests/transcript_ingest_suite/main.rs @@ -18,6 +18,7 @@ mod codex_compaction; mod codex_goals; mod codex_ingest; mod codex_response_items; +mod codex_rollout_sources; mod codex_usage; mod cursor; mod cursor_composer;