fix: preserve test completion and fixture invariants under concurrency

crates/worker-runtime/src/worker_source.rs

@@ -1816,10 +1816,12 @@ mod tests {
     }
 
     #[test]
-    fn runtime_owned_workspace_client_has_no_fixed_request_timeout() {
+    fn runtime_owned_workspace_client_returns_successful_response_without_request_timeout() {
-        // This must exceed reqwest::blocking::Client's default 30-second timeout.
+        // Effective deadline clearing is exercised with a short inherited deadline in
-        // A short delay only checks our Option field, not the effective HTTP deadline.
+        // workspace_request::tests::blocking_workspace_client_clears_initial_request_timeout.
-        let (base_url, server) = delayed_workspace_response(Duration::from_secs(31));
+        let (base_url, server) = binary_loopback(
+            b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}".to_vec(),
+        );
         let client =
             RuntimeOwnedWorkspaceClient::new("workspace-a", base_url, "runtime-a", "worker-a");
         assert_eq!(client.request_timeout, None);

crates/worker-runtime/src/working_directory/cleanup.rs

@@ -1223,6 +1223,7 @@ pub(super) mod tests {
     }
 
     fn sized_fixture(
+        repo: &Path,
         size: u64,
     ) -> (
         tempfile::TempDir,
@@ -1230,7 +1231,9 @@ pub(super) mod tests {
         WorkingDirectoryBinding,
         BodyWatch,
     ) {
-        let (runtime, materializer, binding) = fixture();
+        let runtime = tempfile::tempdir().unwrap();
+        let materializer = RuntimeGitMaterializer::new(runtime.path());
+        let binding = materializer.create(&request(repo)).unwrap();
         fs::write(binding.root.join(".git/info/exclude"), "target/\n").unwrap();
         fs::create_dir(binding.root.join("target")).unwrap();
         // Entry count remains constant, including .git and many ignored files.
@@ -1256,10 +1259,14 @@ pub(super) mod tests {
     #[test]
     fn actual_cleanup_and_restart_retry_never_open_bodies_when_only_file_sizes_grow() {
         use io_observation::observe;
+        // Vary only file sizes, not Git object names/layout. Separate commits
+        // can have different object-prefix directory counts as timestamps change.
+        // Each case still owns an independent materialized checkout and Runtime.
+        let repo = create_clean_repo();
         for failure in [None, Some(0), Some(10)] {
             let mut small_counts = None;
             for size in [256, 128 * 1024 * 1024] {
-                let (runtime, materializer, binding, mut watch) = sized_fixture(size);
+                let (runtime, materializer, binding, mut watch) = sized_fixture(repo.path(), size);
                 let id = &binding.working_directory.id;
                 let (first, first_io) =
                     observe(failure, || materializer.cleanup_working_directory(id));

crates/worker-runtime/src/workspace_request.rs

@@ -262,12 +262,14 @@ impl RuntimeWorkspaceRequestClient {
         request: RuntimeWorkspaceRequest,
     ) -> Result<RuntimeWorkspaceResponse, RuntimeWorkspaceRequestError> {
         let client = self.clone();
-        std::thread::spawn(move || client.execute_blocking_inner(request, true))
+        std::thread::spawn(move || {
-            .join()
+            client.execute_blocking_inner(request, true, reqwest::blocking::Client::builder())
-            .map_err(|_| RuntimeWorkspaceRequestError::Transport {
+        })
-                message: "Workspace request thread panicked".to_string(),
+        .join()
-                timeout: false,
+        .map_err(|_| RuntimeWorkspaceRequestError::Transport {
-            })?
+            message: "Workspace request thread panicked".to_string(),
+            timeout: false,
+        })?
     }
 
     /// Binary requests must retain their exact target and never follow redirects.
@@ -276,18 +278,21 @@ impl RuntimeWorkspaceRequestClient {
         request: RuntimeWorkspaceRequest,
     ) -> Result<RuntimeWorkspaceResponse, RuntimeWorkspaceRequestError> {
         let client = self.clone();
-        std::thread::spawn(move || client.execute_blocking_inner(request, false))
+        std::thread::spawn(move || {
-            .join()
+            client.execute_blocking_inner(request, false, reqwest::blocking::Client::builder())
-            .map_err(|_| RuntimeWorkspaceRequestError::Transport {
+        })
-                message: "Workspace binary request thread panicked".to_string(),
+        .join()
-                timeout: false,
+        .map_err(|_| RuntimeWorkspaceRequestError::Transport {
-            })?
+            message: "Workspace binary request thread panicked".to_string(),
+            timeout: false,
+        })?
     }
 
     fn execute_blocking_inner(
         &self,
         request: RuntimeWorkspaceRequest,
         follow_redirects: bool,
+        client_builder: reqwest::blocking::ClientBuilder,
     ) -> Result<RuntimeWorkspaceResponse, RuntimeWorkspaceRequestError> {
         let prepared = self.prepare(&request)?;
         if !follow_redirects {
@@ -304,7 +309,7 @@ impl RuntimeWorkspaceRequestClient {
         }
         // The blocking client defaults to 30 seconds. Apply None explicitly so
         // provider-owned command waits are not cut short by an HTTP deadline.
-        let mut client_builder = reqwest::blocking::Client::builder().timeout(request.timeout);
+        let mut client_builder = client_builder.timeout(request.timeout);
         if !follow_redirects {
             client_builder = client_builder.redirect(reqwest::redirect::Policy::none());
         }
@@ -425,6 +430,68 @@ mod tests {
     use super::*;
     use crate::auth::decode_runtime_request_source_claims;
 
+    // Component boundary: production request preparation, timeout policy and HTTP execution;
+    // only the initial client deadline and remote Workspace are supplied by the fixture.
+    #[tokio::test]
+    async fn blocking_workspace_client_clears_initial_request_timeout() {
+        use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+        tokio::time::timeout(Duration::from_secs(5), async {
+            let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+                .await
+                .unwrap();
+            let address = listener.local_addr().unwrap();
+            let initial_timeout = Duration::from_millis(20);
+            let mut request = tokio::task::spawn_blocking(move || {
+                let builder = reqwest::blocking::Client::builder()
+                    .no_proxy()
+                    .timeout(initial_timeout);
+                RuntimeWorkspaceRequestClient::new("workspace-a", format!("http://{address}"), "runtime-a")
+                    .execute_blocking_inner(
+                        RuntimeWorkspaceRequest {
+                            method: reqwest::Method::GET,
+                            path_and_query: "/api/test".into(),
+                            body: Vec::new(),
+                            headers: HeaderMap::new(),
+                            permission: "test".into(),
+                            worker_id: None,
+                            timeout: None,
+                            max_response_bytes: 1024,
+                        },
+                        true,
+                        builder,
+                    )
+            });
+
+            let receive_and_cross_deadline = async {
+                let (mut stream, _) = listener.accept().await.unwrap();
+                let mut headers = Vec::new();
+                while !headers.ends_with(b"\r\n\r\n") {
+                    headers.push(stream.read_u8().await.unwrap());
+                    assert!(headers.len() <= 8192, "unbounded request headers");
+                }
+                // Deliberately cross the inherited deadline after request arrival.
+                // This timer supplies the fault stimulus, not a completion signal.
+                tokio::time::sleep(initial_timeout * 2).await;
+                stream
+            };
+            let mut stream = tokio::select! {
+                result = &mut request => panic!("request ended before response: {result:?}"),
+                stream = receive_and_cross_deadline => stream,
+            };
+            stream
+                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 11\r\nConnection: close\r\n\r\n{\"ok\":true}")
+                .await
+                .unwrap();
+            drop(stream);
+            let response = request.await.unwrap().unwrap();
+            assert_eq!(response.status, reqwest::StatusCode::OK);
+            assert_eq!(response.body, br#"{"ok":true}"#);
+        })
+        .await
+        .expect("loopback request did not complete");
+    }
+
     #[test]
     fn route_issues_workspace_scoped_request_proof() {
         let identity = RuntimeIdentityMaterial::generate("runtime-a").unwrap();

crates/worker/tests/controller_test.rs

@@ -1017,14 +1017,45 @@ async fn journal_backend_grant_installs_bound_tools_and_seals_saved_finish_befor
             })
             .unwrap();
         let handle = spawn_controller(worker).await;
+        let mut events = handle.subscribe();
+        let submission_request_id = protocol::new_submission_request_id();
         handle
             .send(Method::submit_text(
-                protocol::new_submission_request_id(),
+                submission_request_id.clone(),
                 "Record only admitted public input.",
             ))
             .await
             .unwrap();
-        wait_for_status(&handle, WorkerStatus::Idle).await;
+        // Initial Idle (or the first captured model request) does not prove the
+        // submitted tool loop completed. The exact Submit teardown fence follows
+        // persist_turn, so coverage and the sealed result retry are committed.
+        // This is a deadlock watchdog, not a turn-latency performance assertion.
+        tokio::time::timeout(std::time::Duration::from_secs(2), async {
+            let mut saw_run_end = false;
+            loop {
+                match events
+                    .recv()
+                    .await
+                    .expect("Journal controller event stream closed")
+                {
+                    Event::RunEnd { result } => {
+                        assert_eq!(result, protocol::RunResult::Finished);
+                        saw_run_end = true;
+                    }
+                    Event::WorkerState { snapshot }
+                        if saw_run_end
+                            && snapshot.catalog_status() == WorkerStatus::Idle
+                            && snapshot.last_finished_submission_request_id.as_deref()
+                                == Some(submission_request_id.as_str()) =>
+                    {
+                        break;
+                    }
+                    _ => {}
+                }
+            }
+        })
+        .await
+        .expect("Journal Submit did not finish and commit its history");
         let request = wait_for_captured_request(&captured).await;
         let tools = request_tool_names(&request);
         assert_eq!(

crates/workspace-server/src/server/tests/protocol_backpressure_tests.rs

@@ -1,6 +1,15 @@
 use super::*;
 
+// Keep the short transport/session-lock liveness bounds separate from full
+// lifecycle cleanup. The 10s output/Close deadlines below are asserted exactly
+// with paused time; these wall-clock limits are deadlock watchdogs, not the
+// normal-suite performance budget.
 const PROTOCOL_TEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2);
+// Stop also runs Host admission, Runtime persistence and linked-Workdir storage
+// publication. The parallel normal run exceeded 2s there after transport had
+// already terminated. Check lock release separately at 2s, but allow this real
+// lifecycle path scheduling/I/O headroom without replacing it with a stub.
+const LIFECYCLE_CLEANUP_WATCHDOG: std::time::Duration = std::time::Duration::from_secs(15);
 
 // Completion replies are produced synchronously by the actual attached execution.
 // Unlike flooding the fixture's 32-entry broadcast bus, this reaches the Server's
@@ -289,8 +298,13 @@ async fn browser_method_queue_overflow_fails_fast_and_does_not_pin_session_clean
     );
     assert_eq!(receiver.len(), WORKER_PROTOCOL_METHOD_CAPACITY - 1);
 
+    let session_lock = current_worker_session_lock(&api, &identity);
+    let guard = tokio::time::timeout(PROTOCOL_TEST_TIMEOUT, session_lock.lock())
+        .await
+        .expect("full method queue retained the session guard");
+    drop(guard);
     let stopped = tokio::time::timeout(
-        PROTOCOL_TEST_TIMEOUT,
+        LIFECYCLE_CLEANUP_WATCHDOG,
         worker.stop(
             &WorkerOperationContext::Backend,
             WorkerLifecycleRequest {
@@ -636,6 +650,7 @@ fn assert_bounded_unknown_delivery_close(messages: &[WsMessage]) {
 }
 
 async fn assert_same_worker_reconnect_and_stop(
+    api: &WorkspaceApi,
     worker: &WorkspaceWorker,
     context: &WorkerOperationContext,
 ) {
@@ -649,8 +664,13 @@ async fn assert_same_worker_reconnect_and_stop(
             .expect("same Worker reconnect did not deliver a fresh snapshot"),
         Some(protocol::Event::Snapshot { .. })
     ));
+    let session_lock = current_worker_session_lock(api, worker.identity());
+    let guard = tokio::time::timeout(PROTOCOL_TEST_TIMEOUT, session_lock.lock())
+        .await
+        .expect("terminated transport retained same Worker's session guard");
+    drop(guard);
     let stopped = tokio::time::timeout(
-        PROTOCOL_TEST_TIMEOUT,
+        LIFECYCLE_CLEANUP_WATCHDOG,
         worker.stop(
             &WorkerOperationContext::Backend,
             WorkerLifecycleRequest {
@@ -780,7 +800,7 @@ async fn assert_stalled_browser_output_reclaimed(explicit_termination: bool) {
     // SQLite jobs run on external threads; paused Tokio time cannot observe their
     // progress and may auto-advance a reconnect timeout before SQL completes.
     tokio::time::resume();
-    assert_same_worker_reconnect_and_stop(&worker, &context).await;
+    assert_same_worker_reconnect_and_stop(&api, &worker, &context).await;
 }
 
 #[tokio::test(start_paused = true)]
@@ -930,5 +950,5 @@ async fn stalled_remote_bridge_browser_cancel_overflow_reclaims_wire_and_allows_
         ));
     }
     assert!(execution.protocol_methods().is_empty());
-    assert_same_worker_reconnect_and_stop(&worker, &context).await;
+    assert_same_worker_reconnect_and_stop(&api, &worker, &context).await;
 }

crates/workspace-server/src/server_assignment_attention_tests.rs

@@ -155,8 +155,17 @@ async fn orchestrator_attention_barrier_callers_share_unsent_reservation() {
 
 #[tokio::test]
 async fn orchestrator_attention_input_rechecks_reserved_noop_before_and_after_comparison() {
+    // This handshake holds a reservation across the real guarded Restore path
+    // (including durable intent/config/Runtime writes). Its release deadline is
+    // a deadlock watchdog, not a 5s Restore performance contract: the parallel
+    // normal run exceeded that old bound before it could release the caller.
+    let restore_watchdog = std::time::Duration::from_secs(15);
     for pause_before_comparison in [true, false] {
-        let (_dir, api, execution) = attention_test_api().await;
+        // This case only needs assigned Ticket metadata, not a checkout. Keep
+        // the real API/storage and configured target, but skip Git init/commit.
+        let _dir = tempfile::tempdir().unwrap();
+        let (api, execution) = test_api_with_recording_backend(_dir.path()).await;
+        attention_assign_ticket(&api, "Attention backlog");
         let started = start_attention_test_orchestrator(&api).await;
         let worker = RuntimeWorkerRef::new(
             EMBEDDED_WORKER_RUNTIME_ID,
@@ -192,8 +201,8 @@ async fn orchestrator_attention_input_rechecks_reserved_noop_before_and_after_co
                         if checks == 0 && pause_before_comparison {
                             entered_tx.send(()).unwrap();
                             release_rx
-                                .recv_timeout(std::time::Duration::from_secs(5))
+                                .recv_timeout(restore_watchdog)
-                                .unwrap();
+                                .expect("guarded Restore did not release the reserved no-op");
                         }
                         check_orchestrator_assignment_attention(
                             &api_for_caller,
@@ -210,8 +219,8 @@ async fn orchestrator_attention_input_rechecks_reserved_noop_before_and_after_co
                             );
                             entered_tx.send(()).unwrap();
                             release_rx
-                                .recv_timeout(std::time::Duration::from_secs(5))
+                                .recv_timeout(restore_watchdog)
-                                .unwrap();
+                                .expect("guarded Restore did not release the reserved no-op");
                         }
                         checks += 1;
                         assert!(attention_pending(&api_for_caller).is_empty());

crates/yoi/src/ticket_cli.rs

@@ -1313,6 +1313,7 @@ mod tests {
                 priority: "normal".into(),
                 labels: vec![],
                 created_at: None,
+                created_by: None,
                 updated_at: None,
                 assignee: None,
                 readiness: None,
@@ -1448,6 +1449,7 @@ mod tests {
             Ok(current.clone())
         }
         unused_backend_methods! {
+            fn edit_post_with_snapshots(&self, id: TicketIdOrSlug, edit: ticket::TicketPostEdit) -> ticket::Result<(ticket::Ticket, ticket::Ticket)>;
             fn list_workspace_tags(&self, request: ticket::TagListRequest) -> ticket::Result<Vec<ticket::WorkspaceTag>>;
             fn mutate_workspace_tag(&self, request: ticket::WorkspaceTagMutation) -> ticket::Result<ticket::WorkspaceTag>;
             fn set_ticket_tags(&self, id: TicketIdOrSlug, request: ticket::TicketTagSetRequest) -> ticket::Result<Vec<ticket::WorkspaceTag>>;