fix(browser): preserve publication errors during failed cleanup (#6871)

* fix(browser): preserve publication errors during failed cleanup

* fix(browser): log atomic output cleanup failures

* fix(browser): keep cleanup logging best effort

* fix(skills): preserve projection manifest publication failures

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
zzr 2026-10-19 16:27:15 +08:00 committed by GitHub
parent bf2cb0e1df
commit 3e6e83c87c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 357 additions and 6 deletions

View File

@ -1527,6 +1527,8 @@ When you install `.skill` archives through the Gateway, DeerFlow accepts standar
Disabling a skill also removes it from the sandbox filesystem view, so shell commands and structured file tools follow the same enabled state. Local, Docker/AIO, hostPath provisioner, and newly created E2B sandboxes source `/mnt/skills` from enabled-only projections that update when public, custom, legacy, or managed integration skills are toggled, edited, created, deleted, or installed. Structured `read_file` calls (including line ranges and read-before-write checks) use the sandbox provider's mount mapping, so the user identity captured when the sandbox was acquired remains authoritative. Managed integration packages remain shared, while their projected filesystem visibility follows each user's enabled state. Multi-worker Gateways re-read on-disk enable state while rebuilding user projections, so a toggle handled by one worker is honored by another worker's next sandbox acquire. Existing E2B sandboxes retain their creation-time snapshot until they are recreated. PVC-backed provisioner skills keep their configured PVC snapshot/layout for now; dynamic PVC materialization is tracked separately.
Skill projection manifest saves preserve the original error or interruption even if temporary-file cleanup or its logging also fails. Manifest files retain their private filesystem permissions.
```
# Paths inside the sandbox container
/mnt/skills/public
@ -2789,6 +2791,10 @@ 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.
If saving fails, the original error is preserved even when temporary-file
cleanup also fails; the previous output remains intact.
Cleanup failures are logged with the temporary-file path for troubleshooting.
Logging-handler errors cannot replace the save error or skip remaining cleanup.
Browserless reserves capture names with empty placeholders to prevent concurrent
captures from overwriting one another. Failed publication releases the placeholder;

View File

@ -20,6 +20,7 @@
- **Shared cache reset**: the prompt-layer caches in `agents/lead_agent/prompt.py` (global enabled-skills list, per-`(app_config, user_id)` entries, rendered `<skill_system>` LRU) are the only skill state that can diverge between Gateway processes sharing one home volume. Every mutation in `routers/skills.py` (install/upload, edit, delete, rollback, enable/disable, reload) refreshes its own process, then calls `publish_skills_cache_reset(user_id=...)`, which atomically replaces `.<extensions config name>.skills-cache-reset.json` beside the shared config under the extensions-config locks (`deerflow.config.shared_reset_marker`). Every cache lookup polls that marker at most once per second and, on a new signature, calls `_invalidate_enabled_skills_cache()` or, for a marker that names one `user_id` as the only publication since the last poll, `invalidate_user_skill_cache(user_id)`; the publishing process skips its own generation. `/skills/reload` runs refresh + publish through the same `await_drained(_drain_skill_mutation(...))` tail as the mutations, so a cancelled caller cannot skip the marker; it answers `scope: shared_config` when the marker was written and `scope: process` when no extensions config path resolves. Publication failures raise `SkillCacheResetPublishError` (a `RuntimeError`, never an `OSError`), so the handlers' `FileNotFoundError -> 404` / `ValueError -> 400` mappings cannot claim an applied change was not found; every handler reports it as a generic 500, and a 200 never means only one process was refreshed. The marker reaches processes mounting that directory, not replicas with independent filesystems. Tests: `tests/test_skills_cache_shared_reset.py`.
- **Tool policy**: Agent `allowed-tools` declarations apply dynamically only to slash-activated skills and skills captured in `ThreadState.skill_context` through configured `read_file` loads; passive enabled skills and skill allowlists do not clamp the baseline tool set. A lead custom Agent's explicit `skills` list is additionally enforced at the sandbox filesystem layer (see Sandbox projection); subagent skill lists still scope discovery and activation only because concurrently delegated subagents share the lead thread sandbox. Subagents render only skill discovery metadata at startup and reuse the same adjacent `SkillActivationMiddleware` + `SkillToolPolicyMiddleware` pair as the lead; their configured `skills` field limits discovery and activation instead of eagerly loading bodies or unioning policies. Slash policy is dominant for its run, preventing subsequently read skills from widening explicit authority; autonomous captured skills use the existing union only when no slash source exists. `tool_search` and `describe_skill` stay available as framework discovery infrastructure, while every discovered or promoted business tool still requires active-policy permission for schema visibility and execution; `task`, `list_background_tasks`, and `cancel_background_task` likewise require explicit declarations. Each active model call intentionally reloads the full live registry so enable/disable changes, frontmatter edits, and custom/public name-shadow winners take effect without a stale TTL or unsafe direct-path cache; all tool calls produced by that model step reuse the resulting source-and-path-signed decision. Registry failures and all-invalid active sets fail closed, while stale individual paths are skipped when another valid skill remains. The dynamic `allowed-tools` policy remains best-effort behavioral scoping: alternate loading paths are not captured and bounded autonomous context may evict entries.
- **Sandbox projection**: `skills/projection.py` materializes enabled-only shared trees at `{base_dir}/skills_view/public` and `{base_dir}/users/{user_id}/skills_view/{custom,legacy,integrations}`. A lead Agent with an explicit `skills` allowlist (including `[]`) gets the intersection of enabled public/user-visible skills and that allowlist at `{base_dir}/users/{user_id}/threads/{thread_id}/skills_view/{public,custom,legacy,integrations}`. `skills=None` keeps the shared zero-copy mount until a thread has used an explicit policy; later unrestricted runs repopulate the same stable thread root with all enabled skills (with host bash enabled, `LocalSandboxProvider` mounts the shared views instead). Rebuilds sign source state, view state, and normalized policy in a manifest, revoke every old category before adding the new policy, stage copies in temporary directories, and atomically replace files. Category root inodes stay stable for live bind mounts; concurrent readers can briefly see fewer skills during a policy change, never a skill revoked by the new policy. Policy-scoped copies reject absolute symlinks and relative symlinks that resolve outside their own skill package, preventing a permitted package from linking back to an omitted source. It copies files into the view (`_copy_into_view`) so a sandbox write cannot mutate the canonical skill inode; the operational trade-off is an O(total bytes) I/O and per-user/thread storage multiplier across rebuilds, prioritized for write isolation over zero-copy hardlinks. Steady-state freshness checks combine source and view metadata tree digests, so in-sandbox view tampering is detected and repaired on the next acquire. Storage writes, archive installs, deletes, and toggles rebuild shared scopes under a cross-process lock; Gateway boot ensures only the shared public view, user views are repaired lazily on acquire, and Agent thread views are recomputed before sandbox reuse. Managed integration packages are global, but their projected category is per-user because enabled state is isolated. User projection rebuilds re-read global enable state from disk instead of the process singleton, so a toggle handled by another Gateway worker is reflected on the next acquire. Gateway public-skill toggles take the public projection lock before the shared `extensions_config_write_lock`, re-read an existing config raw from disk, change only that skill's entry, and rebuild before responding; keep this as one worker-owned critical section so MCP writes cannot interleave and request cancellation cannot release either lock while the worker still runs. The shared public steady-state signature check runs without the global projection lock; stale/error paths take the lock and re-check before rebuilding or clearing. User and thread scope checks are serialized per scope. Projection failures clear the affected view before raising.
- **Projection manifest failures**: `_write_manifest` keeps UTF-8 JSON, private `mkstemp` permissions and same-directory replacement. On ordinary errors or interruptions, reuse `utils.atomic_write.cleanup_failed_atomic_write`; transfer descriptor ownership only after `fdopen` succeeds. Descriptor close, sibling unlink and warning logs must not mask the original error. Coverage: `tests/test_skill_projection_manifest_cleanup.py`.
- **Injection (legacy / default)**: Enabled skills are listed in the agent system prompt with full metadata and container paths (`<available_skills>` block). Controlled by `skills.deferred_discovery: false` (default).
- **Deferred discovery** (`skills.deferred_discovery: true`): Skills are listed by name only in a compact `<skill_index>` block, keeping the system prompt prefix-cache friendly. The agent calls the `describe_skill` tool at runtime to fetch full metadata for skills it wants to use, then loads the SKILL.md via `read_file`. Two new modules support this path:
- `skills/catalog.py` — immutable `SkillCatalog`; query forms: `select:a,b`, `+prefix`, free-text intent terms. Ranked queries use up to 256 characters / 16 unique literal terms, preferring name matches over description-only matches at equal coverage; ties keep catalog order. A lazy per-catalog index caches normalized names/descriptions. Unlike tool search, ranking uses literal intent terms, not regexes. `select:` is parsed before search limits and returns all exact catalog matches without query-length or result caps; other modes cap at `MAX_RESULTS=5`.

View File

@ -18,6 +18,7 @@ from typing import TYPE_CHECKING
from deerflow.skills.parser import parse_skill_file
from deerflow.skills.types import SKILL_MD_FILE, Skill, SkillCategory
from deerflow.utils.atomic_write import cleanup_failed_atomic_write
from deerflow.utils.file_lock import retry_until_locked
if TYPE_CHECKING:
@ -437,11 +438,13 @@ def _write_manifest(scope_root: Path, source_signature: str, view_signature: str
if view_signature is not None:
payload["view_signature"] = view_signature
try:
with os.fdopen(fd, "w", encoding="utf-8") as stream:
stream = os.fdopen(fd, "w", encoding="utf-8")
fd = None # The file object now owns the descriptor.
with stream:
json.dump(payload, stream, sort_keys=True)
temporary.replace(target)
except Exception:
temporary.unlink(missing_ok=True)
except BaseException:
cleanup_failed_atomic_write(temporary, fd)
raise

View File

@ -6,6 +6,16 @@ capture/fetch publication. It writes and fsyncs a hidden exclusive sibling befor
umask; replacements preserve existing permission bits before publication. Never
read the process umask by changing it in this threaded runtime. Behavioral tests
for both tool imports live in `tests/test_browser_outputs_atomic.py`.
`cleanup_failed_atomic_write` shares failure cleanup between this publisher and
`skills/projection.py`'s private manifest writer. Call it while handling the
original `BaseException`; pass a raw descriptor only until `fdopen` succeeds.
Callers retain their own serialization, permissions and publication policy.
Failure cleanup must preserve the original publication exception, including
interruptions. Descriptor close and sibling unlink are separate best-effort
operations: a close failure must not prevent the unlink attempt.
Attempt warning logs for cleanup `OSError`s with the temporary path and exception
details. Guard logging-handler exceptions so diagnostics neither replace the
publication exception (including interruptions) nor skip remaining cleanup.
Both tools take collision names from `output_names.output_name_candidates`
(name, `-1`..`-1000`, five timestamps); claim each atomically, never pre-check.

View File

@ -2,11 +2,37 @@
from __future__ import annotations
import logging
import os
import stat
from contextlib import suppress
from pathlib import Path
from uuid import uuid4
logger = logging.getLogger(__name__)
def cleanup_failed_atomic_write(temporary: Path, fd: int | None = None) -> None:
"""Release an owned temporary without replacing the caller's active error.
Pass *fd* only while the raw descriptor is still owned by the caller;
successful ``fdopen`` transfers that ownership to the file object.
Filesystem and logging-handler errors remain independent best-effort steps.
"""
# Cleanup and diagnostics are best-effort: keep the original error
# and attempt unlink even if closing an untransferred descriptor fails.
if fd is not None:
try:
os.close(fd)
except OSError:
with suppress(Exception):
logger.warning("Failed to close temporary output descriptor for %s", temporary, exc_info=True)
try:
temporary.unlink(missing_ok=True)
except OSError:
with suppress(Exception):
logger.warning("Failed to remove temporary output file %s", temporary, exc_info=True)
def publish_bytes_atomically(path: Path, content: bytes) -> None:
"""Replace *path* with complete bytes, preserving its existing mode.
@ -40,7 +66,5 @@ def publish_bytes_atomically(path: Path, content: bytes) -> None:
os.fsync(handle.fileno())
os.replace(temporary, path)
except BaseException:
if fd is not None:
os.close(fd)
temporary.unlink(missing_ok=True)
cleanup_failed_atomic_write(temporary, fd)
raise

View File

@ -4,6 +4,7 @@
from __future__ import annotations
import inspect
import logging
import os
import stat
from pathlib import Path
@ -119,6 +120,36 @@ def test_publish_failure_preserves_target_and_cleans_temporary(tmp_path, monkeyp
assert list(tmp_path.iterdir()) == []
def test_descriptor_cleanup_failure_still_attempts_unlink(tmp_path, monkeypatch, publisher, caplog):
target = tmp_path / "shot.png"
publish_error = OSError("fdopen failed")
close = os.close
closed = []
def fail_fdopen(*args):
raise publish_error
def fail_close(fd):
close(fd)
closed.append(fd)
raise OSError("close failed")
with monkeypatch.context() as patches:
patches.setattr(os, "fdopen", fail_fdopen)
patches.setattr(os, "close", fail_close)
with pytest.raises(OSError) as caught:
publisher(target, b"new")
assert caught.value is publish_error
assert len(closed) == 1
assert list(tmp_path.iterdir()) == []
records = [record for record in caplog.records if record.name == "deerflow.utils.atomic_write"]
assert len(records) == 1
assert records[0].levelno == logging.WARNING
assert "close" in records[0].getMessage()
assert records[0].exc_info[1].args == ("close failed",)
@pytest.mark.skipif(os.name == "nt", reason="POSIX descriptor permissions")
def test_publish_permission_failure_keeps_previous_file(tmp_path, monkeypatch, publisher):
target = tmp_path / "shot.png"
@ -157,3 +188,132 @@ def test_capture_writes_go_through_the_atomic_publish():
def test_both_tools_share_the_atomic_publisher():
for module in (browser_automation_tools, browserless_tools):
assert module._publish_bytes_atomically is publish_bytes_atomically
@pytest.mark.parametrize("publish_error", [OSError("publication failed"), KeyboardInterrupt("publication interrupted")], ids=["io-error", "interrupt"])
@pytest.mark.parametrize("existing", [False, True], ids=["new", "existing"])
def test_cleanup_failure_preserves_publication_error(tmp_path, monkeypatch, publisher, publish_error, existing, caplog):
target = tmp_path / "shot.png"
if existing:
target.write_bytes(b"old")
cleanup_attempts = []
def fail_publish(*args):
raise publish_error
def fail_cleanup(path, **kwargs):
cleanup_attempts.append(path)
raise PermissionError("cleanup failed")
with monkeypatch.context() as patches:
patches.setattr(os, "replace", fail_publish)
patches.setattr(Path, "unlink", fail_cleanup)
with pytest.raises(type(publish_error)) as caught:
publisher(target, b"new complete capture")
assert caught.value is publish_error
assert len(cleanup_attempts) == 1
assert cleanup_attempts[0].parent == tmp_path
assert cleanup_attempts[0].read_bytes() == b"new complete capture"
records = [record for record in caplog.records if record.name == "deerflow.utils.atomic_write"]
assert len(records) == 1
assert records[0].levelno == logging.WARNING
assert str(cleanup_attempts[0]) in records[0].getMessage()
assert isinstance(records[0].exc_info[1], PermissionError)
if existing:
assert target.read_bytes() == b"old"
else:
assert not target.exists()
def test_both_cleanup_failures_are_logged_without_masking_interruption(tmp_path, monkeypatch, publisher, caplog):
interruption = KeyboardInterrupt("publication interrupted")
close = os.close
def fail_fdopen(*args):
raise interruption
def fail_close(fd):
close(fd)
raise OSError("close failed")
def fail_unlink(*args, **kwargs):
raise PermissionError("unlink failed")
with monkeypatch.context() as patches:
patches.setattr(os, "fdopen", fail_fdopen)
patches.setattr(os, "close", fail_close)
patches.setattr(Path, "unlink", fail_unlink)
with pytest.raises(KeyboardInterrupt) as caught:
publisher(tmp_path / "shot.png", b"new")
assert caught.value is interruption
records = [record for record in caplog.records if record.name == "deerflow.utils.atomic_write"]
assert [record.levelno for record in records] == [logging.WARNING, logging.WARNING]
assert [record.exc_info[1].args for record in records] == [("close failed",), ("unlink failed",)]
assert len(list(tmp_path.glob("*.tmp"))) == 1
@pytest.mark.parametrize("publish_error_type", [OSError, KeyboardInterrupt], ids=["io-error", "interrupt"])
@pytest.mark.parametrize("existing", [False, True], ids=["new", "existing"])
@pytest.mark.parametrize("cleanup_failure", ["close", "unlink", "both"])
def test_failing_log_handler_preserves_error_and_independent_cleanup(tmp_path, monkeypatch, publisher, caplog, publish_error_type, existing, cleanup_failure):
target = tmp_path / "shot.png"
if existing:
target.write_bytes(b"old")
publish_error = publish_error_type("publication failed")
close = os.close
unlink = Path.unlink
unlink_attempts = []
warning_attempts = []
def fail_publish(*args):
raise publish_error
def fail_close(fd):
close(fd)
raise OSError("close failed")
def cleanup(path, **kwargs):
unlink_attempts.append(path)
if cleanup_failure in {"unlink", "both"}:
raise PermissionError("unlink failed")
return unlink(path, **kwargs)
# A standard delayed FileHandler raises from emit when its directory is
# unavailable; this exercises logging itself rather than mocking warning().
handler = logging.FileHandler(tmp_path / "missing-log-directory" / "cleanup.log", delay=True)
emit = handler.emit
def record_warning(record):
warning_attempts.append(record)
emit(record)
log = logging.getLogger("deerflow.utils.atomic_write")
log.addHandler(handler)
try:
with monkeypatch.context() as patches:
patches.setattr(handler, "emit", record_warning)
patches.setattr(Path, "unlink", cleanup)
if cleanup_failure in {"close", "both"}:
patches.setattr(os, "fdopen", fail_publish)
patches.setattr(os, "close", fail_close)
else:
patches.setattr(os, "replace", fail_publish)
with caplog.at_level(logging.WARNING, logger=log.name), pytest.raises(publish_error_type) as caught:
publisher(target, b"new complete capture")
finally:
log.removeHandler(handler)
handler.close()
assert caught.value is publish_error
assert len(unlink_attempts) == 1
temporary = unlink_attempts[0]
assert temporary.parent == tmp_path
assert temporary.exists() == (cleanup_failure in {"unlink", "both"})
assert len(warning_attempts) == (2 if cleanup_failure == "both" else 1)
assert all(str(temporary) in record.getMessage() and record.exc_info is not None for record in warning_attempts)
if existing:
assert target.read_bytes() == b"old"
else:
assert not target.exists()

View File

@ -0,0 +1,147 @@
"""Skill projection manifests preserve publication errors through failed cleanup."""
from __future__ import annotations
import json
import logging
import os
import stat
from contextlib import suppress
from pathlib import Path
from unittest.mock import Mock
import pytest
class _Records(logging.Handler):
def __init__(self):
super().__init__()
self.records = []
def emit(self, record):
self.records.append(record)
@pytest.mark.parametrize("phase", ["fdopen", "fdopen_close", "write", "replace"])
@pytest.mark.parametrize("primary_type", [RuntimeError, KeyboardInterrupt])
@pytest.mark.parametrize("existing", [False, True])
@pytest.mark.parametrize("unlink_fails", [False, True])
@pytest.mark.parametrize("logging_fails", [False, True])
def test_manifest_preserves_primary_error_and_independent_cleanup(tmp_path, monkeypatch, phase, primary_type, existing, unlink_fails, logging_fails):
from deerflow.skills import projection
from deerflow.utils import atomic_write
scope = tmp_path / "scope"
scope.mkdir()
target = scope / ".projection-manifest.json"
if existing:
target.write_bytes(b"previous manifest")
primary = primary_type("manifest publication failed")
allocated = []
unlinks = []
closes = []
real_allocate = projection.tempfile.mkstemp
real_fdopen = os.fdopen
real_close = os.close
real_fstat = os.fstat
real_unlink = Path.unlink
real_replace = Path.replace
def allocate(*args, **kwargs):
fd, name = real_allocate(*args, **kwargs)
allocated.append((fd, Path(name)))
return fd, name
def fdopen(fd, *args, **kwargs):
if allocated and fd == allocated[0][0] and phase.startswith("fdopen"):
raise primary
return real_fdopen(fd, *args, **kwargs)
def close(fd):
if allocated and fd == allocated[0][0]:
closes.append(fd)
real_close(fd)
if phase == "fdopen_close":
raise OSError("descriptor cleanup failed")
else:
real_close(fd)
def dump(payload, stream, **kwargs):
stream.write('{"partial":')
raise primary
def replace(path, destination):
if allocated and path == allocated[0][1]:
raise primary
return real_replace(path, destination)
def unlink(path, *args, **kwargs):
if allocated and path == allocated[0][1]:
unlinks.append(path)
if unlink_fails:
raise PermissionError("temporary unlink failed")
return real_unlink(path, *args, **kwargs)
monkeypatch.setattr(projection.tempfile, "mkstemp", allocate)
monkeypatch.setattr(os, "fdopen", fdopen)
monkeypatch.setattr(os, "close", close)
monkeypatch.setattr(Path, "unlink", unlink)
if phase == "write":
monkeypatch.setattr(projection.json, "dump", dump)
if phase == "replace":
monkeypatch.setattr(Path, "replace", replace)
records = _Records()
# A real stdlib handler: opening this delayed destination raises from emit.
broken = logging.FileHandler(tmp_path / "missing-logs" / "gateway.log", delay=True)
broken.emit = Mock(wraps=broken.emit)
monkeypatch.setattr(atomic_write.logger, "handlers", [records, broken] if logging_fails else [records])
monkeypatch.setattr(atomic_write.logger, "propagate", False)
monkeypatch.setattr(atomic_write.logger, "level", logging.WARNING)
try:
with pytest.raises(primary_type) as raised:
projection._write_manifest(scope, "source", "view")
assert raised.value is primary
assert len(allocated) == 1
fd, temporary = allocated[0]
assert unlinks == [temporary]
assert closes == ([fd] if phase.startswith("fdopen") else [])
with pytest.raises(OSError):
real_fstat(fd)
assert temporary.exists() == unlink_fails
if existing:
assert target.read_bytes() == b"previous manifest"
else:
assert not target.exists()
expected_warnings = int(phase == "fdopen_close") + int(unlink_fails)
assert len(records.records) == expected_warnings
assert all(record.exc_info and str(temporary) in record.getMessage() for record in records.records)
assert broken.emit.call_count == (expected_warnings if logging_fails else 0)
finally:
broken.close()
records.close()
for fd, temporary in allocated:
with suppress(OSError):
real_close(fd)
real_unlink(temporary, missing_ok=True)
@pytest.mark.parametrize("existing", [False, True])
@pytest.mark.parametrize("view_signature", [None, "view-签名"])
def test_manifest_success_keeps_format_and_private_permissions(tmp_path, existing, view_signature):
from deerflow.skills import projection
scope = tmp_path / "scope"
scope.mkdir()
target = scope / ".projection-manifest.json"
if existing:
target.write_bytes(b"old")
target.chmod(0o644)
projection._write_manifest(scope, "source-签名", view_signature)
expected = {"version": projection._MANIFEST_VERSION, "source_signature": "source-签名"}
if view_signature is not None:
expected["view_signature"] = view_signature
assert target.read_text(encoding="utf-8") == json.dumps(expected, sort_keys=True)
if os.name != "nt":
assert stat.S_IMODE(target.stat().st_mode) == 0o600
assert list(scope.iterdir()) == [target]