From 3b28b8a9ad2e0b465bb085828008bfd706d9662e Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:16:29 -0800 Subject: [PATCH 01/10] perf(sessions): avoid empty JSONL replay scope work Refs #4374 --- .../jsonl_observation_admission.rs | 38 +++++++++ .../jsonl_observation_admission/tests.rs | 81 +++++++++++++++++++ 2 files changed, 119 insertions(+) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index 5b76239710b..daa9a7ee3fa 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -2459,6 +2459,44 @@ pub(in crate::runtime) async fn admit_jsonl_observations( return Err(TranscriptIngestError::Cancelled { provider }); } let generation = ObservationSourceGenerationV1::new(raw.new_cursor.file_id)?; + // An unchanged, fully covered source has no context or frames to resolve. + // Keep the durable completion check: a prior interrupted rewrite may still + // have unretired records even though this pass reads zero new bytes. + if had_expected_cursor + && previous_generation == Some(generation) + && !raw.replacement_generation + && raw.start_offset == raw.read_through + && raw.frames.is_empty() + && raw.skipped.is_empty() + && !progress.source_deferred + { + let records_retired = admission + .complete_source_rewrite(&source, &scope, generation) + .await + .map_err(|outcome| { + if is_admission_cancellation(&outcome, &cancellation) { + TranscriptIngestError::Cancelled { provider } + } else { + host_admission_error(provider, outcome) + } + })?; + tracing::debug!( + event = "transcript_admission_batch", + phase = "complete", + provider, + transcript = %transcript_log_identity(path), + total_frames, + retained_frame_bytes, + bytes_consumed = progress.bytes_consumed, + source_deferred = progress.source_deferred, + records_retired, + "transcript admission batch finished" + ); + progress.covered_through = expected_cursor + .as_ref() + .map_or(0, ObservationSourceCursorV1::position); + return Ok(progress); + } let mut state = initialize(JsonlObservationScan { resumed: had_expected_cursor && raw.start_offset > 0, // Derived from the scanned generation rather than this batch's first diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index 4f3c2bd5885..7681f8fa137 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -1971,6 +1971,87 @@ async fn a_refusal_memo_cannot_authorize_uncovered_bytes_or_hide_an_in_place_rew } } +/// A durable unchanged cursor must not pay for a second provider-scope +/// initialization on every discovery re-offer. Rewrite completion remains +/// authoritative even when no frames were read, and the next append must +/// still admit through normal scope/provenance checks. +#[tokio::test] +async fn unchanged_codex_reoffers_skip_empty_initializer_but_complete_rewrites() { + let (_temp, path, len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + let first = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("initial admission"); + assert_eq!(first.covered_through, len); + let previous = stored_cursor(&spy).await.expect("durable cursor"); + let original_rows = spy.inner.observations().len(); + assert!(original_rows > 0, "the cold pass must admit actual evidence"); + let original_captures = spy.capture_count(); + spy.rewrites.lock().unwrap().clear(); + + for _ in 0..16 { + let request = super::JsonlObservationAdmissionRequest::new( + "codex", + &path, + &spy, + crate::runtime::hosts::codex::codex_observation_source_v2(SESSION_ID).unwrap(), + ObservationScopeV1::Profile, + RetentionClass::new("test").unwrap(), + ) + .with_persisted_cursor_update(super::PersistedCursorUpdate::Replace) + .with_lazy_shared_frame_preparation(); + let mut initialized = false; + let idle = super::admit_jsonl_observations( + request, + async |_| { + initialized = true; + Err::<(), TranscriptIngestError>(TranscriptIngestError::InvalidFrameState { + provider: "codex", + }) + }, + |(), _, _, _, _, _| panic!("unchanged source must not normalize frames"), + ) + .await + .expect("unchanged source must avoid the empty initializer"); + assert!(!initialized, "zero-frame re-offer started scope resolution"); + assert_eq!(idle.bytes_consumed, 0); + assert_eq!(idle.frames_persisted, 0); + assert_eq!(idle.covered_through, len); + } + let completions = spy.rewrites.lock().unwrap(); + assert_eq!(completions.len(), 16, "no-op re-offers must still check durable rewrites"); + assert!(completions.iter().all(|call| { + matches!(call, RewriteCall::Complete(generation) if *generation == previous.generation()) + })); + drop(completions); + assert_eq!(spy.inner.observations().len(), original_rows); + assert_eq!(spy.capture_count(), original_captures); + assert_eq!(stored_cursor(&spy).await, Some(previous)); + + let appended = json!({ + "timestamp": "2026-01-01T00:00:02.000Z", + "type": "event_msg", + "payload": crate::runtime::hosts::codex::user_message_payload("newly appended evidence") + }) + .to_string() + + "\n"; + std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap() + .write_all(appended.as_bytes()) + .unwrap(); + spin_until_jsonl_change_settled(&path); + let resumed = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("append must still take the real admission path"); + assert_eq!(resumed.bytes_consumed, appended.len() as u64); + assert_eq!(resumed.frames_persisted, 1); + assert_eq!(spy.inner.observations().len(), original_rows + 1); +} + #[tokio::test] async fn an_unchanged_resumed_rollout_does_not_rebuild_its_prior_context() { let (_temp, path, len) = rollout_fixture(); From 6efd556a537c0cee09706a655980a6e0468ba25f Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:19:05 -0800 Subject: [PATCH 02/10] fix(sessions): make empty-page bypass Codex-specific Refs #4374 --- .../src/runtime/hosts/codex/observation.rs | 1 + .../jsonl_observation_admission.rs | 13 ++++- .../jsonl_observation_admission/tests.rs | 48 ++++++++++++++++++- 3 files changed, 60 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs index cf3816de8df..4978b951fdf 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs @@ -772,6 +772,7 @@ async fn admit_codex_jsonl_page( .with_max_new_bytes(max_new_bytes) .with_persisted_cursor_update(PersistedCursorUpdate::Replace) .with_lazy_shared_frame_preparation() + .with_skip_empty_initialization() .with_cancellation(cancellation.clone()); if let Some(max_frames) = max_frames { request = request.with_max_frames(max_frames); diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index daa9a7ee3fa..dc7e53c4c1e 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -89,6 +89,7 @@ pub(in crate::runtime) struct JsonlObservationAdmissionRequest<'request> { persisted_cursor_update: PersistedCursorUpdate, cancellation: ObservationCancellation, shared_frame_preparation: SharedJsonlFramePreparation, + skip_empty_initialization: bool, } impl<'request> JsonlObservationAdmissionRequest<'request> { @@ -112,6 +113,7 @@ impl<'request> JsonlObservationAdmissionRequest<'request> { persisted_cursor_update: PersistedCursorUpdate::Monotonic, cancellation: ObservationCancellation::default(), shared_frame_preparation: SharedJsonlFramePreparation::None, + skip_empty_initialization: false, } } @@ -145,6 +147,13 @@ impl<'request> JsonlObservationAdmissionRequest<'request> { self.shared_frame_preparation = SharedJsonlFramePreparation::Lazy; self } + + /// Opt in only when an empty page needs no provider-specific validation. + /// Pi, for example, must verify its header even on a no-frame rewrite. + pub(in crate::runtime) fn with_skip_empty_initialization(mut self) -> Self { + self.skip_empty_initialization = true; + self + } } pub(in crate::runtime) enum JsonlFrameAdmission { @@ -2345,6 +2354,7 @@ pub(in crate::runtime) async fn admit_jsonl_observations( persisted_cursor_update, cancellation, shared_frame_preparation, + skip_empty_initialization, } = request; if cancellation.is_cancelled() { return Err(TranscriptIngestError::Cancelled { provider }); @@ -2462,7 +2472,8 @@ pub(in crate::runtime) async fn admit_jsonl_observations( // An unchanged, fully covered source has no context or frames to resolve. // Keep the durable completion check: a prior interrupted rewrite may still // have unretired records even though this pass reads zero new bytes. - if had_expected_cursor + if skip_empty_initialization + && had_expected_cursor && previous_generation == Some(generation) && !raw.replacement_generation && raw.start_offset == raw.read_through diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index 7681f8fa137..734b814fd8a 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -1971,6 +1971,51 @@ async fn a_refusal_memo_cannot_authorize_uncovered_bytes_or_hide_an_in_place_rew } } +/// Other providers may use their initializer to verify proof even with zero +/// frames. Default behavior therefore continues invoking it and preserves a +/// refusal before the pending-rewrite completion check. +#[tokio::test] +async fn unchanged_noop_does_not_skip_initialization_without_provider_opt_in() { + let (_temp, path, len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + let first = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("initial admission"); + assert_eq!(first.covered_through, len); + spy.rewrites.lock().unwrap().clear(); + let original_cursor = stored_cursor(&spy).await; + + let request = super::JsonlObservationAdmissionRequest::new( + "codex", + &path, + &spy, + crate::runtime::hosts::codex::codex_observation_source_v2(SESSION_ID).unwrap(), + ObservationScopeV1::Profile, + RetentionClass::new("test").unwrap(), + ); + let mut initialized = false; + let error = super::admit_jsonl_observations( + request, + async |_| { + initialized = true; + Err::<(), TranscriptIngestError>(TranscriptIngestError::InvalidFrameState { + provider: "codex", + }) + }, + |(), _, _, _, _, _| panic!("empty page cannot normalize a frame"), + ) + .await + .expect_err("without opt-in the initializer must still validate an empty page"); + assert!(initialized); + assert!(matches!( + error, + TranscriptIngestError::InvalidFrameState { provider: "codex" } + )); + assert_eq!(stored_cursor(&spy).await, original_cursor); + assert!(spy.rewrites.lock().unwrap().is_empty()); +} + /// A durable unchanged cursor must not pay for a second provider-scope /// initialization on every discovery re-offer. Rewrite completion remains /// authoritative even when no frames were read, and the next append must @@ -2000,7 +2045,8 @@ async fn unchanged_codex_reoffers_skip_empty_initializer_but_complete_rewrites() RetentionClass::new("test").unwrap(), ) .with_persisted_cursor_update(super::PersistedCursorUpdate::Replace) - .with_lazy_shared_frame_preparation(); + .with_lazy_shared_frame_preparation() + .with_skip_empty_initialization(); let mut initialized = false; let idle = super::admit_jsonl_observations( request, From 1d4e01528aa2fd1434584e770d097fe12438a08a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:19:55 -0800 Subject: [PATCH 03/10] fix(sessions): require exact no-op cursor position Refs #4374 --- .../src/runtime/observation/jsonl_observation_admission.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index dc7e53c4c1e..39b0e2afc8f 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -2475,6 +2475,9 @@ pub(in crate::runtime) async fn admit_jsonl_observations( if skip_empty_initialization && had_expected_cursor && previous_generation == Some(generation) + && expected_cursor + .as_ref() + .is_some_and(|cursor| cursor.position() == raw.start_offset) && !raw.replacement_generation && raw.start_offset == raw.read_through && raw.frames.is_empty() From f2ab365184920fb211c5310665aefd97cb1f0685 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 20:16:12 -0800 Subject: [PATCH 04/10] fix(sessions): release reoffer probe lock before async reads Follow hosted Bazel formatting diagnostics and confine the test spy MutexGuard to its own scope. This preserves all assertions and keeps the later asynchronous cursor and re-offer checks outside the lock. Refs #4374 --- .../jsonl_observation_admission/tests.rs | 22 +++++++++++++------ 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index 734b814fd8a..274349f01e4 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -2031,7 +2031,10 @@ async fn unchanged_codex_reoffers_skip_empty_initializer_but_complete_rewrites() assert_eq!(first.covered_through, len); let previous = stored_cursor(&spy).await.expect("durable cursor"); let original_rows = spy.inner.observations().len(); - assert!(original_rows > 0, "the cold pass must admit actual evidence"); + assert!( + original_rows > 0, + "the cold pass must admit actual evidence" + ); let original_captures = spy.capture_count(); spy.rewrites.lock().unwrap().clear(); @@ -2065,12 +2068,17 @@ async fn unchanged_codex_reoffers_skip_empty_initializer_but_complete_rewrites() assert_eq!(idle.frames_persisted, 0); assert_eq!(idle.covered_through, len); } - let completions = spy.rewrites.lock().unwrap(); - assert_eq!(completions.len(), 16, "no-op re-offers must still check durable rewrites"); - assert!(completions.iter().all(|call| { - matches!(call, RewriteCall::Complete(generation) if *generation == previous.generation()) - })); - drop(completions); + { + let completions = spy.rewrites.lock().unwrap(); + assert_eq!( + completions.len(), + 16, + "no-op re-offers must still check durable rewrites" + ); + assert!(completions.iter().all(|call| { + matches!(call, RewriteCall::Complete(generation) if *generation == previous.generation()) + })); + } assert_eq!(spy.inner.observations().len(), original_rows); assert_eq!(spy.capture_count(), original_captures); assert_eq!(stored_cursor(&spy).await, Some(previous)); From 3386b3c61d8b113af2e19faf28dc10464053f1fd Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 20:40:32 -0800 Subject: [PATCH 05/10] test(sessions): consolidate Codex cold-reoffer regressions Reuse the production-path 16-rollout / 96-reoffer Git-scope and append-transition tests on the original issue-specific PR. Keep rewrite-completion and provider opt-in regressions already present. Refs #4374 --- .../src/runtime/hosts/codex/observation.rs | 2 + .../hosts/codex/observation/reoffer_tests.rs | 357 ++++++++++++++++++ 2 files changed, 359 insertions(+) create mode 100644 crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs index 4978b951fdf..529248090ca 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs @@ -45,6 +45,8 @@ use tracedecay_runtime_core::resident_memory::ProcessSharedMemoryReservationV1; #[cfg(test)] mod meta_cache_tests; #[cfg(test)] +mod reoffer_tests; +#[cfg(test)] mod retired_source_tests; #[cfg(test)] mod rollout_source_tests; diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs new file mode 100644 index 00000000000..d6272604004 --- /dev/null +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation/reoffer_tests.rs @@ -0,0 +1,357 @@ +//! Empty Codex re-offers must not pay repository resolution costs or cache a +//! scope decision that can leak into the next nonempty admission page. + +use std::io::Write as _; +use std::path::{Path, PathBuf}; +use std::sync::Mutex; +use std::time::Instant; + +use serde_json::{Value, json}; +use tempfile::TempDir; +use tracedecay_domain::{ObservationScopeV1, ProjectId}; +use tracedecay_runtime_core::config::ProfileRoot; +use tracedecay_runtime_core::git_discovery::{ + GitRepositoryIdentityOutcome, discover_repository_identity_cli_first, +}; +use tracedecay_runtime_core::path_safety::plain_git_args; + +use super::{ + CodexJsonlAdmissionProgress, CodexObservationAdmission, codex_observation_source_v2, + try_admit_codex_jsonl_observations, +}; +use crate::admission::test_support::MemoryHostAdmission; +use crate::observation::ObservationCancellation; +use crate::runtime::observation::jsonl_observation_admission::install_test_shared_jsonl_preparation_authority; +use crate::runtime::shared::ProjectRootMatcherCache; +use crate::runtime::with_transcript_source_profile; + +static IDENTITY_RESOLUTIONS: Mutex> = Mutex::new(Vec::new()); + +fn recording_identity(path: &Path) -> GitRepositoryIdentityOutcome { + IDENTITY_RESOLUTIONS + .lock() + .unwrap() + .push(path.to_path_buf()); + discover_repository_identity_cli_first(path) +} + +fn resolution_count(root: &Path) -> usize { + IDENTITY_RESOLUTIONS + .lock() + .unwrap() + .iter() + .filter(|path| path.starts_with(root)) + .count() +} + +fn message(session: &str, item: &str) -> Value { + json!({ + "timestamp": "2026-09-04T12:00:01.004Z", + "type": "event_msg", + "payload": { + "type": "item_completed", + "thread_id": session, + "turn_id": item, + "item": { + "type": "UserMessage", + "id": item, + "content": [{"type": "text", "text": item}] + } + } + }) +} + +fn write_rollout(path: &Path, cwd: &Path, session: &str) -> u64 { + let mut file = std::fs::File::create(path).unwrap(); + writeln!( + file, + "{}", + json!({ + "timestamp": "2026-09-04T12:00:00.000Z", + "type": "session_meta", + "payload": {"id": session, "cwd": cwd} + }) + ) + .unwrap(); + writeln!(file, "{}", message(session, "initial-item")).unwrap(); + file.metadata().unwrap().len() +} + +fn append_in_scope(path: &Path, cwd: &Path, session: &str, item: &str) -> u64 { + let mut file = std::fs::OpenOptions::new().append(true).open(path).unwrap(); + writeln!( + file, + "{}", + json!({"type": "turn_context", "payload": {"cwd": cwd, "model": "scope-model"}}) + ) + .unwrap(); + writeln!(file, "{}", message(session, item)).unwrap(); + file.metadata().unwrap().len() +} + +struct Fixture { + _temp: TempDir, + root: PathBuf, + project: PathBuf, + outside: PathBuf, + profile: ProfileRoot, + project_id: ProjectId, + project_admission: MemoryHostAdmission, + profile_admission: MemoryHostAdmission, +} + +impl Fixture { + fn new() -> Self { + install_test_shared_jsonl_preparation_authority(); + let temp = TempDir::new().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let project = root.join("project"); + let outside = root.join("outside"); + for repository in [&project, &outside] { + std::fs::create_dir_all(repository.join("src")).unwrap(); + let output = std::process::Command::new( + tracedecay_runtime_core::git::try_git_program().unwrap(), + ) + .args(plain_git_args(&[ + "init", + "--quiet", + "--initial-branch=main", + ])) + .current_dir(repository) + .output() + .unwrap(); + assert!( + output.status.success(), + "git init: {}", + String::from_utf8_lossy(&output.stderr) + ); + } + Self { + profile: ProfileRoot::under_home(root.join("owner")), + _temp: temp, + root, + project, + outside, + project_id: ProjectId::new("project.codex-idle-reoffer").unwrap(), + project_admission: MemoryHostAdmission::default(), + profile_admission: MemoryHostAdmission::default(), + } + } + + async fn admit( + &self, + path: &Path, + cache: &ProjectRootMatcherCache, + cancellation: &ObservationCancellation, + ) -> [CodexJsonlAdmissionProgress; 2] { + let project = try_admit_codex_jsonl_observations( + path, + CodexObservationAdmission::Project { + root: &self.project, + project_id: self.project_id.clone(), + }, + &self.project_admission, + None, + None, + cancellation, + cache, + ) + .await + .unwrap(); + let profile = try_admit_codex_jsonl_observations( + path, + CodexObservationAdmission::Profile { + session_id: None, + registered_roots: std::slice::from_ref(&self.project), + }, + &self.profile_admission, + None, + None, + cancellation, + cache, + ) + .await + .unwrap(); + [project, profile] + } +} + +fn assert_unchanged(progress: CodexJsonlAdmissionProgress, len: u64) { + assert!(progress.resumed); + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, len); + assert_eq!(progress.bytes_consumed, 0); + assert_eq!(progress.frames_decoded, 0); + assert_eq!(progress.frames_persisted, 0); + assert_eq!(progress.prior_context_bytes, 0); +} + +#[test] +fn unchanged_corpus_reoffers_skip_cold_scope_resolution() { + let fixture = Fixture::new(); + let mut rollouts = Vec::new(); + for index in 0..16 { + let repository = if index % 2 == 0 { + &fixture.project + } else { + &fixture.outside + }; + let cwd = repository.join(format!("cwd-{index}")); + std::fs::create_dir_all(&cwd).unwrap(); + let path = fixture.root.join(format!("rollout-{index}.jsonl")); + let len = write_rollout(&path, &cwd, &format!("reoffer-session-{index}")); + rollouts.push((path, len)); + } + tracedecay_runtime_core::native_runtime::block_on(with_transcript_source_profile( + fixture.profile.clone(), + async { + let cancellation = ObservationCancellation::default(); + let initial = ProjectRootMatcherCache::with_identity_resolver(recording_identity); + for (path, len) in &rollouts { + for progress in fixture.admit(path, &initial, &cancellation).await { + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, *len); + assert_eq!(progress.bytes_consumed, *len); + } + } + let stores = [&fixture.project_admission, &fixture.profile_admission]; + assert_eq!(stores.map(|store| store.observations().len()), [16, 16]); + let captures = stores.map(MemoryHostAdmission::capture_call_counts); + let retained = stores.map(|store| { + store + .observations() + .iter() + .map(|record| { + ( + record.observation().observation_id().clone(), + record.committed_cursor().clone(), + ) + }) + .collect::>() + }); + let before = resolution_count(&fixture.root); + assert!(before >= rollouts.len(), "the resolver probe must be live"); + drop(initial); + + // A cold process cache cannot hide redundant resolution on the + // first empty re-offer. Later rounds also exercise the warm path. + let cache = ProjectRootMatcherCache::with_identity_resolver(recording_identity); + let started = Instant::now(); + let mut reoffers = 0; + for _ in 0..3 { + for (path, len) in &rollouts { + for progress in fixture.admit(path, &cache, &cancellation).await { + assert_unchanged(progress, *len); + reoffers += 1; + } + } + } + let resolutions = resolution_count(&fixture.root) - before; + tracing::info!( + reoffers, + git_identity_resolutions = resolutions, + wall_seconds = started.elapsed().as_secs_f64(), + "unchanged Codex corpus re-offer measurement" + ); + assert_eq!(reoffers, 96); + assert_eq!(resolutions, 0, "empty pages must not resolve Git scope"); + assert_eq!( + stores.map(MemoryHostAdmission::capture_call_counts), + captures + ); + for (store, retained) in stores.into_iter().zip(retained) { + let current = store + .observations() + .iter() + .map(|record| { + ( + record.observation().observation_id().clone(), + record.committed_cursor().clone(), + ) + }) + .collect::>(); + assert_eq!( + current, retained, + "accepted source identity must not change" + ); + } + }, + )) + .unwrap(); +} + +#[test] +fn append_after_empty_reoffer_rechecks_scope_and_source_identity() { + let fixture = Fixture::new(); + let path = fixture.root.join("rollout-live.jsonl"); + let session = "reoffer-live-session"; + let len = write_rollout(&path, &fixture.project.join("src"), session); + tracedecay_runtime_core::native_runtime::block_on(with_transcript_source_profile( + fixture.profile.clone(), + async { + let cancellation = ObservationCancellation::default(); + let initial = ProjectRootMatcherCache::default(); + let admitted = fixture.admit(&path, &initial, &cancellation).await; + assert_eq!(admitted[0].frames_persisted, 2); + assert_eq!(admitted[1].frames_persisted, 0); + drop(initial); + + let cache = ProjectRootMatcherCache::with_identity_resolver(recording_identity); + for progress in fixture.admit(&path, &cache, &cancellation).await { + assert_unchanged(progress, len); + } + assert_eq!(resolution_count(&fixture.root), 0); + + let new_len = + append_in_scope(&path, &fixture.outside.join("src"), session, "outside-item"); + let appended = fixture.admit(&path, &cache, &cancellation).await; + assert_eq!(appended[0].frames_persisted, 0); + assert!(appended[1].frames_persisted > 0); + for progress in appended { + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, new_len); + assert_eq!(progress.bytes_consumed, new_len - len); + } + assert!(resolution_count(&fixture.root) > 0); + let expected_source = codex_observation_source_v2(session).unwrap(); + let profile_records = fixture.profile_admission.observations(); + assert!(!profile_records.is_empty()); + for record in &profile_records { + assert_eq!(record.observation().source(), &expected_source); + assert_eq!(record.observation().scope(), &ObservationScopeV1::Profile); + } + assert_eq!(fixture.project_admission.observations().len(), 2); + + let final_len = append_in_scope( + &path, + &fixture.project.join("src"), + session, + "returned-item", + ); + let returned = fixture.admit(&path, &cache, &cancellation).await; + assert!(returned[0].frames_persisted > 0); + assert_eq!(returned[1].frames_persisted, 0); + for progress in returned { + assert!(!progress.source_deferred); + assert_eq!(progress.covered_through, final_len); + assert_eq!(progress.bytes_consumed, final_len - new_len); + } + assert_eq!( + fixture.profile_admission.observations().len(), + profile_records.len() + ); + for progress in fixture.admit(&path, &cache, &cancellation).await { + assert_unchanged(progress, final_len); + } + + cancellation.cancel(); + let captures = fixture.project_admission.capture_call_counts(); + for progress in fixture.admit(&path, &cache, &cancellation).await { + assert!(progress.source_deferred, "cancellation is not convergence"); + assert_eq!(progress.frames_persisted, 0); + } + assert_eq!(fixture.project_admission.capture_call_counts(), captures); + }, + )) + .unwrap(); +} From 08b01b9fe2abd35c6de57b9386253418a752c575 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 20:40:59 -0800 Subject: [PATCH 06/10] chore(sessions): retain Codex scope-resolution timings Preserve the trace boundary when consolidating the alternate re-offer implementation and behavioral tests. Refs #4374 --- .../src/runtime/hosts/codex/observation.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs index 529248090ca..7b3c317f1f7 100644 --- a/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs +++ b/crates/tracedecay-sessions/src/runtime/hosts/codex/observation.rs @@ -405,6 +405,11 @@ impl CodexObservationAdmission<'_> { /// Prepare this bounded page's context and scope decisions on the blocking /// pool. The page owns its reservation throughout preparation; only these /// decisions, not filesystem work, are used by frame callbacks and retries. + #[tracing::instrument( + name = "sessions.observation.codex.resolve_scope", + level = "trace", + skip_all + )] async fn resolve_scope( &self, matchers: &ProjectRootMatcherCache, From 29a9150bcc13a55fd5818fea8f56f1db79894ced Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:14:56 -0800 Subject: [PATCH 07/10] test(sessions): refuse failed Codex no-op rewrite checks Ensure a failed durable rewrite-completion read remains a typed block on an unchanged empty page, leaves the accepted cursor/rows intact, and is retried without re-reading transcript frames. Refs #4374 --- .../jsonl_observation_admission/tests.rs | 82 +++++++++++++++++++ 1 file changed, 82 insertions(+) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index 274349f01e4..1a21d88d3e9 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -81,6 +81,7 @@ struct SeamSpyAdmission { /// strictly inside its range. peer_covers_advance_through: Mutex>, rewrites: Mutex>, + rewrite_completion_failure: Mutex>, } #[derive(Debug, PartialEq, Eq)] @@ -1053,6 +1054,9 @@ impl HostAdmission for SeamSpyAdmission { .lock() .unwrap() .push(RewriteCall::Complete(generation)); + if let Some(failure) = self.rewrite_completion_failure.lock().unwrap().take() { + return Box::pin(async move { Err(failure) }); + } self.inner .complete_source_rewrite(source, scope, generation) } @@ -2016,6 +2020,84 @@ async fn unchanged_noop_does_not_skip_initialization_without_provider_opt_in() { assert!(spy.rewrites.lock().unwrap().is_empty()); } +/// An empty re-offer still requires a successful durable rewrite-completion +/// check. A transient read failure must remain a typed block, not an empty +/// success that incorrectly declares source convergence. +#[tokio::test] +async fn unchanged_codex_reoffer_propagates_rewrite_completion_failure() { + let (_temp, path, len) = rollout_fixture(); + let spy = SeamSpyAdmission::default(); + let initial = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("initial admission"); + assert_eq!(initial.covered_through, len); + let cursor = stored_cursor(&spy).await.expect("committed source cursor"); + let rows = spy.inner.observations().len(); + let captures = spy.capture_count(); + assert!(rows > 0, "the source must have real committed evidence"); + spy.rewrites.lock().unwrap().clear(); + *spy.rewrite_completion_failure.lock().unwrap() = + Some(HostAdmissionOutcome::retained_unavailable( + "projection_storage_retry_scheduled", + )); + + let request = super::JsonlObservationAdmissionRequest::new( + "codex", + &path, + &spy, + crate::runtime::hosts::codex::codex_observation_source_v2(SESSION_ID).unwrap(), + ObservationScopeV1::Profile, + RetentionClass::new("test").unwrap(), + ) + .with_persisted_cursor_update(super::PersistedCursorUpdate::Replace) + .with_skip_empty_initialization(); + let error = super::admit_jsonl_observations( + request, + async |_| { + Err::<(), TranscriptIngestError>(TranscriptIngestError::InvalidFrameState { + provider: "codex", + }) + }, + |(), _, _, _, _, _| panic!("unchanged source cannot normalize frames"), + ) + .await + .expect_err("failed durable rewrite check must not report convergence"); + assert!(matches!( + error, + TranscriptIngestError::HostAdmission { + provider: "codex", + reason: "projection_storage_retry_scheduled", + retryable: true, + .. + } + )); + assert_eq!(stored_cursor(&spy).await, Some(cursor.clone())); + assert_eq!(spy.inner.observations().len(), rows); + assert_eq!(spy.capture_count(), captures); + assert_eq!( + spy.rewrites.lock().unwrap().as_slice(), + &[RewriteCall::Complete(cursor.generation())] + ); + + let retry = + try_admit_codex_jsonl_observations_for_profile_with_admission(&path, None, &[], &spy, None) + .await + .expect("retry must recheck durable rewrite completion"); + assert_eq!(retry.covered_through, len); + assert_eq!(retry.bytes_consumed, 0); + assert_eq!(retry.frames_persisted, 0); + assert_eq!(stored_cursor(&spy).await, Some(cursor.clone())); + assert_eq!(spy.inner.observations().len(), rows); + assert_eq!( + spy.rewrites.lock().unwrap().as_slice(), + &[ + RewriteCall::Complete(cursor.generation()), + RewriteCall::Complete(cursor.generation()), + ] + ); +} + /// A durable unchanged cursor must not pay for a second provider-scope /// initialization on every discovery re-offer. Rewrite completion remains /// authoritative even when no frames were read, and the next append must From c69148305633159cc828076921ea8ed1039a7742 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:34:34 -0800 Subject: [PATCH 08/10] style(sessions): fix rewrite refusal regression formatting Apply the pinned Bazel rustfmt formatting from repository gates. Preserve the failure injection and nonvacuous assertions. Refs #4374 --- .../observation/jsonl_observation_admission/tests.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index 1a21d88d3e9..a50adb2af82 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -2037,10 +2037,9 @@ async fn unchanged_codex_reoffer_propagates_rewrite_completion_failure() { let captures = spy.capture_count(); assert!(rows > 0, "the source must have real committed evidence"); spy.rewrites.lock().unwrap().clear(); - *spy.rewrite_completion_failure.lock().unwrap() = - Some(HostAdmissionOutcome::retained_unavailable( - "projection_storage_retry_scheduled", - )); + *spy.rewrite_completion_failure.lock().unwrap() = Some( + HostAdmissionOutcome::retained_unavailable("projection_storage_retry_scheduled"), + ); let request = super::JsonlObservationAdmissionRequest::new( "codex", From 01c0b168040e4d826574d6a515849f095b55b87a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:52:34 -0800 Subject: [PATCH 09/10] test(automation): bind scheduler fixtures to backend identities Repair three hosted automation_runner_test failures by supplying the configured executable identity required by the production prefilter. Use the same identity for the prefilter and retained runner, preserving manual-schedule, exact-ledger, live-lock and lock-race assertions. Add a production-prefilter regression for unconfigured refusal, deduplication and same-occurrence recovery after binding, with zero backend calls and no effectful terminal. Do not relax backend admission, change timeouts, or alter the graph publication changes. Refs #4532 --- .../tests/automation_runner_test/jobs.rs | 124 ++++++++++++++++-- 1 file changed, 110 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index 740283d56e8..7342786381b 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs @@ -86,6 +86,7 @@ fn enabled_job_config() -> AutomationSettingsV1 { struct ContentBackend { calls: AtomicUsize, content: &'static str, + executable: Option, } impl ContentBackend { @@ -93,9 +94,15 @@ impl ContentBackend { Self { calls: AtomicUsize::new(0), content, + executable: None, } } + fn with_executable(mut self, executable: &Path) -> Self { + self.executable = Some(executable.to_path_buf()); + self + } + fn calls(&self) -> usize { self.calls.load(Ordering::SeqCst) } @@ -126,7 +133,7 @@ impl AgentTaskBackend for ContentBackend { } fn executable(&self) -> Option<&std::path::Path> { - None + self.executable.as_deref() } } @@ -988,6 +995,10 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let mut job = sample_job("sched-job"); job.schedule = Some("manual".to_string()); let config = enabled_job_config(); + // Reach the schedule gate with a bound identity, not the earlier + // backend-unconfigured refusal. This gate never invokes the backend. + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); // Manual schedule: every scheduler tick of the same occurrence reaches // the prefilter, which records the skip once. @@ -996,7 +1007,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let run = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1019,7 +1030,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let repeated = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1031,6 +1042,76 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { repeated.ledger_record, records[0], "an immediate repeat skip must return the exact durable prior row" ); + assert_eq!(backend.calls(), 0); +} + +#[tokio::test] +async fn scheduler_prefilter_rejects_unconfigured_backend_and_recovers_without_effect() { + let temp = tempdir().unwrap(); + let dashboard_root = temp.path().join("dashboard"); + let job = sample_job("backend-recovery"); + let config = enabled_job_config(); + let occurrence = "backend-recovery-occurrence"; + let backend = ContentBackend::new("unused"); + + let first = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("a due job still requires a configured backend"); + assert_eq!(first.report["reason"], json!("backend_unconfigured")); + assert_eq!(first.ledger_record.status, AutomationRunStatus::Skipped); + assert_ne!(first.run_id, occurrence); + let repeated = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(repeated.ledger_record, first.ledger_record); + + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = backend.with_executable(executable.path()); + assert!( + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), + "binding the backend must admit the same due occurrence without cooldown" + ); + assert_eq!(backend.calls(), 0); + assert_eq!( + load_run_records(&dashboard_root, 10).await.unwrap(), + vec![first.ledger_record], + "advisory admission must not append an effectful terminal" + ); + assert!( + tracedecay_automation_runtime::automation::run_ledger::find_run_record_exact_bounded( + &dashboard_root, + occurrence, + ) + .await + .unwrap() + .is_none() + ); } #[tokio::test] @@ -1206,15 +1287,22 @@ async fn scheduler_prefilter_live_lock_wins_over_not_due_summary() { let config = enabled_job_config(); let occurrence = "locked-prefilter-occurrence"; - let prefilter = - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .expect("the live job lock must produce a diagnostic"); + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); + let prefilter = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("the live job lock must produce a diagnostic"); assert_eq!(prefilter.report["reason"], json!("scheduler_lock_active")); assert_ne!(prefilter.run_id, occurrence); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, @@ -1262,12 +1350,21 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { job.schedule = Some("every 1h".to_owned()); let config = enabled_job_config(); let occurrence = "prefilter-gap-occurrence"; + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); assert!( - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .is_none(), + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), "a new interval job is advisory-due before effect admission" ); let _raced_in = @@ -1276,7 +1373,6 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { .unwrap() .expect("the advisory prefilter must release its reservation lock"); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, From 53ef6807294eaadf4b77f8ad0e9507b6a1cf54a0 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:51:29 -0800 Subject: [PATCH 10/10] chore(sessions): isolate Codex re-offer review from scheduler tests Restore the unrelated scheduler fixture file to the current master blob on this #4374 branch. The identical scheduler repair remains preserved in the separately owned #4815 branch; do not duplicate its CI/review surface here. Preserve all Codex admission and regression changes. Refs #4374 --- .../tests/automation_runner_test/jobs.rs | 124 ++---------------- 1 file changed, 14 insertions(+), 110 deletions(-) diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index 7342786381b..740283d56e8 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs @@ -86,7 +86,6 @@ fn enabled_job_config() -> AutomationSettingsV1 { struct ContentBackend { calls: AtomicUsize, content: &'static str, - executable: Option, } impl ContentBackend { @@ -94,15 +93,9 @@ impl ContentBackend { Self { calls: AtomicUsize::new(0), content, - executable: None, } } - fn with_executable(mut self, executable: &Path) -> Self { - self.executable = Some(executable.to_path_buf()); - self - } - fn calls(&self) -> usize { self.calls.load(Ordering::SeqCst) } @@ -133,7 +126,7 @@ impl AgentTaskBackend for ContentBackend { } fn executable(&self) -> Option<&std::path::Path> { - self.executable.as_deref() + None } } @@ -995,10 +988,6 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let mut job = sample_job("sched-job"); job.schedule = Some("manual".to_string()); let config = enabled_job_config(); - // Reach the schedule gate with a bound identity, not the earlier - // backend-unconfigured refusal. This gate never invokes the backend. - let executable = tempfile::NamedTempFile::new().unwrap(); - let backend = ContentBackend::new("unused").with_executable(executable.path()); // Manual schedule: every scheduler tick of the same occurrence reaches // the prefilter, which records the skip once. @@ -1007,7 +996,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let run = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - backend.executable(), + None, &job, "sched-occurrence", None, @@ -1030,7 +1019,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let repeated = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - backend.executable(), + None, &job, "sched-occurrence", None, @@ -1042,76 +1031,6 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { repeated.ledger_record, records[0], "an immediate repeat skip must return the exact durable prior row" ); - assert_eq!(backend.calls(), 0); -} - -#[tokio::test] -async fn scheduler_prefilter_rejects_unconfigured_backend_and_recovers_without_effect() { - let temp = tempdir().unwrap(); - let dashboard_root = temp.path().join("dashboard"); - let job = sample_job("backend-recovery"); - let config = enabled_job_config(); - let occurrence = "backend-recovery-occurrence"; - let backend = ContentBackend::new("unused"); - - let first = evaluate_and_record_scheduler_skip( - &dashboard_root, - &config, - backend.executable(), - &job, - occurrence, - None, - ) - .await - .unwrap() - .expect("a due job still requires a configured backend"); - assert_eq!(first.report["reason"], json!("backend_unconfigured")); - assert_eq!(first.ledger_record.status, AutomationRunStatus::Skipped); - assert_ne!(first.run_id, occurrence); - let repeated = evaluate_and_record_scheduler_skip( - &dashboard_root, - &config, - backend.executable(), - &job, - occurrence, - None, - ) - .await - .unwrap() - .unwrap(); - assert_eq!(repeated.ledger_record, first.ledger_record); - - let executable = tempfile::NamedTempFile::new().unwrap(); - let backend = backend.with_executable(executable.path()); - assert!( - evaluate_and_record_scheduler_skip( - &dashboard_root, - &config, - backend.executable(), - &job, - occurrence, - None, - ) - .await - .unwrap() - .is_none(), - "binding the backend must admit the same due occurrence without cooldown" - ); - assert_eq!(backend.calls(), 0); - assert_eq!( - load_run_records(&dashboard_root, 10).await.unwrap(), - vec![first.ledger_record], - "advisory admission must not append an effectful terminal" - ); - assert!( - tracedecay_automation_runtime::automation::run_ledger::find_run_record_exact_bounded( - &dashboard_root, - occurrence, - ) - .await - .unwrap() - .is_none() - ); } #[tokio::test] @@ -1287,22 +1206,15 @@ async fn scheduler_prefilter_live_lock_wins_over_not_due_summary() { let config = enabled_job_config(); let occurrence = "locked-prefilter-occurrence"; - let executable = tempfile::NamedTempFile::new().unwrap(); - let backend = ContentBackend::new("unused").with_executable(executable.path()); - let prefilter = evaluate_and_record_scheduler_skip( - &dashboard_root, - &config, - backend.executable(), - &job, - occurrence, - None, - ) - .await - .unwrap() - .expect("the live job lock must produce a diagnostic"); + let prefilter = + evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) + .await + .unwrap() + .expect("the live job lock must produce a diagnostic"); assert_eq!(prefilter.report["reason"], json!("scheduler_lock_active")); assert_ne!(prefilter.run_id, occurrence); + let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, @@ -1350,21 +1262,12 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { job.schedule = Some("every 1h".to_owned()); let config = enabled_job_config(); let occurrence = "prefilter-gap-occurrence"; - let executable = tempfile::NamedTempFile::new().unwrap(); - let backend = ContentBackend::new("unused").with_executable(executable.path()); assert!( - evaluate_and_record_scheduler_skip( - &dashboard_root, - &config, - backend.executable(), - &job, - occurrence, - None, - ) - .await - .unwrap() - .is_none(), + evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) + .await + .unwrap() + .is_none(), "a new interval job is advisory-due before effect admission" ); let _raced_in = @@ -1373,6 +1276,7 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { .unwrap() .expect("the advisory prefilter must release its reservation lock"); + let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root,