Skip to content

fix(codex): give each revert rollout its own observation source #4366

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ScriptedAlchemy merged 6 commits into from
Oct 20, 2026
Merged

fix(codex): give each revert rollout its own observation source #4366

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b79e90c
test(codex): cover the rollouts of a reverted Codex thread
ScriptedAlchemy Oct 19, 2026
ad4afac
fix(codex): give each revert rollout its own observation source
ScriptedAlchemy Oct 19, 2026
20e7616
test(codex): cover v2 upgrades and shared records of reverted threads
ScriptedAlchemy Oct 19, 2026
4181cc3
fix(codex): let both rollouts of a thread own one record
ScriptedAlchemy Oct 19, 2026
7a46938
test(codex): seed rollout upgrades through production ingest
ScriptedAlchemy Oct 20, 2026
bc4d972
fix(codex): re-offer rollouts after the source reading changes
ScriptedAlchemy Oct 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions crates/tracedecay-domain/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
20 changes: 20 additions & 0 deletions crates/tracedecay-domain/src/observation.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
112 changes: 88 additions & 24 deletions crates/tracedecay-global-db/src/observation_adapter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -742,6 +745,7 @@ struct ObservationPreflightSnapshot {
admission_refusals: HashMap<(String, String), PayloadDigestV1>,
stored_observations: HashMap<String, StoredObservation>,
retrieval_aliases: HashMap<(String, String, String), RetrievalAnchorId>,
alias_owners: HashMap<RetrievalAnchorId, CanonicalObservationIdV1>,
cline_supersessions: HashMap<RetrievalAnchorId, RetrievalAnchorId>,
source_cursors:
HashMap<(ObservationSourceIdentityV1, ObservationScopeV1), ObservationSourceCursorV1>,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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())
}
Expand Down Expand Up @@ -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();
Expand All @@ -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 [
Expand All @@ -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)
Expand All @@ -988,6 +1021,7 @@ async fn load_observation_preflight(
admission_refusals,
stored_observations,
retrieval_aliases,
alias_owners,
cline_supersessions,
source_cursors,
})
Expand Down Expand Up @@ -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,
Expand All @@ -1208,15 +1256,18 @@ async fn read_retrieval_aliases_from_snapshot(
snapshot: &DatabaseEngineReadSnapshot,
writes: &[AnchoredObservationWrite],
operation: &'static str,
) -> ObservationStoreResult<HashMap<(String, String, String), RetrievalAnchorId>> {
) -> ObservationStoreResult<(
HashMap<(String, String, String), RetrievalAnchorId>,
HashMap<RetrievalAnchorId, CanonicalObservationIdV1>,
)> {
let mut requested = HashSet::new();
for write in writes {
for alias in write.retrieval_anchor().aliases() {
requested.insert(retrieval_alias_key(write.observation().scope(), alias)?);
}
}
if requested.is_empty() {
return Ok(HashMap::new());
return Ok((HashMap::new(), HashMap::new()));
}
let requested = requested
.into_iter()
Expand All @@ -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')
Expand All @@ -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
Expand All @@ -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::<Option<String>>(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 {
Expand Down
Loading
Loading