fix(community): reserve Browserless capture names before publishing (#6901)

* fix(community): reserve Browserless capture names before publishing

Name selection and publication were two separate operations:
`_dedupe_output_name` probed with `Path.exists()` and then
`publish_bytes_atomically` ran `os.replace`, so two `web_capture` calls in the
same thread outputs directory could both settle on `report.png` before either
wrote, and the second publication replaced the first capture while both reported
success for the same artifact path (#6883). `web_capture_tool` runs the writer
through `asyncio.to_thread`, and same-second default names collide the same way
because the generated stem's timestamp has second resolution.

Claim the name instead of checking for it: an `O_CREAT | O_EXCL` open either
creates an empty placeholder or reports the name held, so the probe walk
(-1, -2, ... then the microsecond timestamp) is race-free. `_write_capture_output`
releases the placeholder when publication fails, catching BaseException so an
interruption also frees the name, and a failing unlink is logged rather than
allowed to mask the original error - the same error-masking family #6871 is
addressing in the shared publisher, which this change deliberately does not
touch.

When even the timestamped names are held, the function now raises instead of
returning a name it cannot own; returning one would reintroduce the silent
overwrite. `tests/test_browser_outputs_atomic.py` pins the publisher call as
source text, so its expectation moved to the new call shape in the same commit.

Refs #6883

* fix(community): open the reserved capture name with new-file permissions

An `os.open` without an explicit mode creates the file with `0o777 & ~umask`, and
`publish_bytes_atomically` preserves an existing destination's mode, so the
reservation added in the previous commit made every capture executable (`0o755`
under the usual umask) where main produced `0o666 & ~umask` (`0o644`) through the
publisher's new-file branch. Pass `0o666` explicitly and pin the mode against a
`Path.write_bytes` reference in the same directory, which is the permission the
publisher documents for new files.

Refs #6883

* fix(browserless): warn when capture reservation cleanup fails

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

README.md

@@ -2592,10 +2592,15 @@ This is the difference between a chatbot with tool access and an agent with an a
 
 ### Agentic Browser Control
 
-Browser screenshots and Browserless capture/fetch outputs are published only
-after the complete file is written. New files use the host's normal creation
-permissions (typically `0644`), and replacements preserve existing permissions,
-so sandbox code can read captures when those permissions allow it.
+Browser screenshots and Browserless captures publish complete content atomically.
+New files use the host's normal creation permissions (typically `0644`), and
+replacements preserve existing permissions, so sandbox code can read captures
+when those permissions allow it.
+
+Browserless reserves capture names with empty placeholders to prevent concurrent
+captures from overwriting one another. Failed publication releases the placeholder;
+if cleanup fails, Gateway logs a warning with its path and preserves the original
+publication error or interruption.
 
 Browser dependency auto-detection accepts `name`, `group`, and `use` in any
 order within a tool entry, with either indented or indentless YAML lists.

backend/packages/harness/deerflow/community/browserless/tools.py

@@ -34,6 +34,9 @@ _OUTPUT_FORMAT_TO_EXTENSION = {
 _SAFE_FILENAME_RE = re.compile(r"[^A-Za-z0-9._-]+")
 # Cap collision-suffix probing so a saturated outputs directory cannot spin forever.
 _MAX_FILENAME_COLLISION_PROBES = 1000
+# Retry bound for the timestamped fallback: the microseconds stamp can only repeat
+# when another writer claimed that exact name within the same microsecond.
+_MAX_TIMESTAMP_CLAIM_ATTEMPTS = 5
 DEFAULT_BASE_URL = "http://localhost:3032"
 _UNSET = object()
 
@@ -221,34 +224,71 @@ def _tool_message(content: str, tool_call_id: str) -> Command:
     return Command(update={"messages": [ToolMessage(content, tool_call_id=tool_call_id)]})
 
 
-def _dedupe_output_name(outputs_path: Path, output_name: str) -> str:
-    """Return a non-colliding filename under ``outputs_path``.
+def _reserve_output_name(outputs_path: Path, name: str) -> bool:
+    """Claim ``name`` for this capture, returning False when it is already held.
 
-    Keeps the original name when free, otherwise appends ``-1``, ``-2``, ...
-    before the extension so an explicit filename never silently overwrites an
-    earlier capture. Falls back to a timestamp suffix if the directory is
-    saturated with the bounded probe range.
+    The claim is an ``O_CREAT | O_EXCL`` open that leaves an empty placeholder
+    behind, not an existence check: two writers that merely *observe* a free name
+    would both go on to publish to it, and the second ``os.replace`` would
+    silently replace the first capture (#6883). ``0o666`` is passed explicitly
+    because ``publish_bytes_atomically`` preserves an existing file's mode, and
+    ``os.open`` would otherwise default the placeholder to ``0o777 & ~umask`` and
+    make every capture executable instead of the ``0o666 & ~umask`` a new file
+    gets from ``Path.write_bytes``.
     """
-    candidate = outputs_path / output_name
-    if not candidate.exists():
-        return output_name
+    try:
+        descriptor = os.open(outputs_path / name, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o666)
+    except FileExistsError:
+        return False
+    os.close(descriptor)
+    return True
 
+
+def _dedupe_output_name(outputs_path: Path, output_name: str) -> str:
+    """Reserve and return a non-colliding filename under ``outputs_path``.
+
+    Keeps the original name when it can be claimed, otherwise appends ``-1``,
+    ``-2``, ... before the extension so an explicit filename never silently
+    overwrites an earlier capture. Falls back to a timestamp suffix if the
+    directory is saturated with the bounded probe range.
+
+    Raises:
+        RuntimeError: when even the timestamped names cannot be claimed, which
+            means another writer is holding them; returning a name anyway would
+            reintroduce the silent overwrite this function exists to prevent.
+    """
     stem = Path(output_name).stem
     suffix = Path(output_name).suffix
+    if _reserve_output_name(outputs_path, output_name):
+        return output_name
+
     for index in range(1, _MAX_FILENAME_COLLISION_PROBES + 1):
         probe = f"{stem}-{index}{suffix}"
-        if not (outputs_path / probe).exists():
+        if _reserve_output_name(outputs_path, probe):
             return probe
 
-    timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S-%f")
-    return f"{stem}-{timestamp}{suffix}"
+    for _ in range(_MAX_TIMESTAMP_CLAIM_ATTEMPTS):
+        timestamp = datetime.now(UTC).strftime("%Y%m%d-%H%M%S-%f")
+        fallback = f"{stem}-{timestamp}{suffix}"
+        if _reserve_output_name(outputs_path, fallback):
+            return fallback
+
+    raise RuntimeError(f"could not reserve an output filename for {output_name!r} in {outputs_path}")
 
 
 def _write_capture_output(outputs_path: Path, output_name: str, content: bytes) -> str:
     """Write ``content`` into ``outputs_path`` and return the actual filename used."""
     outputs_path.mkdir(parents=True, exist_ok=True)
     final_name = _dedupe_output_name(outputs_path, output_name)
-    _publish_bytes_atomically(outputs_path / final_name, content)
+    destination = outputs_path / final_name
+    try:
+        _publish_bytes_atomically(destination, content)
+    except BaseException:
+        try:
+            destination.unlink(missing_ok=True)
+        except OSError:
+            logger.warning("Could not release the reserved capture name %s", destination, exc_info=True)
+        raise
     return final_name
 
 

backend/tests/AGENTS.md

@@ -79,6 +79,10 @@ Browser-asset confinement tests use `support.symlinks.symlink_or_skip` for real
 file and directory symlinks. Keep missing-file, duplicate-key, and size-limit
 checks separate so they still run when the host cannot create symlinks.
 
+Browserless reservation cleanup tests require a warning with the leftover path
+and unlink error while preserving the original publication exception, including
+interruption. Successful cleanup stays quiet.
+
 `blocking_io/test_web_tool_url_validation.py` resolves a synthetic `.invalid`
 hostname to loopback by patching `_socket.getaddrinfo`, below the real
 `socket.getaddrinfo` wrapper. Do not replace that wrapper or the production URL

backend/tests/test_browser_outputs_atomic.py

@@ -150,7 +150,8 @@ def test_capture_writes_go_through_the_atomic_publish():
     source = inspect.getsource(browserless_tools)
 
     assert "(outputs_path / final_name).write_bytes(" not in source
-    assert "_publish_bytes_atomically(outputs_path / final_name, content)" in source
+    assert "destination.unlink(missing_ok=True)" in source
+    assert "_publish_bytes_atomically(destination, content)" in source
 
 
 def test_both_tools_share_the_atomic_publisher():

backend/tests/test_browserless_capture_name_reservation.py

@@ -0,0 +1,197 @@
+"""Browserless capture names must be claimed atomically, not merely checked.
+
+``_dedupe_output_name`` promises that "an explicit filename never silently
+overwrites an earlier capture", but on main name selection (``Path.exists()``)
+and publication (``os.replace`` inside ``publish_bytes_atomically``) are two
+separate operations. Two ``web_capture`` calls in the same thread outputs
+directory can therefore both choose ``report.png`` before either publishes, and
+the second publication replaces the first capture while both tools report
+success for the same artifact path. ``web_capture_tool`` runs its writer through
+``asyncio.to_thread``, so parallel captures are the normal shape rather than a
+corner case. Same-second default names collide the same way, because the
+generated stem's timestamp has second resolution.
+"""
+
+import importlib
+import logging
+import stat
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from datetime import UTC, datetime
+
+import pytest
+
+tools = importlib.import_module("deerflow.community.browserless.tools")
+
+
+@pytest.fixture()
+def outputs(tmp_path):
+    directory = tmp_path / "outputs"
+    directory.mkdir()
+    return directory
+
+
+def test_concurrent_captures_do_not_claim_the_same_name(outputs, monkeypatch):
+    """Two writers that pick names before either publishes still get distinct files.
+
+    The barrier sits after name selection, which is exactly the interleaving the
+    check-then-publish split allows: both writers have committed to a name and
+    only then write bytes.
+    """
+    barrier = threading.Barrier(2, timeout=5)
+    real_dedupe = tools._dedupe_output_name
+
+    def choose_then_meet(outputs_path: object, output_name: str) -> str:
+        chosen = real_dedupe(outputs_path, output_name)
+        barrier.wait()
+        return chosen
+
+    monkeypatch.setattr(tools, "_dedupe_output_name", choose_then_meet)
+    claimed: dict[str, str] = {}
+
+    def capture(tag: str, content: bytes) -> None:
+        claimed[tag] = tools._write_capture_output(outputs, "report.png", content)
+
+    with ThreadPoolExecutor(max_workers=2) as pool:
+        futures = [pool.submit(capture, "A", b"first"), pool.submit(capture, "B", b"second")]
+        for future in futures:
+            future.result()
+
+    assert claimed["A"] != claimed["B"], f"both captures claimed {claimed}"
+    assert (outputs / claimed["A"]).read_bytes() == b"first"
+    assert (outputs / claimed["B"]).read_bytes() == b"second"
+
+
+def test_reserved_names_keep_new_file_permissions(outputs):
+    """Claiming a name must not change the mode the capture would already have had.
+
+    ``publish_bytes_atomically`` preserves an existing file's mode, so a
+    placeholder opened with the wrong default mode would make every capture
+    executable rather than ``0o666 & ~umask``.
+    """
+    reference = outputs / "reference.bin"
+    reference.write_bytes(b"x")
+    expected = stat.S_IMODE(reference.stat().st_mode)
+
+    claimed = tools._write_capture_output(outputs, "report.png", b"payload")
+
+    assert stat.S_IMODE((outputs / claimed).stat().st_mode) == expected
+
+
+def test_reservation_closes_the_descriptor_it_opened(outputs, monkeypatch):
+    """The claim is an ``os.open``, so its descriptor has to be released too."""
+    real_close = tools.os.close
+    closed: list[int] = []
+
+    def spy_close(fd: int) -> None:
+        closed.append(fd)
+        real_close(fd)
+
+    monkeypatch.setattr(tools.os, "close", spy_close)
+
+    tools._write_capture_output(outputs, "report.png", b"payload")
+
+    assert closed, "the reserved name leaked its file descriptor"
+
+
+def test_single_writer_still_uses_the_documented_suffix(outputs):
+    (outputs / "report.png").write_bytes(b"existing")
+
+    assert tools._write_capture_output(outputs, "report.png", b"new") == "report-1.png"
+    assert (outputs / "report.png").read_bytes() == b"existing"
+    assert (outputs / "report-1.png").read_bytes() == b"new"
+
+
+def test_publication_failure_does_not_leave_the_reserved_name(outputs, monkeypatch, caplog):
+    def failing_publish(path: object, content: bytes) -> None:
+        raise OSError("disk full")
+
+    monkeypatch.setattr(tools, "_publish_bytes_atomically", failing_publish)
+
+    with pytest.raises(OSError, match="disk full"):
+        tools._write_capture_output(outputs, "report.png", b"payload")
+
+    assert not (outputs / "report.png").exists()
+    assert not any(record.name == tools.__name__ and record.levelno >= logging.WARNING for record in caplog.records)
+
+
+@pytest.mark.parametrize("publication_error", [OSError("disk full"), KeyboardInterrupt("interrupted")], ids=["publication-error", "interruption"])
+def test_unlink_failure_warns_without_masking_the_publication_error(outputs, monkeypatch, caplog, publication_error):
+    def failing_publish(path: object, content: bytes) -> None:
+        raise publication_error
+
+    def failing_unlink(*args: object, **kwargs: object) -> None:
+        raise OSError("busy")
+
+    monkeypatch.setattr(tools, "_publish_bytes_atomically", failing_publish)
+    monkeypatch.setattr(tools.os, "unlink", failing_unlink)
+
+    with caplog.at_level(logging.WARNING, logger=tools.__name__), pytest.raises(type(publication_error)) as raised:
+        tools._write_capture_output(outputs, "report.png", b"payload")
+    assert raised.value is publication_error
+    destination = outputs / "report.png"
+    assert destination.is_file() and destination.stat().st_size == 0
+    warnings = [record for record in caplog.records if record.name == tools.__name__ and record.levelno == logging.WARNING]
+    assert len(warnings) == 1
+    assert str(destination) in warnings[0].getMessage()
+    assert isinstance(warnings[0].exc_info[1], OSError)
+    assert str(warnings[0].exc_info[1]) == "busy"
+
+
+def test_interruption_propagates_and_frees_the_name(outputs, monkeypatch):
+    def interrupting_publish(path: object, content: bytes) -> None:
+        raise KeyboardInterrupt
+
+    monkeypatch.setattr(tools, "_publish_bytes_atomically", interrupting_publish)
+
+    with pytest.raises(KeyboardInterrupt):
+        tools._write_capture_output(outputs, "report.png", b"payload")
+
+    assert not (outputs / "report.png").exists()
+
+
+def test_probe_bound_is_inclusive(outputs, monkeypatch):
+    """The last probe index is reachable, so a directory full of ``-1``/``-2`` still gets ``-3``."""
+    monkeypatch.setattr(tools, "_MAX_FILENAME_COLLISION_PROBES", 3)
+    for name in ("report.png", "report-1.png", "report-2.png"):
+        (outputs / name).write_bytes(b"earlier")
+
+    assert tools._write_capture_output(outputs, "report.png", b"new") == "report-3.png"
+
+
+def test_saturated_probe_range_still_reserves_a_fresh_name(outputs, monkeypatch):
+    monkeypatch.setattr(tools, "_MAX_FILENAME_COLLISION_PROBES", 2)
+    taken = ["report.png", "report-1.png", "report-2.png", "report-3.png"]
+    for name in taken:
+        (outputs / name).write_bytes(b"earlier")
+
+    claimed = tools._write_capture_output(outputs, "report.png", b"new")
+
+    assert claimed not in taken
+    assert claimed.startswith("report-")
+    assert (outputs / claimed).read_bytes() == b"new"
+    assert all((outputs / name).read_bytes() == b"earlier" for name in taken)
+
+
+def test_exhausted_timestamp_claims_refuse_to_overwrite(outputs, monkeypatch):
+    """When even the fallback name is held, fail loudly rather than returning it.
+
+    A returned name here would be published onto an existing capture, which is
+    the exact silent overwrite the reservation replaces.
+    """
+
+    class _FrozenClock(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            return datetime(2026, 10, 14, 12, 0, 0, 123456, tzinfo=UTC)
+
+    monkeypatch.setattr(tools, "_MAX_FILENAME_COLLISION_PROBES", 1)
+    monkeypatch.setattr(tools, "datetime", _FrozenClock)
+    held = ["report.png", "report-1.png", "report-20261014-120000-123456.png"]
+    for name in held:
+        (outputs / name).write_bytes(b"earlier")
+
+    with pytest.raises(RuntimeError, match="could not reserve an output filename"):
+        tools._dedupe_output_name(outputs, "report.png")
+
+    assert all((outputs / name).read_bytes() == b"earlier" for name in held)