diff --git a/acceptance/bin/fault.py b/acceptance/bin/fault.py
index 30405fbc5a0..3ca72e5fae8 100755
--- a/acceptance/bin/fault.py
+++ b/acceptance/bin/fault.py
@@ -1,13 +1,16 @@
 #!/usr/bin/env python3
 """Set up a fault rule on the testserver for the current test token.
 
-Usage: fault.py PATTERN STATUS_CODE OFFSET TIMES
+Usage: fault.py PATTERN STATUS_CODE OFFSET TIMES [DELAY_MS]
 
   PATTERN     HTTP method and path, supports trailing * wildcard,
               e.g. "PUT /api/2.0/permissions/pipelines/*"
-  STATUS_CODE HTTP status code to return, e.g. 504
+  STATUS_CODE HTTP status code to return, e.g. 504. Use 0 together with
+              DELAY_MS for a delay-only rule that sleeps and then lets the
+              real handler run (a slow-but-successful response).
   OFFSET      number of requests to let through before fault starts
-  TIMES       number of times to return the fault response
+  TIMES       number of times to apply the rule
+  DELAY_MS    optional milliseconds to sleep before responding (default 0)
 
 The rule is scoped to the current DATABRICKS_TOKEN so it only affects
 the test that registers it, even when tests share a server.
@@ -25,18 +28,22 @@
     print("DATABRICKS_HOST not set", file=sys.stderr)
     sys.exit(1)
 
-if len(sys.argv) != 5:
-    print(f"usage: {sys.argv[0]} PATTERN STATUS_CODE OFFSET TIMES", file=sys.stderr)
+if len(sys.argv) not in (5, 6):
+    print(f"usage: {sys.argv[0]} PATTERN STATUS_CODE OFFSET TIMES [DELAY_MS]", file=sys.stderr)
     sys.exit(1)
 
 pattern, status_code, offset, times = sys.argv[1], int(sys.argv[2]), int(sys.argv[3]), int(sys.argv[4])
-body = '{"error_code":"INJECTED","message":"Fault injected by test."}'
+delay_ms = int(sys.argv[5]) if len(sys.argv) == 6 else 0
+# A delay-only rule (status_code 0) has no error body; the testserver falls
+# through to the real handler after sleeping.
+body = "" if status_code == 0 else '{"error_code":"INJECTED","message":"Fault injected by test."}'
 
 data = json.dumps(
     {
         "pattern": pattern,
         "status_code": status_code,
         "body": body,
+        "delay_ms": delay_ms,
         "offset": offset,
         "times": times,
     }
diff --git a/acceptance/bundle/deploy/deploy-race-repro/gap/app/databricks.yml b/acceptance/bundle/deploy/deploy-race-repro/gap/app/databricks.yml
new file mode 100644
index 00000000000..404225a0ea3
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/gap/app/databricks.yml
@@ -0,0 +1,16 @@
+bundle:
+  name: deploy-race-gap
+
+resources:
+  jobs:
+    etl:
+      name: etl
+      tasks:
+        - task_key: run_etl
+          spark_python_task:
+            python_file: ./src/etl.py
+          environment_key: default
+      environments:
+        - environment_key: default
+          spec:
+            client: "2"
diff --git a/acceptance/bundle/deploy/deploy-race-repro/gap/app/src/etl.py b/acceptance/bundle/deploy/deploy-race-repro/gap/app/src/etl.py
new file mode 100644
index 00000000000..007d70fd6e0
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/gap/app/src/etl.py
@@ -0,0 +1 @@
+print("etl task ran")
diff --git a/acceptance/bundle/deploy/deploy-race-repro/gap/out.test.toml b/acceptance/bundle/deploy/deploy-race-repro/gap/out.test.toml
new file mode 100644
index 00000000000..3ef9121aba6
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/gap/out.test.toml
@@ -0,0 +1,3 @@
+Local = false
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/deploy/deploy-race-repro/gap/script b/acceptance/bundle/deploy/deploy-race-repro/gap/script
new file mode 100644
index 00000000000..53b8b09a833
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/gap/script
@@ -0,0 +1,46 @@
+# THROWAWAY investigation harness -- NOT a CI test (disabled via parent test.toml).
+#
+# Behaviour 1: delete-before-put gap raced by a live job.
+#
+# Renaming a task file makes applyDiff DELETE the old remote path and PUT the new
+# one (libs/sync/diff.go addFilesWithRemoteNameChanged). Because the job is only
+# repointed after files.Upload (bundle/phases/deploy.go), a job that runs during
+# the gap sees a transient "file not found". This proves it as a real wall-clock
+# race by delaying the PUT and hammering the job while the window is open.
+#
+# Run it by hand to observe the failure:
+#   go test ./acceptance -run TestAccept/bundle/deploy/deploy-race-repro/gap \
+#     -update -tail -v
+
+cd app
+
+unset MSYS_NO_PATHCONV
+
+trace $CLI bundle deploy
+trace $CLI bundle run etl
+
+FILES_PATH=$($CLI bundle summary --output json | jq -r '.workspace.file_path')
+
+# Rename the task file: applyDiff DELETEs .../src/etl.py, then PUTs .../src/etl_v2.py.
+# Delay that PUT by 3s so the delete-before-put window stays open long enough for a
+# concurrently running job to fall into it.
+mv src/etl.py src/etl_v2.py
+update_file.py databricks.yml "etl.py" "etl_v2.py"
+fault.py "POST /api/2.0/workspace-files/import-file${FILES_PATH}/src/etl_v2.py" 0 0 1 3000
+
+# Start the slow deploy in the background, then hammer the job with runs while the
+# window is open. At least one run should observe the transient "file not found".
+$CLI bundle deploy &> LOG.deploy &
+DEPLOY_PID=$!
+
+for i in $(seq 1 20); do
+    echo ">>> run attempt $i" >> LOG.runs
+    $CLI bundle run etl &>> LOG.runs || echo "RUN $i FAILED (raced the deploy gap)" >> LOG.runs
+done
+
+wait $DEPLOY_PID
+
+echo "--- runs that raced the deploy (transient not-found) ---" >> LOG.summary
+grep -c "FAILED (raced the deploy gap)" LOG.runs >> LOG.summary || true
+
+trace $CLI bundle destroy --auto-approve &>> LOG.destroy
diff --git a/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/app/databricks.yml b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/app/databricks.yml
new file mode 100644
index 00000000000..3a31cc68a76
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/app/databricks.yml
@@ -0,0 +1,16 @@
+bundle:
+  name: deploy-race-permissions
+
+resources:
+  jobs:
+    etl:
+      name: etl
+      tasks:
+        - task_key: run_etl
+          spark_python_task:
+            python_file: ./src/etl.py
+          environment_key: default
+      environments:
+        - environment_key: default
+          spec:
+            client: "2"
diff --git a/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/app/src/etl.py b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/app/src/etl.py
new file mode 100644
index 00000000000..007d70fd6e0
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/app/src/etl.py
@@ -0,0 +1 @@
+print("etl task ran")
diff --git a/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/out.test.toml b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/out.test.toml
new file mode 100644
index 00000000000..3ef9121aba6
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/out.test.toml
@@ -0,0 +1,3 @@
+Local = false
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/script b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/script
new file mode 100644
index 00000000000..36c538a9633
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/script
@@ -0,0 +1,34 @@
+# THROWAWAY investigation harness -- NOT a CI test (disabled via parent test.toml).
+#
+# Behaviour 3: permission ACL reset on the deployed folder after upload.
+#
+# ApplyWorkspaceRootPermissions runs SetPermissions AFTER files.Upload
+# (bundle/phases/deploy.go), replacing the folder's direct ACL with exactly the
+# declared permissions. Recording the calls shows the run-as SP's access is
+# rewritten on every deploy, not merely added to -- so a deploy can transiently
+# narrow access rather than only widening it.
+#
+# Run it by hand to observe the recorded ACL writes:
+#   go test ./acceptance -run TestAccept/bundle/deploy/deploy-race-repro/permissions-reset \
+#     -update -tail -v
+
+cd app
+
+unset MSYS_NO_PATHCONV
+
+trace $CLI bundle deploy
+
+# Declare a permission for a service principal, then re-deploy.
+cat >> databricks.yml <<'YAML'
+
+permissions:
+  - level: CAN_MANAGE
+    service_principal_name: 11111111-2222-3333-4444-555555555555
+YAML
+rm -f out.requests.txt
+$CLI bundle deploy &> LOG.deploy
+# The PUT /api/2.0/permissions/directories/* calls are ApplyWorkspaceRootPermissions
+# replacing the folder ACL with exactly the declared permissions, on every deploy.
+print_requests.py //api/2.0/permissions &>> LOG.perms
+
+trace $CLI bundle destroy --auto-approve &>> LOG.destroy
diff --git a/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/test.toml b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/test.toml
new file mode 100644
index 00000000000..6a8be0e3583
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/permissions-reset/test.toml
@@ -0,0 +1,2 @@
+# Needed for print_requests.py to capture the SetPermissions calls.
+RecordRequests = true
diff --git a/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/app/databricks.yml b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/app/databricks.yml
new file mode 100644
index 00000000000..59081440e55
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/app/databricks.yml
@@ -0,0 +1,16 @@
+bundle:
+  name: deploy-race-snapshot-loss
+
+resources:
+  jobs:
+    etl:
+      name: etl
+      tasks:
+        - task_key: run_etl
+          spark_python_task:
+            python_file: ./src/etl.py
+          environment_key: default
+      environments:
+        - environment_key: default
+          spec:
+            client: "2"
diff --git a/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/app/src/etl.py b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/app/src/etl.py
new file mode 100644
index 00000000000..007d70fd6e0
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/app/src/etl.py
@@ -0,0 +1 @@
+print("etl task ran")
diff --git a/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/out.test.toml b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/out.test.toml
new file mode 100644
index 00000000000..3ef9121aba6
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/out.test.toml
@@ -0,0 +1,3 @@
+Local = false
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/script b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/script
new file mode 100644
index 00000000000..0f22cb1557f
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/snapshot-loss/script
@@ -0,0 +1,34 @@
+# THROWAWAY investigation harness -- NOT a CI test (disabled via parent test.toml).
+#
+# Behaviour 2: snapshot-loss mass re-upload widening the window.
+#
+# A deploy from a runner with no local sync snapshot re-PUTs every file in one
+# deploy (libs/sync/snapshot.go loadOrNewSnapshot -> empty snapshot). Deleting
+# .databricks and delaying all uploads shows the whole files/ tree is in flight at
+# once, so the delete-before-put window (see ../gap) widens across every file
+# rather than just a single renamed one.
+#
+# Run it by hand to observe the failure:
+#   go test ./acceptance -run TestAccept/bundle/deploy/deploy-race-repro/snapshot-loss \
+#     -update -tail -v
+
+cd app
+
+unset MSYS_NO_PATHCONV
+
+trace $CLI bundle deploy
+trace $CLI bundle run etl
+
+FILES_PATH=$($CLI bundle summary --output json | jq -r '.workspace.file_path')
+
+# Drop the local snapshot so the next deploy re-uploads every file, and delay all
+# uploads so the whole files/ tree is in flight at once. A job running now can race
+# the mass re-upload.
+rm -rf .databricks
+fault.py "POST /api/2.0/workspace-files/import-file${FILES_PATH}/*" 0 0 100 1000
+$CLI bundle deploy &> LOG.deploy &
+DEPLOY_PID=$!
+$CLI bundle run etl &>> LOG.runs || echo "RUN raced the mass re-upload" >> LOG.summary
+wait $DEPLOY_PID
+
+trace $CLI bundle destroy --auto-approve &>> LOG.destroy
diff --git a/acceptance/bundle/deploy/deploy-race-repro/test.toml b/acceptance/bundle/deploy/deploy-race-repro/test.toml
new file mode 100644
index 00000000000..497d3fa9057
--- /dev/null
+++ b/acceptance/bundle/deploy/deploy-race-repro/test.toml
@@ -0,0 +1,31 @@
+# THROWAWAY investigation harnesses -- intentionally disabled.
+#
+# Each child test drives a real wall-clock race (a backgrounded bundle run or
+# deploy against an in-flight deploy with a delayed upload), so its output is
+# non-deterministic and it must never run in CI. Run one by hand with -tail -v to
+# observe the behaviour; the committed test ../notebook-not-found-gap proves the
+# core mechanism deterministically.
+#
+# The children split the three DABs deployment behaviours from the investigation:
+#   gap/               delete-before-put gap raced by a live job
+#   snapshot-loss/     snapshot-loss mass re-upload widening the window
+#   permissions-reset/ permission ACL reset on the deployed folder after upload
+#
+# All observed output goes to LOG.* sinks (visible under -v, excluded from the
+# golden diff) so re-running never fails on timing noise.
+Local = false
+Cloud = false
+
+# Bug: bundle deploy is non-atomic (applyDiff deletes before it puts, job repointed
+# only after files.Upload), so a job running during a deploy can transiently fail
+# with "file not found" and self-heal on the next run.
+Badness = "bundle deploy mutates the live files/ tree non-atomically; a concurrent job run can observe a transient not-found"
+
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+Ignore = [
+    ".databricks",
+    "app/.databricks",
+    "app/databricks.yml",
+    "app/src",
+]
diff --git a/acceptance/bundle/deploy/immutable/script b/acceptance/bundle/deploy/immutable/script
index 85e9a908d1d..a3248ed7dee 100644
--- a/acceptance/bundle/deploy/immutable/script
+++ b/acceptance/bundle/deploy/immutable/script
@@ -16,4 +16,6 @@ trace $CLI jobs get $JOB_ID | jq '.settings.tasks' | jq '.[] | select(.notebook_
 trace $CLI jobs get $JOB_ID | jq '.settings.environments[0].spec.dependencies'
 
 # Redirect run output to a log file — the real workspace produces different output than the local test server.
-$CLI bundle run my_job &> LOG.run
+# The local test server cannot resolve the content-addressed immutable snapshot path, so the run fails locally;
+# tolerate its exit code since the run output is intentionally not asserted.
+errcode $CLI bundle run my_job &> LOG.run
diff --git a/acceptance/bundle/deploy/notebook-not-found-gap/app/databricks.yml b/acceptance/bundle/deploy/notebook-not-found-gap/app/databricks.yml
new file mode 100644
index 00000000000..73780d31ae0
--- /dev/null
+++ b/acceptance/bundle/deploy/notebook-not-found-gap/app/databricks.yml
@@ -0,0 +1,16 @@
+bundle:
+  name: notebook-gap
+
+resources:
+  jobs:
+    etl:
+      name: etl
+      tasks:
+        - task_key: run_etl
+          spark_python_task:
+            python_file: ./src/etl.py
+          environment_key: default
+      environments:
+        - environment_key: default
+          spec:
+            client: "2"
diff --git a/acceptance/bundle/deploy/notebook-not-found-gap/app/src/etl.py b/acceptance/bundle/deploy/notebook-not-found-gap/app/src/etl.py
new file mode 100644
index 00000000000..007d70fd6e0
--- /dev/null
+++ b/acceptance/bundle/deploy/notebook-not-found-gap/app/src/etl.py
@@ -0,0 +1 @@
+print("etl task ran")
diff --git a/acceptance/bundle/deploy/notebook-not-found-gap/out.test.toml b/acceptance/bundle/deploy/notebook-not-found-gap/out.test.toml
new file mode 100644
index 00000000000..e90b6d5d1ba
--- /dev/null
+++ b/acceptance/bundle/deploy/notebook-not-found-gap/out.test.toml
@@ -0,0 +1,3 @@
+Local = true
+Cloud = false
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
diff --git a/acceptance/bundle/deploy/notebook-not-found-gap/output.txt b/acceptance/bundle/deploy/notebook-not-found-gap/output.txt
new file mode 100644
index 00000000000..b4f4d069637
--- /dev/null
+++ b/acceptance/bundle/deploy/notebook-not-found-gap/output.txt
@@ -0,0 +1,118 @@
+
+=== Deploy the bundle and confirm the job runs cleanly
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle run etl
+Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
+
+[TIMESTAMP] "etl" RUNNING
+[TIMESTAMP] "etl" TERMINATED SUCCESS
+etl task ran
+
+=== Rename the task file (remote-name change => DELETE old + PUT new)
+>>> mv src/etl.py src/etl_v2.py
+
+=== Re-deploy, but the PUT of the renamed file fails
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files...
+Error: Fault injected by test. (500 INJECTED)
+
+Endpoint: POST [DATABRICKS_URL]/api/2.0/workspace-files/import-file/Workspace%2FUsers%2F[USERNAME]%2F.bundle%2Fnotebook-gap%2Fdefault%2Ffiles%2Fsrc%2Fetl_v2.py?overwrite=true
+HTTP Status: 500 Internal Server Error
+API error_code: INJECTED
+API message: Fault injected by test.
+
+
+Exit code: 1
+
+>>> print_requests.py //src/etl //api/2.0/workspace/delete --sort
+{
+  "method": "POST",
+  "path": "/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files/src/etl.py",
+  "q": {
+    "overwrite": "true"
+  },
+  "raw_body": "print(\"etl task ran\")\n"
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files/src/etl_v2.py",
+  "q": {
+    "overwrite": "true"
+  },
+  "raw_body": "print(\"etl task ran\")\n"
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/workspace/delete",
+  "body": {
+    "path": "/Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/artifacts/.internal",
+    "recursive": true
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/workspace/delete",
+  "body": {
+    "path": "/Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/artifacts/.internal",
+    "recursive": true
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/workspace/delete",
+  "body": {
+    "path": "/Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files/src/etl.py"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/workspace/delete",
+  "body": {
+    "path": "/Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/state/deploy.lock"
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/workspace/delete",
+  "body": {
+    "path": "/Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/state/deploy.lock"
+  }
+}
+
+=== The job now fails: its file was deleted and never re-put
+>>> [CLI] bundle run etl
+Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
+
+[TIMESTAMP] "etl" RUNNING
+[TIMESTAMP] "etl" TERMINATED FAILED python file not found in workspace: /Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files/src/etl.py
+Error: run failed: python file not found in workspace: /Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files/src/etl.py
+
+Exit code: 1
+
+=== Self-heal: a clean re-deploy re-puts the file and the run succeeds
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/notebook-gap/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] bundle run etl
+Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
+
+[TIMESTAMP] "etl" RUNNING
+[TIMESTAMP] "etl" TERMINATED SUCCESS
+etl task ran
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.jobs.etl
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/notebook-gap/default
+
+Deleting files...
+Destroy complete!
diff --git a/acceptance/bundle/deploy/notebook-not-found-gap/script b/acceptance/bundle/deploy/notebook-not-found-gap/script
new file mode 100644
index 00000000000..5c25db313b6
--- /dev/null
+++ b/acceptance/bundle/deploy/notebook-not-found-gap/script
@@ -0,0 +1,58 @@
+# Reproduces a bundle-deployed job intermittently failing at task start because
+# the task file "does not exist", then self-healing on a re-deploy with no config
+# change.
+#
+# Root cause demonstrated here deterministically: libs/sync/watchdog.go
+# applyDiff() deletes remote objects BEFORE it puts (delete -> rmdir -> mkdir ->
+# put), and the job definition is updated in deployCore only AFTER files.Upload
+# (bundle/phases/deploy.go). A remote-name change -- a rename here, or a .py file
+# gaining/losing the "# Databricks notebook source" marker -- queues a DELETE of
+# the old remote path and a PUT of the new one (libs/sync/diff.go
+# addFilesWithRemoteNameChanged). If the PUT does not land while the job still
+# points at the old path, the task sees a genuine "file does not exist". We freeze
+# that window deterministically by faulting the PUT.
+#
+# The bundle lives in ./app so the acceptance harness files (script, output.txt,
+# out.requests.txt, ...) are not swept into the bundle sync root.
+
+cd app
+
+unset MSYS_NO_PATHCONV
+
+title "Deploy the bundle and confirm the job runs cleanly"
+trace $CLI bundle deploy
+trace $CLI bundle run etl
+rm -f out.requests.txt
+
+title "Rename the task file (remote-name change => DELETE old + PUT new)"
+trace mv src/etl.py src/etl_v2.py
+update_file.py databricks.yml "etl.py" "etl_v2.py"
+
+title "Re-deploy, but the PUT of the renamed file fails"
+# The old file is deleted first; the PUT is faulted, so files.Upload aborts BEFORE
+# deployCore updates the job. The job is left pointing at the deleted path.
+# Fault only the renamed file's exact upload path, so the deploy lock (which also
+# uses import-file, under /state/) is unaffected.
+FILES_PATH=$($CLI bundle summary --output json | jq -r '.workspace.file_path')
+fault.py "POST /api/2.0/workspace-files/import-file${FILES_PATH}/src/etl_v2.py" 500 0 1
+errcode trace $CLI bundle deploy
+trace print_requests.py //src/etl //api/2.0/workspace/delete --sort
+
+title "The job now fails: its file was deleted and never re-put"
+# The file "may not exist", even though the job config and permissions were never
+# changed.
+errcode trace $CLI bundle run etl
+
+title "Self-heal: a clean re-deploy re-puts the file and the run succeeds"
+# No config change and no permission change: a re-deploy re-puts the file and the
+# job succeeds again.
+trace $CLI bundle deploy
+trace $CLI bundle run etl
+
+trace $CLI bundle destroy --auto-approve
+
+# Drain the recorded requests from the self-heal and destroy steps into a
+# non-asserted sink so the trailing out.requests.txt (which carries run-to-run
+# ordering of state-file reads) is not compared.
+print_requests.py //api/2.0 --get &> LOG.requests
+rm -f out.requests.txt
diff --git a/acceptance/bundle/deploy/notebook-not-found-gap/test.toml b/acceptance/bundle/deploy/notebook-not-found-gap/test.toml
new file mode 100644
index 00000000000..f1df63fecc8
--- /dev/null
+++ b/acceptance/bundle/deploy/notebook-not-found-gap/test.toml
@@ -0,0 +1,19 @@
+Badness = "bundle deploy is non-atomic: applyDiff deletes remote objects before it puts, and the job is repointed only in deployCore after files.Upload; a failed/slow PUT after the delete leaves a deployed job pointing at a missing file"
+Local = true
+Cloud = false
+RecordRequests = true
+
+# Telemetry carries timing fields that vary run-to-run; drop it from recordings.
+Env.DATABRICKS_CLI_DISABLE_TELEMETRY = "1"
+
+# The delete-before-put gap is in libs/sync (engine-independent); pin to one
+# engine so the recorded state-file requests (resources.json vs terraform.tfstate)
+# don't diverge between variants.
+EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["direct"]
+
+Ignore = [
+    ".databricks",
+    "app/.databricks",
+    "app/databricks.yml",
+    "app/src",
+]
diff --git a/acceptance/bundle/run/state-wiped/output.txt b/acceptance/bundle/run/state-wiped/output.txt
index ff2f702bf28..d8d3102be2e 100644
--- a/acceptance/bundle/run/state-wiped/output.txt
+++ b/acceptance/bundle/run/state-wiped/output.txt
@@ -15,6 +15,7 @@ Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
 [TIMESTAMP] "foo" RUNNING
 [TIMESTAMP] "foo" TERMINATED SUCCESS
 
+
 === Testing that clean state that affect run command -- it'll fetch the state
 >>> rm -fr .databricks
 
@@ -23,3 +24,4 @@ Run URL: [DATABRICKS_URL]/jobs/[NUMID]/runs/[NUMID]?o=[NUMID]
 
 [TIMESTAMP] "foo" RUNNING
 [TIMESTAMP] "foo" TERMINATED SUCCESS
+
diff --git a/libs/testserver/fault.go b/libs/testserver/fault.go
index 49b98b0f25f..21b4cf131a0 100644
--- a/libs/testserver/fault.go
+++ b/libs/testserver/fault.go
@@ -11,10 +11,12 @@ type faultRuleKey struct {
 	pattern string
 }
 
-// FaultRule describes a single injected fault: HTTP status, body, and remaining fire count.
+// FaultRule describes a single injected fault: HTTP status, body, an optional
+// pre-response delay, and remaining fire count.
 type FaultRule struct {
 	StatusCode int
 	Body       string
+	DelayMs    int
 	offset     int
 	times      int
 }
@@ -31,12 +33,13 @@ func NewFaultRules() *FaultRules {
 }
 
 // Set registers or replaces a fault rule for the given token and pattern.
-func (fr *FaultRules) Set(token, pattern string, statusCode int, body string, offset, times int) {
+func (fr *FaultRules) Set(token, pattern string, statusCode int, body string, delayMs, offset, times int) {
 	fr.mu.Lock()
 	defer fr.mu.Unlock()
 	fr.rules[faultRuleKey{token: token, pattern: pattern}] = &FaultRule{
 		StatusCode: statusCode,
 		Body:       body,
+		DelayMs:    delayMs,
 		offset:     offset,
 		times:      times,
 	}
@@ -90,13 +93,14 @@ func faultEndpointHandler(fr *FaultRules) HandlerFunc {
 			Pattern    string `json:"pattern"`
 			StatusCode int    `json:"status_code"`
 			Body       string `json:"body"`
+			DelayMs    int    `json:"delay_ms"`
 			Offset     int    `json:"offset"`
 			Times      int    `json:"times"`
 		}
 		if err := json.Unmarshal(req.Body, &body); err != nil {
 			return Response{StatusCode: 400, Body: map[string]string{"error": err.Error()}}
 		}
-		fr.Set(req.Token, body.Pattern, body.StatusCode, body.Body, body.Offset, body.Times)
+		fr.Set(req.Token, body.Pattern, body.StatusCode, body.Body, body.DelayMs, body.Offset, body.Times)
 		return Response{StatusCode: 200}
 	}
 }
diff --git a/libs/testserver/fault_test.go b/libs/testserver/fault_test.go
index 2366fb733c1..7455ea16688 100644
--- a/libs/testserver/fault_test.go
+++ b/libs/testserver/fault_test.go
@@ -10,7 +10,7 @@ import (
 
 func TestFaultRulesNoMatch(t *testing.T) {
 	fr := testserver.NewFaultRules()
-	fr.Set("tok", "GET /foo", 504, "body", 0, 1)
+	fr.Set("tok", "GET /foo", 504, "body", 0, 0, 1)
 
 	assert.Nil(t, fr.Check("POST", "/foo", "tok"))
 	assert.Nil(t, fr.Check("GET", "/bar", "tok"))
@@ -19,7 +19,7 @@ func TestFaultRulesNoMatch(t *testing.T) {
 
 func TestFaultRulesExactMatch(t *testing.T) {
 	fr := testserver.NewFaultRules()
-	fr.Set("tok", "PUT /api/2.0/jobs/123", 504, "body", 0, 1)
+	fr.Set("tok", "PUT /api/2.0/jobs/123", 504, "body", 0, 0, 1)
 
 	rule := fr.Check("PUT", "/api/2.0/jobs/123", "tok")
 	require.NotNil(t, rule)
@@ -27,9 +27,21 @@ func TestFaultRulesExactMatch(t *testing.T) {
 	assert.Equal(t, "body", rule.Body)
 }
 
+func TestFaultRulesDelay(t *testing.T) {
+	fr := testserver.NewFaultRules()
+	// A delay-only rule (status code 0) carries the delay but no error status, so
+	// the caller sleeps and then falls through to the real handler.
+	fr.Set("tok", "POST /api/2.0/slow", 0, "", 250, 0, 1)
+
+	rule := fr.Check("POST", "/api/2.0/slow", "tok")
+	require.NotNil(t, rule)
+	assert.Equal(t, 0, rule.StatusCode)
+	assert.Equal(t, 250, rule.DelayMs)
+}
+
 func TestFaultRulesWildcardMatch(t *testing.T) {
 	fr := testserver.NewFaultRules()
-	fr.Set("tok", "PUT /api/2.0/permissions/pipelines/*", 504, "body", 0, 2)
+	fr.Set("tok", "PUT /api/2.0/permissions/pipelines/*", 504, "body", 0, 0, 2)
 
 	assert.NotNil(t, fr.Check("PUT", "/api/2.0/permissions/pipelines/abc", "tok"))
 	assert.NotNil(t, fr.Check("PUT", "/api/2.0/permissions/pipelines/xyz", "tok"))
@@ -38,7 +50,7 @@ func TestFaultRulesWildcardMatch(t *testing.T) {
 
 func TestFaultRulesOffset(t *testing.T) {
 	fr := testserver.NewFaultRules()
-	fr.Set("tok", "GET /foo", 504, "body", 2, 1)
+	fr.Set("tok", "GET /foo", 504, "body", 0, 2, 1)
 
 	assert.Nil(t, fr.Check("GET", "/foo", "tok"))    // offset 2→1
 	assert.Nil(t, fr.Check("GET", "/foo", "tok"))    // offset 1→0
@@ -48,7 +60,7 @@ func TestFaultRulesOffset(t *testing.T) {
 
 func TestFaultRulesTimes(t *testing.T) {
 	fr := testserver.NewFaultRules()
-	fr.Set("tok", "GET /foo", 504, "body", 0, 3)
+	fr.Set("tok", "GET /foo", 504, "body", 0, 0, 3)
 
 	for range 3 {
 		assert.NotNil(t, fr.Check("GET", "/foo", "tok"))
diff --git a/libs/testserver/jobs.go b/libs/testserver/jobs.go
index 9097d43c086..3b297232c1e 100644
--- a/libs/testserver/jobs.go
+++ b/libs/testserver/jobs.go
@@ -659,13 +659,14 @@ func (s *FakeWorkspace) executeNotebookTask(task jobs.Task, notebookParams map[s
 	}
 
 	// Try both with and without .py extension (notebooks are stored with .py but referenced without)
-	notebookData := s.files[notebookPath].Data
-	if len(notebookData) == 0 {
-		notebookData = s.files[notebookPath+".py"].Data
+	entry, ok := s.files[notebookPath]
+	if !ok {
+		entry, ok = s.files[notebookPath+".py"]
 	}
-	if len(notebookData) == 0 {
+	if !ok {
 		return "", fmt.Errorf("notebook not found in workspace: %s (also tried .py)", notebookPath)
 	}
+	notebookData := entry.Data
 
 	// Create a temporary Python environment for notebook execution
 	tmpDir, err := os.MkdirTemp("", "notebook-task-*")
@@ -748,10 +749,11 @@ func (s *FakeWorkspace) executeSparkPythonTask(task jobs.Task) (string, error) {
 		pythonPath = "/" + pythonPath
 	}
 
-	pythonData := s.files[pythonPath].Data
-	if len(pythonData) == 0 {
+	entry, ok := s.files[pythonPath]
+	if !ok {
 		return "", fmt.Errorf("python file not found in workspace: %s", pythonPath)
 	}
+	pythonData := entry.Data
 
 	env, cleanup, err := s.getOrCreateClusterEnv(task)
 	if err != nil {
@@ -865,19 +867,30 @@ func (s *FakeWorkspace) JobsGetRun(req Request) Response {
 		return Response{StatusCode: 404}
 	}
 
-	// Simulate cloud behavior: first poll returns RUNNING, next returns TERMINATED SUCCESS.
+	// Simulate cloud behavior: first poll returns RUNNING, next returns TERMINATED.
 	if run.State.LifeCycleState == jobs.RunLifeCycleStateRunning {
-		// Transition stored state to TERMINATED for the next poll.
-		run.State = &jobs.RunState{
-			LifeCycleState: jobs.RunLifeCycleStateTerminated,
-			ResultState:    jobs.RunResultStateSuccess,
-		}
+		// The per-task result states were already computed in JobsRunNow (a task
+		// whose file is missing is marked FAILED). Roll them up into the run-level
+		// result rather than forcing SUCCESS, so a failed task surfaces as a failed
+		// run on the next poll. Carry the failed task's error into the run-level
+		// state_message, mirroring how the cloud reports the failure reason.
+		resultState := jobs.RunResultStateSuccess
+		stateMessage := ""
 		for i := range run.Tasks {
-			run.Tasks[i].State = &jobs.RunState{
-				LifeCycleState: jobs.RunLifeCycleStateTerminated,
-				ResultState:    jobs.RunResultStateSuccess,
+			if run.Tasks[i].State != nil && run.Tasks[i].State.ResultState == jobs.RunResultStateFailed {
+				resultState = jobs.RunResultStateFailed
+				stateMessage = s.JobRunOutputs[run.Tasks[i].RunId].Error
+				break
 			}
 		}
+
+		// Transition stored state to TERMINATED for the next poll, preserving the
+		// existing per-task states.
+		run.State = &jobs.RunState{
+			LifeCycleState: jobs.RunLifeCycleStateTerminated,
+			ResultState:    resultState,
+			StateMessage:   stateMessage,
+		}
 		s.JobRuns[runIdInt] = run
 
 		// Return RUNNING for this poll (before the transition).
diff --git a/libs/testserver/server.go b/libs/testserver/server.go
index 5ae3141bfd2..462c712d165 100644
--- a/libs/testserver/server.go
+++ b/libs/testserver/server.go
@@ -16,6 +16,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/databricks/cli/internal/testutil"
 	"github.com/databricks/cli/libs/testserver/testsql"
@@ -372,7 +373,16 @@ func (s *Server) serve(w http.ResponseWriter, r *http.Request, handler HandlerFu
 
 	var resp EncodedResponse
 
-	if rule := s.faults.Check(r.Method, r.URL.Path, token); rule != nil {
+	rule := s.faults.Check(r.Method, r.URL.Path, token)
+	if rule != nil && rule.DelayMs > 0 {
+		// Hold the request open. A delay-only rule (StatusCode == 0) then falls
+		// through to the real handler, simulating a slow-but-successful call; this
+		// is what widens the delete-before-put window so a concurrently running job
+		// can observe the transient not-found (see FaultRules).
+		time.Sleep(time.Duration(rule.DelayMs) * time.Millisecond)
+	}
+
+	if rule != nil && rule.StatusCode != 0 {
 		resp = EncodedResponse{
 			StatusCode: rule.StatusCode,
 			Body:       []byte(rule.Body),