diff --git a/CHANGELOG.md b/CHANGELOG.md
index ae27f5858..7e29b2fd6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -637,6 +637,14 @@ Compatibility is documented in release notes, not encoded in the version string.
 - The `nextcloud_bridge` block in a generated `profile.yml` described itself as
   turning "a Nextcloud folder" into a trigger source. It answers questions from
   a Talk room; the comment now says so.
+- Deleting an artifact from the gallery, and the dispatch worker's retention
+  sweep, no longer show up as agent actions in the web terminal's activity
+  strip. Only mutations the agent actually performed are reported.
+- The python executor tools now reject `execution_mode` values other than
+  `readonly` and `readwrite`. An unrecognized spelling used to slip past both
+  write gates and run write-pattern code even with
+  `control_system.writes_enabled=false`. The deployment-level kill switch also
+  now covers `execute_file`, which previously had no such check.
 
 ## [2026.8.0]
 
diff --git a/eslint.config.js b/eslint.config.js
index bc7f82d17..c28b77550 100644
--- a/eslint.config.js
+++ b/eslint.config.js
@@ -54,19 +54,6 @@ export default [
     },
   },
 
-  // (4c) max-lines ratchet on panel-manager.js alone. It sits right at the 450
-  // cap above while three in-flight tasks (wire-glow-call-sites,
-  // strip-verbs-and-labels, badge-ack-restore) all land agent-visibility code
-  // in it. Splitting the module mid-flight would be more destabilizing than the
-  // raised ceiling; the split is earmarked for the polish pass, and this block
-  // goes away with it. Scoped to the one file so nothing else drifts upward.
-  {
-    files: ['src/osprey/interfaces/web_terminal/static/js/panel-manager.js'],
-    rules: {
-      'max-lines': ['error', { max: 550, skipComments: true, skipBlankLines: true }],
-    },
-  },
-
   // (5) Root config files: node globals.
   {
     files: ['vitest.config.js', 'eslint.config.js'],
diff --git a/src/osprey/interfaces/artifacts/app.py b/src/osprey/interfaces/artifacts/app.py
index 01eee4772..6d0018d14 100644
--- a/src/osprey/interfaces/artifacts/app.py
+++ b/src/osprey/interfaces/artifacts/app.py
@@ -517,6 +517,7 @@ def create_app(workspace_root: Path | None = None) -> FastAPI:
     from osprey.stores.artifact_store import (
         ArtifactEntry,
         ArtifactStore,
+        artifact_mutation_actor,
         register_artifact_delete_listener,
         register_artifact_listener,
         unregister_artifact_delete_listener,
@@ -850,7 +851,10 @@ async def serve_file(artifact_id: str, filename: str):
 
     @app.delete("/api/artifacts/{artifact_id}")
     async def delete_artifact(artifact_id: str):
-        deleted = store.delete_entry(artifact_id)
+        # This delete is a person clicking in the gallery, not the agent —
+        # tag it so store listeners don't report it as agent activity.
+        with artifact_mutation_actor("human"):
+            deleted = store.delete_entry(artifact_id)
         if not deleted:
             raise HTTPException(status_code=404, detail=f"Artifact {artifact_id} not found")
         return {"status": "ok", "artifact_id": artifact_id}
diff --git a/src/osprey/interfaces/web_terminal/static/js/panel-agent-attention.js b/src/osprey/interfaces/web_terminal/static/js/panel-agent-attention.js
new file mode 100644
index 000000000..0cf6d7eb6
--- /dev/null
+++ b/src/osprey/interfaces/web_terminal/static/js/panel-agent-attention.js
@@ -0,0 +1,120 @@
+/* OSPREY Web Terminal — Agent attention on the panel rail.
+
+   The two ways a rail entry signals agent activity: a transient glow for
+   "the agent just touched this", and a persistent badge for "this panel has
+   activity the operator has not seen". Everything here anchors on the rail
+   element bound once via initAgentAttention().
+
+   A badge must outlive the page: the server's history ring is re-read on every
+   SSE open (restoreAgentBadges) and any panel activity the operator has not
+   seen re-badges its entry. "Seen" is an ACKNOWLEDGMENT — the server ts of the
+   newest badge the operator cleared by surfacing that panel, kept per panel in
+   localStorage under `agent-ack:<panelId>`.
+
+   Only SERVER timestamps are ever stored or compared here. The ring's `ts` is
+   the web-terminal process's clock; a browser clock skewed ahead of it would
+   permanently suppress real badges, and one skewed behind would resurrect
+   cleared ones on every reconnect. So a badge whose frame carried no ts leaves
+   the stored ack untouched rather than substituting Date.now(). */
+
+import { fetchJSON } from './api.js';
+import { getEntry, setEntryAttention } from './panel-rail.js';
+import { flashElement } from '/design-system/js/highlight.js';
+
+/** @type {HTMLElement} */
+let railEl = /** @type {HTMLElement} */ (/** @type {unknown} */ (null));
+
+/**
+ * Bind the rail element every glow/badge call anchors on. Call once, as soon
+ * as the rail exists and before any SSE frame can arrive.
+ * @param {HTMLElement} el
+ */
+export function initAgentAttention(el) { railEl = el; }
+
+/**
+ * Transient agent glow on a rail entry — flash only, never the persistent
+ * badge (that is agent_activity's job via badgePanelActivity). No-op for ids
+ * without an entry.
+ * @param {string} panelId
+ */
+export function flashAgentGlow(panelId) { const entry = getEntry(railEl, panelId); if (entry) flashElement(entry); }
+
+/** Newest badge-causing server ts seen this page lifetime, per panel.
+ *  @type {Map<string, number>} */
+const badgeTs = new Map();
+
+const ACK_KEY_PREFIX = 'agent-ack:';
+
+/**
+ * Record a badge's server ts, keeping the newest. The history ring replays a
+ * panel's older events alongside its newest, so this must not walk backwards.
+ * @param {string} panelId
+ * @param {number} [ts]
+ */
+function noteBadgeTs(panelId, ts) {
+  if (typeof ts !== 'number') return;
+  const prev = badgeTs.get(panelId);
+  if (prev === undefined || ts > prev) badgeTs.set(panelId, ts);
+}
+
+/**
+ * The acknowledged server ts for a panel. A panel that was never acknowledged
+ * (and a storage read that is denied or corrupt) has seen nothing, so it reads
+ * as -Infinity: an unknown ack RESTORES a badge, it never suppresses one.
+ * @param {string} panelId
+ * @returns {number}
+ */
+function ackedTs(panelId) {
+  let raw = null;
+  try { raw = localStorage.getItem(ACK_KEY_PREFIX + panelId); } catch { return -Infinity; }
+  const ts = raw === null ? NaN : Number(raw);
+  return Number.isFinite(ts) ? ts : -Infinity;
+}
+
+/**
+ * Clear a panel's badge and acknowledge it up to that badge's own server ts,
+ * so a reload does not bring it back. With no ts on record the badge is still
+ * cleared but the stored ack is left exactly as it was (see the module note).
+ * @param {string} panelId
+ */
+export function clearBadge(panelId) {
+  setEntryAttention(railEl, panelId, false);
+  const ts = badgeTs.get(panelId);
+  if (ts === undefined) return;
+  try { localStorage.setItem(ACK_KEY_PREFIX + panelId, String(ts)); } catch { /* storage denied */ }
+}
+
+/**
+ * Badge a panel's rail entry for a live agent_activity frame, remembering the
+ * badge's server ts so clearing it can acknowledge exactly this event and the
+ * reload restore can skip it.
+ * @param {string} panelId
+ * @param {number} [ts]
+ * @returns {boolean} true when the rail anchored the badge; false means the
+ *   caller still owns the frame (no entry for this id).
+ */
+export function badgePanelActivity(panelId, ts) {
+  if (!setEntryAttention(railEl, panelId, true, ts)) return false;
+  noteBadgeTs(panelId, ts);
+  return true;
+}
+
+/**
+ * Re-badge rail entries for panel activity this operator has not acknowledged.
+ * Runs after the membership resync on every SSE open — including the first, so
+ * a reload restores badges — which is also why it runs after it: an entry that
+ * the resync delta just added can then carry a badge.
+ *
+ * Only 'panel'-kind rows with a ts can badge anything; the strip owns the rest
+ * and history there is its own concern (the seam is not replayed).
+ */
+export async function restoreAgentBadges() {
+  let body = null;
+  try { body = await fetchJSON('/api/agent-activity/recent'); } catch { return; }
+  for (const ev of (body?.events || [])) {
+    const target = ev?.target;
+    if (target?.kind !== 'panel' || !target.panel || typeof ev.ts !== 'number') continue;
+    if (ev.ts <= ackedTs(target.panel)) continue;
+    badgePanelActivity(target.panel, ev.ts);
+  }
+}
diff --git a/src/osprey/interfaces/web_terminal/static/js/panel-manager.js b/src/osprey/interfaces/web_terminal/static/js/panel-manager.js
index fa7a83d8c..88977875a 100644
--- a/src/osprey/interfaces/web_terminal/static/js/panel-manager.js
+++ b/src/osprey/interfaces/web_terminal/static/js/panel-manager.js
@@ -38,13 +38,15 @@ import {
 import { initDockSync, withEchoSuppressed, setTileCloseHandler, setTileFocusHandler } from './dock-sync.js';
 import { initRailDrag, railDragStart, railDragEnd } from './rail-drag.js';
 import { startHealthPolling as startPolling } from './panel-health.js';
+import { updateStatusBar } from './panel-status-bar.js';
 import { openTerminalPanel, closeTerminalPanel } from './dock-workspace.js';
 import { initRailThemeCoupling } from './rail-position.js';
-import { flashElement } from '/design-system/js/highlight.js';
 import {
-  createRail, addEntry, removeEntry, getEntry, setActive,
-  setEntryEnabled, setEntryAttention,
+  createRail, addEntry, removeEntry, setActive, setEntryEnabled,
 } from './panel-rail.js';
+import {
+  initAgentAttention, flashAgentGlow, clearBadge, badgePanelActivity, restoreAgentBadges,
+} from './panel-agent-attention.js';
 
 // ---- Types ----
 
@@ -180,6 +182,7 @@ export async function initPanelManager(panelId) {
   railEl = /** @type {HTMLElement} */ (document.getElementById('panel-rail'));
   contentEl = /** @type {HTMLElement} */ (containerEl.querySelector('#panel-content') || containerEl.querySelector('.panel-content'));
   if (!railEl || !contentEl) return;
+  initAgentAttention(railEl);
 
   // Hand the iframe adapter its fallback mount host. When the dockview shell is
   // up, panel iframes live in the adapter's overlay layer instead (dockview
@@ -465,14 +468,10 @@ export async function initPanelManager(panelId) {
 
         } else if (data.type === 'agent_activity' && data.target) {
           // kind 'panel' with a live rail entry → persistent badge + glow via
-          // setEntryAttention; everything the rail cannot anchor falls through
+          // badgePanelActivity; everything the rail cannot anchor falls through
           // to the activity-strip seam (no-op until a handler registers).
           const t = data.target;
-          if (t.kind === 'panel' && t.panel && setEntryAttention(railEl, t.panel, true, data.ts)) {
-            // Remember the badge's server ts so clearing it can acknowledge
-            // exactly this event and the reload restore can skip it.
-            noteBadgeTs(t.panel, data.ts);
-          } else {
+          if (!(t.kind === 'panel' && t.panel && badgePanelActivity(t.panel, data.ts))) {
             onAgentActivity(data);
           }
         }
@@ -569,93 +568,6 @@ async function resyncPanelState() {
 
 // ---- Rail Rendering ----
 
-/**
- * Transient agent glow on a rail entry — flash only, never the persistent
- * badge (that is agent_activity's job via setEntryAttention). No-op for ids
- * without an entry.
- * @param {string} panelId
- */
-function flashAgentGlow(panelId) { const entry = getEntry(railEl, panelId); if (entry) flashElement(entry); }
-
-// ---- Agent-attention badges across reloads ----
-//
-// A badge must outlive the page: the server's history ring is re-read on every
-// SSE open (restoreAgentBadges) and any panel activity the operator has not
-// seen re-badges its entry. "Seen" is an ACKNOWLEDGMENT — the server ts of the
-// newest badge the operator cleared by surfacing that panel, kept per panel in
-// localStorage under `agent-ack:<panelId>`.
-//
-// Only SERVER timestamps are ever stored or compared here. The ring's `ts` is
-// the web-terminal process's clock; a browser clock skewed ahead of it would
-// permanently suppress real badges, and one skewed behind would resurrect
-// cleared ones on every reconnect. So a badge whose frame carried no ts leaves
-// the stored ack untouched rather than substituting Date.now().
-
-/** Newest badge-causing server ts seen this page lifetime, per panel.
- *  @type {Map<string, number>} */
-const badgeTs = new Map();
-
-const ACK_KEY_PREFIX = 'agent-ack:';
-
-/**
- * Record a badge's server ts, keeping the newest. The history ring replays a
- * panel's older events alongside its newest, so this must not walk backwards.
- * @param {string} panelId
- * @param {number} [ts]
- */
-function noteBadgeTs(panelId, ts) {
-  if (typeof ts !== 'number') return;
-  const prev = badgeTs.get(panelId);
-  if (prev === undefined || ts > prev) badgeTs.set(panelId, ts);
-}
-
-/**
- * The acknowledged server ts for a panel. A panel that was never acknowledged
- * (and a storage read that is denied or corrupt) has seen nothing, so it reads
- * as -Infinity: an unknown ack RESTORES a badge, it never suppresses one.
- * @param {string} panelId
- * @returns {number}
- */
-function ackedTs(panelId) {
-  let raw = null;
-  try { raw = localStorage.getItem(ACK_KEY_PREFIX + panelId); } catch { return -Infinity; }
-  const ts = raw === null ? NaN : Number(raw);
-  return Number.isFinite(ts) ? ts : -Infinity;
-}
-
-/**
- * Clear a panel's badge and acknowledge it up to that badge's own server ts,
- * so a reload does not bring it back. With no ts on record the badge is still
- * cleared but the stored ack is left exactly as it was (see the section note).
- * @param {string} panelId
- */
-function clearBadge(panelId) {
-  setEntryAttention(railEl, panelId, false);
-  const ts = badgeTs.get(panelId);
-  if (ts === undefined) return;
-  try { localStorage.setItem(ACK_KEY_PREFIX + panelId, String(ts)); } catch { /* storage denied */ }
-}
-
-/**
- * Re-badge rail entries for panel activity this operator has not acknowledged.
- * Runs after the membership resync on every SSE open — including the first, so
- * a reload restores badges — which is also why it runs after it: an entry that
- * the resync delta just added can then carry a badge.
- *
- * Only 'panel'-kind rows with a ts can badge anything; the strip owns the rest
- * and history there is its own concern (the seam is not replayed).
- */
-async function restoreAgentBadges() {
-  let body = null;
-  try { body = await fetchJSON('/api/agent-activity/recent'); } catch { return; }
-  for (const ev of (body?.events || [])) {
-    const target = ev?.target;
-    if (target?.kind !== 'panel' || !target.panel || typeof ev.ts !== 'number') continue;
-    if (ev.ts <= ackedTs(target.panel)) continue;
-    if (setEntryAttention(railEl, target.panel, true, ev.ts)) noteBadgeTs(target.panel, ev.ts);
-  }
-}
-
 /**
  * Interaction closures handed to every rail render/append call. Routing
  * activation and close through here keeps a human click/"×" and an agent MCP
@@ -927,7 +839,7 @@ function ensureActivePanel() {
  * @param {boolean} wasHealthy
  */
 function onHealthSettled(panel, wasHealthy) {
-  updateStatusBar(panel);
+  updateStatusBar(panel, panelState[panel.id]);
   if (panelState[panel.id].healthy && !wasHealthy) {
     setEntryEnabled(railEl, panel.id, true);
     ensureActivePanel();
@@ -952,23 +864,6 @@ function assumeHealthy(panel) {
   setEntryEnabled(railEl, panel.id, true);
 }
 
-/** @param {Panel} panel */
-function updateStatusBar(panel) {
-  if (!panel.statusBarId) return;
-
-  const statusItem = document.getElementById(panel.statusBarId);
-  if (!statusItem) return;
-
-  const state = panelState[panel.id];
-  if (state.url) {
-    statusItem.style.display = '';
-    const dot = statusItem.querySelector('.status-dot');
-    if (dot) {
-      dot.className = 'status-dot' + (state.healthy ? ' live' : ' error');
-    }
-  }
-}
-
 /**
  * Broadcast the current UI mode to every panel iframe — the hub-role fan-out
  * the header toggle fires after it swaps <html data-ui-mode>. Mirrors
diff --git a/src/osprey/interfaces/web_terminal/static/js/panel-status-bar.js b/src/osprey/interfaces/web_terminal/static/js/panel-status-bar.js
new file mode 100644
index 000000000..6fe08c424
--- /dev/null
+++ b/src/osprey/interfaces/web_terminal/static/js/panel-status-bar.js
@@ -0,0 +1,31 @@
+// @ts-check
+/* OSPREY Web Terminal — Status-bar health readout.
+ *
+ * The one DOM consequence of a health poll settling that lives outside the
+ * rail: panels with a `statusBarId` mirror their healthy flag onto the status
+ * bar's dot. Kept out of panel-health.js on purpose — that module owns timing
+ * and fetch only, with no DOM knowledge.
+ */
+
+/** @typedef {import('./panel-catalog.js').Panel} Panel */
+
+/**
+ * Reflect a panel's health on its status-bar item, if it has one. Hidden until
+ * the panel's config has loaded (`state.url` set), then dot class live/error.
+ * @param {Panel} panel
+ * @param {{url: string | null, healthy: boolean}} state
+ */
+export function updateStatusBar(panel, state) {
+  if (!panel.statusBarId) return;
+
+  const statusItem = document.getElementById(panel.statusBarId);
+  if (!statusItem) return;
+
+  if (state.url) {
+    statusItem.style.display = '';
+    const dot = statusItem.querySelector('.status-dot');
+    if (dot) {
+      dot.className = 'status-dot' + (state.healthy ? ' live' : ' error');
+    }
+  }
+}
diff --git a/src/osprey/mcp_server/ariel/tools/entry.py b/src/osprey/mcp_server/ariel/tools/entry.py
index cffa063c6..20b2680f5 100644
--- a/src/osprey/mcp_server/ariel/tools/entry.py
+++ b/src/osprey/mcp_server/ariel/tools/entry.py
@@ -6,7 +6,6 @@
   attachment limits, logbook name conventions
 """
 
-import functools
 import json
 import logging
 import os
@@ -14,12 +13,11 @@
 from datetime import UTC, datetime
 from pathlib import Path
 
-import anyio
 from fastmcp.exceptions import ToolError
 
 from osprey.mcp_server.ariel.server import build_entry_url, make_error, mcp, serialize_entry
 from osprey.mcp_server.ariel.server_context import get_ariel_context
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 
 logger = logging.getLogger("osprey.mcp_server.ariel.tools.entry")
 
@@ -382,17 +380,9 @@ async def entry_create(
         # entry is persisted and before attachments are processed — an
         # attachment failure must not lose the signal for an entry that already
         # exists. Passive on purpose: unlike the draft branch above, a direct
-        # write does not steal focus. notify_agent_activity never raises; the
+        # write does not steal focus. notify_agent_activity_async never raises; the
         # blocking call runs off the event loop.
-        await anyio.to_thread.run_sync(
-            functools.partial(
-                notify_agent_activity,
-                "entry_create",
-                "panel",
-                panel="ariel",
-                detail=entry_id,
-            )
-        )
+        await notify_agent_activity_async("entry_create", "panel", panel="ariel", detail=entry_id)
 
         # Process attachments if provided
         attachment_count = 0
diff --git a/src/osprey/mcp_server/ariel/tools/publish.py b/src/osprey/mcp_server/ariel/tools/publish.py
index 852a36da2..539322c3b 100644
--- a/src/osprey/mcp_server/ariel/tools/publish.py
+++ b/src/osprey/mcp_server/ariel/tools/publish.py
@@ -1,15 +1,13 @@
 """MCP tool: entry_publish — publish an existing ARIEL entry to the facility logbook."""
 
-import functools
 import json
 import logging
 
-import anyio
 from fastmcp.exceptions import ToolError
 
 from osprey.mcp_server.ariel.server import build_entry_url, make_error, mcp
 from osprey.mcp_server.ariel.server_context import get_ariel_context
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 from osprey.services.ariel_search.exceptions import AuthenticationRequiredError
 
 logger = logging.getLogger("osprey.mcp_server.ariel.tools.publish")
@@ -49,16 +47,10 @@ async def entry_publish(
         # Agent-activity highlight for the ARIEL panel. Only reached once the
         # upstream write succeeded — every refusal (not_found, not_supported,
         # auth_required, internal_error) raises out of publish_entry above and
-        # emits nothing. Passive: no focus steal. notify_agent_activity never
+        # emits nothing. Passive: no focus steal. notify_agent_activity_async never
         # raises; the blocking call runs off the event loop.
-        await anyio.to_thread.run_sync(
-            functools.partial(
-                notify_agent_activity,
-                "entry_publish",
-                "panel",
-                panel="ariel",
-                detail=result.entry_id,
-            )
+        await notify_agent_activity_async(
+            "entry_publish", "panel", panel="ariel", detail=result.entry_id
         )
 
         # The just-published entry now carries a facility-assigned id, so the
diff --git a/src/osprey/mcp_server/artifact_activity.py b/src/osprey/mcp_server/artifact_activity.py
index 5d767d906..0b00d41cb 100644
--- a/src/osprey/mcp_server/artifact_activity.py
+++ b/src/osprey/mcp_server/artifact_activity.py
@@ -8,7 +8,13 @@
 ``/api/agent-activity`` frames, so artifact visibility does not have to be
 re-implemented in every tool that writes one.
 
-Two properties shape the implementation:
+Three properties shape the implementation:
+
+**Only agent mutations emit.** The store fires the same listeners for a human
+deleting from the gallery UI and for the dispatch worker's retention sweep;
+those callers tag themselves via
+:func:`osprey.stores.artifact_store.artifact_mutation_actor` and their events
+are dropped here — the frames are *agent* activity.
 
 **The caller is never blocked.** ``notify_agent_activity`` performs a blocking
 HTTP POST, and ``ArtifactStore.delete_all`` fires the delete listener once per
@@ -41,6 +47,7 @@
 from typing import TYPE_CHECKING
 
 from osprey.mcp_server.http import notify_agent_activity
+from osprey.stores.artifact_store import current_artifact_mutation_actor
 
 if TYPE_CHECKING:
     from osprey.stores.artifact_store import ArtifactEntry
@@ -102,6 +109,13 @@ def _drain_pending() -> None:
 
 
 def _enqueue(tool: str, entry: ArtifactEntry) -> None:
+    # Frames are *agent* activity. A gallery click or a retention sweep fires
+    # the same store listener; those callers tag themselves via
+    # artifact_mutation_actor and must not be reported as agent actions. The
+    # actor is only valid here, on the mutating caller's own context — the
+    # worker thread that POSTs later would always see the default.
+    if current_artifact_mutation_actor() != "agent":
+        return
     if _is_bookkeeping(entry):
         return
     try:
diff --git a/src/osprey/mcp_server/bluesky/tools/queue.py b/src/osprey/mcp_server/bluesky/tools/queue.py
index b7637d4fc..a12abfaab 100644
--- a/src/osprey/mcp_server/bluesky/tools/queue.py
+++ b/src/osprey/mcp_server/bluesky/tools/queue.py
@@ -71,7 +71,6 @@
 
 from __future__ import annotations
 
-import functools
 import json
 from typing import NoReturn
 
@@ -86,7 +85,7 @@
     get_server_context,
 )
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 
 # Remediation guidance per bridge refusal code, kept in one table so every
 # tool answers the same code with the same next step. Codes absent here get
@@ -502,13 +501,8 @@ async def queue_add(draft_revision: int) -> str:
         )
 
     run_id = body.get("run_id") if isinstance(body, dict) else None
-    await anyio.to_thread.run_sync(
-        functools.partial(
-            notify_agent_activity,
-            "queue_add",
-            "run",
-            detail=str(run_id) if run_id is not None else None,
-        )
+    await notify_agent_activity_async(
+        "queue_add", "run", detail=str(run_id) if run_id is not None else None
     )
     return json.dumps(body)
 
@@ -610,9 +604,7 @@ async def queue_start() -> str:
             fallback_hints=["Check queue_list for the queue's current state."],
         )
 
-    await anyio.to_thread.run_sync(
-        functools.partial(notify_agent_activity, "queue_start", "run", detail="queue")
-    )
+    await notify_agent_activity_async("queue_start", "run", detail="queue")
     return json.dumps(body)
 
 
@@ -637,9 +629,7 @@ async def _request_panel_start() -> str:
         )
 
     record = body.get("start_request") if isinstance(body, dict) else None
-    await anyio.to_thread.run_sync(
-        functools.partial(notify_agent_activity, "queue_start", "run", detail="start-request")
-    )
+    await notify_agent_activity_async("queue_start", "run", detail="start-request")
     return json.dumps(
         {
             "started": False,
@@ -727,12 +717,7 @@ async def queue_stop(cancel: bool = False) -> str:
     # The two directions are opposite operations, so they must not share a
     # label: rendering a withdrawal as "stop" would tell the operator the queue
     # is halting when it has just been released to keep draining.
-    await anyio.to_thread.run_sync(
-        functools.partial(
-            notify_agent_activity,
-            "queue_stop",
-            "run",
-            detail="stop-withdrawn" if cancel else "stop",
-        )
+    await notify_agent_activity_async(
+        "queue_stop", "run", detail="stop-withdrawn" if cancel else "stop"
     )
     return json.dumps(body)
diff --git a/src/osprey/mcp_server/bluesky/tools/stop.py b/src/osprey/mcp_server/bluesky/tools/stop.py
index 7a77a2f9d..ffc6d00c8 100644
--- a/src/osprey/mcp_server/bluesky/tools/stop.py
+++ b/src/osprey/mcp_server/bluesky/tools/stop.py
@@ -41,7 +41,7 @@
 # abort answers to the same bridge vocabulary, and a second copy here would be
 # a second thing to keep in step.
 from osprey.mcp_server.bluesky.tools.queue import _relay_refusal
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 
 # The abort is the one bridge call whose server-side work is a COMPOSITION of
 # manager calls, so ``server_context._TIMEOUT`` (15s, sized for single-call
@@ -124,7 +124,5 @@ async def stop_run() -> str:
             ],
         )
 
-    await anyio.to_thread.run_sync(
-        functools.partial(notify_agent_activity, "stop_run", "run", detail="abort")
-    )
+    await notify_agent_activity_async("stop_run", "run", detail="abort")
     return json.dumps(body)
diff --git a/src/osprey/mcp_server/control_system/tools/channel_write.py b/src/osprey/mcp_server/control_system/tools/channel_write.py
index 250f37162..7f4709802 100644
--- a/src/osprey/mcp_server/control_system/tools/channel_write.py
+++ b/src/osprey/mcp_server/control_system/tools/channel_write.py
@@ -4,17 +4,14 @@
 Tool docstring is the static prompt visible to Claude Code.
 """
 
-import functools
 import json
 import logging
 
-import anyio
-
 from osprey.errors import ChannelWriteBlockedError
 from osprey.mcp_server.control_system.error_handling import connector_error_handler
 from osprey.mcp_server.control_system.server import mcp
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 
 logger = logging.getLogger("osprey.mcp_server.tools.channel_write")
 
@@ -213,13 +210,8 @@ async def channel_write(
         # never raises; the blocking call runs off the event loop.
         executed_channels = [r["channel"] for r in results_serialised if r["success"]]
         if executed_channels:
-            await anyio.to_thread.run_sync(
-                functools.partial(
-                    notify_agent_activity,
-                    "channel_write",
-                    "channel",
-                    detail=", ".join(executed_channels),
-                )
+            await notify_agent_activity_async(
+                "channel_write", "channel", detail=", ".join(executed_channels)
             )
 
         # Return ephemeral result (no persistent storage for channel writes)
diff --git a/src/osprey/mcp_server/dispatch_worker/retention.py b/src/osprey/mcp_server/dispatch_worker/retention.py
index 2f3573a23..4d3d45af5 100644
--- a/src/osprey/mcp_server/dispatch_worker/retention.py
+++ b/src/osprey/mcp_server/dispatch_worker/retention.py
@@ -154,16 +154,21 @@ def sweep_artifacts(
     cutoff = now - retention_days * _SECONDS_PER_DAY
     deleted = 0
 
+    from osprey.stores.artifact_store import artifact_mutation_actor
+
     # Snapshot the entry ids first: delete_entry mutates the index under a lock,
     # so iterate over a stable list rather than the live entry collection.
-    for entry in list(store.list_entries()):
-        if entry.run_id and entry.run_id in in_flight:
-            continue
-        ts = _parse_iso_timestamp(entry.timestamp)
-        if ts is None or ts >= cutoff:
-            continue
-        if store.delete_entry(entry.id):
-            deleted += 1
+    # These deletes are maintenance, not agent actions — tag them so store
+    # listeners don't report them as agent activity.
+    with artifact_mutation_actor("system"):
+        for entry in list(store.list_entries()):
+            if entry.run_id and entry.run_id in in_flight:
+                continue
+            ts = _parse_iso_timestamp(entry.timestamp)
+            if ts is None or ts >= cutoff:
+                continue
+            if store.delete_entry(entry.id):
+                deleted += 1
 
     return deleted
 
diff --git a/src/osprey/mcp_server/http.py b/src/osprey/mcp_server/http.py
index ec525ec16..bc10f954e 100644
--- a/src/osprey/mcp_server/http.py
+++ b/src/osprey/mcp_server/http.py
@@ -357,3 +357,24 @@ def notify_agent_activity(
         post_json(f"{base}/api/agent-activity", {"tool": tool, "target": target}, timeout=1)
     except Exception as exc:
         logger.warning("agent-activity notify failed (non-fatal): %s", exc)
+
+
+async def notify_agent_activity_async(
+    tool: str,
+    kind: str,
+    panel: str | None = None,
+    detail: str | None = None,
+) -> None:
+    """Awaitable form of :func:`notify_agent_activity` for coroutine emit sites.
+
+    The single thread-hop for async tools: the blocking (bounded ~1s) POST runs
+    on a worker thread so the event loop is never stalled. Same fire-and-forget
+    contract — never raises.
+    """
+    import functools
+
+    import anyio
+
+    await anyio.to_thread.run_sync(
+        functools.partial(notify_agent_activity, tool, kind, panel=panel, detail=detail)
+    )
diff --git a/src/osprey/mcp_server/phoebus/tools/bridge_tools.py b/src/osprey/mcp_server/phoebus/tools/bridge_tools.py
index 60ec801c6..edf856269 100644
--- a/src/osprey/mcp_server/phoebus/tools/bridge_tools.py
+++ b/src/osprey/mcp_server/phoebus/tools/bridge_tools.py
@@ -49,7 +49,6 @@
 """
 
 import asyncio
-import functools
 import json
 import logging
 import os
@@ -67,7 +66,7 @@
 from osprey.mcp_server.errors import make_error
 from osprey.mcp_server.http import (
     _post_json_with_response,
-    notify_agent_activity,
+    notify_agent_activity_async,
     notify_panel_focus,
     phoebus_bridge_url,
 )
@@ -660,17 +659,12 @@ async def phoebus_drive(
     # ``type`` is the exception: it writes the widget's PV through the runtime
     # and so reports fired=false even though the value landed. Every refusal
     # (validation, handle enforcement, unreachable bridge, non-200) returns
-    # above, so those emit nothing. notify_agent_activity never raises; the
+    # above, so those emit nothing. notify_agent_activity_async never raises; the
     # blocking call runs off the event loop.
     fired = bool(body.get("fired"))
     if fired or (verb_l == "type" and mode_l == "semantic"):
-        await anyio.to_thread.run_sync(
-            functools.partial(
-                notify_agent_activity,
-                "phoebus_drive",
-                "channel",
-                detail=f"{verb_l} {widget} on {display}",
-            )
+        await notify_agent_activity_async(
+            "phoebus_drive", "channel", detail=f"{verb_l} {widget} on {display}"
         )
 
     return json.dumps(
diff --git a/src/osprey/mcp_server/python_executor/tools/_execution_gates.py b/src/osprey/mcp_server/python_executor/tools/_execution_gates.py
new file mode 100644
index 000000000..d764e1ff7
--- /dev/null
+++ b/src/osprey/mcp_server/python_executor/tools/_execution_gates.py
@@ -0,0 +1,75 @@
+"""Shared execution-mode gates for the ``execute`` and ``execute_file`` tools.
+
+Both tools take an ``execution_mode`` string and guard control-system writes
+with two independent checks: a per-call readonly gate (pattern detection) and a
+deployment-level kill switch (``control_system.writes_enabled`` in the project
+config). Each gate only recognises one canonical spelling, so any *other*
+string used to fall through both — not "readonly", so write patterns were not
+blocked; not "readwrite", so the kill switch never fired. Rejecting unknown
+modes here closes that hole for every caller at once, and gives the kill
+switch a single implementation instead of one copy per tool.
+"""
+
+from __future__ import annotations
+
+import logging
+
+from osprey.mcp_server.errors import make_error
+
+logger = logging.getLogger("osprey.mcp_server.tools.execution_gates")
+
+#: The closed set of recognised execution modes. Downstream gates may test
+#: equality against either member only because this set is enforced first.
+VALID_EXECUTION_MODES = frozenset({"readonly", "readwrite"})
+
+
+def require_known_execution_mode(execution_mode: str) -> None:
+    """Raise ``ToolError`` (validation_error) unless the mode is recognised.
+
+    Must run before any write gate: the gates branch on string equality, and
+    an unrecognised value would otherwise satisfy neither branch and execute
+    with no write protection at all.
+    """
+    if execution_mode in VALID_EXECUTION_MODES:
+        return
+    make_error(
+        "validation_error",
+        f"Unknown execution_mode {execution_mode!r}.",
+        ['Use "readonly" (default) to block control-system writes, or "readwrite" to allow them.'],
+    )
+
+
+def enforce_deployment_writes_gate(execution_mode: str) -> None:
+    """Raise ``ToolError`` (safety_error) on readwrite runs in a no-writes deployment.
+
+    Fires whenever the caller asks for write mode, regardless of whether the
+    pattern detector recognises specific write syntax — the deployment-level
+    kill switch must not depend on detection accuracy.
+    """
+    if execution_mode != "readwrite":
+        return
+
+    try:
+        from osprey.services.python_executor.execution.control import (
+            get_execution_control_config,
+        )
+
+        exec_control_config = get_execution_control_config()
+    except ImportError:
+        logger.warning(
+            "Execution control config unavailable — skipping deployment-level writes check"
+        )
+        return
+
+    if (
+        exec_control_config is not None
+        and exec_control_config.control_system_writes_enabled is False
+    ):
+        make_error(
+            "safety_error",
+            "Control-system writes are disabled in this deployment "
+            "(control_system.writes_enabled=false in project config).",
+            [
+                "Set control_system.writes_enabled=true in the project config to enable writes.",
+            ],
+        )
diff --git a/src/osprey/mcp_server/python_executor/tools/python_execute.py b/src/osprey/mcp_server/python_executor/tools/python_execute.py
index df455b258..ba5ad39ce 100644
--- a/src/osprey/mcp_server/python_executor/tools/python_execute.py
+++ b/src/osprey/mcp_server/python_executor/tools/python_execute.py
@@ -1,14 +1,15 @@
 """MCP tool: execute — run user-provided Python code with safety checks."""
 
-import functools
 import json
 import logging
 
-import anyio
-
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 from osprey.mcp_server.python_executor.server import mcp
+from osprey.mcp_server.python_executor.tools._execution_gates import (
+    enforce_deployment_writes_gate,
+    require_known_execution_mode,
+)
 from osprey.mcp_server.python_executor.tools._package_inventory import with_live_packages
 
 logger = logging.getLogger("osprey.mcp_server.tools.execute")
@@ -42,7 +43,7 @@ async def execute(
         code: Python source code to execute.
         description: Human-readable description of what the code does.
         execution_mode: "readonly" (default) blocks detected write patterns;
-                        "readwrite" allows them.
+                        "readwrite" allows them. Any other value is rejected.
         save_output: If True, save the code and output to a workspace data file.
 
     Returns:
@@ -55,6 +56,10 @@ async def execute(
             ["Provide Python code to execute."],
         )
 
+    # Reject unrecognised modes before any gate: the write gates branch on
+    # string equality and an unknown value would satisfy neither branch.
+    require_known_execution_mode(execution_mode)
+
     # Pre-execution safety checks (syntax, security, imports)
     try:
         from osprey.services.python_executor.analysis.safety_checks import quick_safety_check
@@ -81,33 +86,7 @@ async def execute(
         patterns = {"has_writes": False, "has_reads": False, "detected_patterns": {}}
 
     # Deployment-level kill switch (independent of pattern detection accuracy).
-    # Fires whenever the caller asks for write mode, regardless of whether the
-    # pattern detector recognises specific write syntax in `code`.
-    if execution_mode == "readwrite":
-        try:
-            from osprey.services.python_executor.execution.control import (
-                get_execution_control_config,
-            )
-
-            exec_control_config = get_execution_control_config()
-        except ImportError:
-            logger.warning(
-                "Execution control config unavailable — skipping deployment-level writes check"
-            )
-            exec_control_config = None
-
-        if (
-            exec_control_config is not None
-            and exec_control_config.control_system_writes_enabled is False
-        ):
-            return make_error(
-                "safety_error",
-                "Control-system writes are disabled in this deployment "
-                "(control_system.writes_enabled=false in project config).",
-                [
-                    "Set control_system.writes_enabled=true in the project config to enable writes.",
-                ],
-            )
+    enforce_deployment_writes_gate(execution_mode)
 
     # Per-call execution-mode gate (uses pattern detection — readonly-mode safety).
     if patterns.get("has_writes") and execution_mode == "readonly":
@@ -138,22 +117,18 @@ async def execute(
     # "never launched" outcome comes back from execute_code's setup handler with
     # it still None. Every pre-execution gate above returns before this point.
     #
-    # The mode test is the complement of the readonly gate, not equality with
-    # "readwrite": any mode string that got past that gate let the writes run,
-    # so it has to be visible. Emitting before build_execution_response keeps
-    # the report independent of artifact/notebook persistence failures.
+    # The mode test is the complement of the readonly gate: with the mode set
+    # closed by require_known_execution_mode, the two spellings are equivalent,
+    # but the complement keeps this emit correct even if the set ever grows.
+    # Emitting before build_execution_response keeps the report independent of
+    # artifact/notebook persistence failures.
     if (
         patterns.get("has_writes")
         and execution_mode != "readonly"
         and exec_result.execution_time_seconds is not None
     ):
-        await anyio.to_thread.run_sync(
-            functools.partial(
-                notify_agent_activity,
-                "execute",
-                "channel",
-                detail="ran a script with control-system writes",
-            )
+        await notify_agent_activity_async(
+            "execute", "channel", detail="ran a script with control-system writes"
         )
 
     from osprey.mcp_server.python_executor.tools._response_builder import build_execution_response
diff --git a/src/osprey/mcp_server/python_executor/tools/python_execute_file.py b/src/osprey/mcp_server/python_executor/tools/python_execute_file.py
index d8feeba79..17128fa4f 100644
--- a/src/osprey/mcp_server/python_executor/tools/python_execute_file.py
+++ b/src/osprey/mcp_server/python_executor/tools/python_execute_file.py
@@ -1,15 +1,16 @@
 """MCP tool: execute_file — run an existing Python file with safety checks."""
 
-import functools
 import json
 import logging
 from pathlib import Path
 
-import anyio
-
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 from osprey.mcp_server.python_executor.server import mcp
+from osprey.mcp_server.python_executor.tools._execution_gates import (
+    enforce_deployment_writes_gate,
+    require_known_execution_mode,
+)
 
 logger = logging.getLogger("osprey.mcp_server.tools.execute_file")
 
@@ -33,7 +34,7 @@ async def execute_file(
                    relative paths resolve against the project root.
         description: Human-readable description of what the script does.
         execution_mode: "readonly" (default) blocks detected write patterns;
-                        "readwrite" allows them.
+                        "readwrite" allows them. Any other value is rejected.
         script_args: Optional command-line arguments for the script
                      (populates ``sys.argv[1:]``).
         save_output: If True, save the code and output to a workspace data file.
@@ -48,6 +49,10 @@ async def execute_file(
             ["Provide a path to a Python (.py) file."],
         )
 
+    # Reject unrecognised modes before any gate: the write gates branch on
+    # string equality and an unknown value would satisfy neither branch.
+    require_known_execution_mode(execution_mode)
+
     # Resolve project root and file path
     from osprey.mcp_server.python_executor.executor import _resolve_project_root
 
@@ -130,6 +135,9 @@ async def execute_file(
         logger.warning("Pattern detection module unavailable — skipping write detection")
         patterns = {"has_writes": False, "has_reads": False, "detected_patterns": {}}
 
+    # Deployment-level kill switch (independent of pattern detection accuracy).
+    enforce_deployment_writes_gate(execution_mode)
+
     if patterns.get("has_writes") and execution_mode == "readonly":
         return make_error(
             "safety_error",
@@ -158,20 +166,14 @@ async def execute_file(
 
     # Same emit contract as the ``execute`` tool — see the comment there for why
     # `execution_time_seconds` is the launch discriminator and why the mode test
-    # is the complement of the readonly gate rather than equality with
-    # "readwrite".
+    # is the complement of the readonly gate.
     if (
         patterns.get("has_writes")
         and execution_mode != "readonly"
         and exec_result.execution_time_seconds is not None
     ):
-        await anyio.to_thread.run_sync(
-            functools.partial(
-                notify_agent_activity,
-                "execute_file",
-                "channel",
-                detail="ran a script with control-system writes",
-            )
+        await notify_agent_activity_async(
+            "execute_file", "channel", detail="ran a script with control-system writes"
         )
 
     # Build response using original code (not augmented) for metadata/notebook
diff --git a/src/osprey/mcp_server/workspace/tools/focus_tools.py b/src/osprey/mcp_server/workspace/tools/focus_tools.py
index 043ccce40..75a6c17bc 100644
--- a/src/osprey/mcp_server/workspace/tools/focus_tools.py
+++ b/src/osprey/mcp_server/workspace/tools/focus_tools.py
@@ -13,15 +13,16 @@
   gallery was actually notified via ``gallery_notified``.
 """
 
-import functools
 import json
 import logging
 import urllib.error
 
-import anyio
-
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import _post_json_with_response, gallery_url, notify_agent_activity
+from osprey.mcp_server.http import (
+    _post_json_with_response,
+    gallery_url,
+    notify_agent_activity_async,
+)
 from osprey.mcp_server.workspace.server import mcp
 
 logger = logging.getLogger("osprey.mcp_server.tools.focus")
@@ -88,14 +89,9 @@ async def artifact_focus(artifact_id: str, fullscreen: bool = False) -> str:
 
     # Agent-activity highlight for the host activity strip. The gallery itself
     # is unchanged — it already self-signals via its own focus SSE above.
-    # notify_agent_activity never raises; the blocking call runs off the loop.
-    await anyio.to_thread.run_sync(
-        functools.partial(
-            notify_agent_activity,
-            "artifact_focus",
-            "artifact",
-            detail=entry.title or artifact_id,
-        )
+    # notify_agent_activity_async never raises; the blocking call runs off the loop.
+    await notify_agent_activity_async(
+        "artifact_focus", "artifact", detail=entry.title or artifact_id
     )
 
     return json.dumps(
@@ -151,14 +147,7 @@ async def artifact_pin(artifact_id: str, pinned: bool = True) -> str:
         logger.warning("Gallery pin notification failed (non-fatal): %s", exc)
 
     # Agent-activity highlight (same contract as artifact_focus above).
-    await anyio.to_thread.run_sync(
-        functools.partial(
-            notify_agent_activity,
-            "artifact_pin",
-            "artifact",
-            detail=entry.title or artifact_id,
-        )
-    )
+    await notify_agent_activity_async("artifact_pin", "artifact", detail=entry.title or artifact_id)
 
     return json.dumps(
         {
diff --git a/src/osprey/mcp_server/workspace/tools/lattice_tools.py b/src/osprey/mcp_server/workspace/tools/lattice_tools.py
index 5675c5d7b..93777d09e 100644
--- a/src/osprey/mcp_server/workspace/tools/lattice_tools.py
+++ b/src/osprey/mcp_server/workspace/tools/lattice_tools.py
@@ -13,17 +13,15 @@
   - ``lattice_update_settings``: Deep-merge new settings (partial update).
 """
 
-import functools
 import json
 import logging
 import os
 
-import anyio
 import httpx
 from fastmcp.exceptions import ToolError
 
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 from osprey.mcp_server.workspace.server import mcp
 from osprey.utils.workspace import load_osprey_config
 
@@ -73,15 +71,7 @@ async def _notify_lattice(tool: str, detail: str) -> None:
         tool: Name of the lattice tool that mutated the dashboard.
         detail: Short human-readable description of what changed.
     """
-    await anyio.to_thread.run_sync(
-        functools.partial(
-            notify_agent_activity,
-            tool,
-            "panel",
-            panel="lattice",
-            detail=detail,
-        )
-    )
+    await notify_agent_activity_async(tool, "panel", panel="lattice", detail=detail)
 
 
 @mcp.tool()
diff --git a/src/osprey/mcp_server/workspace/tools/screen_capture.py b/src/osprey/mcp_server/workspace/tools/screen_capture.py
index c705dabb9..8b7a90b19 100644
--- a/src/osprey/mcp_server/workspace/tools/screen_capture.py
+++ b/src/osprey/mcp_server/workspace/tools/screen_capture.py
@@ -5,17 +5,15 @@
 The tool functions below are thin validation + delegation layers.
 """
 
-import functools
 import json
 import logging
 from datetime import UTC, datetime
 from pathlib import Path
 
-import anyio
 from fastmcp.exceptions import ToolError
 
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 from osprey.mcp_server.workspace.server import mcp
 from osprey.mcp_server.workspace.tools.screen_capture_backends import (
     BackendUnavailableError,
@@ -267,14 +265,7 @@ async def manage_window(
 
         # Only once the backend moved the window: a refused or failed action
         # raises out of make_error above and reports nothing.
-        await anyio.to_thread.run_sync(
-            functools.partial(
-                notify_agent_activity,
-                "manage_window",
-                "ui",
-                detail=f"{action} {app}",
-            )
-        )
+        await notify_agent_activity_async("manage_window", "ui", detail=f"{action} {app}")
 
         return json.dumps(
             {
diff --git a/src/osprey/mcp_server/workspace/tools/setup.py b/src/osprey/mcp_server/workspace/tools/setup.py
index fc8f45940..3a4ee61c6 100644
--- a/src/osprey/mcp_server/workspace/tools/setup.py
+++ b/src/osprey/mcp_server/workspace/tools/setup.py
@@ -4,18 +4,16 @@
 Used by the /setup-mode skill to help operators troubleshoot setup issues.
 """
 
-import functools
 import json
 import logging
 import os
 import re
 from pathlib import Path
 
-import anyio
 from fastmcp.exceptions import ToolError
 
 from osprey.mcp_server.errors import make_error
-from osprey.mcp_server.http import notify_agent_activity
+from osprey.mcp_server.http import notify_agent_activity_async
 from osprey.mcp_server.workspace.server import mcp
 from osprey.utils.workspace import load_osprey_config, resolve_config_path
 
@@ -240,13 +238,8 @@ async def _notify_patch(file: str, key_path: str) -> None:
         file: Target file name.
         key_path: Dot-notation path that was patched.
     """
-    await anyio.to_thread.run_sync(
-        functools.partial(
-            notify_agent_activity,
-            "setup_patch",
-            "config",
-            detail=_activity_detail(file, key_path),
-        )
+    await notify_agent_activity_async(
+        "setup_patch", "config", detail=_activity_detail(file, key_path)
     )
 
 
diff --git a/src/osprey/stores/artifact_store.py b/src/osprey/stores/artifact_store.py
index a86ce1191..fe07ea19f 100644
--- a/src/osprey/stores/artifact_store.py
+++ b/src/osprey/stores/artifact_store.py
@@ -13,12 +13,14 @@
 
 from __future__ import annotations
 
+import contextvars
 import json
 import logging
 import mimetypes
 import os
 import uuid
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
 from dataclasses import asdict, dataclass, field
 from datetime import UTC, datetime
 from pathlib import Path
@@ -28,6 +30,35 @@
 
 logger = logging.getLogger("osprey.stores.artifact_store")
 
+#: Who is performing the current store mutation. The store itself cannot tell
+#: an agent tool call from a gallery click or a retention sweep — the frames a
+#: listener emits from these events are *agent* activity, so non-agent callers
+#: declare themselves via :func:`artifact_mutation_actor` and listeners read
+#: :func:`current_artifact_mutation_actor` at event time. Defaults to "agent"
+#: because every MCP-tool path mutates the store on the agent's behalf.
+_MUTATION_ACTOR: contextvars.ContextVar[str] = contextvars.ContextVar(
+    "artifact_mutation_actor", default="agent"
+)
+
+
+@contextmanager
+def artifact_mutation_actor(actor: str) -> Iterator[None]:
+    """Attribute store mutations in this scope to *actor* ("human", "system").
+
+    Listener callbacks fire synchronously inside the mutation call, so the
+    scope only needs to cover the ``save_*``/``delete_*`` call itself.
+    """
+    token = _MUTATION_ACTOR.set(actor)
+    try:
+        yield
+    finally:
+        _MUTATION_ACTOR.reset(token)
+
+
+def current_artifact_mutation_actor() -> str:
+    """Return the actor of the mutation currently firing listeners."""
+    return _MUTATION_ACTOR.get()
+
 
 def register_artifact_listener(fn: Callable[[ArtifactEntry], None]) -> None:
     """Register a callback invoked after every artifact save."""
diff --git a/tests/dispatch_worker/test_retention.py b/tests/dispatch_worker/test_retention.py
index 0e83b1bdf..bb5a8d9d3 100644
--- a/tests/dispatch_worker/test_retention.py
+++ b/tests/dispatch_worker/test_retention.py
@@ -218,6 +218,34 @@ def test_artifact_of_in_flight_run_survives(tmp_path):
     assert store.get_entry(art) is not None
 
 
+def test_artifact_sweep_deletes_run_under_the_system_actor(tmp_path):
+    """Retention deletes are maintenance: the store-level activity listener
+    must be able to tell them apart from agent deletes."""
+    from osprey.stores.artifact_store import (
+        current_artifact_mutation_actor,
+        register_artifact_delete_listener,
+        unregister_artifact_delete_listener,
+    )
+
+    store = ArtifactStore(workspace_root=tmp_path)
+    old = _save_artifact(store, "old")
+    _set_artifact_age_days(store, old, 100)
+
+    actors = []
+
+    def record_actor(_entry):
+        actors.append(current_artifact_mutation_actor())
+
+    register_artifact_delete_listener(record_actor)
+    try:
+        deleted = retention.sweep_artifacts(store, retention_days=5, now=_NOW)
+    finally:
+        unregister_artifact_delete_listener(record_actor)
+
+    assert deleted == 1
+    assert actors == ["system"]
+
+
 def test_artifact_sweep_disabled(tmp_path):
     store = ArtifactStore(workspace_root=tmp_path)
     art = _save_artifact(store, "old")
diff --git a/tests/interfaces/artifacts/test_app_endpoints.py b/tests/interfaces/artifacts/test_app_endpoints.py
index 30e5a1ca3..ebba99bb7 100644
--- a/tests/interfaces/artifacts/test_app_endpoints.py
+++ b/tests/interfaces/artifacts/test_app_endpoints.py
@@ -66,6 +66,31 @@ def test_get_artifact_returns_entry(self, app_client):
         assert resp.json()["id"] == entry.id
         assert resp.json()["title"] == "Fetch Me"
 
+    def test_delete_route_runs_under_the_human_actor(self, app_client):
+        """A gallery delete is a human action: the store-level activity
+        listener must be able to tell it apart from an agent delete."""
+        from osprey.stores.artifact_store import (
+            current_artifact_mutation_actor,
+            register_artifact_delete_listener,
+            unregister_artifact_delete_listener,
+        )
+
+        client, _ = app_client
+        entry = _save_text_artifact(client.app.state.artifact_store)
+
+        actors = []
+
+        def record_actor(_entry):
+            actors.append(current_artifact_mutation_actor())
+
+        register_artifact_delete_listener(record_actor)
+        try:
+            assert client.delete(f"/api/artifacts/{entry.id}").status_code == 200
+        finally:
+            unregister_artifact_delete_listener(record_actor)
+
+        assert actors == ["human"]
+
     def test_delete_artifact_removes_entry(self, app_client):
         client, _ = app_client
         entry = _save_text_artifact(client.app.state.artifact_store)
diff --git a/tests/mcp_server/bluesky/test_draft_tools_emit.py b/tests/mcp_server/bluesky/test_draft_tools_emit.py
index 1a551934b..f1bba0554 100644
--- a/tests/mcp_server/bluesky/test_draft_tools_emit.py
+++ b/tests/mcp_server/bluesky/test_draft_tools_emit.py
@@ -251,7 +251,7 @@ async def test_queue_stop_plain_stop_emits_detail_stop(tmp_path, monkeypatch):
     _queue_stop_posture(tmp_path, monkeypatch, writes=False, token=None)
     with (
         patch(f"{_QUEUE_MOD}._http_post_json", return_value=(200, _STOP_RESP)),
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _queue_stop_fn()()
 
@@ -265,7 +265,7 @@ async def test_queue_stop_withdrawal_emits_detail_stop_withdrawn(tmp_path, monke
     _queue_stop_posture(tmp_path, monkeypatch, writes=True, token=_QUEUE_STOP_TOKEN)
     with (
         patch(f"{_QUEUE_MOD}._http_post_json", return_value=(200, _WITHDRAWN_RESP)),
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _queue_stop_fn()(cancel=True)
 
@@ -280,7 +280,7 @@ async def test_queue_stop_withdrawal_refused_for_writes_disabled_does_not_emit(
     _queue_stop_posture(tmp_path, monkeypatch, writes=False, token=_QUEUE_STOP_TOKEN)
     with (
         patch(f"{_QUEUE_MOD}._http_post_json") as post,
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="writes_disabled"):
             await _queue_stop_fn()(cancel=True)
@@ -293,7 +293,7 @@ async def test_queue_stop_withdrawal_refused_without_a_token_does_not_emit(tmp_p
     _queue_stop_posture(tmp_path, monkeypatch, writes=True, token=None)
     with (
         patch(f"{_QUEUE_MOD}._http_post_json") as post,
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="launch_token_required"):
             await _queue_stop_fn()(cancel=True)
@@ -308,7 +308,7 @@ async def test_queue_stop_bridge_refusal_does_not_emit(tmp_path, monkeypatch):
     body = {"detail": {"code": "queue_request_rejected", "detail": "no stop is pending"}}
     with (
         patch(f"{_QUEUE_MOD}._http_post_json", return_value=(409, body)),
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="queue_request_rejected"):
             await _queue_stop_fn()()
diff --git a/tests/mcp_server/bluesky/test_queue_tools.py b/tests/mcp_server/bluesky/test_queue_tools.py
index 527122848..cd1a64357 100644
--- a/tests/mcp_server/bluesky/test_queue_tools.py
+++ b/tests/mcp_server/bluesky/test_queue_tools.py
@@ -235,7 +235,7 @@ async def test_queue_add_posts_the_pinned_revision_with_the_token_when_armed(tmp
     _armed(tmp_path, monkeypatch)
     body = {"run_id": "abc123", "revision": 7, "item": {"item_uid": "u1"}}
     with patch(f"{_MOD}._http_post_json", return_value=(200, body)) as m:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             result = await _add_fn()(draft_revision=7)
 
     assert m.call_args.args[0] == "/queue/items"
@@ -255,7 +255,7 @@ async def test_queue_add_withholds_the_token_when_writes_are_disabled(tmp_path,
     """
     _configure(tmp_path, monkeypatch, writes=False, token=_TOKEN)
     with patch(f"{_MOD}._http_post_json", return_value=(200, {"run_id": "r1"})) as m:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             await _add_fn()(draft_revision=3)
 
     assert m.call_args.kwargs["headers"] is None
@@ -265,7 +265,7 @@ async def test_queue_add_missing_config_fails_closed_and_withholds_the_token(tmp
     """No config.yml at all is not "writes enabled" by omission."""
     _configure(tmp_path, monkeypatch, writes=None, token=_TOKEN)
     with patch(f"{_MOD}._http_post_json", return_value=(200, {"run_id": "r1"})) as m:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             await _add_fn()(draft_revision=3)
 
     assert m.call_args.kwargs["headers"] is None
@@ -275,7 +275,7 @@ async def test_queue_add_without_a_configured_token_still_composes(tmp_path, mon
     """An unarmed deployment may still build a queue; only starting it is gated."""
     _configure(tmp_path, monkeypatch, writes=True, token=None)
     with patch(f"{_MOD}._http_post_json", return_value=(200, {"run_id": "r1"})) as m:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             await _add_fn()(draft_revision=3)
 
     m.assert_called_once()
@@ -444,13 +444,13 @@ async def test_queue_add_emits_agent_activity_only_after_a_confirmed_enqueue(tmp
     with patch(
         f"{_MOD}._http_post_json", return_value=(409, _refusal("stale_draft_revision", "no"))
     ):
-        with patch(f"{_MOD}.notify_agent_activity") as notify:
+        with patch(f"{_MOD}.notify_agent_activity_async") as notify:
             with assert_raises_error(error_type="stale_draft_revision"):
                 await _add_fn()(draft_revision=7)
     notify.assert_not_called()
 
     with patch(f"{_MOD}._http_post_json", return_value=(200, {"run_id": "abc123"})):
-        with patch(f"{_MOD}.notify_agent_activity") as notify:
+        with patch(f"{_MOD}.notify_agent_activity_async") as notify:
             await _add_fn()(draft_revision=7)
     assert notify.call_args.kwargs["detail"] == "abc123"
 
@@ -507,7 +507,7 @@ async def test_queue_start_without_a_token_files_a_panel_start_request(tmp_path,
         "items_in_queue": 2,
     }
     with patch(f"{_MOD}._http_post_json", return_value=(200, {"start_request": record})) as m:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             result = await _start_fn()()
 
     assert m.call_args.args[0] == "/queue/start-request"
@@ -570,7 +570,7 @@ async def test_queue_start_armed_posts_with_the_token(tmp_path, monkeypatch):
     """Contrast case: proves the refusals above are gated, not vacuous."""
     _armed(tmp_path, monkeypatch)
     with patch(f"{_MOD}._http_post_json", return_value=(200, {"started": True, "msg": ""})) as m:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             result = await _start_fn()()
 
     assert m.call_args.args[0] == "/queue/start"
diff --git a/tests/mcp_server/test_artifact_activity_listener.py b/tests/mcp_server/test_artifact_activity_listener.py
index 6343dce59..4e48c5b51 100644
--- a/tests/mcp_server/test_artifact_activity_listener.py
+++ b/tests/mcp_server/test_artifact_activity_listener.py
@@ -204,6 +204,65 @@ def test_delete_all_emits_per_surviving_entry(project, notified):
     assert "Orbit X" in details and "Orbit Y" in details
 
 
+@pytest.mark.unit
+@pytest.mark.parametrize("actor", ["human", "system"])
+def test_non_agent_deletes_never_emit(project, notified, actor):
+    """The frames are *agent* activity: a delete by the gallery user or a
+    retention sweep must not be reported as an agent action."""
+    from osprey.mcp_server.startup import initialize_workspace_singletons
+    from osprey.stores.artifact_store import artifact_mutation_actor
+
+    initialize_workspace_singletons()
+    store = ArtifactStore(workspace_root=project / "_agent_data")
+
+    save_figure(store, title="Orbit X")
+    wait_drained()
+    notified.clear()
+
+    with artifact_mutation_actor(actor):
+        store.delete_all()
+    wait_drained()
+
+    assert notified == []
+
+
+@pytest.mark.unit
+def test_actor_tag_is_scoped_to_the_context(project, notified):
+    """The agent default is restored when a non-agent scope exits."""
+    from osprey.mcp_server.startup import initialize_workspace_singletons
+    from osprey.stores.artifact_store import artifact_mutation_actor
+
+    initialize_workspace_singletons()
+    store = ArtifactStore(workspace_root=project / "_agent_data")
+
+    doomed = store.save_file(
+        file_content=b"\x89PNG fake",
+        filename="doomed.png",
+        artifact_type="image",
+        title="Doomed",
+        mime_type="image/png",
+        tool_source="execute",
+    )
+    kept = store.save_file(
+        file_content=b"\x89PNG fake",
+        filename="kept.png",
+        artifact_type="image",
+        title="Kept",
+        mime_type="image/png",
+        tool_source="execute",
+    )
+    wait_drained()
+    notified.clear()
+
+    with artifact_mutation_actor("human"):
+        store.delete_entry(doomed.id)
+    store.delete_entry(kept.id)
+    wait_drained()
+
+    assert [kwargs["tool"] for kwargs, _ in notified] == ["artifact_delete"]
+    assert "Kept" in notified[0][0]["detail"]
+
+
 @pytest.mark.unit
 def test_notify_runs_on_the_worker_not_the_caller(project, notified):
     """The store callback must not do HTTP on the thread that saved."""
diff --git a/tests/mcp_server/test_backend_emit_sites.py b/tests/mcp_server/test_backend_emit_sites.py
index 017a8c0c1..56b1ec01e 100644
--- a/tests/mcp_server/test_backend_emit_sites.py
+++ b/tests/mcp_server/test_backend_emit_sites.py
@@ -1,18 +1,20 @@
 """Agent-activity emit sites for backend-direct tools.
 
 Verifies that the mutating backend tools report agent activity via
-``notify_agent_activity`` — and, just as important, that refusal paths emit
-NOTHING and that tool results are unchanged when the web terminal is down.
+``notify_agent_activity_async`` — and, just as important, that refusal paths
+emit NOTHING and that tool results are unchanged when the web terminal is down.
 
 Layout: one ``# ── <tool family> ──`` section per tool family, each owning its
 own helpers and fixtures, followed by a trailing cross-cutting "terminal down"
 section. New emit sites append a new section immediately BEFORE that trailing
 section; shared helpers stay generic and live at module top.
 
-Patch seam: each tool module imports ``notify_agent_activity`` directly, so
-the mock must target the caller's namespace (e.g.
-``osprey.mcp_server.control_system.tools.channel_write.notify_agent_activity``),
-not ``osprey.mcp_server.http``.
+Patch seam: each tool module imports ``notify_agent_activity_async`` directly,
+so the mock must target the caller's namespace (e.g.
+``osprey.mcp_server.control_system.tools.channel_write.notify_agent_activity_async``),
+not ``osprey.mcp_server.http``. The helper owns the thread-hop off the event
+loop; that property is pinned once in ``test_notify_agent_activity.py``, not
+per tool.
 """
 
 import contextlib
@@ -114,7 +116,7 @@ async def test_channel_write_limits_violation_no_emit(tmp_path, monkeypatch):
             "osprey.connectors.control_system.limits_validator.LimitsValidator.from_config",
             return_value=mock_validator,
         ),
-        patch(f"{_CW_MOD}.notify_agent_activity") as notify,
+        patch(f"{_CW_MOD}.notify_agent_activity_async") as notify,
     ):
         fn = _get_channel_write()
         with assert_raises_error(error_type="limits_violation"):
@@ -146,7 +148,7 @@ async def test_channel_write_partial_success_emits_executed_only(tmp_path, monke
     mock_connector.write_multiple_channels.return_value = results
 
     conn_patch, validator_patch = _channel_write_patches(mock_connector)
-    with conn_patch, validator_patch, patch(f"{_CW_MOD}.notify_agent_activity") as notify:
+    with conn_patch, validator_patch, patch(f"{_CW_MOD}.notify_agent_activity_async") as notify:
         fn = _get_channel_write()
         result = await fn(
             operations=[
@@ -199,7 +201,7 @@ async def test_channel_write_all_blocked_no_emit(tmp_path, monkeypatch):
     mock_connector.write_multiple_channels.return_value = results
 
     conn_patch, validator_patch = _channel_write_patches(mock_connector)
-    with conn_patch, validator_patch, patch(f"{_CW_MOD}.notify_agent_activity") as notify:
+    with conn_patch, validator_patch, patch(f"{_CW_MOD}.notify_agent_activity_async") as notify:
         fn = _get_channel_write()
         with assert_raises_error(error_type="write_refused"):
             await fn(
@@ -225,7 +227,7 @@ async def test_channel_write_full_success_single_emit(tmp_path, monkeypatch):
     mock_connector.write_channel.return_value = write_result
 
     conn_patch, validator_patch = _channel_write_patches(mock_connector)
-    with conn_patch, validator_patch, patch(f"{_CW_MOD}.notify_agent_activity") as notify:
+    with conn_patch, validator_patch, patch(f"{_CW_MOD}.notify_agent_activity_async") as notify:
         fn = _get_channel_write()
         result = await fn(operations=[{"channel": "SR01:HCM1:SP", "value": 42.0}])
 
@@ -264,7 +266,7 @@ async def test_queue_add_success_emits_run_id(_bluesky_context):
     body = {"run_id": "abc123", "revision": 7, "item": {"item_uid": "u1"}}
     with (
         patch(f"{_QUEUE_MOD}._http_post_json", return_value=(200, body)),
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _get_queue_tool("queue_add")(draft_revision=7)
 
@@ -281,7 +283,7 @@ async def test_queue_add_failure_no_emit(_bluesky_context):
             f"{_QUEUE_MOD}._http_post_json",
             return_value=(500, {"detail": "enqueue failed: boom"}),
         ),
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="bluesky_bridge_error"):
             await _get_queue_tool("queue_add")(draft_revision=7)
@@ -292,7 +294,7 @@ async def test_queue_add_failure_no_emit(_bluesky_context):
 async def test_queue_start_success_emits(_bluesky_context):
     with (
         patch(f"{_QUEUE_MOD}._http_post_json", return_value=(200, {"started": True, "msg": ""})),
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         await _get_queue_tool("queue_start")()
 
@@ -310,7 +312,7 @@ async def test_queue_start_client_side_refusal_no_emit(_bluesky_context, monkeyp
 
     with (
         patch(f"{_QUEUE_MOD}._http_post_json") as post,
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="writes_disabled"):
             await _get_queue_tool("queue_start")()
@@ -334,7 +336,7 @@ async def test_queue_start_tokenless_emits_the_start_request(_bluesky_context, m
     body = {"start_request": {"request_id": "r1", "requested_by": "agent"}}
     with (
         patch(f"{_QUEUE_MOD}._http_post_json", return_value=(200, body)) as post,
-        patch(f"{_QUEUE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_QUEUE_MOD}.notify_agent_activity_async") as notify,
     ):
         await _get_queue_tool("queue_start")()
 
@@ -365,7 +367,7 @@ async def test_artifact_focus_emits_artifact(tmp_path, monkeypatch):
     # errors, and this test is about the notify seam, not the gallery.
     with (
         patch(f"{_FOCUS_MOD}._post_json_with_response", return_value=(200, {"status": "ok"})),
-        patch(f"{_FOCUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_FOCUS_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await get_tool_fn(artifact_focus)(artifact_id=artifact_id)
 
@@ -381,7 +383,7 @@ async def test_artifact_focus_not_found_no_emit(tmp_path, monkeypatch):
 
     from osprey.mcp_server.workspace.tools.focus_tools import artifact_focus
 
-    with patch(f"{_FOCUS_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_FOCUS_MOD}.notify_agent_activity_async") as notify:
         with assert_raises_error(error_type="not_found"):
             await get_tool_fn(artifact_focus)(artifact_id="nonexistent-id")
 
@@ -434,7 +436,7 @@ async def test_ariel_entry_create_direct_emits_panel(_ariel_context):
 
     with (
         _patch_ariel_service(mock_service),
-        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity") as notify,
+        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _get_ariel_tool("entry", "entry_create")(
             subject="Beam lost", details="Injector trip at 03:12", draft=False
@@ -460,7 +462,7 @@ async def test_ariel_entry_create_direct_does_not_steal_focus(_ariel_context):
 
     with (
         _patch_ariel_service(mock_service),
-        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity"),
+        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity_async"),
         patch("osprey.mcp_server.http.notify_panel_focus") as focus,
     ):
         await _get_ariel_tool("entry", "entry_create")(
@@ -488,7 +490,7 @@ async def test_ariel_entry_create_emits_before_attachment_failure(_ariel_context
             "osprey.services.ariel_search.attachments.process_attachments_for_entry",
             new=AsyncMock(side_effect=RuntimeError("attachment store offline")),
         ),
-        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity") as notify,
+        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="internal_error"):
             await _get_ariel_tool("entry", "entry_create")(
@@ -504,7 +506,7 @@ async def test_ariel_entry_create_emits_before_attachment_failure(_ariel_context
 
 async def test_ariel_entry_create_validation_refusal_no_emit(_ariel_context):
     """Argument validation refuses before any write, so nothing is emitted."""
-    with patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity_async") as notify:
         with assert_raises_error(error_type="validation_error"):
             await _get_ariel_tool("entry", "entry_create")(
                 subject="   ", details="Injector trip at 03:12", draft=False
@@ -520,7 +522,7 @@ async def test_ariel_entry_create_upsert_failure_no_emit(_ariel_context):
 
     with (
         _patch_ariel_service(mock_service),
-        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity") as notify,
+        patch(f"{_ARIEL_ENTRY_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="internal_error"):
             await _get_ariel_tool("entry", "entry_create")(
@@ -544,7 +546,7 @@ async def test_ariel_entry_publish_success_emits_facility_id(_ariel_context):
 
     with (
         _patch_ariel_service(mock_service),
-        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity") as notify,
+        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _get_ariel_tool("publish", "entry_publish")(
             entry_id="e1", logbook="Operations"
@@ -560,7 +562,7 @@ async def test_ariel_entry_publish_success_emits_facility_id(_ariel_context):
 
 async def test_ariel_entry_publish_validation_refusal_no_emit(_ariel_context):
     """An empty entry_id is refused before the service is touched."""
-    with patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity_async") as notify:
         with assert_raises_error(error_type="validation_error"):
             await _get_ariel_tool("publish", "entry_publish")(entry_id="")
 
@@ -574,7 +576,7 @@ async def test_ariel_entry_publish_not_found_no_emit(_ariel_context):
 
     with (
         _patch_ariel_service(mock_service),
-        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity") as notify,
+        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="not_found"):
             await _get_ariel_tool("publish", "entry_publish")(entry_id="e99")
@@ -593,7 +595,7 @@ async def test_ariel_entry_publish_auth_required_no_emit(_ariel_context):
 
     with (
         _patch_ariel_service(mock_service),
-        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity") as notify,
+        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="auth_required"):
             await _get_ariel_tool("publish", "entry_publish")(entry_id="e1")
@@ -610,7 +612,7 @@ async def test_ariel_entry_publish_not_supported_no_emit(_ariel_context):
 
     with (
         _patch_ariel_service(mock_service),
-        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity") as notify,
+        patch(f"{_ARIEL_PUBLISH_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="not_supported"):
             await _get_ariel_tool("publish", "entry_publish")(entry_id="e1")
@@ -653,7 +655,7 @@ async def test_phoebus_drive_click_fired_emits(_phoebus_active_display_allowed):
     """A click that fired a control wrote to the machine — emit once, kind channel."""
     with (
         _phoebus_bridge(200, {"fired": True, "detail": "fired via ButtonBase.fire()"}),
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _get_phoebus_drive()(widget="SetButton", verb="click")
 
@@ -665,7 +667,7 @@ async def test_phoebus_drive_synthetic_type_fired_emits():
     """A synthetic type that fired echoes the normalized verb and the display ref."""
     with (
         _phoebus_bridge(200, {"fired": True, "detail": "committed 42"}),
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         await _get_phoebus_drive()(widget="Setpoint", verb="TYPE", value="42", display="handle:d-3")
 
@@ -676,7 +678,7 @@ async def test_phoebus_drive_synthetic_200_not_fired_no_emit(_phoebus_active_dis
     """Bridge contract: a synthetic 200 with fired=false resolved no control — no write."""
     with (
         _phoebus_bridge(200, {"fired": False, "detail": "no interactive control resolved"}),
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _get_phoebus_drive()(widget="Readback", verb="type", value="42")
 
@@ -696,7 +698,7 @@ async def test_phoebus_drive_semantic_type_not_fired_emits(
     """
     with (
         _phoebus_bridge(200, {"fired": False, "detail": "wrote PV SR:CORR:SP"}),
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         await _get_phoebus_drive()(widget="Setpoint", verb=verb, value="1.5", mode=mode)
 
@@ -707,7 +709,7 @@ async def test_phoebus_drive_semantic_click_bypass_no_emit(_phoebus_active_displ
     """Semantic mode bypasses click entirely — nothing was driven, so emit nothing."""
     with (
         _phoebus_bridge(200, {"fired": False, "detail": "semantic mode bypasses click"}),
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         await _get_phoebus_drive()(widget="SetButton", verb="click", mode="semantic")
 
@@ -726,7 +728,7 @@ async def test_phoebus_drive_validation_refusal_no_emit(kwargs, _phoebus_active_
     """Argument validation refuses before the bridge is contacted — emit nothing."""
     with (
         patch(f"{_PHOEBUS_MOD}._http_post_drive") as post,
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="validation_error"):
             await _get_phoebus_drive()(**kwargs)
@@ -741,7 +743,7 @@ async def test_phoebus_drive_handle_required_refusal_no_emit(monkeypatch):
 
     with (
         patch(f"{_PHOEBUS_MOD}._http_post_drive") as post,
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="phoebus_handle_required"):
             await _get_phoebus_drive()(widget="SetButton", verb="click")
@@ -754,7 +756,7 @@ async def test_phoebus_drive_bridge_rejected_no_emit(_phoebus_active_display_all
     """A non-200 from the bridge means no widget was driven — emit nothing."""
     with (
         _phoebus_bridge(400, {"error": "Unknown widget '0'", "status": 400}),
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="phoebus_rejected"):
             await _get_phoebus_drive()(widget="0", verb="click")
@@ -768,7 +770,7 @@ async def test_phoebus_drive_unreachable_no_emit(_phoebus_active_display_allowed
 
     with (
         patch(f"{_PHOEBUS_MOD}._http_post_drive", side_effect=urllib.error.URLError("refused")),
-        patch(f"{_PHOEBUS_MOD}.notify_agent_activity") as notify,
+        patch(f"{_PHOEBUS_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="phoebus_unreachable"):
             await _get_phoebus_drive()(widget="SetButton", verb="click")
@@ -776,24 +778,6 @@ async def test_phoebus_drive_unreachable_no_emit(_phoebus_active_display_allowed
     notify.assert_not_called()
 
 
-async def test_phoebus_drive_emit_runs_off_the_event_loop(_phoebus_active_display_allowed):
-    """The blocking notify must be thread-wrapped, never awaited inline."""
-    import threading
-
-    seen: list[int] = []
-
-    with (
-        _phoebus_bridge(200, {"fired": True, "detail": "ok"}),
-        patch(
-            f"{_PHOEBUS_MOD}.notify_agent_activity",
-            side_effect=lambda *a, **k: seen.append(threading.get_ident()),
-        ),
-    ):
-        await _get_phoebus_drive()(widget="SetButton", verb="click")
-
-    assert seen and seen[0] != threading.get_ident()
-
-
 # ── python executor: execute / execute_file ─────────────────────────────────
 #
 # Launch discriminator this section pins: ``ExecutionResult.execution_time_seconds``
@@ -898,7 +882,7 @@ def _execute_env(mod, tmp_path, *, exec_result=None, has_writes=True, writes_ena
             return_value=ExecutionControlConfig(control_system_writes_enabled=writes_enabled),
         ),
         patch("osprey.mcp_server.python_executor.executor.execute_code", exec_code),
-        patch(f"{mod}.notify_agent_activity") as notify,
+        patch(f"{mod}.notify_agent_activity_async") as notify,
     ):
         yield notify, exec_code
 
@@ -1000,31 +984,29 @@ async def test_execute_readwrite_without_write_patterns_no_emit(tool_name, tmp_p
 
 
 @pytest.mark.parametrize("tool_name", _EXECUTE_TOOLS)
-async def test_execute_uncanonical_write_mode_still_emits(tool_name, tmp_path, monkeypatch):
-    """A non-canonical mode spelling clears the readonly gate, so its writes must report.
+async def test_execute_uncanonical_write_mode_rejected_no_emit(tool_name, tmp_path, monkeypatch):
+    """A non-canonical mode spelling is refused at the boundary — emit nothing.
 
-    ``execution_mode`` is an unvalidated free string and is never normalized, so
-    "READWRITE" runs the script exactly as "readwrite" does. Testing equality
-    with "readwrite" instead of the gate's complement would silence this run.
+    "READWRITE" used to clear both write gates as an unvalidated free string
+    and run the script; it is now rejected before any gate, so nothing runs
+    and nothing reports.
     """
     monkeypatch.chdir(tmp_path)
     mod, call = _execute_tool_call(tool_name, tmp_path)
 
     with _execute_env(mod, tmp_path) as (notify, exec_code):
-        await call(execution_mode="READWRITE")
-
-    exec_code.assert_called_once()
-    notify.assert_called_once_with(tool_name, "channel", detail=_EXECUTE_DETAIL)
+        with assert_raises_error(error_type="validation_error"):
+            await call(execution_mode="READWRITE")
 
+    exec_code.assert_not_called()
+    notify.assert_not_called()
 
-async def test_execute_deployment_writes_disabled_no_emit(tmp_path, monkeypatch):
-    """The deployment kill switch refuses before launch — emit nothing.
 
-    Only the ``execute`` tool carries this gate; ``execute_file`` has no
-    equivalent, so this case is not parametrized.
-    """
+@pytest.mark.parametrize("tool_name", _EXECUTE_TOOLS)
+async def test_execute_deployment_writes_disabled_no_emit(tool_name, tmp_path, monkeypatch):
+    """The deployment kill switch refuses before launch — emit nothing."""
     monkeypatch.chdir(tmp_path)
-    mod, call = _execute_tool_call("execute", tmp_path)
+    mod, call = _execute_tool_call(tool_name, tmp_path)
 
     with _execute_env(mod, tmp_path, writes_enabled=False) as (notify, exec_code):
         with assert_raises_error(error_type="safety_error"):
@@ -1034,22 +1016,6 @@ async def test_execute_deployment_writes_disabled_no_emit(tmp_path, monkeypatch)
     notify.assert_not_called()
 
 
-@pytest.mark.parametrize("tool_name", _EXECUTE_TOOLS)
-async def test_execute_emit_runs_off_the_event_loop(tool_name, tmp_path, monkeypatch):
-    """The blocking notify must be thread-wrapped, never awaited inline."""
-    import threading
-
-    monkeypatch.chdir(tmp_path)
-    mod, call = _execute_tool_call(tool_name, tmp_path)
-    seen: list[int] = []
-
-    with _execute_env(mod, tmp_path) as (notify, _):
-        notify.side_effect = lambda *a, **k: seen.append(threading.get_ident())
-        await call(execution_mode="readwrite")
-
-    assert seen and seen[0] != threading.get_ident()
-
-
 # ── lattice dashboard mutators ──────────────────────────────────────────────
 #
 # The six mutators share one refusal shape: every failure arrives as an
@@ -1129,7 +1095,7 @@ async def test_lattice_mutator_emits(tool_name, kwargs, detail):
     """Each acknowledged mutation reports once against the 'lattice' panel."""
     with (
         _lattice_request(),
-        patch(f"{_LATTICE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_LATTICE_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _get_lattice_tool(tool_name)(**kwargs)
 
@@ -1142,7 +1108,7 @@ async def test_lattice_mutator_unreachable_dashboard_no_emit(tool_name, kwargs):
     """A dashboard that never answered changed nothing — emit nothing."""
     with (
         _lattice_request(side_effect=_lattice_unreachable()),
-        patch(f"{_LATTICE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_LATTICE_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="service_unavailable"):
             await _get_lattice_tool(tool_name)(**kwargs)
@@ -1155,7 +1121,7 @@ async def test_lattice_mutator_rejected_request_no_emit(tool_name, kwargs):
     """A non-2xx answer means the dashboard refused the change — emit nothing."""
     with (
         _lattice_request(side_effect=_lattice_http_error()),
-        patch(f"{_LATTICE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_LATTICE_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="lattice_error"):
             await _get_lattice_tool(tool_name)(**kwargs)
@@ -1168,31 +1134,13 @@ async def test_lattice_read_only_tool_never_emits(tool_name, kwargs):
     """Reading state or a figure mutates nothing, so it stays out of the feed."""
     with (
         _lattice_request(return_value={"status": "ok"}),
-        patch(f"{_LATTICE_MOD}.notify_agent_activity") as notify,
+        patch(f"{_LATTICE_MOD}.notify_agent_activity_async") as notify,
     ):
         await _get_lattice_tool(tool_name)(**kwargs)
 
     notify.assert_not_called()
 
 
-async def test_lattice_emit_runs_off_the_event_loop():
-    """The blocking notify must be thread-wrapped, never awaited inline."""
-    import threading
-
-    seen: list[int] = []
-
-    with (
-        _lattice_request(),
-        patch(
-            f"{_LATTICE_MOD}.notify_agent_activity",
-            side_effect=lambda *a, **k: seen.append(threading.get_ident()),
-        ),
-    ):
-        await _get_lattice_tool("lattice_set_param")(family="QF", value=1.25)
-
-    assert seen and seen[0] != threading.get_ident()
-
-
 # ── setup_patch / manage_window ─────────────────────────────────────────────
 #
 # setup_patch edits the two files that hold the agent's credentials, so its
@@ -1264,7 +1212,7 @@ def _backend_unavailable():
 
 async def test_setup_patch_emits_config_activity(setup_project):
     """A landed patch reports the file and key path under the 'config' kind."""
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         result = await _get_setup_patch()(
             file="config.yml", key_path="agent_data.base_dir", value="./_agent_data"
         )
@@ -1277,7 +1225,7 @@ async def test_setup_patch_emits_config_activity(setup_project):
 
 async def test_setup_patch_emits_for_json_target(setup_project):
     """The `.mcp.json` branch reports too — it is the same mutation."""
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         await _get_setup_patch()(
             file=".mcp.json", key_path="mcpServers.demo.env.API_KEY", value=_NEW_SENTINEL
         )
@@ -1291,7 +1239,7 @@ async def test_setup_patch_detail_never_carries_the_patched_values(setup_project
     """Hard security bound: neither the old nor the new value may reach the feed."""
     import json
 
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         await _get_setup_patch()(
             file=".mcp.json", key_path="mcpServers.demo.env.API_KEY", value=_NEW_SENTINEL
         )
@@ -1310,7 +1258,7 @@ async def test_setup_patch_detail_never_carries_the_patched_values(setup_project
 
 async def test_setup_patch_marks_control_system_keys_as_safety_config(setup_project):
     """A safety-relevant key path is distinguishable at a glance in the feed."""
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         await _get_setup_patch()(
             file="config.yml", key_path="control_system.writes_enabled", value="true"
         )
@@ -1329,7 +1277,7 @@ async def test_setup_patch_safety_prefix_is_exact_case(setup_project):
     lowercase one carries the marker and a differently-cased path — which names
     a different key, and gets no hot/cold note either — does not.
     """
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         await _get_setup_patch()(
             file="config.yml", key_path="Control_System.writes_enabled", value="true"
         )
@@ -1341,7 +1289,7 @@ async def test_setup_patch_safety_prefix_is_exact_case(setup_project):
 
 async def test_setup_patch_unpatchable_file_no_emit(setup_project):
     """A file outside the whitelist was never opened — emit nothing."""
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         with assert_raises_error(error_type="validation_error"):
             await _get_setup_patch()(file="settings.json", key_path="permissions.deny", value="[]")
 
@@ -1351,7 +1299,7 @@ async def test_setup_patch_unpatchable_file_no_emit(setup_project):
 @pytest.mark.parametrize("key_path", ["", "../../etc/passwd", "/abs/path", "has space"])
 async def test_setup_patch_invalid_key_path_no_emit(setup_project, key_path):
     """A rejected key path changed nothing — emit nothing."""
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         with assert_raises_error(error_type="validation_error"):
             await _get_setup_patch()(file="config.yml", key_path=key_path, value="1")
 
@@ -1362,7 +1310,7 @@ async def test_setup_patch_missing_file_no_emit(setup_project):
     """Nothing to patch means nothing to report."""
     (setup_project / ".mcp.json").unlink()
 
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         with assert_raises_error(error_type="not_found"):
             await _get_setup_patch()(file=".mcp.json", key_path="mcpServers.x", value="1")
 
@@ -1373,34 +1321,19 @@ async def test_setup_patch_unparseable_file_no_emit(setup_project):
     """A file that could not be read back was never rewritten — emit nothing."""
     (setup_project / ".mcp.json").write_text("{ this is not json")
 
-    with patch(f"{_SETUP_MOD}.notify_agent_activity") as notify:
+    with patch(f"{_SETUP_MOD}.notify_agent_activity_async") as notify:
         with assert_raises_error(error_type="internal_error"):
             await _get_setup_patch()(file=".mcp.json", key_path="mcpServers.x", value="1")
 
     notify.assert_not_called()
 
 
-async def test_setup_patch_emit_runs_off_the_event_loop(setup_project):
-    """The blocking notify must be thread-wrapped, never awaited inline."""
-    import threading
-
-    seen: list[int] = []
-
-    with patch(
-        f"{_SETUP_MOD}.notify_agent_activity",
-        side_effect=lambda *a, **k: seen.append(threading.get_ident()),
-    ):
-        await _get_setup_patch()(file="config.yml", key_path="control_system.type", value="mock")
-
-    assert seen and seen[0] != threading.get_ident()
-
-
 @pytest.mark.parametrize("action,kwargs", _MANAGE_WINDOW_ACTIONS)
 async def test_manage_window_emits_ui_activity(action, kwargs):
     """Each completed window action reports once under the 'ui' kind."""
     with (
         _screen_backend(),
-        patch(f"{_SCREEN_MOD}.notify_agent_activity") as notify,
+        patch(f"{_SCREEN_MOD}.notify_agent_activity_async") as notify,
     ):
         result = await _get_manage_window()(app="Phoebus", action=action, **kwargs)
 
@@ -1412,7 +1345,7 @@ async def test_manage_window_detail_preserves_app_name_case():
     """Nothing normalises `app`, so the feed shows the name the operator sees."""
     with (
         _screen_backend(),
-        patch(f"{_SCREEN_MOD}.notify_agent_activity") as notify,
+        patch(f"{_SCREEN_MOD}.notify_agent_activity_async") as notify,
     ):
         await _get_manage_window()(app="google Chrome", action="bring_to_front")
 
@@ -1424,7 +1357,7 @@ async def test_manage_window_unknown_action_no_emit(action):
     """Action matching is exact-case; every unmatched spelling is refused, not reported."""
     with (
         _screen_backend(),
-        patch(f"{_SCREEN_MOD}.notify_agent_activity") as notify,
+        patch(f"{_SCREEN_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="validation_error"):
             await _get_manage_window()(app="Phoebus", action=action, x=1, y=2)
@@ -1445,7 +1378,7 @@ async def test_manage_window_incomplete_parameters_no_emit(action, kwargs):
     """A refusal before the backend call moved no window — emit nothing."""
     with (
         _screen_backend(),
-        patch(f"{_SCREEN_MOD}.notify_agent_activity") as notify,
+        patch(f"{_SCREEN_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="validation_error"):
             await _get_manage_window()(app="Phoebus", action=action, **kwargs)
@@ -1458,7 +1391,7 @@ async def test_manage_window_backend_failure_no_emit(action, kwargs):
     """A backend that could not act leaves the window alone — emit nothing."""
     with (
         _screen_backend(side_effect=_backend_unavailable()),
-        patch(f"{_SCREEN_MOD}.notify_agent_activity") as notify,
+        patch(f"{_SCREEN_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="platform_error"):
             await _get_manage_window()(app="Phoebus", action=action, **kwargs)
@@ -1470,7 +1403,7 @@ async def test_manage_window_unknown_app_no_emit():
     """A window the backend cannot find was never touched — emit nothing."""
     with (
         _screen_backend(side_effect=ValueError("no window for app 'Nope'")),
-        patch(f"{_SCREEN_MOD}.notify_agent_activity") as notify,
+        patch(f"{_SCREEN_MOD}.notify_agent_activity_async") as notify,
     ):
         with assert_raises_error(error_type="validation_error"):
             await _get_manage_window()(app="Nope", action="bring_to_front")
@@ -1478,24 +1411,6 @@ async def test_manage_window_unknown_app_no_emit():
     notify.assert_not_called()
 
 
-async def test_manage_window_emit_runs_off_the_event_loop():
-    """The blocking notify must be thread-wrapped, never awaited inline."""
-    import threading
-
-    seen: list[int] = []
-
-    with (
-        _screen_backend(),
-        patch(
-            f"{_SCREEN_MOD}.notify_agent_activity",
-            side_effect=lambda *a, **k: seen.append(threading.get_ident()),
-        ),
-    ):
-        await _get_manage_window()(app="Phoebus", action="bring_to_front")
-
-    assert seen and seen[0] != threading.get_ident()
-
-
 # ── terminal down: real helper against a dead port ──────────────────────────
 
 
diff --git a/tests/mcp_server/test_notify_agent_activity.py b/tests/mcp_server/test_notify_agent_activity.py
index 29c0a561d..89b294b34 100644
--- a/tests/mcp_server/test_notify_agent_activity.py
+++ b/tests/mcp_server/test_notify_agent_activity.py
@@ -116,6 +116,30 @@ def test_none_fields_omitted(self, capture_server):
         assert "detail" not in body["target"]
 
 
+class TestAsyncWrapper:
+    async def test_posts_off_the_event_loop_with_args_passed_through(self):
+        """The async helper is the one thread-hop for every coroutine emit site.
+
+        Tools await it instead of hand-rolling
+        ``anyio.to_thread.run_sync(functools.partial(...))``; the blocking POST
+        must run on a worker thread with all arguments forwarded unchanged.
+        """
+        from osprey.mcp_server.http import notify_agent_activity_async
+
+        calls: list[tuple[int, str, str, str | None, str | None]] = []
+
+        def record(tool, kind, panel=None, detail=None):
+            calls.append((threading.get_ident(), tool, kind, panel, detail))
+
+        with patch(f"{_MODULE}.notify_agent_activity", side_effect=record):
+            await notify_agent_activity_async(
+                "channel_write", "channel", panel="controls", detail="SR:HC1:SP"
+            )
+
+        assert [c[1:] for c in calls] == [("channel_write", "channel", "controls", "SR:HC1:SP")]
+        assert calls[0][0] != threading.get_ident()
+
+
 class TestTimeout:
     def test_hanging_socket_returns_quickly_without_raising(self):
         # Socket that accepts connections but never responds.
diff --git a/tests/mcp_server/test_python_execute_file_tool.py b/tests/mcp_server/test_python_execute_file_tool.py
index 8a7e7691e..acd6fece9 100644
--- a/tests/mcp_server/test_python_execute_file_tool.py
+++ b/tests/mcp_server/test_python_execute_file_tool.py
@@ -277,6 +277,10 @@ async def test_execute_file_readwrite_allows_writes(tmp_path, monkeypatch):
 
     mock_exec = _mock_execute_code(success=True, stdout="done\n")
 
+    # The deployment-level writes gate is independent of execution_mode; this
+    # test asserts the readwrite path, so the gate must be allow-through.
+    from osprey.services.python_executor.execution.control import ExecutionControlConfig
+
     with (
         patch(
             "osprey.mcp_server.python_executor.executor._resolve_project_root",
@@ -290,6 +294,10 @@ async def test_execute_file_readwrite_allows_writes(tmp_path, monkeypatch):
                 "detected_patterns": {"writes": ["caput"], "reads": []},
             },
         ),
+        patch(
+            "osprey.services.python_executor.execution.control.get_execution_control_config",
+            return_value=ExecutionControlConfig(control_system_writes_enabled=True),
+        ),
         patch(
             "osprey.mcp_server.python_executor.executor.execute_code",
             mock_exec,
@@ -307,6 +315,104 @@ async def test_execute_file_readwrite_allows_writes(tmp_path, monkeypatch):
     assert data["summary"]["status"] == "Success"
 
 
+@pytest.mark.unit
+async def test_execute_file_deployment_writes_disabled_blocks(tmp_path, monkeypatch):
+    """The deployment kill switch refuses readwrite runs when writes are disabled.
+
+    ``execute_file`` used to carry no deployment-level gate at all: a script
+    full of write patterns ran under execution_mode="readwrite" even with
+    control_system.writes_enabled=false in the project config.
+    """
+    monkeypatch.chdir(tmp_path)
+
+    script = tmp_path / "writer.py"
+    script.write_text("epics.caput('PV', 1)\n")
+
+    mock_exec = _mock_execute_code()
+
+    from osprey.services.python_executor.execution.control import ExecutionControlConfig
+
+    with (
+        patch(
+            "osprey.mcp_server.python_executor.executor._resolve_project_root",
+            return_value=tmp_path,
+        ),
+        patch(
+            "osprey.services.python_executor.analysis.pattern_detection.detect_control_system_operations",
+            return_value={
+                "has_writes": True,
+                "has_reads": False,
+                "detected_patterns": {"writes": ["caput"], "reads": []},
+            },
+        ),
+        patch(
+            "osprey.services.python_executor.execution.control.get_execution_control_config",
+            return_value=ExecutionControlConfig(control_system_writes_enabled=False),
+        ),
+        patch(
+            "osprey.mcp_server.python_executor.executor.execute_code",
+            mock_exec,
+        ),
+    ):
+        fn = _get_python_execute_file()
+        with assert_raises_error(error_type="safety_error") as ctx:
+            await fn(
+                file_path=str(script),
+                description="kill switch test",
+                execution_mode="readwrite",
+            )
+
+    mock_exec.assert_not_called()
+    assert "writes_enabled" in ctx["envelope"]["error_message"]
+
+
+@pytest.mark.unit
+@pytest.mark.parametrize("mode", ["ReadWrite", "READWRITE", "write", "read_write"])
+async def test_execute_file_rejects_unknown_execution_mode(tmp_path, monkeypatch, mode):
+    """Modes outside {readonly, readwrite} are rejected before any gate runs."""
+    monkeypatch.chdir(tmp_path)
+
+    script = tmp_path / "writer.py"
+    script.write_text("epics.caput('PV', 1)\n")
+
+    mock_exec = _mock_execute_code()
+
+    from osprey.services.python_executor.execution.control import ExecutionControlConfig
+
+    with (
+        patch(
+            "osprey.mcp_server.python_executor.executor._resolve_project_root",
+            return_value=tmp_path,
+        ),
+        patch(
+            "osprey.services.python_executor.analysis.pattern_detection.detect_control_system_operations",
+            return_value={
+                "has_writes": True,
+                "has_reads": False,
+                "detected_patterns": {"writes": ["caput"], "reads": []},
+            },
+        ),
+        patch(
+            "osprey.services.python_executor.execution.control.get_execution_control_config",
+            return_value=ExecutionControlConfig(control_system_writes_enabled=False),
+        ),
+        patch(
+            "osprey.mcp_server.python_executor.executor.execute_code",
+            mock_exec,
+        ),
+    ):
+        fn = _get_python_execute_file()
+        with assert_raises_error(error_type="validation_error") as ctx:
+            await fn(
+                file_path=str(script),
+                description="unknown mode bypass",
+                execution_mode=mode,
+            )
+
+    mock_exec.assert_not_called()
+    assert "execution_mode" in ctx["envelope"]["error_message"]
+
+
 @pytest.mark.unit
 async def test_execute_file_script_args(tmp_path, monkeypatch):
     """Script args are injected into sys.argv preamble."""
diff --git a/tests/mcp_server/test_python_execute_tool.py b/tests/mcp_server/test_python_execute_tool.py
index 2d2718fca..7762e8502 100644
--- a/tests/mcp_server/test_python_execute_tool.py
+++ b/tests/mcp_server/test_python_execute_tool.py
@@ -247,6 +247,52 @@ async def test_python_execute_readwrite_mode(tmp_path, monkeypatch):
     assert data["summary"]["status"] == "Success"
 
 
+@pytest.mark.unit
+@pytest.mark.parametrize("mode", ["ReadWrite", "READWRITE", "write", "read_write"])
+async def test_python_execute_rejects_unknown_execution_mode(tmp_path, monkeypatch, mode):
+    """Modes outside {readonly, readwrite} are rejected before any gate runs.
+
+    An unrecognized string used to fall through BOTH write gates: it is not
+    "readonly" (so the pattern block never fired) and not "readwrite" (so the
+    deployment kill switch never fired), letting write patterns execute even
+    with control_system.writes_enabled=false.
+    """
+    monkeypatch.chdir(tmp_path)
+
+    from osprey.services.python_executor.execution.control import ExecutionControlConfig
+
+    mock_exec = _mock_execute_code()
+
+    with (
+        patch(
+            "osprey.services.python_executor.analysis.pattern_detection.detect_control_system_operations",
+            return_value={
+                "has_writes": True,
+                "has_reads": False,
+                "detected_patterns": {"caput": ["caput('TEST:PV', 1.0)"]},
+            },
+        ),
+        patch(
+            "osprey.services.python_executor.execution.control.get_execution_control_config",
+            return_value=ExecutionControlConfig(control_system_writes_enabled=False),
+        ),
+        patch(
+            "osprey.mcp_server.python_executor.executor.execute_code",
+            mock_exec,
+        ),
+    ):
+        fn = _get_python_execute()
+        with assert_raises_error(error_type="validation_error") as ctx:
+            await fn(
+                code="caput('TEST:PV', 1.0)",
+                description="unknown mode bypass",
+                execution_mode=mode,
+            )
+
+    mock_exec.assert_not_called()
+    assert "execution_mode" in ctx["envelope"]["error_message"]
+
+
 @pytest.mark.unit
 async def test_python_execute_empty_code(tmp_path, monkeypatch):
     """Empty code returns validation error."""
diff --git a/tests/mcp_server/test_stop_run.py b/tests/mcp_server/test_stop_run.py
index 54c3c0936..933d71536 100644
--- a/tests/mcp_server/test_stop_run.py
+++ b/tests/mcp_server/test_stop_run.py
@@ -71,7 +71,7 @@ def _locked_down(tmp_path, monkeypatch) -> None:
 
 async def test_stop_run_aborts_the_running_plan():
     with patch(f"{_MOD}._http_post_json", return_value=(200, _ABORT_OK)) as post:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             result = await _fn()()
 
     assert post.call_args.args[0] == "/queue/abort"
@@ -92,7 +92,7 @@ async def test_stop_run_budgets_above_the_bridges_composed_abort():
     actually in flight, and told to retry into a race.
     """
     with patch(f"{_MOD}._http_post_json", return_value=(200, _ABORT_OK)) as post:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             await _fn()()
 
     assert post.call_args.kwargs["timeout"] == stop._ABORT_TIMEOUT
@@ -118,7 +118,7 @@ async def test_stop_run_is_ungated_with_writes_off_and_no_token(tmp_path, monkey
     _locked_down(tmp_path, monkeypatch)
 
     with patch(f"{_MOD}._http_post_json", return_value=(200, _ABORT_OK)) as post:
-        with patch(f"{_MOD}.notify_agent_activity"):
+        with patch(f"{_MOD}.notify_agent_activity_async"):
             result = await _fn()()
 
     assert extract_response_dict(result)["aborted"] is True
@@ -143,7 +143,7 @@ async def test_stop_run_does_not_read_the_writes_kill_switch(tmp_path, monkeypat
 
     with patch("osprey.mcp_server.bluesky.tools.queue._writes_enabled") as writes:
         with patch(f"{_MOD}._http_post_json", return_value=(200, _ABORT_OK)):
-            with patch(f"{_MOD}.notify_agent_activity"):
+            with patch(f"{_MOD}.notify_agent_activity_async"):
                 await _fn()()
 
     assert writes.call_count == 0