diff --git a/CHANGELOG.md b/CHANGELOG.md
index ef8bea6b..7e9b7ea6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -26,6 +26,9 @@ and versions follow [Semantic Versioning](https://semver.org/).
   approvals and rules, and Stop everything stops it. It only answers your chats: tasks, suggestions and other
   background work keep using your other models, and it can't be the memory search model. Anthropic may change how
   this is counted or allowed.
+- **Browser downloads:** files Sentient downloads while using a website are saved to `downloads/` in your Files
+  folder, and the browser result says where. In an attached browser, only downloads from Sentient's own tabs are
+  saved.
 - **Push to talk and dictation into any app:** hold Ctrl+Alt+Shift+T (Cmd+Option+Shift+T on a Mac) anywhere, speak,
   and let go: what you said goes to Sentient as a chat message, and the answer is read aloud. Press Ctrl+Alt+Shift+D,
   speak, and press it again (or just pause): your words are typed where your cursor is, in any app. A small bar at the
@@ -234,6 +237,7 @@ and versions follow [Semantic Versioning](https://semver.org/).
 - Everyday phrases like "If you don't know, say so" no longer make Sentient suggest a Never rule for unrelated
   tools. A suggested rule for a single tool now needs your words to name an action, the app or the tool, so
   "Never write files for me" still suggests blocking file writing.
+- Edge's downloads hub no longer replaces the active page or appears as a browser tab when a download starts.
 
 ### Security
 - Once Sentient has read an email, a web page, a message or anything else other people wrote, it asks before
diff --git a/docs/API.md b/docs/API.md
index 9a9b68d7..b913d0a0 100644
--- a/docs/API.md
+++ b/docs/API.md
@@ -1439,9 +1439,26 @@ Every new tool declares a `Risk`; approvals behave as in section 1.
   idle_minutes, allow_domains, block_domains, max_snapshot_chars, max_extract_chars, confirm_purchases, live_view,
   profiles`. Every safety rule below applies the same in every profile, attached ones included.
 - Tools (plugin `browser`). Failures return `{error}` with a message the model can act on.
+  - Downloads from launched profiles are accepted by Playwright; in an attached profile, only downloads from the tab
+    Sentient opens and popups opened from Sentient-owned tabs are handled (not the user's existing tabs). Every handled
+    download is saved under `downloads/` in Sentient's Files folder. Browser tool results (including `browser_scroll`)
+    may include an optional `downloads` array of relative paths such as `["downloads/report.pdf"]`.
+    `browser_open`, `browser_click`, `browser_type`, `browser_press`, `browser_select` and `browser_back` wait up to
+    0.3 s for a download to start during the action. Opening a URL that directly starts a download returns a normal
+    browser result with its download status rather than a navigation error. Edge's internal downloads hub is excluded
+    from browser tabs and does not replace the active page. If a download starts later, it remains tracked and is
+    reported on a later browser result after saving finishes. Other browser results collect finished downloads
+    immediately without that start wait. Actions do not wait for downloads to finish; unfinished downloads are listed
+    by filename in an optional `downloads_in_progress` array and reported under `downloads` when a later result observes
+    that they have finished. Saves reserve unique filenames under a short lock, then transfer concurrently. Each save
+    is limited to 120 s total; when the limit is reached, Sentient asks the browser to cancel it, with a 5 s limit for
+    that cancellation request. A cancellation failure is included in `download_errors`.
+    Failed saves are reported in an optional `download_errors` array; they don't discard the browser action result or
+    successful downloads from the same batch. Completed downloads that no browser result consumes become eligible for
+    cleanup after 5 minutes, and active browser actions get the chance to report them before cleanup.
   - `browser_open(url, profile="")` read → same as `browser_snapshot` plus `profile` (http/https only; allow/block lists
-    apply, also after redirects). `profile` switches to that profile for this and the following calls of the run;
-    empty uses the run's profile, else `default`.
+    apply, also after redirects). `profile` switches to that profile for this and the following calls of the run; empty
+    uses the run's profile, else `default`.
   - `browser_snapshot()` read → `{url, title, text, truncated?}`. `text` is `Page:`/`URL:`/`Scroll:` header, interactive
     elements one per line (`[e12] button "Sign in"`, `[e4] textbox "Search" value=""`, `[e7] combobox "Country"
     value="India" options: India | Japan`, `[e3] link "Docs" -> /docs`, flags `checked`, `disabled`, `focused`) and the
diff --git a/sentient/browser/service.py b/sentient/browser/service.py
index bf885c6d..73909a89 100644
--- a/sentient/browser/service.py
+++ b/sentient/browser/service.py
@@ -64,6 +64,11 @@
 STORAGE_FLAG = "--enable-aggressive-domstorage-flushing"
 STORAGE_FLUSH_S = 1.5
 PAGE_CLOSE_TIMEOUT_S = 5.0
+DOWNLOAD_APPEAR_GRACE_S = 0.3
+DOWNLOAD_MAX_DURATION_S = 120.0
+DOWNLOAD_CANCEL_TIMEOUT_S = 5.0
+DOWNLOAD_TASK_RETENTION_S = 300.0
+DOWNLOAD_TASK_CLEANUP_INTERVAL_S = 30.0
 # models often pass the whole snapshot line ("[e4] button \"Place order\"") instead of just "e4"
 _REF_RE = re.compile(r"\b(e\d+)\b", re.IGNORECASE)
 
@@ -204,6 +209,8 @@ def __init__(self, app):
         self._context: Any = None
         self._browser: Any = None  # attach profiles: the user's browser we are connected to
         self._attached = False
+        self._download_pages: set[Any] = set()
+        self._internal_pages: set[Any] = set()
         self._profile = DEFAULT_PROFILE  # the running profile, or the next one to start
         self._for_user = False  # the open window was shown for the user (Open to sign in): don't switch under them
         self._engine: str | None = None
@@ -226,6 +233,11 @@ def __init__(self, app):
         self._site_seen = False  # a tab of the open browser showed a website (its storage may need writing out)
         self._last_tabs_sig: Any = None
         self.idle_check_s = 30.0
+        self._download_tasks: set[asyncio.Task[str]] = set()
+        self._download_names: dict[asyncio.Task[str], str] = {}
+        self._download_completed_at: dict[asyncio.Task[str], float] = {}
+        self._download_cleanup_task: asyncio.Task | None = None
+        self._download_lock = asyncio.Lock()
 
     # ------------------------------------------------------------------ lifecycle
     async def start(self) -> None:
@@ -298,13 +310,19 @@ async def _tabs(self) -> list[dict]:
         if ctx is None:
             return []
         out = []
-        for i, page in enumerate(list(ctx.pages)):
+        for i, page in enumerate(self._usable_pages(ctx)):
             title = ""
             with contextlib.suppress(Exception):
                 title = await asyncio.wait_for(page.title(), timeout=2)
             out.append({"index": i, "url": page.url, "title": title, "active": page is self._active})
         return out
 
+    def _usable_pages(self, context: Any) -> list[Any]:
+        return [
+            page for page in list(context.pages)
+            if not page.is_closed() and page not in self._internal_pages
+        ]
+
     async def status(self) -> dict:
         running = self._context is not None
         name = self._profile if running else DEFAULT_PROFILE
@@ -520,10 +538,13 @@ async def _launch(self, headless: bool) -> None:
         context.on("page", self._on_page)
         for page in context.pages:
             self._wire_page(page)
+            if self._is_downloads_hub(page.url):
+                self._internal_pages.add(page)
         if self._attached:  # work in a tab of our own, never in one the user is using
-            self._active = await context.new_page()
+            self._active = await self._new_page(context)
         else:
-            self._active = context.pages[0] if context.pages else await context.new_page()
+            pages = self._usable_pages(context)
+            self._active = pages[0] if pages else await context.new_page()
         self._last_used = time.monotonic()
         if self._idle_task is None or self._idle_task.done():
             self._idle_task = asyncio.create_task(self._idle_watch(), name="browser:idle")
@@ -538,6 +559,7 @@ async def _launch_persistent(self, prof: Any, headless: bool) -> Any:
             kwargs: dict[str, Any] = {
                 "user_data_dir": str(folder),
                 "headless": headless,
+                "accept_downloads": True,
                 "timeout": 45_000,
                 "args": ["--no-first-run", "--no-default-browser-check", "--hide-crash-restore-bubble", STORAGE_FLAG],
             }
@@ -601,10 +623,16 @@ async def _ensure(self, ctx: Any = None) -> Any:
     async def _page(self, ctx: Any = None) -> Any:
         c = await self._ensure(ctx)
         if self._active is None or self._active.is_closed():
-            pages = [p for p in c.pages if not p.is_closed()]
-            self._active = pages[-1] if pages and not self._attached else await c.new_page()
+            pages = self._usable_pages(c)
+            self._active = pages[-1] if pages and not self._attached else await self._new_page(c)
         return self._active
 
+    async def _new_page(self, context: Any) -> Any:
+        page = await context.new_page()
+        if self._attached:
+            self._enable_attached_page_downloads(page)
+        return page
+
     async def _close_context(self) -> None:
         """Close the running profile. An attached browser is only disconnected, never closed. Hold ``_life_lock``."""
         ctx, browser, attached = self._context, self._browser, self._attached
@@ -627,6 +655,8 @@ async def _close_context(self) -> None:
             self._active = None
             self._snap = None
             self._focused = None
+            self._download_pages.clear()
+            self._internal_pages.clear()
 
     async def _flush_storage(self, ctx: Any) -> None:
         """Let a launched browser write site storage to disk before it closes: close its tabs with their unload
@@ -719,6 +749,7 @@ def _on_context_closed(self, context: Any) -> None:
         self._active = None
         self._snap = None
         self._focused = None
+        self._download_pages.clear()
         if not self._closing:  # the user closed the visible window (or their attached browser)
             self._spawn(self._publish_status())
 
@@ -727,30 +758,65 @@ def _wire_page(self, page: Any) -> None:
         page.on("dialog", self._on_dialog)
         page.on("popup", lambda popup: self._on_popup(page, popup))
         page.on("framenavigated", lambda frame: self._on_navigated(page, frame))
+        if not self._attached:
+            page.on("download", self._on_download)
+
+    def _enable_attached_page_downloads(self, page: Any) -> None:
+        if page not in self._download_pages:
+            self._download_pages.add(page)
+            page.on("download", self._on_download)
 
     def _on_navigated(self, page: Any, frame: Any) -> None:
+        if frame is page.main_frame and self._is_downloads_hub(frame.url):
+            self._mark_internal_page(page)
+            return
         # any website shown, by the assistant or by the user in a visible window
         if frame is page.main_frame and str(frame.url or "").startswith(("http://", "https://")):
             self._site_seen = True
 
+    @staticmethod
+    def _is_downloads_hub(url: Any) -> bool:
+        return str(url or "").lower().startswith("edge://downloads-hub")
+
+    def _mark_internal_page(self, page: Any) -> None:
+        self._internal_pages.add(page)
+        if self._active is page:
+            context = self._context
+            pages = [
+                candidate for candidate in (self._usable_pages(context) if context is not None else [])
+                if not self._attached or candidate in self._download_pages
+            ]
+            self._active = pages[-1] if pages else None
+
     def _on_page(self, page: Any) -> None:
         self._wire_page(page)
-        if not self._attached:  # a link opened a new tab: keep working in it
+        if self._is_downloads_hub(page.url):
+            self._mark_internal_page(page)
+        elif not self._attached:  # a link opened a new tab: keep working in it
             self._active = page
         self._spawn(self._publish_status())
 
     def _on_popup(self, opener: Any, popup: Any) -> None:
-        # attached: follow only tabs our own tab opened, never ones the user opens
-        if self._attached and self._active is opener:
-            self._active = popup
+        if self._is_downloads_hub(popup.url):
+            self._mark_internal_page(popup)
+            return
+        if self._attached and opener in self._download_pages:
+            self._enable_attached_page_downloads(popup)
+            if self._active is opener:
+                self._active = popup
 
     def _on_page_closed(self, page: Any) -> None:
+        self._download_pages.discard(page)
+        self._internal_pages.discard(page)
         if self._snap and self._snap.get("page") is page:
             self._snap = None
         ctx = self._context
         if self._active is page:
-            remaining = [p for p in (ctx.pages if ctx else []) if p is not page and not p.is_closed()]
-            self._active = remaining[-1] if remaining and not self._attached else None
+            remaining = [
+                p for p in (self._usable_pages(ctx) if ctx else [])
+                if p is not page and (not self._attached or p in self._download_pages)
+            ]
+            self._active = remaining[-1] if remaining else None
         if ctx is not None and not self._closing:
             self._spawn(self._publish_status())
 
@@ -762,6 +828,194 @@ async def _on_dialog(self, dialog: Any) -> None:
             else:
                 await dialog.dismiss()
 
+    def _on_download(self, download: Any) -> None:
+        task = asyncio.create_task(self._save_download(download), name="browser:download")
+        self._download_tasks.add(task)
+        self._download_names[task] = self._download_filename(download)
+        task.add_done_callback(self._on_download_task_done)
+
+    def _on_download_task_done(self, task: asyncio.Task[str]) -> None:
+        if not task.cancelled():
+            with contextlib.suppress(Exception):
+                task.exception()
+        if task in self._download_tasks:
+            self._download_completed_at[task] = time.monotonic()
+            if self._download_cleanup_task is None or self._download_cleanup_task.done():
+                self._download_cleanup_task = asyncio.create_task(
+                    self._cleanup_download_tasks(), name="browser:download-cleanup"
+                )
+                self._bg.add(self._download_cleanup_task)
+                self._download_cleanup_task.add_done_callback(self._bg.discard)
+
+    async def _cleanup_download_tasks(self) -> None:
+        try:
+            while self._download_completed_at:
+                await asyncio.sleep(DOWNLOAD_TASK_CLEANUP_INTERVAL_S)
+                expired_before = time.monotonic() - DOWNLOAD_TASK_RETENTION_S
+                async with self._lock:
+                    expired = [
+                        task
+                        for task, completed_at in self._download_completed_at.items()
+                        if completed_at <= expired_before and task.done()
+                    ]
+                    for task in expired:
+                        self._download_tasks.discard(task)
+                        self._download_names.pop(task, None)
+                        self._download_completed_at.pop(task, None)
+        finally:
+            self._download_cleanup_task = None
+
+    @staticmethod
+    def _download_filename(download: Any) -> str:
+        raw_name = str(getattr(download, "suggested_filename", None) or "download")
+        name = raw_name.replace("\\", "/").rsplit("/", 1)[-1]
+        return re.sub(r"[^\w.\- ()]+", "_", name).strip(" .")[:160] or "download"
+
+    async def _save_download(self, download: Any) -> str:
+        name = self._download_filename(download)
+
+        def reject_symlinked_folder(folder: Path) -> None:
+            if folder.is_symlink():
+                raise BrowserError("The downloads folder must not be a symbolic link.")
+
+        async def reserve_target() -> Path:
+            async with self._download_lock:
+                folder = paths.files_dir() / "downloads"
+                reject_symlinked_folder(folder)
+                folder.mkdir(parents=True, exist_ok=True)
+                stem = Path(name).stem
+                suffix = Path(name).suffix
+                index = 0
+                while True:
+                    candidate_name = f"{stem}({index}){suffix}" if index else name
+                    candidate = folder / candidate_name
+                    try:
+                        with candidate.open("xb"):
+                            pass
+                    except FileExistsError:
+                        index += 1
+                        continue
+                    return candidate
+
+        async def request_download_cancel() -> str | None:
+            cancel_task = asyncio.create_task(download.cancel())
+            done, _ = await asyncio.wait({cancel_task}, timeout=DOWNLOAD_CANCEL_TIMEOUT_S)
+            if not done:
+                cancel_task.cancel()
+                cancel_task.add_done_callback(
+                    lambda task: task.exception() if not task.cancelled() else None
+                )
+                return f"timed out after {DOWNLOAD_CANCEL_TIMEOUT_S:g} seconds"
+            try:
+                cancel_task.result()
+            except asyncio.CancelledError:
+                return "the cancellation request was cancelled"
+            except Exception as cancel_exc:
+                return str(cancel_exc).strip() or type(cancel_exc).__name__
+            return None
+
+        def save_failure(exc: BaseException) -> str:
+            reason = str(exc).strip().splitlines()[0][:300] or type(exc).__name__
+            return f"Couldn't save downloaded file '{name}': {reason}"
+
+        timeout = asyncio.timeout(DOWNLOAD_MAX_DURATION_S)
+        target: Path | None = None
+        try:
+            async with timeout:
+                target = await reserve_target()
+                await download.save_as(str(target))
+        except asyncio.CancelledError as exc:
+            try:
+                cancel_error = await request_download_cancel()
+            finally:
+                cleanup_error = self._remove_download_reservation(target)
+            if cancel_error:
+                exc.add_note(f"Browser cancellation failed: {cancel_error}")
+            if cleanup_error:
+                exc.add_note(f"Couldn't remove the reserved file: {cleanup_error}")
+            raise
+        except TimeoutError as exc:
+            if timeout.expired():
+                cancel_error = await request_download_cancel()
+                cleanup_error = self._remove_download_reservation(target)
+                message = f"Download '{name}' did not finish within {DOWNLOAD_MAX_DURATION_S:g} seconds."
+                if cancel_error:
+                    message += f" Browser cancellation failed: {cancel_error}"
+                if cleanup_error:
+                    message += f" Couldn't remove the reserved file: {cleanup_error}"
+                raise BrowserError(message) from exc
+            cleanup_error = self._remove_download_reservation(target)
+            message = save_failure(exc)
+            if cleanup_error:
+                message += f"; couldn't remove the reserved file: {cleanup_error}"
+            raise BrowserError(message) from exc
+        except Exception as exc:
+            cleanup_error = self._remove_download_reservation(target)
+            if isinstance(exc, BrowserError):
+                if cleanup_error:
+                    raise BrowserError(f"{exc}; couldn't remove the reserved file: {cleanup_error}") from exc
+                raise
+            message = save_failure(exc)
+            if cleanup_error:
+                message += f"; couldn't remove the reserved file: {cleanup_error}"
+            raise BrowserError(message) from exc
+        assert target is not None
+        return target.relative_to(paths.files_dir()).as_posix()
+
+    @staticmethod
+    def _remove_download_reservation(target: Path | None) -> str | None:
+        if target is None:
+            return None
+        try:
+            target.unlink(missing_ok=True)
+        except OSError as exc:
+            return str(exc).strip() or type(exc).__name__
+        return None
+
+    async def _include_downloads(
+        self,
+        result: dict,
+        before: set[asyncio.Task[str]],
+        wait_for_download_start: bool = False,
+    ) -> dict:
+        if wait_for_download_start:
+            deadline = time.monotonic() + DOWNLOAD_APPEAR_GRACE_S
+            while time.monotonic() < deadline:
+                if self._download_tasks - before or any(task.done() for task in self._download_tasks):
+                    break
+                await asyncio.sleep(0.05)
+
+        # Collect finished downloads, but never wait for a save to complete inside a browser action.
+        tasks = (self._download_tasks - before) | {task for task in self._download_tasks if task.done()}
+        done = {task for task in tasks if task.done()}
+        if done:
+            try:
+                outcomes = await asyncio.gather(*done, return_exceptions=True)
+            finally:
+                self._download_tasks.difference_update(done)
+                for task in done:
+                    self._download_names.pop(task, None)
+                    self._download_completed_at.pop(task, None)
+            names = []
+            errors = []
+            for outcome in outcomes:
+                if isinstance(outcome, BaseException):
+                    message = str(outcome).strip() or f"{type(outcome).__name__} while saving download"
+                    errors.append(message)
+                else:
+                    names.append(outcome)
+            if names:
+                result["downloads"] = names
+            if errors:
+                result["download_errors"] = errors
+
+        pending = [task for task in self._download_tasks if not task.done()]
+        if pending:
+            result["downloads_in_progress"] = [
+                self._download_names.get(task, "download") for task in pending
+            ]
+        return result
+
     async def _idle_watch(self) -> None:
         while self._context is not None:
             await asyncio.sleep(self.idle_check_s)
@@ -979,20 +1233,34 @@ async def open(self, ctx: Any, url: str, profile: str = "") -> dict:
             raise BrowserError(problem)
         self.use_profile(ctx, profile)
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page = await self._page(ctx)
-            await page.goto(target, wait_until="domcontentloaded")
-            await self._settle(page)
+            download_started = False
+            try:
+                await page.goto(target, wait_until="domcontentloaded")
+            except Exception as exc:
+                if "Download is starting" not in str(exc):
+                    raise
+                download_started = True
+            if not download_started:
+                await self._settle(page)
             await self._enforce_domains(page)
             result = await self._snapshot_locked(page)
             result["profile"] = self._profile
             await self._after_action(ctx, page)
+            result = await self._include_downloads(
+                result, downloads_before, wait_for_download_start=True
+            )
             return self._take_dialogs(result)
 
     async def snapshot(self, ctx: Any) -> dict:
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page = await self._page(ctx)
             self._last_used = time.monotonic()
-            return self._take_dialogs(await self._snapshot_locked(page))
+            result = await self._snapshot_locked(page)
+            result = await self._include_downloads(result, downloads_before)
+            return self._take_dialogs(result)
 
     async def _snapshot_locked(self, page: Any) -> dict:
         cfg = self.app.config.browser
@@ -1013,6 +1281,7 @@ async def _snapshot_locked(self, page: Any) -> dict:
 
     async def click(self, ctx: Any, ref: str) -> dict:
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page, loc, info = await self._locate(ctx, ref)
             if info.get("disabled"):
                 raise BrowserError(f"{format_element(info)} is disabled. Something else on the page may need to be done first.")
@@ -1020,7 +1289,7 @@ async def click(self, ctx: Any, ref: str) -> dict:
             refusal = self._guard("click", info["ref"], live_risk)
             if refusal:
                 return {"error": refusal}
-            pages_before = len(self._context.pages)
+            pages_before = len(self._usable_pages(self._context))
             self._accept_confirm = live_risk >= Risk.send or self.app.config.tools.approvals.mode == "off"
             try:
                 await loc.click(timeout=10_000)
@@ -1030,7 +1299,11 @@ async def click(self, ctx: Any, ref: str) -> dict:
             # a new tab arrives as a separate event; links with target=_blank get longer to show up
             limit = 3.0 if str(info.get("target", "")).lower() == "_blank" else 0.3
             waited = 0.0
-            while self._context is not None and len(self._context.pages) <= pages_before and waited < limit:
+            while (
+                self._context is not None
+                and len(self._usable_pages(self._context)) <= pages_before
+                and waited < limit
+            ):
                 await asyncio.sleep(0.05)
                 waited += 0.05
             if self._context is None:
@@ -1044,14 +1317,18 @@ async def click(self, ctx: Any, ref: str) -> dict:
                 "url": active.url,
                 "message": "Clicked. Call browser_snapshot to see the page now.",
             }
-            if len(self._context.pages) > pages_before:
+            if len(self._usable_pages(self._context)) > pages_before:
                 out["new_tab"] = True
                 out["message"] = "Clicked; it opened a new tab, which is now active. Call browser_snapshot to see it."
             await self._after_action(ctx, active)
+            out = await self._include_downloads(
+                out, downloads_before, wait_for_download_start=True
+            )
             return self._take_dialogs(out)
 
     async def type(self, ctx: Any, ref: str, text: str, submit: bool = False) -> dict:
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page, loc, info = await self._locate(ctx, ref)
             kind = safety.sensitive_field(info) or safety.sensitive_field(info.get("live"))
             if kind:
@@ -1082,10 +1359,14 @@ async def type(self, ctx: Any, ref: str, text: str, submit: bool = False) -> dic
             described = format_element({k: v for k, v in info.items() if k != "value"})
             out = {"ok": True, "typed_into": described, "submitted": bool(submit), "url": active.url}
             await self._after_action(ctx, active)
+            out = await self._include_downloads(
+                out, downloads_before, wait_for_download_start=True
+            )
             return self._take_dialogs(out)
 
     async def select(self, ctx: Any, ref: str, option: str) -> dict:
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page, loc, info = await self._locate(ctx, ref)
             if str(info.get("tag", "")).lower() != "select":
                 raise BrowserError(
@@ -1108,11 +1389,15 @@ async def select(self, ctx: Any, ref: str, option: str) -> dict:
             await self._settle(page, 1_000)
             out = {"ok": True, "selected": option, "in": format_element(info), "url": page.url}
             await self._after_action(ctx, page)
+            out = await self._include_downloads(
+                out, downloads_before, wait_for_download_start=True
+            )
             return self._take_dialogs(out)
 
     async def press(self, ctx: Any, key: str) -> dict:
         name = normalize_key(key)
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page = await self._page(ctx)
             focused = None
             with contextlib.suppress(Exception):
@@ -1135,6 +1420,9 @@ async def press(self, ctx: Any, key: str) -> dict:
             await self._enforce_domains(active)
             out = {"ok": True, "pressed": name, "url": active.url}
             await self._after_action(ctx, active)
+            out = await self._include_downloads(
+                out, downloads_before, wait_for_download_start=True
+            )
             return self._take_dialogs(out)
 
     async def scroll(self, ctx: Any, direction: str = "down") -> dict:
@@ -1150,6 +1438,7 @@ async def scroll(self, ctx: Any, direction: str = "down") -> dict:
         if d not in scripts:
             raise BrowserError("direction must be one of: down, up, top, bottom, left, right.")
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page = await self._page(ctx)
             pos = await page.evaluate(
                 "() => { " + scripts[d] + "; return [Math.round(scrollY), Math.round(document.documentElement.scrollHeight"
@@ -1159,10 +1448,11 @@ async def scroll(self, ctx: Any, direction: str = "down") -> dict:
             out = {"ok": True, "scrolled": d, "from_top": pos[0], "more_below": max(0, pos[1]),
                    "message": "Call browser_snapshot to see what is visible now."}
             await self._after_action(ctx, page)
-            return out
+            return await self._include_downloads(out, downloads_before)
 
     async def back(self, ctx: Any) -> dict:
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page = await self._page(ctx)
             resp = await page.go_back(wait_until="domcontentloaded")
             if resp is None and page.url in {"about:blank", ""}:
@@ -1174,19 +1464,24 @@ async def back(self, ctx: Any) -> dict:
                 title = await page.title()
             out = {"ok": True, "url": page.url, "title": title}
             await self._after_action(ctx, page)
-            return out
+            return await self._include_downloads(
+                out, downloads_before, wait_for_download_start=True
+            )
 
     async def tabs(self, ctx: Any, profile: str = "") -> dict:
         self.use_profile(ctx, profile)
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             await self._ensure(ctx)
             self._last_used = time.monotonic()
-            return {"profile": self._profile, "tabs": await self._tabs()}
+            result = {"profile": self._profile, "tabs": await self._tabs()}
+            return await self._include_downloads(result, downloads_before)
 
     async def switch_tab(self, ctx: Any, index: int) -> dict:
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             c = await self._ensure(ctx)
-            pages = list(c.pages)
+            pages = self._usable_pages(c)
             if not 0 <= index < len(pages):
                 raise BrowserError(f"There is no tab {index}. Open tabs are numbered 0 to {len(pages) - 1}.")
             url = pages[index].url or ""
@@ -1202,11 +1497,12 @@ async def switch_tab(self, ctx: Any, index: int) -> dict:
             out = {"ok": True, "index": index, "url": self._active.url, "title": title,
                    "message": "Switched. Call browser_snapshot to see this tab."}
             await self._after_action(ctx, self._active)
-            return out
+            return await self._include_downloads(out, downloads_before)
 
     async def extract(self, ctx: Any, question: str = "") -> dict:
         cfg = self.app.config.browser
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page = await self._page(ctx)
             self._last_used = time.monotonic()
             data = await page.evaluate(EXTRACT_JS, {"maxText": cfg.max_extract_chars * 5})
@@ -1216,10 +1512,11 @@ async def extract(self, ctx: Any, question: str = "") -> dict:
                 out["question"] = question
             if truncated:
                 out["truncated"] = True
-            return out
+            return await self._include_downloads(out, downloads_before)
 
     async def screenshot(self, ctx: Any) -> dict:
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
             page = await self._page(ctx)
             self._last_used = time.monotonic()
             folder = paths.files_dir() / "outputs" / "browser"
@@ -1229,13 +1526,51 @@ async def screenshot(self, ctx: Any) -> dict:
             title = ""
             with contextlib.suppress(Exception):
                 title = await page.title()
-            return {"file": f"outputs/browser/{name}", "url": page.url, "title": title}
+            result = {"file": f"outputs/browser/{name}", "url": page.url, "title": title}
+            return await self._include_downloads(result, downloads_before)
 
     async def close_tool(self, ctx: Any) -> dict:
-        was_running = self._context is not None
         async with self._lock:
+            downloads_before = self._download_tasks.copy()
+            was_running = self._context is not None
+            shutdown_errors = []
+            download_deadline = time.monotonic() + DOWNLOAD_MAX_DURATION_S
+            shutdown_deadline = download_deadline + DOWNLOAD_CANCEL_TIMEOUT_S
+            while pending := {task for task in self._download_tasks if not task.done()}:
+                _, pending = await asyncio.wait(
+                    pending,
+                    timeout=max(0.0, download_deadline - time.monotonic()),
+                )
+                if not pending:
+                    continue
+
+                interrupted_names = {
+                    task: self._download_names.get(task, "download") for task in pending
+                }
+                for task in pending:
+                    task.cancel()
+                _, still_pending = await asyncio.wait(
+                    pending, timeout=max(0.0, shutdown_deadline - time.monotonic())
+                )
+                for task, name in interrupted_names.items():
+                    if task.cancelled():
+                        self._download_tasks.discard(task)
+                        self._download_names.pop(task, None)
+                        self._download_completed_at.pop(task, None)
+                        shutdown_errors.append(
+                            f"Download '{name}' was cancelled because the browser was closing."
+                        )
+                    elif task in still_pending:
+                        shutdown_errors.append(
+                            f"Download '{name}' did not stop before browser shutdown."
+                        )
+                break
             await self._shutdown()
-        return {"ok": True, "message": "Browser closed." if was_running else "The browser wasn't open."}
+        result = {"ok": True, "message": "Browser closed." if was_running else "The browser wasn't open."}
+        result = await self._include_downloads(result, downloads_before)
+        if shutdown_errors:
+            result.setdefault("download_errors", []).extend(shutdown_errors)
+        return result
 
 
 def _clean_ref(ref: Any) -> str | None:
diff --git a/tests/browser/conftest.py b/tests/browser/conftest.py
index 343ee714..604fb9c3 100644
--- a/tests/browser/conftest.py
+++ b/tests/browser/conftest.py
@@ -47,6 +47,8 @@
     "index.html": INDEX,
     "search.html": "<!doctype html><title>Results</title><main><h1>Results page</h1><p>Search results here.</p></main>",
     "help.html": "<!doctype html><title>Help</title><main><h1>Help center</h1></main>",
+    "download.html": '<!doctype html><title>Downloads</title><a href="/report.txt" download>Download report</a>',
+    "report.txt": "Sentient browser download test.\n",
 }
 
 
@@ -54,6 +56,18 @@ class _Quiet(http.server.SimpleHTTPRequestHandler):
     def log_message(self, *args):  # keep test output clean
         return
 
+    def do_GET(self):
+        if self.path == "/direct-download":
+            body = b"Direct browser-open download test.\n"
+            self.send_response(200)
+            self.send_header("Content-Type", "application/octet-stream")
+            self.send_header("Content-Disposition", 'attachment; filename="direct-report.txt"')
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+            return
+        super().do_GET()
+
 
 @pytest.fixture(scope="session")
 def site(tmp_path_factory):
diff --git a/tests/browser/test_live_browser.py b/tests/browser/test_live_browser.py
index 978f778a..fa0da27b 100644
--- a/tests/browser/test_live_browser.py
+++ b/tests/browser/test_live_browser.py
@@ -57,6 +57,41 @@ async def test_open_type_click_extract_screenshot(browser, site):
     assert jpeg and jpeg[:2] == b"\xff\xd8"
 
 
+async def test_download_is_saved_and_reported(browser, site):
+    ctx = make_ctx(browser.app)
+    snap = await bt.browser_open.call(ctx, {"url": f"{site}/download.html"})
+    link = ref_of(snap["text"], r'link "Download report"')
+    result = await bt.browser_click.call(ctx, {"ref": link})
+    download_paths = result.get("downloads", [])
+    if not download_paths:
+        assert result["downloads_in_progress"] == ["report.txt"]
+        pending = [task for task in browser._download_tasks if not task.done()]
+        assert pending
+        await asyncio.wait_for(asyncio.gather(*pending), timeout=10)
+        later_result = await bt.browser_snapshot.call(ctx, {})
+        download_paths = later_result["downloads"]
+    assert download_paths == ["downloads/report.txt"]
+    saved = paths.files_dir() / download_paths[0]
+    assert saved.read_text(encoding="utf-8") == "Sentient browser download test.\n"
+
+
+async def test_open_direct_download_is_reported_as_normal_result(browser, site):
+    ctx = make_ctx(browser.app)
+    result = await bt.browser_open.call(ctx, {"url": f"{site}/direct-download"})
+
+    download_paths = result.get("downloads", [])
+    if not download_paths:
+        assert result["downloads_in_progress"] == ["direct-report.txt"]
+        pending = [task for task in browser._download_tasks if not task.done()]
+        assert pending
+        await asyncio.wait_for(asyncio.gather(*pending), timeout=10)
+        later_result = await bt.browser_snapshot.call(ctx, {})
+        download_paths = later_result["downloads"]
+    assert download_paths == ["downloads/direct-report.txt"]
+    saved = paths.files_dir() / download_paths[0]
+    assert saved.read_text(encoding="utf-8") == "Direct browser-open download test.\n"
+
+
 async def test_safety_in_real_pages(browser, site):
     app = browser.app
     app.config.tools.approvals.mode = "ask"
diff --git a/tests/browser/test_profiles.py b/tests/browser/test_profiles.py
index 15d58d6b..3bfbb52f 100644
--- a/tests/browser/test_profiles.py
+++ b/tests/browser/test_profiles.py
@@ -6,6 +6,7 @@
 import http.server
 import json
 import threading
+from pathlib import Path
 
 import pytest
 from fastapi.testclient import TestClient
@@ -406,3 +407,501 @@ async def launch_persistent_context(self, **kwargs):
     with pytest.raises(BrowserError):
         await svc._launch_persistent(BrowserProfileConfig(), headless=True)
     assert browser_service.STORAGE_FLAG in seen["args"]
+    assert seen["accept_downloads"] is True
+
+
+async def test_finished_download_is_reported_on_the_next_result():
+    svc = BrowserService(make_app())
+    task = asyncio.create_task(asyncio.sleep(0, result="downloads/late-report.txt"))
+    svc._download_tasks.add(task)
+    await task
+
+    result = await svc._include_downloads({"ok": True}, {task})
+
+    assert result["downloads"] == ["downloads/late-report.txt"]
+    assert task not in svc._download_tasks
+
+
+async def test_download_collection_without_start_wait_returns_immediately(monkeypatch):
+    svc = BrowserService(make_app())
+    monkeypatch.setattr(browser_service, "DOWNLOAD_APPEAR_GRACE_S", 0.01)
+    original_sleep = asyncio.sleep
+    sleeps = []
+
+    async def unexpected_sleep(delay):
+        sleeps.append(delay)
+        await original_sleep(delay)
+
+    monkeypatch.setattr(asyncio, "sleep", unexpected_sleep)
+
+    assert await svc._include_downloads({"ok": True}, set()) == {"ok": True}
+    assert sleeps == []
+
+    assert await svc._include_downloads(
+        {"ok": True}, set(), wait_for_download_start=True
+    ) == {"ok": True}
+    assert sleeps
+
+
+async def test_unfinished_download_stays_tracked_until_a_later_result(monkeypatch):
+    svc = BrowserService(make_app())
+    completion = asyncio.get_running_loop().create_future()
+
+    class SlowDownload:
+        suggested_filename = "slow-report.pdf"
+
+    async def save_download(download):
+        return await completion
+
+    monkeypatch.setattr(svc, "_save_download", save_download)
+    svc._on_download(SlowDownload())
+    task = next(iter(svc._download_tasks))
+
+    result = await asyncio.wait_for(svc._include_downloads({"ok": True}, set()), timeout=0.5)
+
+    assert result == {"ok": True, "downloads_in_progress": ["slow-report.pdf"]}
+    assert task in svc._download_tasks
+    assert not task.done()
+
+    completion.set_result("downloads/slow-report.pdf")
+    await task
+    result = await svc._include_downloads({"ok": True}, set())
+
+    assert result["downloads"] == ["downloads/slow-report.pdf"]
+    assert task not in svc._download_tasks
+    assert task not in svc._download_names
+    if svc._download_cleanup_task:
+        svc._download_cleanup_task.cancel()
+        await asyncio.gather(svc._download_cleanup_task, return_exceptions=True)
+
+
+async def test_close_tool_waits_for_unfinished_download_before_shutdown(monkeypatch):
+    svc = BrowserService(make_app())
+    svc._context = object()
+    started = asyncio.Event()
+    completion = asyncio.get_running_loop().create_future()
+
+    async def save_download():
+        started.set()
+        return await completion
+
+    task = asyncio.create_task(save_download())
+    svc._download_tasks.add(task)
+    svc._download_names[task] = "stalled.pdf"
+    await started.wait()
+
+    shutdown_started = asyncio.Event()
+
+    async def shutdown():
+        shutdown_started.set()
+        svc._context = None
+
+    monkeypatch.setattr(svc, "_shutdown", shutdown)
+
+    close_task = asyncio.create_task(svc.close_tool(None))
+    await asyncio.sleep(0)
+    assert not close_task.done()
+    assert not shutdown_started.is_set()
+
+    completion.set_result("downloads/stalled.pdf")
+    result = await close_task
+
+    assert shutdown_started.is_set()
+    assert result["downloads"] == ["downloads/stalled.pdf"]
+    assert task not in svc._download_tasks
+
+
+async def test_close_tool_bounds_wait_for_download_that_ignores_cancellation(monkeypatch):
+    svc = BrowserService(make_app())
+    svc._context = object()
+    started = asyncio.Event()
+    cancellation_seen = asyncio.Event()
+    release = asyncio.Event()
+
+    async def stubborn_download():
+        started.set()
+        try:
+            await asyncio.Future()
+        except asyncio.CancelledError:
+            cancellation_seen.set()
+            await release.wait()
+            return "downloads/stalled.pdf"
+
+    task = asyncio.create_task(stubborn_download())
+    svc._download_tasks.add(task)
+    svc._download_names[task] = "stalled.pdf"
+    await started.wait()
+
+    shutdown_started = asyncio.Event()
+
+    async def shutdown():
+        shutdown_started.set()
+        svc._context = None
+
+    monkeypatch.setattr(svc, "_shutdown", shutdown)
+    monkeypatch.setattr(browser_service, "DOWNLOAD_MAX_DURATION_S", 0.01)
+    monkeypatch.setattr(browser_service, "DOWNLOAD_CANCEL_TIMEOUT_S", 0.01)
+
+    result = await asyncio.wait_for(svc.close_tool(None), timeout=0.5)
+
+    assert cancellation_seen.is_set()
+    assert shutdown_started.is_set()
+    assert result["download_errors"] == [
+        "Download 'stalled.pdf' did not stop before browser shutdown."
+    ]
+    assert result["downloads_in_progress"] == ["stalled.pdf"]
+    assert task in svc._download_tasks and not task.done()
+
+    release.set()
+    await task
+    later_result = await svc._include_downloads({"ok": True}, set())
+    assert later_result["downloads"] == ["downloads/stalled.pdf"]
+    assert task not in svc._download_tasks
+    if svc._download_cleanup_task:
+        svc._download_cleanup_task.cancel()
+        await asyncio.gather(svc._download_cleanup_task, return_exceptions=True)
+
+
+async def test_completed_download_is_reported_while_another_download_is_stuck(monkeypatch):
+    svc = BrowserService(make_app())
+    stalled_started = asyncio.Event()
+    finish_stalled = asyncio.Event()
+
+    class Download:
+        def __init__(self, suggested_filename):
+            self.suggested_filename = suggested_filename
+
+    async def save_download(download):
+        if download.suggested_filename == "stalled.pdf":
+            stalled_started.set()
+            await finish_stalled.wait()
+            return "downloads/stalled.pdf"
+        return "downloads/ready.pdf"
+
+    monkeypatch.setattr(svc, "_save_download", save_download)
+    svc._on_download(Download("stalled.pdf"))
+    stalled_task = next(iter(svc._download_tasks))
+    await stalled_started.wait()
+    svc._on_download(Download("ready.pdf"))
+    ready_task = next(task for task in svc._download_tasks if task is not stalled_task)
+    await ready_task
+
+    result = await asyncio.wait_for(svc._include_downloads({"ok": True}, set()), timeout=0.5)
+
+    assert result == {
+        "ok": True,
+        "downloads": ["downloads/ready.pdf"],
+        "downloads_in_progress": ["stalled.pdf"],
+    }
+    assert stalled_task in svc._download_tasks
+
+    finish_stalled.set()
+    await stalled_task
+    later_result = await svc._include_downloads({"ok": True}, set())
+    assert later_result == {"ok": True, "downloads": ["downloads/stalled.pdf"]}
+    if svc._download_cleanup_task:
+        svc._download_cleanup_task.cancel()
+        await asyncio.gather(svc._download_cleanup_task, return_exceptions=True)
+
+
+async def test_attached_downloads_are_wired_only_for_sentient_pages_and_their_popups():
+    svc = BrowserService(make_app())
+    svc._attached = True
+
+    class Page:
+        def __init__(self):
+            self.url = "about:blank"
+            self.handlers = {}
+
+        def on(self, event, handler):
+            self.handlers.setdefault(event, []).append(handler)
+
+    class Context:
+        async def new_page(self):
+            return sentient_page
+
+    user_page = Page()
+    sentient_page = Page()
+    user_popup = Page()
+    sentient_popup = Page()
+    nested_popup = Page()
+
+    svc._wire_page(user_page)
+    svc._wire_page(sentient_page)
+    svc._active = await svc._new_page(Context())
+    svc._wire_page(user_popup)
+    user_page.handlers["popup"][0](user_popup)
+    svc._wire_page(sentient_popup)
+    sentient_page.handlers["popup"][0](sentient_popup)
+    assert svc._active is sentient_popup
+    svc._wire_page(nested_popup)
+    sentient_popup.handlers["popup"][0](nested_popup)
+
+    assert "download" not in user_page.handlers
+    assert "download" not in user_popup.handlers
+    assert sentient_page.handlers["download"] == [svc._on_download]
+    assert sentient_popup.handlers["download"] == [svc._on_download]
+    assert nested_popup.handlers["download"] == [svc._on_download]
+    assert nested_popup in svc._download_pages
+
+
+def test_edge_downloads_hub_is_ignored_and_active_page_is_restored():
+    svc = BrowserService(make_app())
+
+    class Page:
+        def __init__(self, url):
+            self.url = url
+
+        @property
+        def main_frame(self):
+            return self
+
+        def is_closed(self):
+            return False
+
+    original = Page("https://example.com/")
+    downloads_hub = Page("edge://downloads-hub/")
+    context = type("Context", (), {"pages": [original, downloads_hub]})()
+    svc._context = context
+    svc._active = downloads_hub
+
+    svc._on_navigated(downloads_hub, downloads_hub)
+
+    assert svc._active is original
+    assert svc._usable_pages(context) == [original]
+
+
+def test_attached_downloads_hub_fallback_stays_on_sentient_owned_tab():
+    svc = BrowserService(make_app())
+    svc._attached = True
+
+    class Page:
+        def __init__(self, url):
+            self.url = url
+
+        @property
+        def main_frame(self):
+            return self
+
+        def is_closed(self):
+            return False
+
+    sentient_page = Page("https://sentient.example/")
+    user_page = Page("https://user.example/")
+    downloads_hub = Page("edge://downloads-hub/")
+    context = type("Context", (), {"pages": [sentient_page, user_page, downloads_hub]})()
+    svc._context = context
+    svc._active = downloads_hub
+    svc._download_pages.update((sentient_page, downloads_hub))
+
+    svc._on_navigated(downloads_hub, downloads_hub)
+
+    assert svc._active is sentient_page
+
+
+async def test_failed_download_does_not_discard_action_or_successful_download():
+    svc = BrowserService(make_app())
+
+    async def fail_download():
+        raise BrowserError("Couldn't save downloaded file 'broken.pdf'")
+
+    successful = asyncio.create_task(asyncio.sleep(0, result="downloads/report.pdf"))
+    failed = asyncio.create_task(fail_download())
+    svc._download_tasks.update((successful, failed))
+    await asyncio.gather(successful, failed, return_exceptions=True)
+
+    result = await svc._include_downloads({"ok": True, "clicked": "Download"}, set())
+
+    assert result == {
+        "ok": True,
+        "clicked": "Download",
+        "downloads": ["downloads/report.pdf"],
+        "download_errors": ["Couldn't save downloaded file 'broken.pdf'"],
+    }
+    assert successful not in svc._download_tasks
+    assert failed not in svc._download_tasks
+
+
+async def test_stalled_download_is_timed_out_and_reported(tmp_path, monkeypatch):
+    svc = BrowserService(make_app())
+
+    class StalledDownload:
+        suggested_filename = "stalled.pdf"
+        cancelled = False
+
+        async def save_as(self, target):
+            Path(target).write_text("partial", encoding="utf-8")
+            await asyncio.Future()
+
+        async def cancel(self):
+            self.cancelled = True
+
+    download = StalledDownload()
+
+    monkeypatch.setattr(browser_service.paths, "files_dir", lambda: tmp_path)
+    monkeypatch.setattr(browser_service, "DOWNLOAD_MAX_DURATION_S", 0.01)
+    task = asyncio.create_task(svc._save_download(download))
+    svc._download_tasks.add(task)
+
+    with pytest.raises(BrowserError, match="did not finish within"):
+        await task
+    assert download.cancelled
+    assert not (tmp_path / "downloads" / "stalled.pdf").exists()
+    result = await svc._include_downloads({"ok": True}, set())
+
+    assert result == {
+        "ok": True,
+        "download_errors": ["Download 'stalled.pdf' did not finish within 0.01 seconds."],
+    }
+    assert task not in svc._download_tasks
+
+
+async def test_download_save_error_includes_underlying_reason(tmp_path, monkeypatch):
+    svc = BrowserService(make_app())
+
+    class FailedDownload:
+        suggested_filename = "broken.pdf"
+
+        async def save_as(self, target):
+            raise OSError("The connection was lost")
+
+        async def cancel(self):
+            raise AssertionError("a failed save should not be cancelled")
+
+    monkeypatch.setattr(browser_service.paths, "files_dir", lambda: tmp_path)
+
+    with pytest.raises(
+        BrowserError,
+        match=r"Couldn't save downloaded file 'broken\.pdf': The connection was lost",
+    ):
+        await svc._save_download(FailedDownload())
+
+    assert not (tmp_path / "downloads" / "broken.pdf").exists()
+
+
+async def test_stalled_download_cancel_is_bounded_and_partial_file_is_removed(tmp_path, monkeypatch):
+    svc = BrowserService(make_app())
+    cancel_started = asyncio.Event()
+
+    class StalledDownload:
+        suggested_filename = "stalled.pdf"
+
+        async def save_as(self, target):
+            Path(target).write_text("partial", encoding="utf-8")
+            await asyncio.Future()
+
+        async def cancel(self):
+            cancel_started.set()
+            await asyncio.Future()
+
+    monkeypatch.setattr(browser_service.paths, "files_dir", lambda: tmp_path)
+    monkeypatch.setattr(browser_service, "DOWNLOAD_MAX_DURATION_S", 0.01)
+    monkeypatch.setattr(browser_service, "DOWNLOAD_CANCEL_TIMEOUT_S", 0.01)
+
+    with pytest.raises(BrowserError, match=r"Browser cancellation failed: timed out after 0.01 seconds"):
+        await asyncio.wait_for(svc._save_download(StalledDownload()), timeout=0.5)
+
+    assert cancel_started.is_set()
+    assert not (tmp_path / "downloads" / "stalled.pdf").exists()
+
+
+async def test_cancelled_save_cancels_download_and_removes_partial_file(tmp_path, monkeypatch):
+    svc = BrowserService(make_app())
+    save_started = asyncio.Event()
+
+    class StalledDownload:
+        suggested_filename = "cancelled.pdf"
+        cancelled = False
+
+        async def save_as(self, target):
+            Path(target).write_text("partial", encoding="utf-8")
+            save_started.set()
+            await asyncio.Future()
+
+        async def cancel(self):
+            self.cancelled = True
+
+    download = StalledDownload()
+    monkeypatch.setattr(browser_service.paths, "files_dir", lambda: tmp_path)
+    task = asyncio.create_task(svc._save_download(download))
+    await save_started.wait()
+    task.cancel()
+
+    with pytest.raises(asyncio.CancelledError):
+        await task
+
+    assert download.cancelled
+    assert not (tmp_path / "downloads" / "cancelled.pdf").exists()
+
+
+async def test_stalled_download_does_not_block_another_save(tmp_path, monkeypatch):
+    svc = BrowserService(make_app())
+    stalled_started = asyncio.Event()
+    finish_stalled = asyncio.Event()
+
+    class StalledDownload:
+        suggested_filename = "report.txt"
+
+        async def save_as(self, target):
+            stalled_started.set()
+            await finish_stalled.wait()
+            Path(target).write_text("first", encoding="utf-8")
+
+        async def cancel(self):
+            finish_stalled.set()
+
+    class QuickDownload:
+        suggested_filename = "report.txt"
+
+        async def save_as(self, target):
+            Path(target).write_text("second", encoding="utf-8")
+
+        async def cancel(self):
+            raise AssertionError("a completed download should not be cancelled")
+
+    monkeypatch.setattr(browser_service.paths, "files_dir", lambda: tmp_path)
+    monkeypatch.setattr(browser_service, "DOWNLOAD_MAX_DURATION_S", 1)
+
+    stalled = asyncio.create_task(svc._save_download(StalledDownload()))
+    await stalled_started.wait()
+    quick = asyncio.create_task(svc._save_download(QuickDownload()))
+
+    assert await asyncio.wait_for(quick, timeout=0.2) == "downloads/report(1).txt"
+    finish_stalled.set()
+    assert await stalled == "downloads/report.txt"
+    assert (tmp_path / "downloads" / "report.txt").read_text(encoding="utf-8") == "first"
+    assert (tmp_path / "downloads" / "report(1).txt").read_text(encoding="utf-8") == "second"
+
+
+async def test_unreported_completed_download_expires_after_reporting_window(monkeypatch):
+    svc = BrowserService(make_app())
+    monkeypatch.setattr(browser_service, "DOWNLOAD_TASK_RETENTION_S", 0.01)
+    monkeypatch.setattr(browser_service, "DOWNLOAD_TASK_CLEANUP_INTERVAL_S", 0.01)
+
+    async def save_download(download):
+        return f"downloads/{download}.pdf"
+
+    monkeypatch.setattr(svc, "_save_download", save_download)
+    await svc._lock.acquire()
+    try:
+        downloads = list(range(100))
+        for download in downloads:
+            svc._on_download(download)
+        tasks = set(svc._download_tasks)
+        await asyncio.gather(*tasks)
+        await asyncio.sleep(0)
+        assert svc._download_cleanup_task is not None
+        assert svc._download_cleanup_task.get_name() == "browser:download-cleanup"
+        assert len(svc._download_tasks) == len(downloads)
+        await asyncio.sleep(0.02)
+        assert len(svc._download_tasks) == len(downloads)
+    finally:
+        svc._lock.release()
+
+    for _ in range(20):
+        if not svc._download_tasks:
+            break
+        await asyncio.sleep(0.01)
+    assert not svc._download_tasks
+    assert not svc._download_completed_at
+    assert not svc._download_names
diff --git a/tests/browser/test_profiles_live.py b/tests/browser/test_profiles_live.py
index 0f40ef86..6a6da6ae 100644
--- a/tests/browser/test_profiles_live.py
+++ b/tests/browser/test_profiles_live.py
@@ -11,6 +11,7 @@
 import httpx
 import pytest
 
+from sentient import paths
 from sentient.browser import tools as bt
 from sentient.browser.service import _engine_paths, profile_dir
 from sentient.config.schema import BrowserProfileConfig
@@ -164,7 +165,37 @@ async def test_attach_to_a_running_browser_and_detach_leaves_it_running(browser,
     assert status["attached"] and status["profile"] == "mine" and status["engine"] is None
     assert len(status["tabs"]) >= 2  # the user's own tab plus the one Sentient opened for itself
 
+    user_page = next(page for page in browser._context.pages if page is not browser._active)
+    await user_page.goto(f"{site}/download.html")
+    downloads_before = browser._download_tasks.copy()
+    async with user_page.expect_download(timeout=10_000) as download_info:
+        await user_page.get_by_role("link", name="Download report").click()
+    user_download = await download_info.value
+    assert user_download.suggested_filename == "report.txt"
+    assert browser._download_tasks == downloads_before
+    await user_download.cancel()
+
+    download_page = await bt.browser_open.call(ctx, {"url": f"{site}/download.html", "profile": "mine"})
+    download_ref = next(
+        line.split("]")[0][1:]
+        for line in download_page["text"].splitlines()
+        if 'link "Download report"' in line
+    )
+    download_result = await bt.browser_click.call(ctx, {"ref": download_ref})
+    download_paths = download_result.get("downloads", [])
+    if not download_paths:
+        assert download_result["downloads_in_progress"] == ["report.txt"]
+        pending = [task for task in browser._download_tasks if not task.done()]
+        assert pending
+        await asyncio.wait_for(asyncio.gather(*pending), timeout=10)
+        later_result = await bt.browser_snapshot.call(ctx, {})
+        download_paths = later_result["downloads"]
+    assert download_paths == ["downloads/report.txt"]
+    saved = paths.files_dir() / download_paths[0]
+    assert saved.read_text(encoding="utf-8") == "Sentient browser download test.\n"
+
     # the same safety applies inside the user's browser
+    snap = await bt.browser_open.call(ctx, {"url": f"{site}/index.html", "profile": "mine"})
     app.config.tools.approvals.mode = "ask"
     text = snap["text"]
     pw = next(line.split("]")[0][1:] for line in text.splitlines() if 'textbox "Password"' in line)