fix: don't fail hybrid jobs when input symlinks are changed

src/braket/jobs/_entry_point_template.py

@@ -8,6 +8,7 @@
 # load and run serialized entry point function
 recovered = cloudpickle.loads({serialized})
 def {function_name}():
+    links = {{}}
     try:
         # set working directory to results dir
         os.chdir(get_results_dir())
@@ -66,14 +67,14 @@ def link_input():
 
 def clean_links(links):
     for link, target in links.items():
-        if link.is_symlink and link.readlink() == target:
+        if link.is_symlink() and link.readlink() == target:
             link.unlink()
 
         if link.is_relative_to(Path()):
             for dir in link.parents[:-1]:
                 try:
                     dir.rmdir()
-                except:
+                except OSError:
                     # directory not empty
                     pass
 '''

test/unit_tests/braket/jobs/test_hybrid_job.py

@@ -37,6 +37,7 @@
     S3DataSourceConfig,
     StoppingCondition,
 )
+from braket.jobs._entry_point_template import symlink_input_data
 from braket.jobs.hybrid_job import _resolve_data_format, _sanitize, _serialize_entry_point
 from braket.jobs.local import LocalQuantumJob
 
@@ -754,3 +755,49 @@ def my_job():
 )
 def test_sanitize_hyperparameters(hyperparameter, expected):
     assert _sanitize(hyperparameter) == expected
+
+
+def _clean_links():
+    namespace = {}
+    exec(
+        symlink_input_data.format(
+            prefix_matches={},
+            input_data_items=[],
+            prefix_channels=set(),
+            directory_channels=set(),
+        ),
+        namespace,
+    )
+    return namespace["clean_links"]
+
+
+@pytest.mark.parametrize("replace_with_file", (True, False))
+def test_clean_links_skips_non_symlinks(tmp_path, monkeypatch, replace_with_file):
+    monkeypatch.chdir(tmp_path)
+    target = tmp_path / "target.txt"
+    target.write_text("input")
+    link = Path("data", "input.txt")
+    link.parent.mkdir()
+    link.symlink_to(target)
+
+    link.unlink()
+    if replace_with_file:
+        link.write_text("overwritten by user")
+
+    _clean_links()({link: target})
+    assert link.exists() == replace_with_file
+
+
+def test_entry_point_link_failure_not_masked(tmp_path, monkeypatch):
+    monkeypatch.chdir(tmp_path)
+
+    def my_entry():
+        pass
+
+    def link_input():
+        raise FileExistsError("link exists")
+
+    namespace = {"link_input": link_input, "clean_links": MagicMock()}
+    exec(_serialize_entry_point(my_entry, (), {}), namespace)
+    with pytest.raises(FileExistsError, match="link exists"):
+        namespace["my_entry"]()