Fix flaky agent-crate tests and five cloud e2e tests

crates/freshell-freshagent/src/claude.rs

@@ -13950,6 +13950,40 @@ rl.on('line', (line) => {
         }
     }
 
+    /// Wait until the fresh create's adoption has COMMITTED ownership under
+    /// `durable_id`. The adoption publishes the `cli_index` alias and then
+    /// writes the identity binding BEFORE its commit (ext r27 F4: a failed
+    /// binding abandons it pre-commit), and the commit-to-Live plus the
+    /// retained release stamp are one critical section under the stamps
+    /// lock (ext r30 F1) — so neither the alias nor `was_recorded` proves
+    /// the commit: both can be observed while the claim is still Starting
+    /// and the stamp is absent. The stamp is the real condition.
+    async fn await_create_ownership_committed(st: &FreshClaudeState, durable_id: &str) {
+        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
+        while crate::ownership_lane::peek_retained_stamp(&st.ownership_stamps, durable_id).is_none()
+        {
+            assert!(
+                tokio::time::Instant::now() < deadline,
+                "the create's adoption never committed ownership (no retained stamp)"
+            );
+            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
+        }
+    }
+
+    /// [`await_create_ownership_committed`] for a create with a RECORDABLE
+    /// binding, then pins the binding-first order.
+    async fn await_create_adoption_committed(
+        st: &FreshClaudeState,
+        sink: &crate::identity_sink::FakeIdentitySink,
+        durable_id: &str,
+    ) {
+        await_create_ownership_committed(st, durable_id).await;
+        assert!(
+            sink.was_recorded("claude", durable_id),
+            "the adoption records the binding before it commits ownership"
+        );
+    }
+
     /// b8ke delta round-3 F1: a coordinator-REFUSED kill mutates NOTHING
     /// durable — the pane-ledger row stays BOUND, the session stays live,
     /// and the refusal broadcast is TYPED (code + message, never
@@ -14051,20 +14085,7 @@ rl.on('line', (line) => {
             .await;
         let created = await_claude_created(&mut rx, "req-e3r1-f1").await;
         let placeholder = created["sessionId"].as_str().unwrap().to_string();
-        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
-        while st
-            .cli_index
-            .lock()
-            .await
-            .get(FRESH_CREATE_DURABLE_ID)
-            .is_none()
-        {
-            assert!(
-                tokio::time::Instant::now() < deadline,
-                "the create's adoption never published"
-            );
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
+        await_create_ownership_committed(&st, FRESH_CREATE_DURABLE_ID).await;
 
         // THE COMPLETED HANDOFF (simulated at the coordinator level): the
         // stamp is consumed and a TERMINAL owner is committed under the
@@ -14134,14 +14155,7 @@ rl.on('line', (line) => {
         st.handle_create(create, None).await;
         let created = await_claude_created(&mut rx, "req-e3r1-f2").await;
         let placeholder = created["sessionId"].as_str().unwrap().to_string();
-        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
-        while !sink.was_recorded("claude", FRESH_CREATE_DURABLE_ID) {
-            assert!(
-                tokio::time::Instant::now() < deadline,
-                "the create's adoption never recorded the binding"
-            );
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
+        await_create_adoption_committed(&st, &sink, FRESH_CREATE_DURABLE_ID).await;
         assert!(
             crate::ownership_lane::peek_retained_stamp(
                 &st.ownership_stamps,
@@ -14218,14 +14232,7 @@ rl.on('line', (line) => {
         st.handle_create(create, None).await;
         let created = await_claude_created(&mut rx, "req-e3r3-f8").await;
         let placeholder = created["sessionId"].as_str().unwrap().to_string();
-        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
-        while !sink.was_recorded("claude", FRESH_CREATE_DURABLE_ID) {
-            assert!(
-                tokio::time::Instant::now() < deadline,
-                "the create's adoption never recorded the binding"
-            );
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
+        await_create_adoption_committed(&st, &sink, FRESH_CREATE_DURABLE_ID).await;
         let stamp = crate::ownership_lane::peek_retained_stamp(
             &st.ownership_stamps,
             FRESH_CREATE_DURABLE_ID,
@@ -14329,14 +14336,7 @@ rl.on('line', (line) => {
         st.handle_create(create, None).await;
         let created = await_claude_created(&mut rx, "req-e3r4-f1").await;
         let placeholder = created["sessionId"].as_str().unwrap().to_string();
-        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
-        while !sink.was_recorded("claude", FRESH_CREATE_DURABLE_ID) {
-            assert!(
-                tokio::time::Instant::now() < deadline,
-                "the create's adoption never recorded the binding"
-            );
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
+        await_create_adoption_committed(&st, &sink, FRESH_CREATE_DURABLE_ID).await;
 
         // THE NO-STAMP SHAPE: consume the retained stamp (the handoff's
         // kill_for_handoff consumption) — the kill runs the no-stamp
@@ -14444,14 +14444,7 @@ rl.on('line', (line) => {
         st.handle_create(create, None).await;
         let created = await_claude_created(&mut rx, "req-e3r2-f1").await;
         let placeholder = created["sessionId"].as_str().unwrap().to_string();
-        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
-        while !sink.was_recorded("claude", FRESH_CREATE_DURABLE_ID) {
-            assert!(
-                tokio::time::Instant::now() < deadline,
-                "the create's adoption never recorded the binding"
-            );
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
+        await_create_adoption_committed(&st, &sink, FRESH_CREATE_DURABLE_ID).await;
 
         // THE MID-HANDOFF WINDOW: kill_for_handoff consumed the stamp
         // while the coordinator record stays Handoff.
@@ -14607,14 +14600,7 @@ rl.on('line', (line) => {
         st.handle_create(create, None).await;
         let created = await_claude_created(&mut rx, "req-e3r2-f4").await;
         let placeholder = created["sessionId"].as_str().unwrap().to_string();
-        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
-        while !sink.was_recorded("claude", FRESH_CREATE_DURABLE_ID) {
-            assert!(
-                tokio::time::Instant::now() < deadline,
-                "the create's adoption never recorded the binding"
-            );
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
+        await_create_adoption_committed(&st, &sink, FRESH_CREATE_DURABLE_ID).await;
 
         // THE RUNTIME EXITS (the sidecar's recorded pid is killed outright
         // — it is dead by the time the kill's unwind verdict runs).
@@ -14691,20 +14677,7 @@ rl.on('line', (line) => {
             .await;
         let created = await_claude_created(&mut rx, "req-e3r1-f3").await;
         let _placeholder = created["sessionId"].as_str().unwrap().to_string();
-        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(15);
-        while st
-            .cli_index
-            .lock()
-            .await
-            .get(FRESH_CREATE_DURABLE_ID)
-            .is_none()
-        {
-            assert!(
-                tokio::time::Instant::now() < deadline,
-                "the create's adoption never published"
-            );
-            tokio::time::sleep(std::time::Duration::from_millis(10)).await;
-        }
+        await_create_ownership_committed(&st, FRESH_CREATE_DURABLE_ID).await;
 
         // THE UNOWNED-LIVE shape: the coordinator key goes Vacant (a
         // confirmed-kill release) while the lane map still holds the live
@@ -16699,6 +16672,10 @@ rl.on('line', (line) => {
             .await;
         let created = await_claude_created(&mut rx, "req-postcap-f3").await;
         let _placeholder = created["sessionId"].as_str().unwrap().to_string();
+        // The first create's adoption must COMMIT before the pre-commit
+        // park arms — otherwise it parks there too, and both the `reached`
+        // flag and the stale-start recovery below see the WRONG claim.
+        await_create_ownership_committed(&st, FRESH_CREATE_DURABLE_ID).await;
 
         // The pre-commit park: a SECOND session (fresh map key — the
         // ordinary-create shape, no prior durable id) whose adoption

crates/freshell-freshagent/src/claude_snapshot.rs

@@ -2732,11 +2732,15 @@ pub(crate) mod tests {
     /// under the `CLAUDE_CONFIG_DIR` candidate, and agrees with
     /// [`locate_transcript`] on the same ordered candidates (one definition
     /// of "the transcript exists"). Env vars are process-global — serialize
-    /// under the shared claude env lock.
+    /// under the shared claude env lock. Only the claude-specific roots are
+    /// cleared: `HOME` is never touched, because tests that do not take the
+    /// claude lock read it concurrently (the OpenCode home harness asserts
+    /// the parent's `HOME` is unchanged) and the platform home candidate
+    /// resolves through the passwd entry without it anyway.
     #[tokio::test]
     async fn locate_transcript_selected_answers_root_path_and_cwd_under_config_dir() {
         let _guard = crate::claude::tests::CLAUDE_ENV_LOCK.lock().await;
-        let _restore = EnvVarsRestore::remove_all(&["CLAUDE_CONFIG_DIR", "CLAUDE_HOME", "HOME"]);
+        let _restore = EnvVarsRestore::remove_all(&["CLAUDE_CONFIG_DIR", "CLAUDE_HOME"]);
         let store = temp_home();
         std::env::set_var("CLAUDE_CONFIG_DIR", store.path());
         let id = "66666666-6666-4666-8666-666666666666";

crates/freshell-freshagent/src/lib.rs

@@ -2221,10 +2221,11 @@ pub struct FreshAgentState {
     /// b8ke focused round-4 R4-2 test seam: when set, the REST opencode
     /// drive PARKS at the exact interleaving point under review — after
     /// the gate-held ownership/condemned re-check, BEFORE the prompt POST
-    /// — until the Notify fires. Never armed in production. Installed via
+    /// — until released (see [`RestTurnTestPause`]). Never armed in
+    /// production. Installed via
     /// [`Self::set_rest_turn_test_pause_for_test`].
     #[cfg(test)]
-    rest_turn_test_pause: Arc<Mutex<Option<Arc<tokio::sync::Notify>>>>,
+    rest_turn_test_pause: Arc<Mutex<Option<Arc<RestTurnTestPause>>>>,
     /// The ONE server-wide runtime-ownership coordinator (kata b8ke Task 3),
     /// wired from freshell-server::main next to `fresh_agent_leases`.
     /// `None` (every pre-existing test) = the lane skips coordinator
@@ -2323,6 +2324,24 @@ pub(crate) struct RestOpencodeTurn {
     pub(crate) route: Option<String>,
 }
 
+/// b8ke focused round-4 R4-2 test seam (cfg(test) only): the handshake a
+/// test uses to hold the REST opencode drive at the interleaving point
+/// under review and to observe a lifecycle quiesce queueing behind it.
+/// Every signal is `notify_one` (a stored permit), so a waiter that
+/// arrives late still sees it — no step depends on scheduling order.
+#[cfg(test)]
+#[derive(Default)]
+pub(crate) struct RestTurnTestPause {
+    /// Signalled by the drive once it is parked — the gate held, the
+    /// ownership/condemned re-check done, the prompt POST not issued.
+    pub(crate) parked: tokio::sync::Notify,
+    /// Awaited by the parked drive before it issues its prompt POST.
+    pub(crate) release: tokio::sync::Notify,
+    /// Signalled by a lifecycle quiesce that found the session's
+    /// dispatch/condemn gate held and is about to wait on it.
+    pub(crate) quiesce_blocked: tokio::sync::Notify,
+}
+
 /// A Slice-1 terminal pane's record: just enough to dispatch `send-keys` /
 /// `capture` / `wait-for` to the right `terminal_id` in the shared registry.
 #[derive(Clone)]
@@ -3113,16 +3132,31 @@ impl FreshAgentState {
     /// b8ke focused round-4 R4-2 test seam installer: see
     /// [`Self::rest_turn_test_pause`].
     #[cfg(test)]
-    pub(crate) fn set_rest_turn_test_pause_for_test(
-        &self,
-        pause: Option<Arc<tokio::sync::Notify>>,
-    ) {
+    pub(crate) fn set_rest_turn_test_pause_for_test(&self, pause: Option<Arc<RestTurnTestPause>>) {
         *self
             .rest_turn_test_pause
             .lock()
             .expect("rest_turn_test_pause mutex") = pause;
     }
 
+    /// R4-2 test seam probe, called by the lifecycle quiesce paths right
+    /// before they wait on the session's dispatch/condemn gate: when the
+    /// seam is installed and the gate is held (the parked drive owns it),
+    /// signal [`RestTurnTestPause::quiesce_blocked`].
+    #[cfg(test)]
+    pub(crate) fn note_rest_turn_gate_wait_for_test(&self, gate: &tokio::sync::Mutex<()>) {
+        let pause = self
+            .rest_turn_test_pause
+            .lock()
+            .expect("rest_turn_test_pause mutex")
+            .clone();
+        if let Some(pause) = pause {
+            if gate.try_lock().is_err() {
+                pause.quiesce_blocked.notify_one();
+            }
+        }
+    }
+
     /// The REST resume probe's bounded `get_session` budget, in milliseconds:
     /// cfg(test) state override > `FRESHELL_OPENCODE_GET_SESSION_TIMEOUT_MS`
     /// env parse > 10_000ms default — resolved purely by
@@ -6845,7 +6879,8 @@ async fn send_keys(
             .expect("rest_turn_test_pause mutex")
             .clone();
         if let Some(pause) = pause {
-            pause.notified().await;
+            pause.parked.notify_one();
+            pause.release.notified().await;
         }
     }
     let rx = manager.subscribe(&durable_id);