fix(session-temporal): refuse refresh superseded by a session reset

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_daemon_identity::current_daemon_connection;
 #[cfg(unix)]
 use tracedecay_daemon_protocol::DaemonEndpoint;
@@ -382,20 +388,50 @@ 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<Child>,
+    owned_socket_directory: Option<PathBuf>,
+}
 
 #[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());
+            .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);
-        // Deep worker paths can move the socket outside the profile. Follow
-        // the same profile-owned authority that production clients discover.
+        // Follow the exact isolated profile's published endpoint, retaining
+        // child ownership and startup diagnostics if admission fails.
         while !current_daemon_connection(&cli.profile).is_ok_and(|connection| {
             match connection.endpoint() {
                 DaemonEndpoint::Unix(socket) => {
@@ -404,10 +440,18 @@ impl ProfileDaemon {
                 DaemonEndpoint::Loopback(address) => std::net::TcpStream::connect(address).is_ok(),
             }
         }) {
+            if let Some(status) = daemon.child.as_mut().unwrap().try_wait().unwrap() {
+                panic!(
+                    "profile daemon exited before listening for {}: {status}; stderr:\n{}",
+                    cli.profile.display(),
+                    stderr()
+                );
+            }
             assert!(
                 std::time::Instant::now() < deadline,
-                "the registered daemon never listened for profile {}",
-                cli.profile.display()
+                "the registered daemon never listened for profile {}; stderr:\n{}",
+                cli.profile.display(),
+                stderr()
             );
             std::thread::sleep(std::time::Duration::from_millis(25));
         }
@@ -418,9 +462,37 @@ 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

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(),

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?

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)]

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`.