@@ -1037,6 +1037,16 @@ async def rm(
         except OSError as e:
             raise WorkspaceArchiveWriteError(path=normalized, cause=e) from e
 
+    async def _remove_workspace_entry_on_resume(self, path: Path) -> None:
+        # Snapshot cleanup owns the entry, not the target of a stale symlink.
+        target = self.normalize_path(path.parent, for_write=True) / path.name
+        try:
+            await run_blocking_workspace_io(partial(self._files.rm, target, recursive=True))
+        except FileNotFoundError:
+            return
+        except OSError as e:
+            raise WorkspaceArchiveWriteError(path=target, cause=e) from e
+
     async def _read_bounded(self, path: Path, *, max_bytes: int) -> bytes:
         return self._files.read_bounded(self.normalize_path(path), max_bytes)
 
 
@@ -1473,6 +1473,10 @@ async def _clear_workspace_root_on_resume(self) -> None:
 
         await snapshot_lifecycle.clear_workspace_root_on_resume(self)
 
+    async def _remove_workspace_entry_on_resume(self, path: Path) -> None:
+        """Remove a stale workspace entry before restoring a snapshot."""
+        await self.rm(path, recursive=True)
+
     def _workspace_resume_mount_skip_relpaths(self) -> set[Path]:
         return snapshot_lifecycle.workspace_resume_mount_skip_relpaths(self)
 
 
@@ -224,7 +224,7 @@ async def clear_workspace_dir_on_resume_pruned(
         try:
             child_rel = child.relative_to(root)
         except ValueError:
-            await session.rm(child, recursive=True)
+            await session._remove_workspace_entry_on_resume(child)
             continue
 
         if child_rel in skip_rel_paths:
@@ -236,10 +236,10 @@ async def clear_workspace_dir_on_resume_pruned(
                     skip_rel_paths=skip_rel_paths,
                 )
             else:
-                await session.rm(child, recursive=True)
+                await session._remove_workspace_entry_on_resume(child)
             continue
         # `parse_ls_la` filters "." and ".." already; remove everything else recursively.
-        await session.rm(child, recursive=True)
+        await session._remove_workspace_entry_on_resume(child)
 
 
 def _close_best_effort(stream: io.IOBase) -> None:
 
@@ -112,6 +112,48 @@ async def test_unix_local_snapshot_round_trips_hardlinks(
         await session.shutdown()
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("target_kind", ["directory", "file", "missing"])
+async def test_unix_local_snapshot_resume_removes_stale_link_without_following_target(
+    tmp_path: Path, target_kind: str
+) -> None:
+    workspace = tmp_path / "workspace"
+    external = tmp_path / "external"
+    external.mkdir()
+    sentinel = external / "keep.txt"
+    sentinel.write_bytes(b"external data")
+    target = external if target_kind == "directory" else external / target_kind
+    if target_kind == "file":
+        target.write_bytes(b"external file")
+    client = UnixLocalSandboxClient(inherit_host_environment=False)
+    session = await client.create(
+        manifest=Manifest(
+            root=str(workspace), extra_path_grants=(SandboxPathGrant(path=str(external)),)
+        ),
+        snapshot=LocalSnapshotSpec(base_path=tmp_path / "snapshots"),
+    )
+    await session.start()
+    (workspace / "original.txt").write_bytes(b"snapshot content")
+    await session.stop()
+    (workspace / "original.txt").write_bytes(b"changed after snapshot")
+    stale_link = workspace / "stale-link"
+    stale_link.symlink_to(target, target_is_directory=target_kind == "directory")
+    resumed = await client.resume(session.state)
+    try:
+        await resumed.start()
+        assert (workspace / "original.txt").read_bytes() == b"snapshot content"
+        assert sentinel.read_bytes() == b"external data"
+        if target_kind == "file":
+            assert target.read_bytes() == b"external file"
+        elif target_kind == "missing":
+            assert not target.exists()
+        assert not stale_link.is_symlink()
+        assert not stale_link.exists()
+    finally:
+        await resumed.shutdown()
+        await session.shutdown()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("invalid_kind", ["hardlink", "external_symlink", "invalid_tar"])
 async def test_unix_local_resume_rejects_invalid_snapshot_before_clearing_workspace(