From 5fee01e6f64b5264196347c789b2c6595fee82fe Mon Sep 17 00:00:00 2001 From: "zackary.l.jackson" Date: Tue, 20 Oct 2026 20:30:43 +0000 Subject: [PATCH 1/3] fix(session-temporal): refuse refresh superseded by a session reset A source rewrite that retires an observation while a refresh is running requests a session reset. Completion then failed with a retryable storage error (frontier mismatch), so the scheduler retried forever and the reset, which only applies once no refresh runs, never did. Completion now refuses such a candidate as a terminal state transition so the scheduler fails it durably and the next refresh applies the reset. Fixes #4373 Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../src/refresh.rs | 9 ++ .../temporal_projection/activation.rs | 86 +++++++++++++++++++ 2 files changed, 95 insertions(+) diff --git a/crates/tracedecay-session-temporal-store/src/refresh.rs b/crates/tracedecay-session-temporal-store/src/refresh.rs index e26976a0b7..3b6ae3fad2 100644 --- a/crates/tracedecay-session-temporal-store/src/refresh.rs +++ b/crates/tracedecay-session-temporal-store/src/refresh.rs @@ -3225,6 +3225,15 @@ async fn complete_session_refresh_in_transaction( COMPLETE_REFRESH, ) .await?; + // A begin applies any pending reset, so one pending now arrived while + // this refresh ran: its candidate covers effects the reset retired and can + // never match the live frontier. Refusing it terminally ends the operation + // so the next begin can apply the reset. + if session_reset_is_pending(transaction, request.session_id()).await? { + return Err(SessionStoreError::InvalidStateTransition { + context: "refresh candidate was superseded by a requested session reset", + }); + } let progress = require_exact_terminal_progress( transaction, request.session_id(), diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs index 72782a6be3..4f81293b0a 100644 --- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs +++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs @@ -315,6 +315,92 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() { ); } +/// A source rewrite that retires an observation a running refresh already +/// projected asks for a session reset. That refresh can never prove its +/// frontier again, so completion refuses it as a terminal state transition +/// (never a retryable storage fault that would block the reset forever), and +/// the next refresh applies the reset and activates the rewritten history. +#[tokio::test] +async fn reset_requested_during_a_refresh_supersedes_its_candidate() { + let tmp = TempDir::new().unwrap(); + let (runtime, path, stale) = frontier_digest_fixture(&tmp, true).await; + let session_id = stale.session_id().clone(); + rusqlite::Connection::open(&path) + .unwrap() + .execute( + "INSERT INTO observation_projection_dispositions + (projector_version, observation_id, receipt_id, reason) + SELECT ?1, observation_id, receipt_id, ?2 + FROM observations + WHERE observation_id = ( + SELECT observation_id FROM session_temporal_observation_effects + WHERE session_id = ?3 AND observation_sequence = 2 + )", + rusqlite::params![ + tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION, + tracedecay_store::ProjectionSkipReason::SourceRecordRetired.as_str(), + session_id.as_str(), + ], + ) + .unwrap(); + // The reset request `request_session_temporal_reset` records for a + // retired observation that produced session output. + let requested = rusqlite::Connection::open(&path) + .unwrap() + .execute( + "INSERT INTO session_temporal_resets (session_id, requested_at) + SELECT effect.session_id, unixepoch() * 1000000 + FROM session_temporal_observation_effects AS effect + JOIN observation_projection_dispositions AS retired + ON retired.observation_id = effect.observation_id + WHERE effect.output_count > 0", + [], + ) + .unwrap(); + assert_eq!(requested, 1, "the retired observation requests one reset"); + + let store = runtime + .session_temporal_store(HostAdmissionScope::Profile) + .unwrap(); + let refused = complete_candidate(&store, &stale).await; + assert!( + matches!( + refused, + Err(SessionStoreError::InvalidStateTransition { + context: "refresh candidate was superseded by a requested session reset" + }) + ), + "{refused:?}" + ); + assert_eq!(active_generations(&path).await, vec!["1:active"]); + + fail_candidate(&store, &stale).await; + let rebuilt = begin_candidate(&store, &session_id, 2).await; + assert_ne!(rebuilt.operation_id(), stale.operation_id()); + let first = occurrence(&session_id, &observation(&session_id, 0, "one")); + persist_batch( + &store, + &rebuilt, + batch(&rebuilt, vec![first], vec![], vec![]), + ) + .await + .unwrap(); + complete_candidate(&store, &rebuilt).await.unwrap(); + assert_eq!( + active_generations(&path).await, + vec![format!("{}:active", rebuilt.candidate_generation().value())] + ); + assert_eq!( + rows( + &path, + "SELECT CAST(reset_generation IS NOT NULL AND requested_at IS NULL AS TEXT) + FROM session_temporal_resets" + ) + .await, + vec!["1"] + ); +} + /// Persists observations `one` and `two` in a fresh profile and projects the /// refresh candidate through the frozen frontier of 2: both occurrences when /// `complete`, otherwise only `one`. From 6c5870b1d2b1a9b0dc28648f3c8aebf4146af7df Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 04:53:24 -0700 Subject: [PATCH 2/3] fix(daemon): join warming opens before identity resolution --- crates/tracedecay/src/daemon/engine.rs | 13 ++ .../src/daemon/tests/route_discovery.rs | 118 ++++++++++++++++++ 2 files changed, 131 insertions(+) diff --git a/crates/tracedecay/src/daemon/engine.rs b/crates/tracedecay/src/daemon/engine.rs index 6c97fd1d06..213d5319ac 100644 --- a/crates/tracedecay/src/daemon/engine.rs +++ b/crates/tracedecay/src/daemon/engine.rs @@ -9,6 +9,8 @@ use std::collections::HashSet; #[cfg(unix)] use tracedecay_code_index_runtime::{GitWatchSyncConfigV1, git_watch}; #[cfg(unix)] +use tracedecay_contracts::project_open::ProjectOpenStatusStateV1; +#[cfg(unix)] use tracedecay_daemon_identity::profile_identity; #[cfg(unix)] use tracedecay_daemon_protocol::{client_version_skew, version_skew_action}; @@ -468,6 +470,17 @@ impl DaemonEngine { } self.ensure_registered_project_route(&project_path, handshake.allow_init) .await?; + // The admitted open owns identity resolution for this exact route. + // Re-probing on a warming retry can contend with that open on a + // non-Git root, whose absent topology is intentionally not retained. + // Join its publication; terminal tasks still use normal resolution. + if project_open_tasks(&self.project_open_gates) + .await + .status(&route) + .is_some_and(|status| status.state == ProjectOpenStatusStateV1::Converging) + { + return Ok(None); + } let Some(key) = resolved_project_server_key(&self.store_administration, &project_path, handshake) .await? diff --git a/crates/tracedecay/src/daemon/tests/route_discovery.rs b/crates/tracedecay/src/daemon/tests/route_discovery.rs index f616fc965d..35f3bace27 100644 --- a/crates/tracedecay/src/daemon/tests/route_discovery.rs +++ b/crates/tracedecay/src/daemon/tests/route_discovery.rs @@ -225,6 +225,124 @@ async fn concurrent_routes_resolve_one_repository_topology() { ); } +/// A warming request joins the exact route's admitted open without starting +/// another topology walk that could make the open's no-wait probe refuse it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn warming_route_joins_its_open_without_rediscovering_repository() { + let home = TempDir::new().expect("isolated home"); + let profile_root = home.path().join("profile"); + let project = home.path().join("project"); + std::fs::create_dir_all(&project).expect("non-Git project"); + let client_identity = test_client_identity_for(profile_root.clone()); + initialize_test_project(&project, &client_identity).await; + let _database_scope = + enter_test_daemon_database_scope(&profile_root, "warming route discovery"); + let engine = test_daemon_engine_for_profile(&profile_root); + let handshake = handshake_for(&project, &profile_root); + let (_, route) = engine.project_route(&handshake).expect("registered route"); + let tasks = crate::daemon::project_open_tasks(&engine.project_open_gates).await; + let (release, blocked) = tokio::sync::oneshot::channel(); + let state = match tasks.start_cancellable(route, move |cancellation| async move { + tokio::select! { + biased; + () = cancellation.cancelled() => { + Err(crate::daemon::project_open_cancellation_error()) + } + opened = blocked => { + opened.expect("release controlled open"); + Err(tracedecay_domain::errors::TraceDecayError::project_open( + tracedecay_domain::errors::ProjectOpenFailureKind::IndexMissing, + "the controlled open refused a missing index", + )) + } + } + }) { + crate::daemon::ProjectOpenTaskClaim::InFlight(state) => state, + crate::daemon::ProjectOpenTaskClaim::Failed(_) + | crate::daemon::ProjectOpenTaskClaim::Saturated => { + panic!("the controlled open must own its route"); + } + }; + + observe_repository_discovery_for_test(&project); + let other_scope = DaemonHandshake { + scope_prefix: Some("another-scope".to_owned()), + ..handshake.clone() + }; + assert!( + engine + .cached_project_server(&other_scope) + .await + .expect("another scope remains a separate route") + .is_none() + ); + assert!( + repository_topology_resolution_count_for_test(&project) > 0, + "a pending open must not suppress identity resolution for another route" + ); + + observe_repository_discovery_for_test(&project); + let mut joins = tasks.joins(); + let next_join = *joins.borrow_and_update() + 1; + let request = engine + .project_server_for_request(&handshake, crate::daemon::ProjectServerRequirement::Core); + tokio::pin!(request); + tokio::select! { + biased; + joined = joins.wait_for(|joins| *joins >= next_join) => { + joined.expect("request must join the controlled open"); + } + answered = &mut request => panic!( + "request answered before joining its open: {:?}", + answered.err() + ), + } + assert_eq!( + repository_topology_resolution_count_for_test(&project), + 0, + "a warming retry must not start a competing repository walk" + ); + assert_eq!( + engine + .project_open_attempts + .load(std::sync::atomic::Ordering::Relaxed), + 0, + "the request must join instead of starting another project open" + ); + + release.send(()).expect("release controlled open"); + let recorded = crate::daemon::ProjectOpenTasks::wait_for_completion(state) + .await + .expect_err("the controlled open must publish its failure"); + let refusal = match request.await { + Ok(_) => panic!("a failed open must not publish a server"), + Err(error) => error, + }; + assert_eq!( + refusal.project_open_failure_kind(), + Some(tracedecay_domain::errors::ProjectOpenFailureKind::IndexMissing), + "the joined request must preserve the open's typed refusal" + ); + assert_eq!(refusal.to_string(), recorded.to_string()); + + assert!(tasks.shutdown().await, "the controlled open must retire"); + assert!( + engine + .cached_project_server(&handshake) + .await + .expect("registered route remains valid") + .is_none(), + "a failed open must leave no mounted server" + ); + let resolutions = repository_topology_resolution_count_for_test(&project); + reset_repository_discovery_for_test(&project); + assert!( + resolutions > 0, + "a retired open must no longer suppress normal identity resolution" + ); + engine.shutdown_all().await; +} + /// A discovery that outlives its budget is deferred, never a hang and never /// "this project is not enrolled". #[tokio::test(flavor = "multi_thread", worker_threads = 2)] From b98a83c10daa0f2e104c9e7a7daa77f223a9d80c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Wed, 21 Oct 2026 10:33:06 -0700 Subject: [PATCH 3/3] test(cli): use canonical daemon socket paths in host journeys --- .../sweep_outcomes.rs | 91 +++++++++++++++++-- 1 file changed, 81 insertions(+), 10 deletions(-) diff --git a/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs b/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs index e447a37183..54e7188487 100644 --- a/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs +++ b/crates/tracedecay-cli/tests/host_journeys_suite/host_lifecycle_cli_acceptance/sweep_outcomes.rs @@ -3,10 +3,16 @@ use std::fs; use std::path::Path; +#[cfg(unix)] +use std::path::PathBuf; use std::process::Output; +#[cfg(unix)] +use std::process::{Child, Stdio}; use tracedecay_agent_hosts::agents::host_bundle::{HostComponentV1, HostKindV1}; #[cfg(unix)] +use tracedecay_daemon_control::{SOCKET_ENV, default_socket_path}; +#[cfg(unix)] use tracedecay_runtime_core::test_executable::write_executable_script; #[cfg(unix)] @@ -376,24 +382,61 @@ fn a_signed_out_kiro_cli_never_blocks_the_kiro_lifecycle() { /// canonical report from the sole daemon owner and counts its absence as an /// issue. #[cfg(unix)] -pub(super) struct ProfileDaemon(std::process::Child); +pub(super) struct ProfileDaemon { + child: Option, + owned_socket_directory: Option, +} #[cfg(unix)] impl ProfileDaemon { pub(super) fn start(cli: &IsolatedCli) -> Self { + // The daemon relocates sockets that exceed the platform path limit. + // Readiness must use that same authority, including on remote workers + // whose TMPDIR makes an otherwise ordinary profile path too long. + let socket = default_socket_path(&cli.profile).unwrap(); + let mut daemon = Self { + child: None, + owned_socket_directory: None, + }; + let socket_override = std::env::var_os(SOCKET_ENV).filter(|path| !path.is_empty()); + if socket_override.is_none() && socket != cli.profile.join("daemon.sock") { + let parent = socket.parent().unwrap().to_path_buf(); + // Claim only a freshly created directory. Earlier fixture work + // may already own a private parent, which this guard must retain. + match tracedecay_private_fs::create_private_directory(&parent) { + Ok(()) => daemon.owned_socket_directory = Some(parent), + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + tracedecay_private_fs::validate_private_directory(&parent).unwrap(); + } + Err(error) => panic!("could not create fixture socket directory: {error}"), + } + } + let log_path = cli.home.path().join("profile-daemon.stderr"); + let stderr_log = fs::File::create(&log_path).unwrap(); let mut command = cli.command(&["daemon", "run"]); command .env("TRACEDECAY_TEST_ALLOW_INCOMPLETE_HOLDER_SCAN", "1") - .stdout(std::process::Stdio::null()) - .stderr(std::process::Stdio::null()); - let daemon = Self(command.spawn().unwrap()); - let socket = cli.profile.join("daemon.sock"); + .stdout(Stdio::null()) + .stderr(stderr_log); + daemon.child = Some(command.spawn().unwrap()); + let stderr = || { + fs::read_to_string(&log_path) + .unwrap_or_else(|error| format!("could not read daemon stderr: {error}")) + }; let deadline = std::time::Instant::now() + std::time::Duration::from_secs(30); - while std::os::unix::net::UnixStream::connect(&socket).is_err() { + while let Err(error) = std::os::unix::net::UnixStream::connect(&socket) { + if let Some(status) = daemon.child.as_mut().unwrap().try_wait().unwrap() { + panic!( + "profile daemon exited before listening at {}: {status}; stderr:\n{}", + socket.display(), + stderr() + ); + } assert!( std::time::Instant::now() < deadline, - "the profile daemon never listened at {}", - socket.display() + "the profile daemon never listened at {}: {error}; stderr:\n{}", + socket.display(), + stderr() ); std::thread::sleep(std::time::Duration::from_millis(25)); } @@ -404,11 +447,39 @@ impl ProfileDaemon { #[cfg(unix)] impl Drop for ProfileDaemon { fn drop(&mut self) { - let _ = self.0.kill(); - let _ = self.0.wait(); + if let Some(child) = self.child.as_mut() { + let _ = child.kill(); + if child.wait().is_err() { + return; + } + } + if let Some(directory) = self.owned_socket_directory.take() { + let _ = fs::remove_dir_all(directory); + } } } +#[cfg(unix)] +#[test] +fn doctor_reaches_the_canonical_daemon_socket_for_a_long_profile_path() { + let mut cli = IsolatedCli::new(); + cli.profile = cli.home.path().join("p".repeat(120)).join("profile"); + install_cline(&cli); + let socket = default_socket_path(&cli.profile).unwrap(); + assert_ne!(socket, cli.profile.join("daemon.sock")); + + { + let _daemon = ProfileDaemon::start(&cli); + assert_success("profile", "doctor", cli.run(&["doctor"])); + } + + assert!( + !socket.parent().unwrap().exists(), + "the stopped fixture must remove its owned fallback socket directory" + ); + assert!(cli.profile.join("config.toml").exists()); +} + /// Doctor classifies hosts the way `update-plugin` does: Kimi's pending /// `/plugins install` exits 75, a tracked host without its CLI is skipped as /// not installed, and the run converges to 0 once the operator installs