feat(ci): faster docusaurus builds :)

Reserve API requests before their first await, include every drain-owned work type in runtime status, and pause local cron dispatch while a gateway drain is active.

acp_adapter/permissions.py

@@ -38,19 +38,22 @@ def _permission_option_supports_kind(kind: str) -> bool:
     return True
 
 
-def _build_permission_options(*, allow_permanent: bool) -> list[PermissionOption]:
+def _build_permission_options(
+    *, allow_permanent: bool, smart_denied: bool = False,
+) -> list[PermissionOption]:
     """Return ACP options that match Hermes approval semantics."""
-    options = [
-        PermissionOption(option_id="allow_once", kind="allow_once", name="Allow once"),
-        PermissionOption(
+    options = [PermissionOption(
+        option_id="allow_once", kind="allow_once", name="Allow once",
+    )]
+    if not smart_denied:
+        options.append(PermissionOption(
             option_id="allow_session",
             # ACP has no session-scoped kind, so use the closest persistent
             # hint while keeping Hermes semantics in the option id.
             kind="allow_always",
             name="Allow for session",
-        ),
-    ]
-    if allow_permanent:
+        ))
+    if allow_permanent and not smart_denied:
         options.append(
             PermissionOption(
                 option_id="allow_always",
@@ -59,7 +62,7 @@ def _build_permission_options(*, allow_permanent: bool) -> list[PermissionOption
             ),
         )
     options.append(PermissionOption(option_id="deny", kind="reject_once", name="Deny"))
-    if _permission_option_supports_kind("reject_always"):
+    if not smart_denied and _permission_option_supports_kind("reject_always"):
         options.append(
             PermissionOption(
                 option_id="deny_always",
@@ -129,11 +132,15 @@ def make_approval_callback(
         description: str,
         *,
         allow_permanent: bool = True,
+        smart_denied: bool = False,
         **_: object,
     ) -> str:
         from agent.async_utils import safe_schedule_threadsafe
 
-        options = _build_permission_options(allow_permanent=allow_permanent)
+        options = _build_permission_options(
+            allow_permanent=allow_permanent,
+            smart_denied=smart_denied,
+        )
 
         tool_call = _build_permission_tool_call(command, description)
         coro = request_permission_fn(

agent/agent_init.py

@@ -187,10 +187,26 @@ def _normalized_custom_base_url(value: Any) -> str:
 
 
 def _custom_provider_model_matches(agent_model: str, entry: Dict[str, Any]) -> bool:
-    provider_model = str(entry.get("model", "") or "").strip().lower()
-    if not provider_model:
+    agent_model_norm = str(agent_model or "").strip().lower()
+    # Multi-model entries (v12+ `providers.<name>.models` mapping / legacy
+    # `models:` list): the agent's model matching ANY catalog entry counts.
+    # Without this, a provider whose `model`/`default_model` differs from the
+    # session model silently fails to match and per-provider request settings
+    # (extra_body, e.g. OpenAI service_tier) are dropped — billing the whole
+    # session at the wrong tier (July 2026 sweeper incident: flex config
+    # ignored, ~2.3x overbilling).
+    models = entry.get("models")
+    catalog: List[str] = []
+    if isinstance(models, dict):
+        catalog = [str(k).strip().lower() for k in models.keys()]
+    elif isinstance(models, (list, tuple)):
+        catalog = [str(m).strip().lower() for m in models]
+    if catalog and agent_model_norm in catalog:
         return True
-    return provider_model == str(agent_model or "").strip().lower()
+    provider_model = str(entry.get("model", "") or "").strip().lower()
+    if not provider_model and not catalog:
+        return True
+    return provider_model == agent_model_norm
 
 
 def _custom_provider_extra_body_for_agent(

agent/credential_pool.py

@@ -128,6 +128,17 @@ _EXTRA_KEYS = frozenset({
 })
 
 
+def _normalize_pool_auth_type(provider: str, token: Any, auth_type: Any) -> str:
+    """Infer pool auth metadata for token formats with one unambiguous meaning."""
+    if (
+        provider == "anthropic"
+        and isinstance(token, str)
+        and token.startswith("sk-ant-oat")
+    ):
+        return AUTH_TYPE_OAUTH
+    return str(auth_type or AUTH_TYPE_API_KEY)
+
+
 @dataclass
 class PooledCredential:
     provider: str
@@ -157,6 +168,11 @@ class PooledCredential:
     def __post_init__(self):
         if self.extra is None:
             self.extra = {}
+        self.auth_type = _normalize_pool_auth_type(
+            self.provider,
+            self.access_token,
+            self.auth_type,
+        )
 
     def __getattr__(self, name: str):
         if name in _EXTRA_KEYS:
@@ -1416,6 +1432,11 @@ class CredentialPool:
         entries_to_prune: List[str] = []
         available: List[PooledCredential] = []
         for entry in self._entries:
+            # Borrowed credentials persist as metadata-only references and are
+            # hydrated from their live source on load.  A stale duplicate row
+            # can remain unhydrated; never lease or select it as an empty key.
+            if entry.auth_type == AUTH_TYPE_API_KEY and not entry.runtime_api_key:
+                continue
             # For anthropic claude_code entries, sync from the credentials file
             # before any status/refresh checks. This picks up tokens refreshed
             # by other processes (Claude Code CLI, other Hermes profiles).
@@ -1732,11 +1753,15 @@ class CredentialPool:
 
 
 def _upsert_entry(entries: List[PooledCredential], provider: str, source: str, payload: Dict[str, Any]) -> bool:
-    existing_idx = None
+    matching_indices = []
     for idx, entry in enumerate(entries):
         if entry.source == source:
-            existing_idx = idx
-            break
+            matching_indices.append(idx)
+
+    existing_idx = matching_indices[0] if matching_indices else None
+    duplicate_indices = set(matching_indices[1:])
+    if duplicate_indices:
+        entries[:] = [entry for idx, entry in enumerate(entries) if idx not in duplicate_indices]
 
     if existing_idx is None:
         payload.setdefault("id", uuid.uuid4().hex[:6])
@@ -1768,8 +1793,8 @@ def _upsert_entry(entries: List[PooledCredential], provider: str, source: str, p
         # Runtime-only borrowed secret updates should refresh the in-memory
         # entry without forcing auth.json churn when the disk-safe payload is
         # unchanged (for example env keys with the same fingerprint).
-        return existing.to_dict() != updated.to_dict()
-    return False
+        return bool(duplicate_indices) or existing.to_dict() != updated.to_dict()
+    return bool(duplicate_indices)
 
 
 def _normalize_pool_priorities(provider: str, entries: List[PooledCredential]) -> bool:
@@ -2243,12 +2268,6 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
         if _is_source_suppressed(provider, source):
             continue
         active_sources.add(source)
-        # Claude Code OAuth tokens are the only Anthropic credentials that should flow into the OAuth refresh path.
-        auth_type = (
-            AUTH_TYPE_OAUTH
-            if provider == "anthropic" and token.startswith("sk-ant-oat")
-            else AUTH_TYPE_API_KEY
-        )
         base_url = env_url or pconfig.inference_base_url
         if provider == "kimi-coding":
             base_url = _resolve_kimi_base_url(token, pconfig.inference_base_url, env_url)
@@ -2263,7 +2282,6 @@ def _seed_from_env(provider: str, entries: List[PooledCredential]) -> Tuple[bool
                 env_var=env_var,
                 token=token,
                 base_url=base_url,
-                auth_type=auth_type,
             ),
         )
     return changed, active_sources
@@ -2391,16 +2409,37 @@ def load_pool(provider: str) -> CredentialPool:
         for payload in raw_entries
     )
     entries = [PooledCredential.from_dict(provider, payload) for payload in raw_entries]
+    raw_needs_auth_normalization = any(
+        isinstance(payload, dict)
+        and _normalize_pool_auth_type(
+            provider,
+            payload.get("access_token"),
+            payload.get("auth_type", AUTH_TYPE_API_KEY),
+        ) != payload.get("auth_type", AUTH_TYPE_API_KEY)
+        for payload in raw_entries
+    )
+    if raw_needs_auth_normalization:
+        # A profile may be reading this provider from the global-root fallback.
+        # Keep that fallback read-only: only the store that owns these rows may
+        # rewrite them. Loading the default/root profile will heal global rows.
+        active_pool = _load_auth_store().get("credential_pool")
+        active_entries = active_pool.get(provider) if isinstance(active_pool, dict) else None
+        raw_needs_auth_normalization = bool(active_entries)
 
     if provider.startswith(CUSTOM_POOL_PREFIX):
         # Custom endpoint pool — seed from custom_providers config and model config
         custom_changed, custom_sources = _seed_custom_pool(provider, entries)
-        changed = raw_needs_sanitization or custom_changed
+        changed = raw_needs_sanitization or raw_needs_auth_normalization or custom_changed
         changed |= _prune_stale_seeded_entries(entries, custom_sources)
     else:
         singleton_changed, singleton_sources = _seed_from_singletons(provider, entries)
         env_changed, env_sources = _seed_from_env(provider, entries)
-        changed = raw_needs_sanitization or singleton_changed or env_changed
+        changed = (
+            raw_needs_sanitization
+            or raw_needs_auth_normalization
+            or singleton_changed
+            or env_changed
+        )
         # ``load_pool()`` is a non-destructive read for env-seeded entries: a
         # process missing a provider env var must not delete the persisted
         # pool entry for every other process (#9331). File-backed singletons

agent/turn_finalizer.py

@@ -449,6 +449,11 @@ def finalize_turn(
         "estimated_cost_usd": agent.session_estimated_cost_usd,
         "cost_status": agent.session_cost_status,
         "cost_source": agent.session_cost_source,
+        # Requested service tier (from request_overrides.extra_body), for
+        # billing audits by callers like `hermes -z --usage-file`.
+        "service_tier": (
+            (getattr(agent, "request_overrides", {}) or {}).get("extra_body") or {}
+        ).get("service_tier"),
         "session_id": agent.session_id,
     }
     if agent._tool_guardrail_halt_decision is not None:

apps/desktop/scripts/stage-native-deps.mjs

@@ -13,6 +13,7 @@ import { createRequire } from 'node:module'
 import { fileURLToPath } from 'node:url'
 import { dirname, resolve, join } from 'node:path'
 import {
+  chmodSync,
   cpSync,
   existsSync,
   mkdirSync,
@@ -115,6 +116,7 @@ export function stageNodePty({ platform = process.platform, arch = process.arch
       }
       if (entry.name === 'spawn-helper') {
         cpSync(join(prebuildDir, entry.name), join(destPrebuild, entry.name))
+        chmodSync(join(destPrebuild, entry.name), 0o775)
       }
     }
   } else {

apps/desktop/src/app/chat/composer/enter-submit-dom-race.test.tsx

@@ -54,6 +54,10 @@ function Harness({
   }
 
   const submitDraft = () => {
+    if (disabled) {
+      return
+    }
+
     const editor = editorRef.current
 
     if (editor) {
@@ -68,16 +72,6 @@ function Harness({
     const text = draftRef.current
     const payloadPresent = text.trim().length > 0 || attachments.length > 0
 
-    if (disabled) {
-      // Gateway down: queue the draft instead of dropping it (mirrors the real
-      // queueCurrentDraft, which no-ops on an empty payload).
-      if (payloadPresent) {
-        onQueue(text)
-      }
-
-      return
-    }
-
     if (busy) {
       if (payloadPresent) {
         onQueue(text)
@@ -99,10 +93,6 @@ function Harness({
       const hasLivePayload = editorText.trim().length > 0 || attachments.length > 0
 
       if (disabled) {
-        if (hasLivePayload) {
-          submitDraft()
-        }
-
         return
       }
 
@@ -217,17 +207,16 @@ describe('composer Enter submit — live DOM vs stale composer state (#39630)',
     expect(onSubmit).not.toHaveBeenCalled()
   })
 
-  it('queues a reconnect draft on Enter (not send/drain) so it flushes when the gateway returns', async () => {
+  it('keeps reconnect drafts editable but blocks Enter submit until the gateway returns', async () => {
     const onSubmit = vi.fn()
     const onDrain = vi.fn()
-    const onQueue = vi.fn()
 
     const { getByTestId } = render(
       <Harness
         disabled
         onCancel={vi.fn()}
         onDrain={onDrain}
-        onQueue={onQueue}
+        onQueue={vi.fn()}
         onSubmit={onSubmit}
         queued={['queued-1']}
       />
@@ -241,32 +230,8 @@ describe('composer Enter submit — live DOM vs stale composer state (#39630)',
       fireEvent.keyDown(editor, { key: 'Enter' })
     })
 
-    // The gateway is down, so the message can't send — but it must NOT be
-    // silently dropped (the #-bug: "type, hit Enter, nothing happens, no
-    // error"). It queues, and the gateway-open-gated auto-drain sends it later.
-    expect(onQueue).toHaveBeenCalledWith('draft while reconnecting')
-    expect(onSubmit).not.toHaveBeenCalled()
+    expect(editor.textContent).toBe('draft while reconnecting')
     expect(onDrain).not.toHaveBeenCalled()
-  })
-
-  it('treats an empty Enter while reconnecting as a no-op (no phantom queue entry)', async () => {
-    const onQueue = vi.fn()
-    const onSubmit = vi.fn()
-    const onDrain = vi.fn()
-
-    const { getByTestId } = render(
-      <Harness disabled onCancel={vi.fn()} onDrain={onDrain} onQueue={onQueue} onSubmit={onSubmit} />
-    )
-
-    const editor = getByTestId('editor')
-
-    await act(async () => {
-      editor.textContent = ''
-      fireEvent.keyDown(editor, { key: 'Enter' })
-    })
-
-    expect(onQueue).not.toHaveBeenCalled()
     expect(onSubmit).not.toHaveBeenCalled()
-    expect(onDrain).not.toHaveBeenCalled()
   })
 })

apps/desktop/src/app/chat/composer/hooks/use-composer-queue.ts

@@ -28,7 +28,6 @@ interface UseComposerQueueArgs {
   clearDraft: () => void
   draftRef: RefObject<string>
   focusInput: () => void
-  gatewayConnected: boolean
   loadIntoComposer: (text: string, attachments: ComposerAttachment[]) => void
   onCancel: ChatBarProps['onCancel']
   onSubmit: ChatBarProps['onSubmit']
@@ -53,7 +52,6 @@ export function useComposerQueue({
   clearDraft,
   draftRef,
   focusInput,
-  gatewayConnected,
   loadIntoComposer,
   onCancel,
   onSubmit,
@@ -260,7 +258,7 @@ export function useComposerQueue({
   // a stale-session 404) can't strand the entry permanently nor spin-loop. The
   // drain lock serializes sends; a remount/reconnect resets the failure counts.
   const autoDrainNext = useCallback(() => {
-    if (busy || !gatewayConnected || drainingQueueRef.current || !activeQueueSessionKey) {
+    if (busy || drainingQueueRef.current || !activeQueueSessionKey) {
       return
     }
 
@@ -291,7 +289,7 @@ export function useComposerQueue({
         }
       })
       .catch(onFail)
-  }, [activeQueueSessionKey, busy, gatewayConnected, pickDrainHead, queuedPrompts, runDrain, t])
+  }, [activeQueueSessionKey, busy, pickDrainHead, queuedPrompts, runDrain, t])
 
   // Re-key on a runtime session-id change. A stable stored id (queueSessionKey)
   // never churns, so a change there is a real session switch and must NOT
@@ -308,15 +306,14 @@ export function useComposerQueue({
     migrateQueuedPrompts(prev, activeQueueSessionKey)
   }, [activeQueueSessionKey, queueSessionKey])
 
-  // Queued turns flow whenever the session is idle AND the gateway is open — on
-  // the busy→false settle edge, on mount/reconnect, on the socket reopening, and
-  // after a re-key — so a swallowed edge can't strand them. To cancel queued
-  // turns, the user deletes them from the panel.
+  // Queued turns flow whenever the session is idle — on the busy→false settle
+  // edge, on mount/reconnect, and after a re-key — so a swallowed edge can't
+  // strand them. To cancel queued turns, the user deletes them from the panel.
   useEffect(() => {
-    if (shouldAutoDrain({ isBusy: busy, isConnected: gatewayConnected, queueLength: queuedPrompts.length })) {
+    if (shouldAutoDrain({ isBusy: busy, queueLength: queuedPrompts.length })) {
       autoDrainNext()
     }
-  }, [autoDrainNext, busy, gatewayConnected, queuedPrompts.length])
+  }, [autoDrainNext, busy, queuedPrompts.length])
 
   // Queue-edit cleanup: on session swap the scope effect already stashed the
   // edit snapshot; only restore into the composer when still on the same scope.

apps/desktop/src/app/chat/composer/hooks/use-composer-submit.ts

@@ -108,6 +108,10 @@ export function useComposerSubmit({
   )
 
   const submitDraft = () => {
+    if (disabled) {
+      return
+    }
+
     // Source the text from the DOM editor, not React state. The AUI composer
     // state (`draft`) and the derived `hasComposerPayload` lag the DOM by a
     // render, so on fast typing or IME composition the final keystroke(s) may
@@ -127,17 +131,6 @@ export function useComposerSubmit({
       }
     }
 
-    // Gateway isn't open (a post-boot reconnect keeps the composer editable by
-    // design). Don't silently drop the Enter — queue the draft so it shows as
-    // pending and the bounded auto-drain flushes it the instant the socket
-    // reopens, instead of the message vanishing with no feedback.
-    if (disabled) {
-      queueCurrentDraft()
-      focusInput()
-
-      return
-    }
-
     const text = draftRef.current
     const payloadPresent = text.trim().length > 0 || attachments.length > 0
 

apps/desktop/src/app/chat/composer/index.tsx

@@ -128,7 +128,6 @@ export function ChatBar({
 
   const { t } = useI18n()
   const gatewayState = useStore($gatewayState)
-  const gatewayConnected = gatewayState === 'open'
   const reconnecting = gatewayState === 'closed' || gatewayState === 'error'
   const inputDisabled = disabled && !reconnecting
 
@@ -179,7 +178,6 @@ export function ChatBar({
     clearDraft,
     draftRef,
     focusInput,
-    gatewayConnected,
     loadIntoComposer,
     onCancel,
     onSubmit,
@@ -569,14 +567,7 @@ export function ChatBar({
       const editorText = editorRef.current ? composerPlainText(editorRef.current) : draftRef.current
       const hasLivePayload = editorText.trim().length > 0 || attachments.length > 0
 
-      // Gateway down (a post-boot reconnect keeps the composer editable): don't
-      // silently swallow the Enter. Route a real draft into submitDraft, which
-      // queues it so the auto-drain flushes it the instant the socket reopens.
       if (disabled) {
-        if (hasLivePayload) {
-          submitDraft()
-        }
-
         return
       }
 

apps/desktop/src/app/gateway/hooks/use-gateway-boot.ts

@@ -313,6 +313,7 @@ export function useGatewayBoot({
 
       applyDesktopBootProgress(payload)
     })
+
     void desktop
       .getBootProgress()
       .then(snapshot => applyDesktopBootProgress(snapshot))
@@ -356,7 +357,8 @@ export function useGatewayBoot({
       }
     })
 
-    const offEvent = gateway.onEvent(event => callbacksRef.current.handleGatewayEvent(event))
+    const sourceProfile = normalizeProfileKey($activeGatewayProfile.get())
+    const offEvent = gateway.onEvent(event => callbacksRef.current.handleGatewayEvent({ ...event, profile: sourceProfile }))
 
     // Wake signals: power resume (macOS/Windows), network coming back, and the
     // window regaining focus/visibility. Each nudges an immediate reconnect.

apps/desktop/src/app/session/hooks/use-message-stream/approval-mode-event.test.tsx

@@ -0,0 +1,104 @@
+import { QueryClient } from '@tanstack/react-query'
+import { act, cleanup, render, waitFor } from '@testing-library/react'
+import { useEffect, useRef } from 'react'
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+
+import type { ClientSessionState } from '@/app/types'
+import { createClientSessionState } from '@/lib/chat-runtime'
+import { $approvalModes, approvalModeForProfile } from '@/store/approval-mode'
+import { $activeGatewayProfile } from '@/store/profile'
+import type { RpcEvent } from '@/types/hermes'
+
+import { useMessageStream } from './index'
+
+const ACTIVE_SID = 'session-active'
+let handleEvent: ((event: RpcEvent) => void) | null = null
+
+function Harness() {
+  const activeSessionIdRef = useRef<string | null>(ACTIVE_SID)
+  const sessionStateByRuntimeIdRef = useRef(new Map<string, ClientSessionState>())
+  const queryClientRef = useRef(new QueryClient())
+
+  const stream = useMessageStream({
+    activeSessionIdRef,
+    hydrateFromStoredSession: vi.fn(async () => undefined),
+    queryClient: queryClientRef.current,
+    refreshHermesConfig: vi.fn(async () => undefined),
+    refreshSessions: vi.fn(async () => undefined),
+    sessionStateByRuntimeIdRef,
+    updateSessionState: (sessionId, updater) => {
+      const current = sessionStateByRuntimeIdRef.current.get(sessionId) ?? createClientSessionState()
+      const next = updater(current)
+      sessionStateByRuntimeIdRef.current.set(sessionId, next)
+
+      return next
+    }
+  })
+
+  useEffect(() => {
+    handleEvent = stream.handleGatewayEvent
+  }, [stream.handleGatewayEvent])
+
+  return null
+}
+
+async function mountStream() {
+  render(<Harness />)
+  await waitFor(() => expect(handleEvent).not.toBeNull())
+}
+
+describe('live session.info approval mode reconciliation', () => {
+  beforeEach(() => {
+    handleEvent = null
+    $approvalModes.set({})
+    $activeGatewayProfile.set('work')
+  })
+
+  afterEach(() => {
+    cleanup()
+    vi.restoreAllMocks()
+  })
+
+  it('reconciles an active-session event under its source gateway profile', async () => {
+    await mountStream()
+
+    act(() =>
+      handleEvent!({
+        payload: { approval_mode: 'off' },
+        profile: 'work',
+        session_id: ACTIVE_SID,
+        type: 'session.info'
+      })
+    )
+
+    expect(approvalModeForProfile('work')).toBe('off')
+    expect(approvalModeForProfile('default')).toBe('smart')
+  })
+
+  it('ignores stale session.info from a non-active session on the active gateway', async () => {
+    await mountStream()
+
+    act(() =>
+      handleEvent!({
+        payload: { approval_mode: 'off' },
+        profile: 'work',
+        session_id: 'session-stale',
+        type: 'session.info'
+      })
+    )
+
+    expect(approvalModeForProfile('work')).toBe('smart')
+  })
+
+  it('does not cache an event under a different active profile when its source profile is absent', async () => {
+    await mountStream()
+    $activeGatewayProfile.set('personal')
+
+    act(() =>
+      handleEvent!({ payload: { approval_mode: 'off' }, session_id: ACTIVE_SID, type: 'session.info' })
+    )
+
+    expect(approvalModeForProfile('personal')).toBe('smart')
+    expect(approvalModeForProfile('work')).toBe('smart')
+  })
+})

apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts

@@ -12,6 +12,7 @@ import { playCompletionSound } from '@/lib/completion-sound'
 import { gatewayEventRequiresSessionId } from '@/lib/gateway-events'
 import { triggerHaptic } from '@/lib/haptics'
 import { isProviderSetupErrorMessage } from '@/lib/provider-setup-errors'
+import { reconcileApprovalModeForProfile } from '@/store/approval-mode'
 import { clearClarifyRequest, setClarifyRequest } from '@/store/clarify'
 import { setSessionCompacting } from '@/store/compaction'
 import { refreshBackgroundProcesses } from '@/store/composer-status'
@@ -20,6 +21,7 @@ import { dispatchNativeNotification } from '@/store/native-notifications'
 import { notify } from '@/store/notifications'
 import { requestDesktopOnboarding } from '@/store/onboarding'
 import { flashPetActivity, markPetUnread, setPetActivity } from '@/store/pet'
+import { $activeGatewayProfile, normalizeProfileKey } from '@/store/profile'
 import { followActiveSessionCwd } from '@/store/projects'
 import { clearAllPrompts, setApprovalRequest, setSecretRequest, setSudoRequest } from '@/store/prompts'
 import {
@@ -117,6 +119,20 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
         const providerChanged = typeof payload?.provider === 'string'
         const runningChanged = typeof payload?.running === 'boolean'
 
+        // Config is profile-scoped, but session.info also arrives for background
+        // sessions. Only an active-session event from the currently active
+        // gateway may reconcile the foreground cache. Requiring the renderer's
+        // source tag prevents an event queued before a profile swap from being
+        // attributed to the newly active profile.
+        if (
+          isActiveEvent &&
+          typeof payload?.approval_mode === 'string' &&
+          event.profile &&
+          normalizeProfileKey(event.profile) === normalizeProfileKey($activeGatewayProfile.get())
+        ) {
+          reconcileApprovalModeForProfile(event.profile, payload.approval_mode)
+        }
+
         if (apply) {
           if (modelChanged) {
             setCurrentModel(payload!.model || '')
@@ -473,9 +489,11 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) {
         setApprovalRequest({
           // false only when a tirith warning forbids it; backend omits the field otherwise.
           allowPermanent: payload?.allow_permanent !== false,
+          choices: Array.isArray(payload?.choices) ? payload.choices.filter(choice => typeof choice === 'string') : undefined,
           command,
           description,
-          sessionId: sessionId ?? null
+          sessionId: sessionId ?? null,
+          smartDenied: payload?.smart_denied === true
         })
 
         if (sessionId) {

apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx

@@ -1294,7 +1294,18 @@ describe('usePromptActions sleep/wake session recovery', () => {
   })
 
   it('still creates a new session for a genuine new-chat draft (no stored session selected)', async () => {
-    const createBackendSessionForSend = vi.fn(async () => RUNTIME_SESSION_ID)
+    const activeSessionIdRef: MutableRefObject<string | null> = { current: null }
+
+    // Mirror the real createBackendSessionForSend: a successful create
+    // re-homes the active runtime ref to the session it minted BEFORE
+    // returning. An inert stub here is what let the new-chat drift-abort
+    // regression ship green.
+    const createBackendSessionForSend = vi.fn(async () => {
+      activeSessionIdRef.current = RUNTIME_SESSION_ID
+
+      return RUNTIME_SESSION_ID
+    })
+
     const calls: string[] = []
 
     const requestGateway = vi.fn(async (method: string) => {
@@ -1307,6 +1318,7 @@ describe('usePromptActions sleep/wake session recovery', () => {
     render(
       <Harness
         activeSessionId={null}
+        activeSessionIdRef={activeSessionIdRef}
         createBackendSessionForSend={createBackendSessionForSend}
         onReady={h => (handle = h)}
         refreshSessions={async () => undefined}
@@ -1440,6 +1452,107 @@ describe('usePromptActions submit session-context isolation (#54527)', () => {
       session_id: STORED_SESSION_A
     })
   })
+
+  it('submits the first prompt of a new chat — the create pipeline re-homing selection/route is not user drift', async () => {
+    // Regression for the #54527 guard breaking every NEW chat: on a fresh draft
+    // (no stored session, no runtime session) createBackendSessionForSend
+    // legitimately sets selectedStoredSessionIdRef + navigates to the new
+    // session's route. Comparing against the pre-create (null) baseline made
+    // the guard read that self-inflicted move as a user switch and abort, so
+    // prompt.submit never fired: the message vanished, no DB row was ever
+    // persisted, and the desktop stranded on a route whose REST reads 404
+    // ("Session not found").
+    const calls: { method: string; params?: Record<string, unknown> }[] = []
+    const selectedStoredSessionIdRef: MutableRefObject<string | null> = { current: null }
+    const activeSessionIdRef: MutableRefObject<string | null> = { current: null }
+    let routeToken = '/'
+
+    const requestGateway = vi.fn(async (method: string, params?: Record<string, unknown>) => {
+      calls.push({ method, params })
+
+      return {} as never
+    })
+
+    // Mirror the real createBackendSessionForSend: on success it re-homes the
+    // refs AND the route to the session it just created.
+    const createBackendSessionForSend = vi.fn(async () => {
+      activeSessionIdRef.current = 'rt-new-chat'
+      selectedStoredSessionIdRef.current = 'stored-new-chat'
+      routeToken = '/stored-new-chat'
+
+      return 'rt-new-chat'
+    })
+
+    let handle: HarnessHandle | null = null
+    render(
+      <Harness
+        activeSessionId={null}
+        activeSessionIdRef={activeSessionIdRef}
+        createBackendSessionForSend={createBackendSessionForSend}
+        getRouteToken={() => routeToken}
+        onReady={h => (handle = h)}
+        refreshSessions={async () => undefined}
+        requestGateway={requestGateway}
+        selectedStoredSessionIdRef={selectedStoredSessionIdRef}
+        storedSessionId={null}
+      />
+    )
+    await waitFor(() => expect(handle).not.toBeNull())
+
+    expect(await handle!.submitText('first message of a brand-new chat')).toBe(true)
+    expect(createBackendSessionForSend).toHaveBeenCalledTimes(1)
+    expect(calls.find(c => c.method === 'prompt.submit')?.params).toMatchObject({
+      session_id: 'rt-new-chat'
+    })
+  })
+
+  it('aborts when the user switches sessions during the tail of a successful create', async () => {
+    // createBackendSessionForSend awaits once more (armed-YOLO apply) AFTER
+    // committing the refs and returning a real id, so a switch in that window
+    // escapes its internal null-return drift check. The active ref is the
+    // tell: every switch path retargets it synchronously, so it no longer
+    // equals the id create returned. The submit must abort, not adopt the
+    // switched-to context as its re-pinned baseline.
+    const calls: { method: string; params?: Record<string, unknown> }[] = []
+    const selectedStoredSessionIdRef: MutableRefObject<string | null> = { current: null }
+    const activeSessionIdRef: MutableRefObject<string | null> = { current: null }
+    let routeToken = '/'
+
+    const requestGateway = vi.fn(async (method: string, params?: Record<string, unknown>) => {
+      calls.push({ method, params })
+
+      return {} as never
+    })
+
+    const createBackendSessionForSend = vi.fn(async () => {
+      // The user switched to Session B during the post-commit await: the
+      // switch path re-homed all three context markers before create returned.
+      activeSessionIdRef.current = RUNTIME_SESSION_B
+      selectedStoredSessionIdRef.current = STORED_SESSION_B
+      routeToken = `/${STORED_SESSION_B}`
+
+      return 'rt-new-chat'
+    })
+
+    let handle: HarnessHandle | null = null
+    render(
+      <Harness
+        activeSessionId={null}
+        activeSessionIdRef={activeSessionIdRef}
+        createBackendSessionForSend={createBackendSessionForSend}
+        getRouteToken={() => routeToken}
+        onReady={h => (handle = h)}
+        refreshSessions={async () => undefined}
+        requestGateway={requestGateway}
+        selectedStoredSessionIdRef={selectedStoredSessionIdRef}
+        storedSessionId={null}
+      />
+    )
+    await waitFor(() => expect(handle).not.toBeNull())
+
+    expect(await handle!.submitText('message that must not land in session B')).toBe(false)
+    expect(calls.some(c => c.method === 'prompt.submit')).toBe(false)
+  })
 })
 
 describe('usePromptActions eager attachment upload (drop-time)', () => {

apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts

@@ -117,10 +117,12 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) {
 
       // Pin the session context for the whole async submit pipeline. Without
       // this, a fast session switch during session.resume / file.attach can
-      // redirect the user's text into a different chat (#54527).
+      // redirect the user's text into a different chat (#54527). Mutable —
+      // not const — because a new-chat submit legitimately re-homes to the
+      // session it creates (see the re-pin after createBackendSessionForSend).
       const startingActiveSessionId = activeSessionIdRef.current
-      const startingStoredSessionId = selectedStoredSessionIdRef.current
-      const startingRouteToken = getRouteToken()
+      let startingStoredSessionId = selectedStoredSessionIdRef.current
+      let startingRouteToken = getRouteToken()
 
       const sessionContextDrifted = (): boolean =>
         selectedStoredSessionIdRef.current !== startingStoredSessionId ||
@@ -281,11 +283,14 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) {
           return false
         }
 
-        if (sessionContextDrifted()) {
-          return abortForSessionSwitch(sessionId)
-        }
-
         if (!sessionId) {
+          // createBackendSessionForSend returns null when the user switched
+          // sessions mid-create (it closes the orphaned session itself) —
+          // abort silently. Anything else is a real failure worth a toast.
+          if (sessionContextDrifted()) {
+            return abortForSessionSwitch(null)
+          }
+
           dropOptimistic(null)
           releaseBusy()
           notify({ kind: 'error', title: copy.sessionUnavailable, message: copy.createSessionFailed })
@@ -293,6 +298,23 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) {
           return false
         }
 
+        // A successful create re-homes selection + route to the chat it just
+        // minted, so the pre-create baseline can't tell our own re-home from
+        // a user switch (judging it drift aborted EVERY first send of a new
+        // chat: no prompt.submit, no DB row, a stranded route that 404s
+        // "Session not found"). The drift signal for this window is the
+        // active ref instead: every switch path re-nulls or retargets it
+        // synchronously, so it only still equals the id create returned when
+        // nobody re-homed since.
+        if (activeSessionIdRef.current !== sessionId) {
+          return abortForSessionSwitch(sessionId)
+        }
+
+        // Re-pin the baseline to the created chat for the rest of the
+        // pipeline; the closures (seedOptimistic et al) see the new value.
+        startingStoredSessionId = selectedStoredSessionIdRef.current
+        startingRouteToken = getRouteToken()
+
         seedOptimistic(sessionId)
       }
 

apps/desktop/src/app/session/hooks/use-session-actions/utils.test.ts

@@ -1,9 +1,12 @@
-import { describe, expect, it } from 'vitest'
+import { beforeEach, describe, expect, it } from 'vitest'
 
 import type { ChatMessage } from '@/lib/chat-messages'
+import { $approvalModes, approvalModeForProfile } from '@/store/approval-mode'
+import { $activeGatewayProfile } from '@/store/profile'
 import type { SessionInfo } from '@/types/hermes'
 
 import {
+  applyRuntimeInfo,
   chatMessageArraysEquivalent,
   isSessionGoneError,
   reconcileResumeMessages,
@@ -17,6 +20,20 @@ const msg = (id: string, role: ChatMessage['role'], text: string, extra: Partial
 
 const session = (over: Partial<SessionInfo>): SessionInfo => over as SessionInfo
 
+describe('applyRuntimeInfo approval mode', () => {
+  beforeEach(() => {
+    $approvalModes.set({})
+    $activeGatewayProfile.set('work')
+  })
+
+  it('reconciles session.info against the gateway profile', () => {
+    applyRuntimeInfo({ approval_mode: 'smart', desktop_contract: 3 })
+
+    expect(approvalModeForProfile('work')).toBe('smart')
+    expect(approvalModeForProfile('default')).toBe('smart')
+  })
+})
+
 describe('isSessionGoneError', () => {
   it('is true for 404 / session-not-found, false otherwise', () => {
     expect(isSessionGoneError(new Error('Request failed 404'))).toBe(true)

apps/desktop/src/app/session/hooks/use-session-actions/utils.ts

@@ -2,6 +2,7 @@ import { getSession } from '@/hermes'
 import { type ChatMessage, chatMessageText } from '@/lib/chat-messages'
 import { normalizePersonalityValue } from '@/lib/chat-runtime'
 import { embeddedImageUrls, textWithoutEmbeddedImages } from '@/lib/embedded-images'
+import { reconcileApprovalModeForProfile } from '@/store/approval-mode'
 import { requestDesktopOnboarding } from '@/store/onboarding'
 import { $activeGatewayProfile, $profiles, normalizeProfileKey } from '@/store/profile'
 import {
@@ -266,6 +267,10 @@ export function applyRuntimeInfo(info: SessionRuntimeInfo | undefined): SessionR
 
   reportBackendContract(info.desktop_contract)
 
+  if (info.approval_mode !== undefined) {
+    reconcileApprovalModeForProfile($activeGatewayProfile.get(), info.approval_mode)
+  }
+
   if (info.credential_warning) {
     requestDesktopOnboarding(info.credential_warning)
   }

apps/desktop/src/app/settings/fallback-models-field.test.tsx

@@ -110,4 +110,19 @@ describe('FallbackModelsField', () => {
 
     await waitFor(() => expect(screen.getAllByLabelText('Remove')).toHaveLength(1))
   })
+
+  it('keeps a draft row visible after autosave re-renders the same persisted chain', async () => {
+    const onChange = vi.fn()
+    const rerender = await renderFieldWithRerender([], onChange)
+
+    fireEvent.click(screen.getByText('Add fallback'))
+
+    expect(onChange.mock.calls.at(-1)?.[0]).toEqual([])
+    expect(screen.getAllByLabelText('Remove')).toHaveLength(1)
+
+    // Parent autosave echo — same complete chain, new array identity.
+    rerender([])
+
+    await waitFor(() => expect(screen.getAllByLabelText('Remove')).toHaveLength(1))
+  })
 })

apps/desktop/src/app/settings/fallback-models-field.tsx

@@ -1,5 +1,5 @@
 import { useQuery } from '@tanstack/react-query'
-import { useEffect, useState } from 'react'
+import { useEffect, useRef, useState } from 'react'
 
 import { Button } from '@/components/ui/button'
 import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from '@/components/ui/select'
@@ -40,6 +40,14 @@ function normalizeEntries(value: unknown): FallbackEntry[] {
   })
 }
 
+function completeEntries(rows: FallbackEntry[]): FallbackEntry[] {
+  return rows.filter(entry => entry.provider && entry.model)
+}
+
+function entriesEqual(a: FallbackEntry[], b: FallbackEntry[]): boolean {
+  return a.length === b.length && a.every((entry, index) => entry.provider === b[index]?.provider && entry.model === b[index]?.model)
+}
+
 /**
  * Structured editor for the top-level `fallback_providers` config list — a
  * chain of `{provider, model}` pairs tried in order when the default model
@@ -69,16 +77,29 @@ export function FallbackModelsField({
   const providers = (modelOptions.data?.providers ?? []).filter(provider => provider.slug)
 
   const [rows, setRows] = useState<FallbackEntry[]>(() => normalizeEntries(value))
+  // Last complete chain we emitted (or seeded). Autosave echoes the same
+  // filtered list back through `value`; ignore that echo so draft rows stay.
+  const lastEmittedRef = useRef(normalizeEntries(value))
 
-  // Settings can reload after a profile/config change while this component
-  // stays mounted. Avoid displaying or saving the previous profile's chain.
+  // Resync on real external changes (profile switch / config reload). Skip
+  // when `value` is just our own commit echoing through the parent.
   useEffect(() => {
-    setRows(normalizeEntries(value))
+    const persisted = normalizeEntries(value)
+
+    if (entriesEqual(persisted, lastEmittedRef.current)) {
+      return
+    }
+
+    lastEmittedRef.current = persisted
+    setRows(persisted)
   }, [value])
 
   const commit = (next: FallbackEntry[]) => {
+    const complete = completeEntries(next)
+
     setRows(next)
-    onChange(next.filter(entry => entry.provider && entry.model))
+    lastEmittedRef.current = complete
+    onChange(complete)
   }
 
   const updateRow = (index: number, patch: Partial<FallbackEntry>) =>

apps/desktop/src/app/shell/approval-mode-menu.test.tsx

@@ -0,0 +1,89 @@
+import { cleanup, fireEvent, render, screen, waitFor, within } from '@testing-library/react'
+import { MemoryRouter } from 'react-router-dom'
+import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest'
+
+import { StatusbarControls } from '@/app/shell/statusbar-controls'
+import { I18nProvider } from '@/i18n'
+import { $approvalModes } from '@/store/approval-mode'
+
+import { useApprovalModeStatusbarItem } from './approval-mode-menu'
+
+class TestResizeObserver {
+  observe() {}
+  unobserve() {}
+  disconnect() {}
+}
+
+beforeAll(() => {
+  vi.stubGlobal('ResizeObserver', TestResizeObserver)
+  Element.prototype.hasPointerCapture ??= () => false
+  Element.prototype.setPointerCapture ??= () => undefined
+  Element.prototype.releasePointerCapture ??= () => undefined
+  HTMLElement.prototype.scrollIntoView ??= () => undefined
+})
+
+afterEach(() => {
+  cleanup()
+  $approvalModes.set({})
+})
+
+function Harness({
+  profile = 'default',
+  requestGateway
+}: {
+  profile?: string
+  requestGateway: (method: string, params?: Record<string, unknown>) => Promise<unknown>
+}) {
+  const item = useApprovalModeStatusbarItem(profile, requestGateway)
+
+  return (
+    <MemoryRouter>
+      <StatusbarControls items={[item]} />
+    </MemoryRouter>
+  )
+}
+
+describe('approval mode statusbar item', () => {
+  it('uses the shared statusbar menu trigger without a nested bespoke button', async () => {
+    const response = new Promise<never>(() => undefined)
+    render(<Harness requestGateway={vi.fn(() => response)} />)
+
+    const statusbar = screen.getByRole('contentinfo')
+    const trigger = within(statusbar).getByRole('button', { name: /smart/i })
+    expect(within(statusbar).getAllByRole('button')).toHaveLength(1)
+
+    fireEvent.pointerDown(trigger, { button: 0 })
+
+    expect(await screen.findByRole('menuitemradio', { name: /manual/i })).toBeTruthy()
+    expect(trigger.getAttribute('aria-haspopup')).toBe('menu')
+    expect(screen.getByRole('menuitemradio', { name: /smart/i })).toBeTruthy()
+    expect(screen.getByRole('menuitemradio', { name: /off/i })).toBeTruthy()
+  })
+
+  it('writes the selected mode through the gateway and updates its shared trigger label', async () => {
+    const requestGateway = vi.fn(async (_method, params) => ({ value: params?.value ?? 'smart' }))
+    render(<Harness profile="work" requestGateway={requestGateway} />)
+
+    fireEvent.pointerDown(screen.getByRole('button', { name: /smart/i }), { button: 0 })
+    fireEvent.click(await screen.findByRole('menuitemradio', { name: /manual/i }))
+
+    await waitFor(() => {
+      expect(requestGateway).toHaveBeenCalledWith('config.set', { key: 'approvals.mode', value: 'manual' })
+      expect(screen.getByRole('button', { name: /manual/i })).toBeTruthy()
+    })
+  })
+
+  it('renders the shared trigger and menu in the active locale', async () => {
+    const response = new Promise<never>(() => undefined)
+    render(
+      <I18nProvider configClient={null} initialLocale="ja">
+        <Harness requestGateway={vi.fn(() => response)} />
+      </I18nProvider>
+    )
+
+    fireEvent.pointerDown(screen.getByRole('button', { name: 'スマート' }), { button: 0 })
+
+    expect(await screen.findByText('必要な場合にのみ確認します')).toBeTruthy()
+    expect(screen.getByText('承認プロンプトなしで実行します')).toBeTruthy()
+  })
+})

apps/desktop/src/app/shell/approval-mode-menu.tsx

@@ -0,0 +1,81 @@
+import { useStore } from '@nanostores/react'
+import { useEffect, useMemo } from 'react'
+
+import type { StatusbarItem } from '@/app/shell/statusbar-controls'
+import {
+  DropdownMenuLabel,
+  DropdownMenuRadioGroup,
+  DropdownMenuRadioItem,
+  DropdownMenuSeparator
+} from '@/components/ui/dropdown-menu'
+import { useI18n } from '@/i18n'
+import { Zap, ZapFilled } from '@/lib/icons'
+import {
+  $approvalModes,
+  type ApprovalMode,
+  type ApprovalModeRequester,
+  setApprovalModeForProfile,
+  syncApprovalModeForProfile
+} from '@/store/approval-mode'
+
+export function useApprovalModeStatusbarItem(
+  profile: string,
+  requestGateway: ApprovalModeRequester
+): StatusbarItem {
+  const { t } = useI18n()
+  const copy = t.shell.approvalMode
+  const modes = useStore($approvalModes)
+  const mode = modes[profile.trim() || 'default'] ?? 'smart'
+
+  const labels = useMemo<Record<ApprovalMode, string>>(
+    () => ({ manual: copy.manual, smart: copy.smart, off: copy.off }),
+    [copy.manual, copy.off, copy.smart]
+  )
+
+  const descriptions = useMemo<Record<ApprovalMode, string>>(
+    () => ({
+      manual: copy.manualDescription,
+      smart: copy.smartDescription,
+      off: copy.offDescription
+    }),
+    [copy.manualDescription, copy.offDescription, copy.smartDescription]
+  )
+
+  useEffect(() => {
+    void syncApprovalModeForProfile(requestGateway, profile).catch(() => undefined)
+  }, [profile, requestGateway])
+
+  return {
+    className: mode === 'off' ? 'bg-(--chrome-action-hover) text-foreground' : undefined,
+    icon: mode === 'off' ? <ZapFilled className="size-3.5" /> : <Zap className="size-3.5 opacity-70" />,
+    id: 'approval-mode',
+    label: labels[mode],
+    menuAlign: 'end',
+    menuClassName: 'w-72 p-1',
+    menuContent: (
+      <>
+        <DropdownMenuLabel>{copy.title}</DropdownMenuLabel>
+        <DropdownMenuSeparator />
+        <DropdownMenuRadioGroup
+          onValueChange={value => {
+            void setApprovalModeForProfile(requestGateway, profile, value as ApprovalMode).catch(() => undefined)
+          }}
+          value={mode}
+        >
+          {(['manual', 'smart', 'off'] as const).map(value => (
+            <DropdownMenuRadioItem className="items-start gap-2" key={value} value={value}>
+              <span className="flex min-w-0 flex-col gap-0.5">
+                <span className="text-xs text-foreground">{labels[value]}</span>
+                <span className="text-[0.6875rem] leading-snug text-(--ui-text-tertiary)">
+                  {descriptions[value]}
+                </span>
+              </span>
+            </DropdownMenuRadioItem>
+          ))}
+        </DropdownMenuRadioGroup>
+      </>
+    ),
+    title: copy.ariaLabel(labels[mode]),
+    variant: 'menu'
+  }
+}

apps/desktop/src/app/shell/hooks/use-statusbar-items.tsx

@@ -1,20 +1,21 @@
 import { useStore } from '@nanostores/react'
-import { useCallback, useMemo } from 'react'
+import { useMemo } from 'react'
 
 import type { CommandCenterSection } from '@/app/command-center'
 import { $terminalTakeover, setTerminalTakeover } from '@/app/right-sidebar/store'
+import { useApprovalModeStatusbarItem } from '@/app/shell/approval-mode-menu'
 import { ContextUsagePanel } from '@/app/shell/context-usage-panel'
 import { GatewayMenuPanel } from '@/app/shell/gateway-menu-panel'
 import { Codicon } from '@/components/ui/codicon'
 import { GlyphSpinner } from '@/components/ui/glyph-spinner'
 import { useI18n } from '@/i18n'
-import { Activity, AlertCircle, Clock, Command, FolderOpen, Hash, Loader2, Terminal, Zap, ZapFilled } from '@/lib/icons'
+import { Activity, AlertCircle, Clock, Command, FolderOpen, Hash, Loader2, Terminal } from '@/lib/icons'
 import type { RuntimeReadinessResult } from '@/lib/runtime-readiness'
 import { contextBarLabel, LiveDuration, usageContextLabel } from '@/lib/statusbar'
 import { cn } from '@/lib/utils'
-import { setGlobalYolo, setSessionYolo } from '@/lib/yolo-session'
 import { copyFilePath, revealFile } from '@/store/file-actions'
 import { revealFileInTree } from '@/store/layout'
+import { $activeGatewayProfile } from '@/store/profile'
 import {
   $activeSessionId,
   $busy,
@@ -23,8 +24,6 @@ import {
   $currentUsage,
   $sessionStartedAt,
   $turnStartedAt,
-  $yoloActive,
-  setYoloActive
 } from '@/store/session'
 import { $subagentsBySession, activeSubagentCount, failedSubagentCount } from '@/store/subagents'
 import { $gatewayRestarting } from '@/store/system-actions'
@@ -39,7 +38,7 @@ import {
 import type { StatusResponse } from '@/types/hermes'
 
 import { CRON_ROUTE } from '../../routes'
-import type { StatusbarItem, StatusbarSelectModifiers } from '../statusbar-controls'
+import type { StatusbarItem } from '../statusbar-controls'
 
 function workspaceLabel(cwd: string): string {
   const normalized = cwd.replace(/[\\/]+$/, '')
@@ -74,7 +73,6 @@ export function useStatusbarItems({
   inferenceStatus,
   openAgents,
   openCommandCenterSection,
-  freshDraftReady,
   requestGateway,
   statusSnapshot,
   toggleCommandCenter
@@ -83,8 +81,8 @@ export function useStatusbarItems({
   const copy = t.shell.statusbar
   const fileMenu = t.fileMenu
   const activeSessionId = useStore($activeSessionId)
+  const activeGatewayProfile = useStore($activeGatewayProfile)
   const terminalTakeover = useStore($terminalTakeover)
-  const yoloActive = useStore($yoloActive)
   const busy = useStore($busy)
   const currentCwd = useStore($currentCwd)
   const currentUsage = useStore($currentUsage)
@@ -101,45 +99,8 @@ export function useStatusbarItems({
 
   const contextUsage = useMemo(() => usageContextLabel(currentUsage), [currentUsage])
   const contextBar = useMemo(() => contextBarLabel(currentUsage), [currentUsage])
+  const approvalModeItem = useApprovalModeStatusbarItem(activeGatewayProfile, requestGateway)
 
-  // Per-session approval bypass (same scope as the TUI's Shift+Tab). On a
-  // new-chat draft (no runtime session yet) we arm locally; the session-create
-  // path applies it once the backend session exists.
-  //
-  // Shift+click flips the GLOBAL approvals.mode instead — a persistent,
-  // all-sessions/CLI/TUI/cron bypass that survives restarts.
-  const toggleYolo = useCallback(
-    async (modifiers?: StatusbarSelectModifiers) => {
-      const next = !$yoloActive.get()
-
-      setYoloActive(next)
-
-      if (modifiers?.shiftKey) {
-        try {
-          await setGlobalYolo(requestGateway, next)
-        } catch {
-          setYoloActive(!next)
-        }
-
-        return
-      }
-
-      const sid = $activeSessionId.get()
-
-      if (!sid) {
-        return
-      }
-
-      try {
-        await setSessionYolo(requestGateway, sid, next)
-      } catch {
-        setYoloActive(!next)
-      }
-    },
-    [requestGateway]
-  )
-
-  const showYoloToggle = gatewayState === 'open' && (!!activeSessionId || freshDraftReady)
 
   const gatewayMenuContent = useMemo(
     () => (close: () => void) => (
@@ -429,17 +390,8 @@ export function useStatusbarItems({
         variant: 'text'
       },
       {
-        className: cn('px-1', yoloActive && 'bg-(--chrome-action-hover)'),
-        hidden: !showYoloToggle,
-        icon: yoloActive ? (
-          <ZapFilled className="size-3.5 shrink-0" />
-        ) : (
-          <Zap className="size-3.5 shrink-0 opacity-70" />
-        ),
-        id: 'yolo',
-        onSelect: modifiers => void toggleYolo(modifiers),
-        title: yoloActive ? copy.yoloOn : copy.yoloOff,
-        variant: 'action'
+        ...approvalModeItem,
+        hidden: gatewayState !== 'open',
       },
       {
         className: `w-7 justify-center px-0${terminalTakeover ? ' bg-accent/55 text-foreground' : ''}`,
@@ -455,6 +407,7 @@ export function useStatusbarItems({
     ],
     [
       activeSessionId,
+      approvalModeItem,
       backendVersionItem,
       busy,
       chatOpen,
@@ -465,11 +418,9 @@ export function useStatusbarItems({
       currentUsage,
       requestGateway,
       sessionStartedAt,
-      showYoloToggle,
+      gatewayState,
       terminalTakeover,
-      toggleYolo,
       turnStartedAt,
-      yoloActive
     ]
   )
 

apps/desktop/src/components/assistant-ui/tool/approval.test.tsx

@@ -30,9 +30,13 @@ function part(toolName: string): ToolPart {
   return { toolName, type: `tool-${toolName}` } as unknown as ToolPart
 }
 
-function setRequest(command = 'rm -rf /tmp/x', allowPermanent?: boolean) {
+function setRequest(
+  command = 'rm -rf /tmp/x',
+  allowPermanent?: boolean,
+  extra: { choices?: string[]; smartDenied?: boolean } = {}
+) {
   $activeSessionId.set('sess-1')
-  setApprovalRequest({ allowPermanent, command, description: 'dangerous command', sessionId: 'sess-1' })
+  setApprovalRequest({ allowPermanent, command, description: 'dangerous command', sessionId: 'sess-1', ...extra })
 }
 
 function mockGateway() {
@@ -131,6 +135,26 @@ describe('PendingToolApproval', () => {
     expect(screen.queryByRole('menuitem', { name: /Always allow/ })).toBeNull()
   })
 
+  it('renders only Once and Deny for a Smart DENY owner override', () => {
+    setRequest('rm -rf /tmp/x', true, { smartDenied: true })
+    render(<PendingToolApproval part={part('terminal')} />)
+
+    expect(screen.getByRole('button', { name: /Run/ })).toBeTruthy()
+    expect(screen.getByRole('button', { name: /Reject/ })).toBeTruthy()
+    expect(screen.queryByRole('button', { name: /More approval options/ })).toBeNull()
+    expect(screen.queryByText(/Allow this session/)).toBeNull()
+    expect(screen.queryByText(/Always allow/)).toBeNull()
+  })
+
+  it('renders only choices explicitly supplied by the gateway event', () => {
+    setRequest('rm -rf /tmp/x', true, { choices: ['once', 'deny'] })
+    render(<PendingToolApproval part={part('terminal')} />)
+
+    expect(screen.getByRole('button', { name: /Run/ })).toBeTruthy()
+    expect(screen.getByRole('button', { name: /Reject/ })).toBeTruthy()
+    expect(screen.queryByRole('button', { name: /More approval options/ })).toBeNull()
+  })
+
   it('renders a floating fallback when no pending tool row is mounted', () => {
     setRequest('rm /tmp/hermes_approval_test.txt')
     const { container } = render(<PendingApprovalFallback />)

apps/desktop/src/components/assistant-ui/tool/approval.tsx

@@ -110,6 +110,10 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
   const busy = submitting !== null
   // false when the backend won't honor a permanent allow (tirith warning) → hide "Always allow".
   const allowPermanent = request.allowPermanent !== false
+  const choices = request.choices ?? (request.smartDenied ? ['once', 'deny'] : undefined)
+  const allowSession = choices ? choices.includes('session') : true
+  const allowAlways = choices ? choices.includes('always') : allowPermanent
+  const hasMoreOptions = allowSession || allowAlways
   const hasCommand = request.command.trim().length > 0
 
   const respond = useCallback(
@@ -183,8 +187,8 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
             {submitting === 'once' ? <Loader2 className="size-3 animate-spin" /> : copy.run}
             {submitting !== 'once' && <span className="text-[0.625rem] text-primary/60">{isMac ? '⌘⏎' : 'Ctrl⏎'}</span>}
           </Button>
-          <span aria-hidden className="w-px self-stretch bg-primary/20" />
-          <DropdownMenu>
+          {hasMoreOptions && <span aria-hidden className="w-px self-stretch bg-primary/20" />}
+          {hasMoreOptions && <DropdownMenu>
             <DropdownMenuTrigger asChild>
               <Button
                 aria-label={copy.moreOptions}
@@ -197,8 +201,8 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
               </Button>
             </DropdownMenuTrigger>
             <DropdownMenuContent align="start" className="min-w-44">
-              <DropdownMenuItem onSelect={() => void respond('session')}>{copy.allowSession}</DropdownMenuItem>
-              {allowPermanent && (
+              {allowSession && <DropdownMenuItem onSelect={() => void respond('session')}>{copy.allowSession}</DropdownMenuItem>}
+              {allowAlways && (
                 <DropdownMenuItem
                   onSelect={() => {
                     // Defer one tick so the menu fully unmounts before the dialog
@@ -214,7 +218,7 @@ const ApprovalBar: FC<{ request: ApprovalRequest; surface: 'floating' | 'inline'
                 {copy.reject}
               </DropdownMenuItem>
             </DropdownMenuContent>
-          </DropdownMenu>
+          </DropdownMenu>}
         </div>
 
         <Button

apps/desktop/src/i18n/en.ts

@@ -2068,6 +2068,16 @@ export const en: Translations = {
       viewAllLogs: 'View all logs →',
       messagingPlatforms: 'Messaging platforms'
     },
+    approvalMode: {
+      title: 'Approval mode',
+      ariaLabel: mode => `Approval mode: ${mode}`,
+      manual: 'Manual',
+      manualDescription: 'Ask before actions that require approval',
+      smart: 'Smart',
+      smartDescription: 'Automatically assess actions and ask when needed',
+      off: 'Off',
+      offDescription: 'Run without approval prompts'
+    },
     statusbar: {
       unknown: 'unknown',
       restart: 'restart',

apps/desktop/src/i18n/ja.ts

@@ -2009,6 +2009,16 @@ export const ja = defineLocale({
       viewAllLogs: 'すべてのログを見る →',
       messagingPlatforms: 'メッセージングプラットフォーム'
     },
+    approvalMode: {
+      title: '承認モード',
+      ariaLabel: mode => `承認モード: ${mode}`,
+      manual: '手動',
+      manualDescription: '承認が必要な操作の前に確認します',
+      smart: 'スマート',
+      smartDescription: '必要な場合にのみ確認します',
+      off: 'オフ',
+      offDescription: '承認プロンプトなしで実行します'
+    },
     statusbar: {
       unknown: '不明',
       restart: '再起動',

apps/desktop/src/i18n/types.ts

@@ -1699,6 +1699,16 @@ export interface Translations {
       viewAllLogs: string
       messagingPlatforms: string
     }
+    approvalMode: {
+      title: string
+      ariaLabel: (mode: string) => string
+      manual: string
+      manualDescription: string
+      smart: string
+      smartDescription: string
+      off: string
+      offDescription: string
+    }
     statusbar: {
       unknown: string
       restart: string

apps/desktop/src/i18n/zh-hant.ts

@@ -1945,6 +1945,16 @@ export const zhHant = defineLocale({
       viewAllLogs: '查看全部記錄 →',
       messagingPlatforms: '訊息平台'
     },
+    approvalMode: {
+      title: '核准模式',
+      ariaLabel: mode => `核准模式：${mode}`,
+      manual: '手動',
+      manualDescription: '執行需要核准的操作前詢問',
+      smart: '智慧',
+      smartDescription: '自動評估操作，並在需要時詢問',
+      off: '關閉',
+      offDescription: '不顯示核准提示，直接執行'
+    },
     statusbar: {
       unknown: '未知',
       restart: '重新啟動',

apps/desktop/src/i18n/zh.ts

@@ -2232,6 +2232,16 @@ export const zh: Translations = {
       viewAllLogs: '查看全部日志 →',
       messagingPlatforms: '消息平台'
     },
+    approvalMode: {
+      title: '审批模式',
+      ariaLabel: mode => `审批模式：${mode}`,
+      manual: '手动',
+      manualDescription: '执行需要审批的操作前询问',
+      smart: '智能',
+      smartDescription: '自动评估操作，并在需要时询问',
+      off: '关闭',
+      offDescription: '不显示审批提示，直接运行'
+    },
     statusbar: {
       unknown: '未知',
       restart: '重启',

apps/desktop/src/lib/chat-messages.ts

@@ -46,6 +46,7 @@ export type GatewayEventPayload = {
   reasoning_effort?: string
   service_tier?: string
   fast?: boolean
+  approval_mode?: string
   yolo?: boolean
   running?: boolean
   cwd?: string
@@ -66,6 +67,7 @@ export type GatewayEventPayload = {
   description?: string
   // False when a tirith content-security warning forbids a permanent allow.
   allow_permanent?: boolean
+  smart_denied?: boolean
   // secret.request (skill credential capture)
   env_var?: string
   prompt?: string

apps/desktop/src/store/approval-mode.test.ts

@@ -0,0 +1,95 @@
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+
+import {
+  $approvalModes,
+  approvalModeForProfile,
+  reconcileApprovalModeForProfile,
+  setApprovalModeForProfile,
+  syncApprovalModeForProfile
+} from './approval-mode'
+
+function deferred<T>() {
+  let resolve!: (value: T) => void
+  let reject!: (error: unknown) => void
+
+  const promise = new Promise<T>((res, rej) => {
+    resolve = res
+    reject = rej
+  })
+
+  return { promise, reject, resolve }
+}
+
+describe('profile-scoped approval mode cache', () => {
+  beforeEach(() => $approvalModes.set({}))
+
+  it('labels an unread profile Smart by default and adopts backend truth', async () => {
+    expect(approvalModeForProfile('default')).toBe('smart')
+
+    const request = vi.fn(async () => ({ value: 'manual' }))
+    await syncApprovalModeForProfile(request, 'default')
+
+    expect(request).toHaveBeenCalledWith('config.get', { key: 'approvals.mode' })
+    expect(approvalModeForProfile('default')).toBe('manual')
+  })
+
+  it('keeps profile values isolated', async () => {
+    await syncApprovalModeForProfile(vi.fn(async () => ({ value: 'manual' })), 'work')
+    await syncApprovalModeForProfile(vi.fn(async () => ({ value: 'off' })), 'personal')
+
+    expect(approvalModeForProfile('work')).toBe('manual')
+    expect(approvalModeForProfile('personal')).toBe('off')
+    expect(approvalModeForProfile('default')).toBe('smart')
+  })
+
+  it('rolls consecutive failed writes back to the last authoritative value', async () => {
+    await syncApprovalModeForProfile(vi.fn(async () => ({ value: 'smart' })), 'default')
+    const first = deferred<{ value: string }>()
+    const second = deferred<{ value: string }>()
+
+    const request = vi
+      .fn()
+      .mockImplementationOnce(() => first.promise)
+      .mockImplementationOnce(() => second.promise)
+
+    const staleWrite = setApprovalModeForProfile(request, 'default', 'manual')
+    const currentWrite = setApprovalModeForProfile(request, 'default', 'off')
+    expect(approvalModeForProfile('default')).toBe('off')
+
+    first.reject(new Error('old failure'))
+    await expect(staleWrite).rejects.toThrow('old failure')
+    expect(approvalModeForProfile('default')).toBe('off')
+
+    second.reject(new Error('current failure'))
+    await expect(currentWrite).rejects.toThrow('current failure')
+    expect(approvalModeForProfile('default')).toBe('smart')
+  })
+
+  it('lets a backend event supersede an optimistic write and its later failure', async () => {
+    const write = deferred<{ value: string }>()
+    const pending = setApprovalModeForProfile(vi.fn(() => write.promise), 'work', 'off')
+
+    reconcileApprovalModeForProfile('work', 'smart')
+    expect(approvalModeForProfile('work')).toBe('smart')
+
+    write.reject(new Error('late failure'))
+    await expect(pending).rejects.toThrow('late failure')
+    expect(approvalModeForProfile('work')).toBe('smart')
+  })
+
+  it('ignores a stale initial read after a newer write succeeds', async () => {
+    const read = deferred<{ value: string }>()
+
+    const request = vi
+      .fn()
+      .mockImplementationOnce(() => read.promise)
+      .mockResolvedValueOnce({ value: 'off' })
+
+    const staleRead = syncApprovalModeForProfile(request, 'default')
+    await setApprovalModeForProfile(request, 'default', 'off')
+    read.resolve({ value: 'manual' })
+    await staleRead
+
+    expect(approvalModeForProfile('default')).toBe('off')
+  })
+})

apps/desktop/src/store/approval-mode.ts

@@ -0,0 +1,100 @@
+import { atom } from 'nanostores'
+
+export type ApprovalMode = 'manual' | 'off' | 'smart'
+export type ApprovalModeRequester = (
+  method: string,
+  params?: Record<string, unknown>
+) => Promise<unknown>
+
+const APPROVAL_MODES = new Set<ApprovalMode>(['manual', 'smart', 'off'])
+const revisions = new Map<string, number>()
+const confirmedModes = new Map<string, ApprovalMode>()
+
+export const $approvalModes = atom<Record<string, ApprovalMode>>({})
+
+function profileKey(profile: string): string {
+  return profile.trim() || 'default'
+}
+
+function nextRevision(profile: string): number {
+  const revision = (revisions.get(profile) ?? 0) + 1
+  revisions.set(profile, revision)
+
+  return revision
+}
+
+function normalizeApprovalMode(value: unknown): ApprovalMode {
+  const normalized = String(value ?? '')
+    .trim()
+    .toLowerCase() as ApprovalMode
+
+  return APPROVAL_MODES.has(normalized) ? normalized : 'manual'
+}
+
+export function approvalModeForProfile(profile: string): ApprovalMode {
+  return $approvalModes.get()[profileKey(profile)] ?? 'smart'
+}
+
+function cacheApprovalMode(profile: string, mode: ApprovalMode): void {
+  const key = profileKey(profile)
+  $approvalModes.set({ ...$approvalModes.get(), [key]: mode })
+}
+
+export function reconcileApprovalModeForProfile(profile: string, value: unknown): ApprovalMode {
+  const key = profileKey(profile)
+  const mode = normalizeApprovalMode(value)
+  nextRevision(key)
+  confirmedModes.set(key, mode)
+  cacheApprovalMode(key, mode)
+
+  return mode
+}
+
+export async function syncApprovalModeForProfile(
+  requestGateway: ApprovalModeRequester,
+  profile: string
+): Promise<ApprovalMode> {
+  const key = profileKey(profile)
+  const revision = nextRevision(key)
+  const result = (await requestGateway('config.get', { key: 'approvals.mode' })) as { value?: string }
+  const mode = normalizeApprovalMode(result?.value)
+
+  if (revisions.get(key) === revision) {
+    confirmedModes.set(key, mode)
+    cacheApprovalMode(key, mode)
+  }
+
+  return mode
+}
+
+export async function setApprovalModeForProfile(
+  requestGateway: ApprovalModeRequester,
+  profile: string,
+  mode: ApprovalMode
+): Promise<ApprovalMode> {
+  const key = profileKey(profile)
+  const revision = nextRevision(key)
+  cacheApprovalMode(key, mode)
+
+  try {
+    const result = (await requestGateway('config.set', {
+      key: 'approvals.mode',
+      value: mode
+    })) as { value?: string }
+
+    const authoritative = normalizeApprovalMode(result?.value)
+
+    if (revisions.get(key) === revision) {
+      confirmedModes.set(key, authoritative)
+      cacheApprovalMode(key, authoritative)
+    }
+
+    return authoritative
+  } catch (error) {
+    if (revisions.get(key) === revision) {
+      cacheApprovalMode(key, confirmedModes.get(key) ?? 'smart')
+    }
+
+    throw error
+  }
+}

apps/desktop/src/store/composer-queue.test.ts

@@ -150,28 +150,21 @@ describe('migrateQueuedPrompts', () => {
 })
 
 describe('shouldAutoDrain', () => {
-  it('drains whenever idle and connected with a non-empty queue', () => {
-    expect(shouldAutoDrain({ isBusy: false, isConnected: true, queueLength: 1 })).toBe(true)
+  it('drains whenever idle with a non-empty queue', () => {
+    expect(shouldAutoDrain({ isBusy: false, queueLength: 1 })).toBe(true)
   })
 
   it('drains on mount/reconnect with no observed busy edge', () => {
     // The whole point of dropping the edge: a remount resets the busy ref, so an
     // edge-gated drain would strand the entry. Idle + non-empty must still fire.
-    expect(shouldAutoDrain({ isBusy: false, isConnected: true, queueLength: 2 })).toBe(true)
+    expect(shouldAutoDrain({ isBusy: false, queueLength: 2 })).toBe(true)
   })
 
   it('does not drain mid-turn', () => {
-    expect(shouldAutoDrain({ isBusy: true, isConnected: true, queueLength: 1 })).toBe(false)
+    expect(shouldAutoDrain({ isBusy: true, queueLength: 1 })).toBe(false)
   })
 
   it('does not drain an empty queue', () => {
-    expect(shouldAutoDrain({ isBusy: false, isConnected: true, queueLength: 0 })).toBe(false)
-  })
-
-  it('does not drain while the gateway is closed, then flushes when it reopens', () => {
-    // A draft queued during a post-boot reconnect must wait — not spin failed
-    // sends — and drain the moment the socket is open again.
-    expect(shouldAutoDrain({ isBusy: false, isConnected: false, queueLength: 1 })).toBe(false)
-    expect(shouldAutoDrain({ isBusy: false, isConnected: true, queueLength: 1 })).toBe(true)
+    expect(shouldAutoDrain({ isBusy: false, queueLength: 0 })).toBe(false)
   })
 })

apps/desktop/src/store/composer-queue.ts

@@ -243,7 +243,6 @@ export const migrateQueuedPrompts = (fromKey: string | null | undefined, toKey:
 /** Inputs to {@link shouldAutoDrain}. */
 export interface AutoDrainInput {
   isBusy: boolean
-  isConnected: boolean
   queueLength: number
 }
 
@@ -256,13 +255,8 @@ export interface AutoDrainInput {
  * busy ref to the current value, swallowing the settle edge — an edge-gated
  * drain would then strand the entry forever. The caller's drain lock
  * (`drainingQueueRef`) serializes sends so being edge-free can't double-submit.
- *
- * Gated on `isConnected` so a draft queued while the gateway is closed (a
- * post-boot reconnect keeps the composer editable) waits instead of spinning
- * failed sends, then flushes the moment the socket reopens.
  */
-export const shouldAutoDrain = ({ isBusy, isConnected, queueLength }: AutoDrainInput): boolean =>
-  !isBusy && isConnected && queueLength > 0
+export const shouldAutoDrain = ({ isBusy, queueLength }: AutoDrainInput): boolean => !isBusy && queueLength > 0
 
 /** Auto-drain attempts for one entry before we stop retrying and toast. The
  * entry stays queued for a manual send; a remount/reconnect resets the count. */

apps/desktop/src/store/gateway.ts

@@ -164,7 +164,7 @@ function createSecondary(profile: string): Secondary {
     wantOpen: true
   }
 
-  entry.offEvent = gateway.onEvent(event => config?.onEvent(event))
+  entry.offEvent = gateway.onEvent(event => config?.onEvent({ ...event, profile }))
   entry.offState = gateway.onState(state => {
     reportGatewayState(profile, state)
 

apps/desktop/src/store/prompts.ts

@@ -71,8 +71,10 @@ function keyedPromptStore<T extends KeyedPrompt>(): PromptStore<T> {
 export interface ApprovalRequest extends KeyedPrompt {
   // false when the backend won't honor a permanent allow (tirith warning) → hide "Always allow".
   allowPermanent?: boolean
+  choices?: string[]
   command: string
   description: string
+  smartDenied?: boolean
 }
 
 export interface SudoRequest extends KeyedPrompt {

apps/desktop/src/store/updates.test.ts

@@ -120,7 +120,7 @@ describe('reportBackendContract', () => {
   })
 
   it('dismisses the toast when the backend meets the contract', () => {
-    reportBackendContract(2)
+    reportBackendContract(3)
     expect(dismissSpy).toHaveBeenCalledWith('backend-contract-skew')
     expect(notifySpy).not.toHaveBeenCalled()
   })
@@ -160,8 +160,8 @@ describe('reportBackendContract', () => {
     lastToast().onDismiss()
     notifySpy.mockClear()
 
-    reportBackendContract(2) // backend updated → satisfied, snooze cleared
-    reportBackendContract(1) // a later regression must warn immediately
+    reportBackendContract(3) // backend updated → satisfied, snooze cleared
+    reportBackendContract(2) // a later regression must warn immediately
     expect(notifySpy).toHaveBeenCalledTimes(1)
   })
 })

apps/desktop/src/store/updates.ts

@@ -91,7 +91,8 @@ function isUpdateToastSnoozed(): boolean {
 // against. The backend reports its own value in session runtime info; a lower
 // value (or none — a pre-GUI checkout) means GUI<->backend skew.
 // v2: requires the file.attach RPC (remote-gateway non-image file upload).
-const REQUIRED_BACKEND_CONTRACT = 2
+// v3: requires approvals.mode config RPCs and session.info reconciliation.
+const REQUIRED_BACKEND_CONTRACT = 3
 const SKEW_TOAST_ID = 'backend-contract-skew'
 // The contract check runs on every session.resume (applyRuntimeInfo), so
 // without a snooze the warning re-popped on every thread the user opened, even

apps/desktop/src/types/hermes.ts

@@ -303,6 +303,7 @@ export interface PaginatedSessions {
 
 export interface RpcEvent<T = unknown> {
   payload?: T
+  profile?: string
   session_id?: string
   type: string
 }
@@ -393,6 +394,7 @@ export interface SessionResumeResponse {
 }
 
 export interface SessionRuntimeInfo {
+  approval_mode?: 'manual' | 'off' | 'smart'
   branch?: string
   config_warning?: string
   credential_warning?: string

apps/shared/src/json-rpc-gateway.ts

@@ -23,6 +23,8 @@ export type GatewayEventName =
 
 export interface GatewayEvent<P = unknown> {
   payload?: P
+  /** Renderer-side source tag added by the Desktop gateway registry. */
+  profile?: string
   session_id?: string
   type: GatewayEventName
 }

cli.py

@@ -11683,13 +11683,15 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
         return ""
 
     def _approval_callback(self, command: str, description: str,
-                           *, allow_permanent: bool = True) -> str:
+                           *, allow_permanent: bool = True,
+                           smart_denied: bool = False) -> str:
         """
         Prompt for dangerous command approval through the prompt_toolkit UI.
 
         Called from the agent thread. Shows a selection UI similar to clarify
-        with choices: once / session / always / deny. When allow_permanent
-        is False (tirith warnings present), the 'always' option is hidden.
+        with choices: once / session / always / deny. Smart DENY owner
+        overrides show only once / deny. When allow_permanent is False for
+        another reason (for example tirith), only 'always' is hidden.
         Long commands also get a 'view' option so the full command can be
         expanded before deciding.
 
@@ -11706,7 +11708,11 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             self._approval_state = {
                 "command": command,
                 "description": description,
-                "choices": self._approval_choices(command, allow_permanent=allow_permanent),
+                "choices": self._approval_choices(
+                    command,
+                    allow_permanent=allow_permanent,
+                    smart_denied=smart_denied,
+                ),
                 "selected": 0,
                 "response_queue": response_queue,
             }
@@ -11752,9 +11758,13 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
             _cprint(f"\n{_DIM}  ⏱ Timeout — denying command{_RST}")
             return "deny"
 
-    def _approval_choices(self, command: str, *, allow_permanent: bool = True) -> list[str]:
+    def _approval_choices(self, command: str, *, allow_permanent: bool = True,
+                          smart_denied: bool = False) -> list[str]:
         """Return approval choices for a dangerous command prompt."""
-        choices = ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"]
+        if smart_denied:
+            choices = ["once", "deny"]
+        else:
+            choices = ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"]
         if len(command) > 70:
             choices.append("view")
         return choices
@@ -15179,7 +15189,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
                                 from tools.approval import get_current_session_key
                                 _drain_sk = get_current_session_key(default="")
                                 for _evt, _synth in process_registry.drain_notifications(session_key=_drain_sk):
+                                    from tools.async_delegation import (
+                                        claim_event_delivery, complete_event_delivery,
+                                    )
+                                    _claim = claim_event_delivery(_evt, "cli-idle")
+                                    if _claim is None:
+                                        continue
                                     self._pending_input.put(_synth)
+                                    complete_event_delivery(_evt, _claim)
                             except Exception:
                                 pass
                         continue
@@ -15341,7 +15358,14 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin):
                         try:
                             from tools.process_registry import process_registry
                             for _evt, _synth in process_registry.drain_notifications():
+                                from tools.async_delegation import (
+                                    claim_event_delivery, complete_event_delivery,
+                                )
+                                _claim = claim_event_delivery(_evt, "cli-post-turn")
+                                if _claim is None:
+                                    continue
                                 self._pending_input.put(_synth)
+                                complete_event_delivery(_evt, _claim)
                         except Exception:
                             pass  # Non-fatal — don't break the main loop
 

cron/scheduler.py

@@ -3557,7 +3557,14 @@ def _notify_provider_jobs_changed() -> None:
         logger.debug("on_jobs_changed notify failed: %s", e)
 
 
-def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> int:
+def tick(
+    verbose: bool = True,
+    adapters=None,
+    loop=None,
+    sync: bool = True,
+    *,
+    can_dispatch=None,
+):
     """
     Check and run all due jobs.
     
@@ -3568,7 +3575,9 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
         verbose: Whether to print status messages
         adapters: Optional dict mapping Platform → live adapter (from gateway)
         loop: Optional asyncio event loop (from gateway) for live adapter sends
-    
+        can_dispatch: Optional synchronous gate; false leaves due jobs untouched
+            for the next allowed tick
+
     Returns:
         Number of jobs executed (0 if another tick is already running)
     """
@@ -3590,6 +3599,10 @@ def tick(verbose: bool = True, adapters=None, loop=None, sync: bool = True) -> i
         return 0
 
     try:
+        if can_dispatch is not None and not can_dispatch():
+            logger.debug("Cron dispatch paused while gateway drains existing work")
+            return 0
+
         due_jobs = get_due_jobs()
 
         if verbose and not due_jobs:

cron/scheduler_provider.py

@@ -156,14 +156,16 @@ class InProcessCronScheduler(CronScheduler):
 
     ``start()`` blocks in the tick loop until ``stop_event`` is set, identical
     to the pre-refactor ``_start_cron_ticker`` core loop. The caller runs it in
-    a daemon thread.
+    a daemon thread. ``can_dispatch`` is an optional synchronous gate supplied
+    by GatewayRunner during external drain; skipped ticks leave due jobs intact
+    for the next allowed tick.
     """
 
     @property
     def name(self) -> str:
         return "builtin"
 
-    def start(self, stop_event, *, adapters=None, loop=None, interval=60):
+    def start(self, stop_event, *, adapters=None, loop=None, interval=60, can_dispatch=None):
         import logging
         from cron.scheduler import tick as cron_tick
         from cron.jobs import record_ticker_heartbeat
@@ -176,7 +178,16 @@ class InProcessCronScheduler(CronScheduler):
         while not stop_event.is_set():
             ok = False
             try:
-                cron_tick(verbose=False, adapters=adapters, loop=loop, sync=False)
+                if can_dispatch is not None and not can_dispatch():
+                    logger.debug("Cron dispatch paused while gateway drains existing work")
+                else:
+                    cron_tick(
+                        verbose=False,
+                        adapters=adapters,
+                        loop=loop,
+                        sync=False,
+                        can_dispatch=can_dispatch,
+                    )
                 ok = True
             except BaseException as e:
                 # Catch BaseException (not just Exception) so a SystemExit from

gateway/platforms/api_server.py

@@ -35,6 +35,9 @@ import asyncio
 import hashlib
 import hmac
 import json
+from contextlib import contextmanager
+from contextvars import ContextVar
+from functools import wraps
 import logging
 import os
 import socket as _socket
@@ -45,6 +48,12 @@ import uuid
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
+def _approval_event_choices(*, smart_denied: bool, allow_permanent: bool) -> list[str]:
+    if smart_denied:
+        return ["once", "deny"]
+    return ["once", "session", "always", "deny"] if allow_permanent else ["once", "session", "deny"]
+
+
 try:
     from aiohttp import web
     AIOHTTP_AVAILABLE = True
@@ -658,6 +667,66 @@ def _openai_error(message: str, err_type: str = "invalid_request_error", param:
     }
 
 
+_api_agent_request_reservation: ContextVar[Optional[dict[str, bool]]] = ContextVar(
+    "api_agent_request_reservation", default=None
+)
+
+
+def _admit_api_agent_request(handler):
+    """Reserve an authenticated API turn before its handler first awaits.
+
+    Gateway shutdown and aiohttp requests share an event loop. Keeping the
+    drain check and reservation in one non-awaiting block prevents a request
+    admitted immediately before shutdown from becoming invisible while it is
+    still parsing its body or resolving session state. The mutable reservation
+    is intentionally shared with child tasks so agent/task bookkeeping releases
+    this one slot exactly once.
+    """
+    @wraps(handler)
+    async def _wrapped(self, request, *args, **kwargs):
+        auth_err = self._check_auth(request)
+        if auth_err:
+            return auth_err
+        draining = self._draining_response()
+        if draining is not None:
+            return draining
+        reservation = {"active": True}
+        token = _api_agent_request_reservation.set(reservation)
+        self._pending_agent_requests += 1
+        try:
+            return await handler(self, request, *args, **kwargs)
+        finally:
+            if reservation["active"]:
+                reservation["active"] = False
+                self._pending_agent_requests = max(0, self._pending_agent_requests - 1)
+            _api_agent_request_reservation.reset(token)
+
+    return _wrapped
+
+
+def _release_pending_api_work(adapter, reservation: dict[str, bool]) -> None:
+    """Release a pending-work reservation exactly once."""
+    if reservation["active"]:
+        reservation["active"] = False
+        adapter._pending_agent_requests = max(0, adapter._pending_agent_requests - 1)
+
+
+@contextmanager
+def _reserve_pending_api_work(adapter):
+    """Keep externally-triggered background work visible across awaits.
+
+    A handler can detach the reservation to an asyncio task; its done callback
+    then owns release so shutdown cannot miss the handoff to background work.
+    """
+    reservation = {"active": True, "detached": False}
+    adapter._pending_agent_requests += 1
+    try:
+        yield reservation
+    finally:
+        if not reservation["detached"]:
+            _release_pending_api_work(adapter, reservation)
+
+
 if AIOHTTP_AVAILABLE:
     @web.middleware
     async def body_limit_middleware(request, handler):
@@ -906,6 +975,64 @@ class APIServerAdapter(BasePlatformAdapter):
         # (the /v1/runs path tracks its own in-flight set via
         # _active_run_tasks).
         self._inflight_agent_runs: int = 0
+        # Requests admitted before their handler reaches agent bookkeeping.
+        # Shutdown counts this reservation so the request cannot slip through
+        # the drain between its first await and _run_agent()/task registration.
+        self._pending_agent_requests: int = 0
+
+    def active_agent_work_count(self) -> int:
+        """Return all live agent work owned by this API adapter.
+
+        ``/v1/runs`` registers an asyncio task before it constructs and stores
+        its agent, so ``_active_run_agents`` has a real queued-before-agent gap.
+        Reuse the task-based accounting used by the concurrent-run limit: it
+        covers that gap and excludes completed tasks retained until cleanup.
+        """
+        try:
+            return (
+                int(getattr(self, "_pending_agent_requests", 0))
+                + int(self._inflight_agent_runs)
+                + sum(not task.done() for task in self._active_run_tasks.values())
+            )
+        except Exception:
+            return 0
+
+    @staticmethod
+    def _gateway_is_draining() -> bool:
+        """Whether the owning gateway currently refuses new agent turns."""
+        try:
+            from gateway.run import _gateway_runner_ref
+
+            runner = _gateway_runner_ref()
+            return bool(
+                runner
+                and (
+                    getattr(runner, "_draining", False)
+                    or getattr(runner, "_external_drain_active", False)
+                )
+            )
+        except Exception:
+            return False
+
+    def _draining_response(self) -> Optional["web.Response"]:
+        """Return a retryable response while the gateway drains existing work."""
+        if not self._gateway_is_draining():
+            return None
+        return web.json_response(
+            _openai_error(
+                "Gateway is draining existing work; retry shortly.",
+                code="gateway_draining",
+            ),
+            status=503,
+            headers={"Retry-After": "1"},
+        )
+
+    def _activate_admitted_request(self) -> None:
+        """Transfer this request's drain reservation to agent bookkeeping."""
+        reservation = _api_agent_request_reservation.get()
+        if reservation and reservation["active"]:
+            reservation["active"] = False
+            self._pending_agent_requests = max(0, self._pending_agent_requests - 1)
 
     def _readiness_work_counts(self) -> tuple[int, int, int]:
         """Return bounded work counts from each subsystem's public state."""
@@ -1913,11 +2040,9 @@ class APIServerAdapter(BasePlatformAdapter):
         fork = db.get_session(fork_id) or {"id": fork_id, "parent_session_id": source_id}
         return web.json_response({"object": "hermes.session", "session": self._session_response(fork)}, status=201)
 
+    @_admit_api_agent_request
     async def _handle_session_chat(self, request: "web.Request") -> "web.Response":
         """POST /api/sessions/{session_id}/chat — one synchronous agent turn."""
-        auth_err = self._check_auth(request)
-        if auth_err:
-            return auth_err
         gateway_session_key, key_err = self._parse_session_key_header(request)
         if key_err is not None:
             return key_err
@@ -1957,11 +2082,9 @@ class APIServerAdapter(BasePlatformAdapter):
             headers=headers,
         )
 
+    @_admit_api_agent_request
     async def _handle_session_chat_stream(self, request: "web.Request") -> "web.StreamResponse":
         """POST /api/sessions/{session_id}/chat/stream — SSE wrapper over _run_agent."""
-        auth_err = self._check_auth(request)
-        if auth_err:
-            return auth_err
         gateway_session_key, key_err = self._parse_session_key_header(request)
         if key_err is not None:
             return key_err
@@ -2098,12 +2221,9 @@ class APIServerAdapter(BasePlatformAdapter):
             logger.debug("[api_server] session SSE stream error: %s", exc)
         return response
 
+    @_admit_api_agent_request
     async def _handle_chat_completions(self, request: "web.Request") -> "web.Response":
         """POST /v1/chat/completions — OpenAI Chat Completions format."""
-        auth_err = self._check_auth(request)
-        if auth_err:
-            return auth_err
-
         # Bound total in-flight agent runs (configurable; #7483).
         limited = self._concurrency_limited_response()
         if limited is not None:
@@ -2305,8 +2425,8 @@ class APIServerAdapter(BasePlatformAdapter):
             # ``tool_progress_callback`` is intentionally not wired here:
             # it would duplicate every emit because ``run_agent`` fires it
             # side-by-side with ``tool_start_callback``/``tool_complete_callback``.
-            # The structured callbacks are strictly richer (they carry the
-            # tool_call id), so they own the chat-completions SSE channel.
+            # The structured callbacks are strictly richer (they carry
+            # the tool_call id), so they own the chat-completions SSE channel.
             agent_ref = [None]
             agent_task = asyncio.ensure_future(self._run_agent(
                 user_message=user_message,
@@ -3234,12 +3354,9 @@ class APIServerAdapter(BasePlatformAdapter):
 
         return response
 
+    @_admit_api_agent_request
     async def _handle_responses(self, request: "web.Request") -> "web.Response":
         """POST /v1/responses — OpenAI Responses API format."""
-        auth_err = self._check_auth(request)
-        if auth_err:
-            return auth_err
-
         # Bound total in-flight agent runs (configurable; #7483).
         limited = self._concurrency_limited_response()
         if limited is not None:
@@ -3782,6 +3899,9 @@ class APIServerAdapter(BasePlatformAdapter):
         auth_err = self._check_auth(request)
         if auth_err:
             return auth_err
+        draining = self._draining_response()
+        if draining is not None:
+            return draining
         cron_err = self._check_jobs_available()
         if cron_err:
             return cron_err
@@ -3827,31 +3947,39 @@ class APIServerAdapter(BasePlatformAdapter):
                 self._request_audit_log_suffix(request),
             )
             return web.json_response({"error": "invalid fire token"}, status=401)
+        draining = self._draining_response()
+        if draining is not None:
+            return draining
 
-        try:
-            body = await request.json()
-        except Exception:
-            body = {}
-        job_id = (body or {}).get("job_id")
-        if not job_id:
-            return web.json_response({"error": "missing job_id"}, status=400)
+        with _reserve_pending_api_work(self) as reservation:
+            try:
+                body = await request.json()
+            except Exception:
+                body = {}
+            job_id = (body or {}).get("job_id")
+            if not job_id:
+                return web.json_response({"error": "missing job_id"}, status=400)
 
-        from cron.scheduler_provider import resolve_cron_scheduler
-        provider = resolve_cron_scheduler()
+            from cron.scheduler_provider import resolve_cron_scheduler
+            provider = resolve_cron_scheduler()
 
-        loop = asyncio.get_running_loop()
-        # Fire in the background (202 immediately). fire_due claims via the
-        # store CAS, so a retry while this is in flight is de-duped.
-        task = asyncio.create_task(
-            asyncio.to_thread(provider.fire_due, job_id, adapters=None, loop=loop)
-        )
-        try:
-            self._background_tasks.add(task)
-            task.add_done_callback(self._background_tasks.discard)
-        except (TypeError, AttributeError):
-            pass
+            loop = asyncio.get_running_loop()
+            # Fire in the background (202 immediately). fire_due claims via the
+            # store CAS, so a retry while this is in flight is de-duped.
+            task = asyncio.create_task(
+                asyncio.to_thread(provider.fire_due, job_id, adapters=None, loop=loop)
+            )
+            reservation["detached"] = True
+            task.add_done_callback(
+                lambda _task: _release_pending_api_work(self, reservation)
+            )
+            try:
+                self._background_tasks.add(task)
+                task.add_done_callback(self._background_tasks.discard)
+            except (TypeError, AttributeError):
+                pass
 
-        return web.json_response({"status": "accepted", "job_id": job_id}, status=202)
+            return web.json_response({"status": "accepted", "job_id": job_id}, status=202)
 
 
     # ------------------------------------------------------------------
@@ -4005,19 +4133,22 @@ class APIServerAdapter(BasePlatformAdapter):
         """Return a 429 response if the concurrent-run cap is reached, else None.
 
         The cap bounds total in-flight agent activity across every
-        agent-serving endpoint: the non-streaming chat/responses paths
-        (tracked by ``_inflight_agent_runs``) plus the ``/v1/runs`` path
-        (tracked by live entries in ``_active_run_tasks``).  Stream queues are
-        transport state and may disappear while their underlying run remains
-        active, so they must not define run concurrency. A configured value of
-        0 disables the cap entirely.
+        agent-serving endpoint. Reuse the same adapter-owned work count that
+        shutdown draining uses, including an admitted request before it reaches
+        agent/task bookkeeping. Stream queues are transport state and may
+        disappear while their underlying run remains active, so they must not
+        define run concurrency. A configured value of 0 disables the cap.
         """
         limit = self._max_concurrent_runs
         if limit <= 0:
             return None
-        inflight = self._inflight_agent_runs + sum(
-            not task.done() for task in self._active_run_tasks.values()
-        )
+        inflight = self.active_agent_work_count()
+        # The current request owns one reservation until it hands off to
+        # _run_agent() or /v1/runs task registration. It must not consume its
+        # own last available slot; other admitted requests remain counted.
+        reservation = _api_agent_request_reservation.get()
+        if reservation and reservation["active"]:
+            inflight -= 1
         if inflight >= limit:
             return web.json_response(
                 _openai_error(
@@ -4135,6 +4266,7 @@ class APIServerAdapter(BasePlatformAdapter):
             finally:
                 clear_session_vars(tokens)
 
+        self._activate_admitted_request()
         self._inflight_agent_runs += 1
         try:
             return await loop.run_in_executor(None, _run)
@@ -4209,12 +4341,9 @@ class APIServerAdapter(BasePlatformAdapter):
 
         return _callback
 
+    @_admit_api_agent_request
     async def _handle_runs(self, request: "web.Request") -> "web.Response":
         """POST /v1/runs — start an agent run, return run_id immediately."""
-        auth_err = self._check_auth(request)
-        if auth_err:
-            return auth_err
-
         # Long-term memory scope header (see chat_completions for details).
         gateway_session_key, key_err = self._parse_session_key_header(request)
         if key_err is not None:
@@ -4375,7 +4504,10 @@ class APIServerAdapter(BasePlatformAdapter):
                         "event": "approval.request",
                         "run_id": run_id,
                         "timestamp": time.time(),
-                        "choices": ["once", "session", "always", "deny"],
+                        "choices": _approval_event_choices(
+                            smart_denied=bool(event.get("smart_denied")),
+                            allow_permanent=event.get("allow_permanent") is not False,
+                        ),
                     })
                     self._set_run_status(
                         run_id,
@@ -4533,6 +4665,7 @@ class APIServerAdapter(BasePlatformAdapter):
                 self._run_approval_sessions.pop(run_id, None)
                 self._stopping_run_ids.discard(run_id)
 
+        self._activate_admitted_request()
         task = asyncio.create_task(_run_and_close())
         self._active_run_tasks[run_id] = task
         try:

gateway/platforms/base.py

@@ -1072,10 +1072,33 @@ def _profile_cache_roots() -> List[Path]:
     return roots
 
 
+def _kanban_attachment_roots() -> List[Path]:
+    """Return durable Kanban attachment roots without importing kanban_db."""
+    override = os.environ.get("HERMES_KANBAN_ATTACHMENTS_ROOT", "").strip()
+    if override:
+        return [Path(override).expanduser()]
+    home_override = os.environ.get("HERMES_KANBAN_HOME", "").strip()
+    root = Path(home_override).expanduser() if home_override else _HERMES_ROOT
+    roots = [root / "kanban" / "attachments"]
+    boards_root = root / "kanban" / "boards"
+    try:
+        board_dirs = [
+            path for path in boards_root.iterdir()
+            if path.is_dir() and not path.is_symlink()
+            and re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,63}", path.name)
+            and (path / "kanban.db").is_file()
+        ]
+    except OSError:
+        return roots
+    roots.extend(path / "attachments" for path in board_dirs)
+    return roots
+
+
 def _media_delivery_allowed_roots() -> List[Path]:
     """Return roots from which model-emitted local media may be delivered."""
     roots = [Path(root) for root in MEDIA_DELIVERY_SAFE_ROOTS]
     roots.extend(_profile_cache_roots())
+    roots.extend(_kanban_attachment_roots())
     extra_roots = os.environ.get(MEDIA_DELIVERY_ALLOW_DIRS_ENV, "")
     for chunk in extra_roots.split(os.pathsep):
         for raw_root in chunk.split(","):

gateway/platforms/qqbot/adapter.py

@@ -2648,7 +2648,10 @@ class QQAdapter(BasePlatformAdapter):
         return await self.send_with_keyboard(
             chat_id,
             build_approval_text(req),
-            build_approval_keyboard(req.session_key),
+            build_approval_keyboard(
+                req.session_key,
+                allow_permanent=getattr(req, "allow_permanent", True),
+            ),
             reply_to=reply_to,
         )
 
@@ -2668,6 +2671,8 @@ class QQAdapter(BasePlatformAdapter):
             session_key: str,
             description: str = "dangerous command",
             metadata: Optional[Dict[str, Any]] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """Send a button-based exec-approval prompt for a dangerous command.
 
@@ -2677,6 +2682,8 @@ class QQAdapter(BasePlatformAdapter):
         adapter's interaction callback (:meth:`_default_interaction_dispatch`).
         """
         del metadata  # QQ doesn't have thread_id / DM targeting overrides.
+        if smart_denied:
+            description += " Owner override applies to this one operation only."
 
         # Use the reply-to message for passive-message context when we have one.
         # QQ requires a msg_id on outbound messages to a user we've never
@@ -2689,6 +2696,7 @@ class QQAdapter(BasePlatformAdapter):
             description=description,
             command_preview=command,
             timeout_sec=self._APPROVAL_TIMEOUT_SECONDS,
+            allow_permanent=allow_permanent and not smart_denied,
         )
         return await self.send_approval_request(
             chat_id, req, reply_to=msg_id,

gateway/platforms/qqbot/keyboards.py

@@ -201,8 +201,8 @@ def _make_callback_button(
     )
 
 
-def build_approval_keyboard(session_key: str) -> InlineKeyboard:
-    """Build the 3-button approval keyboard.
+def build_approval_keyboard(session_key: str, *, allow_permanent: bool = True) -> InlineKeyboard:
+    """Build the approval keyboard, hiding persistent scope when unavailable.
 
     Layout: ``[✅ 允许一次] [⭐ 始终允许] [❌ 拒绝]`` — all three share
     ``group_id='approval'`` so clicking one greys out the rest.
@@ -210,38 +210,25 @@ def build_approval_keyboard(session_key: str) -> InlineKeyboard:
     :param session_key: Embedded into ``button_data`` so the decision
         routes back to the right pending approval.
     """
-    return InlineKeyboard(
-        content=KeyboardContent(
-            rows=[
-                KeyboardRow(buttons=[
-                    _make_callback_button(
-                        btn_id="allow",
-                        label="✅ 允许一次",
-                        visited_label="已允许",
-                        data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-once",
-                        style=1,
-                        group_id="approval",
-                    ),
-                    _make_callback_button(
-                        btn_id="always",
-                        label="⭐ 始终允许",
-                        visited_label="已始终允许",
-                        data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-always",
-                        style=1,
-                        group_id="approval",
-                    ),
-                    _make_callback_button(
-                        btn_id="deny",
-                        label="❌ 拒绝",
-                        visited_label="已拒绝",
-                        data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:deny",
-                        style=0,
-                        group_id="approval",
-                    ),
-                ]),
-            ]
+    buttons = [
+        _make_callback_button(
+            btn_id="allow", label="✅ 允许一次", visited_label="已允许",
+            data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-once",
+            style=1, group_id="approval",
         )
-    )
+    ]
+    if allow_permanent:
+        buttons.append(_make_callback_button(
+            btn_id="always", label="⭐ 始终允许", visited_label="已始终允许",
+            data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:allow-always",
+            style=1, group_id="approval",
+        ))
+    buttons.append(_make_callback_button(
+        btn_id="deny", label="❌ 拒绝", visited_label="已拒绝",
+        data=f"{APPROVAL_BUTTON_PREFIX}{session_key}:deny",
+        style=0, group_id="approval",
+    ))
+    return InlineKeyboard(content=KeyboardContent(rows=[KeyboardRow(buttons=buttons)]))
 
 
 def build_update_prompt_keyboard() -> InlineKeyboard:
@@ -295,6 +282,7 @@ class ApprovalRequest:
     tool_name: str = ""
     severity: str = ""
     timeout_sec: int = 120
+    allow_permanent: bool = True
 
 
 def build_approval_text(req: ApprovalRequest) -> str:

gateway/platforms/whatsapp_cloud.py

@@ -387,6 +387,20 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
             return True
         return super()._open_dm_opted_in()
 
+    def _is_interactive_sender_authorized(self, sender_id: str) -> bool:
+        """Authorize inbound button/list taps before running resolvers.
+
+        Interactive replies bypass the normal ``_build_message_event_from_cloud``
+        path (which calls ``_should_process_message``), so approval /
+        slash-confirm / clarify taps must re-check DM policy here. Uses the
+        strict ``_is_dm_allowed`` gate (not intake/pairing) so a stale prompt
+        cannot be answered after the sender is removed from the allowlist.
+        """
+        principal = str(sender_id or "").strip()
+        if not principal:
+            return False
+        return self._is_dm_allowed(principal)
+
     # ------------------------------------------------------------------ lifecycle
     async def connect(self, *, is_reconnect: bool = False) -> bool:
         if not check_whatsapp_cloud_requirements():
@@ -805,6 +819,8 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
         session_key: str,
         description: str = "dangerous command",
         metadata: Optional[Dict[str, Any]] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """Render a dangerous-command approval prompt with native buttons.
 
@@ -816,6 +832,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
         if self._http_client is None:
             return SendResult(success=False, error="Not connected")
 
+        del allow_permanent  # This adapter already offers one-shot Approve / Deny only.
         # WhatsApp body caps at 1024 chars; reserve room for the
         # framing prose around the command.
         cmd = command or ""
@@ -824,6 +841,7 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
             f"⚠️ *Command Approval Required*\n\n"
             f"```\n{cmd_preview}\n```\n\n"
             f"Reason: {description}"
+            + ("\n\nSmart DENY: owner override applies to this one operation only." if smart_denied else "")
         )
 
         approval_id = uuid.uuid4().hex[:12]
@@ -1635,6 +1653,18 @@ class WhatsAppCloudAdapter(WhatsAppBehaviorMixin, BasePlatformAdapter):
         if not button_id:
             return False
 
+        sender_id = str(raw_message.get("from") or "").strip()
+        if not self._is_interactive_sender_authorized(sender_id):
+            logger.warning(
+                "[whatsapp_cloud] Rejected unauthorized interactive tap "
+                "from %s (button_id=%r)",
+                sender_id or "<unknown>",
+                button_id,
+            )
+            # Claim the webhook entry so the tap is not re-dispatched as
+            # plain text (which could re-enter the agent loop).
+            return True
+
         # Clarify: cl:<clarify_id>:<idx|other>
         if button_id.startswith("cl:"):
             parts = button_id.split(":", 2)

gateway/run.py

@@ -353,6 +353,34 @@ def _redact_approval_command(cmd: "str | None") -> str:
     return redact_sensitive_text(str(cmd or ""), force=True)
 
 
+def _format_exec_approval_fallback(
+    command: str,
+    description: str,
+    command_prefix: str,
+    *,
+    allow_permanent: bool = True,
+    smart_denied: bool = False,
+) -> str:
+    """Render the text fallback from approval capabilities, not platform names."""
+    cmd_preview = command[:200] + "..." if len(command) > 200 else command
+    heading = "⚠️ **Dangerous command requires approval:**"
+    if smart_denied:
+        heading = "⚠️ **Smart DENY — owner override for one operation:**"
+
+    choices = [f"Reply `{command_prefix}approve` to execute this one operation"]
+    if not smart_denied:
+        choices.append(
+            f"`{command_prefix}approve session` to approve this pattern for the session"
+        )
+        if allow_permanent:
+            choices.append(f"`{command_prefix}approve always` to approve permanently")
+    choices.append(f"`{command_prefix}deny` to cancel")
+    return (
+        f"{heading}\n```\n{cmd_preview}\n```\nReason: {description}\n\n"
+        + ", ".join(choices[:-1]) + f", or {choices[-1]}."
+    )
+
+
 def _gateway_provider_error_reply(text: str) -> str:
     """Map raw provider/API errors to a short user-safe Telegram reply."""
     if _GATEWAY_AUTH_ERROR_RE.search(text):
@@ -2951,6 +2979,15 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # cannot grow unbounded over a long-running gateway lifetime.
         self._session_sources: "OrderedDict[str, SessionSource]" = OrderedDict()
         self._session_sources_max = 512
+        # Completion delivery is intentionally lifecycle-scoped. This closes
+        # duplicate queue/watcher races inside one gateway without pretending
+        # the adapter call and a persistence write can be exactly-once across
+        # a process crash. Any durable async-delegation replay state remains
+        # owned by tools.async_delegation, not a parallel gateway ledger.
+        self._completion_delivery_lock = threading.Lock()
+        self._completion_deliveries_inflight: set[tuple[str, str, object]] = set()
+        self._completion_deliveries_delivered: "OrderedDict[tuple[str, str, object], None]" = OrderedDict()
+        self._completion_delivery_retention = 2048
 
         # Cache AIAgent instances per session to preserve prompt caching.
         # Without this, a new AIAgent is created per message, rebuilding the
@@ -4085,6 +4122,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
     def _running_agent_count(self) -> int:
         return len(self._running_agents)
 
+    def _active_work_count(self) -> int:
+        """All agent work the gateway must expose and drain as one total."""
+        return (
+            self._running_agent_count()
+            + self._active_cron_job_count()
+            + self._active_api_run_count()
+        )
+
     def _active_cron_job_count(self) -> int:
         """Count of cron jobs currently executing, from the cron scheduler's
         own in-flight tracking (``cron.scheduler._running_job_ids``).
@@ -4105,6 +4150,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         except Exception:
             return 0
 
+    def _active_api_run_count(self) -> int:
+        """Count API-server work that is outside ``_running_agents``.
+
+        The primary API server owns the sole HTTP listener. Secondary multiplex
+        profiles cannot create an ``api_server`` adapter because it binds a port,
+        so only the primary registry is a supported source of this work.
+        """
+        try:
+            adapter = getattr(self, "adapters", {}).get(Platform.API_SERVER)
+            helper = getattr(adapter, "active_agent_work_count", None)
+            return max(0, int(helper())) if callable(helper) else 0
+        except Exception:
+            return 0
+
     # ── scale-to-zero idle detection / dormant-quiesce (Phase 0) ──────────────
     # The gateway-side BEHAVIOUR that consumes the relay scale-to-zero primitives
     # (gateway-gateway Phase 5). Pure logic lives in gateway/scale_to_zero.py; the
@@ -4492,7 +4551,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 gateway_state=gateway_state,
                 exit_reason=exit_reason,
                 restart_requested=self._restart_requested,
-                active_agents=self._running_agent_count(),
+                active_agents=self._active_work_count(),
             )
         except Exception:
             pass
@@ -4515,7 +4574,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         """
         try:
             from gateway.status import write_runtime_status
-            write_runtime_status(active_agents=self._running_agent_count())
+            write_runtime_status(active_agents=self._active_work_count())
         except Exception:
             pass
 
@@ -4542,7 +4601,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         logger.info(
             "External drain ENGAGED (.drain_request.json present) — refusing "
             "new turns; %d in-flight turn(s) will finish. Process stays up.",
-            self._running_agent_count(),
+            self._active_work_count(),
         )
         # Flip the persisted lifecycle state so /api/status.gateway_busy /
         # gateway_drainable track the drain. Preserve active_agents (the
@@ -4593,6 +4652,10 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             try:
                 if drain_requested():
                     self._enter_external_drain()
+                    # API and cron work live outside messaging's
+                    # _running_agents map. Refresh the aggregate while an
+                    # external caller polls this reversible drain state.
+                    self._persist_active_agents()
                 else:
                     self._exit_external_drain()
             except asyncio.CancelledError:
@@ -5659,22 +5722,26 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         snapshot = self._snapshot_running_agents()
         last_active_count = self._running_agent_count()
         last_cron_count = self._active_cron_job_count()
+        last_api_count = self._active_api_run_count()
         last_status_at = 0.0
 
         def _maybe_update_status(force: bool = False) -> None:
-            nonlocal last_active_count, last_cron_count, last_status_at
+            nonlocal last_active_count, last_cron_count, last_api_count, last_status_at
             now = asyncio.get_running_loop().time()
             active_count = self._running_agent_count()
             cron_count = self._active_cron_job_count()
+            api_count = self._active_api_run_count()
             if (
                 force
                 or active_count != last_active_count
                 or cron_count != last_cron_count
+                or api_count != last_api_count
                 or (now - last_status_at) >= 1.0
             ):
                 self._update_runtime_status("draining")
                 last_active_count = active_count
                 last_cron_count = cron_count
+                last_api_count = api_count
                 last_status_at = now
 
         # Cron jobs run on the scheduler's own thread pool, outside
@@ -5682,7 +5749,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
         # same wait/timeout this method already applies to chat sessions,
         # or a cron job's tool work gets killed with zero warning the
         # instant it's the only active thing running (#60432).
-        if not self._running_agents and last_cron_count == 0:
+        # API-server / desk sessions have the same structural gap (#63529).
+        if not self._running_agents and last_cron_count == 0 and last_api_count == 0:
             _maybe_update_status(force=True)
             return snapshot, False
 
@@ -5692,12 +5760,20 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
         deadline = asyncio.get_running_loop().time() + timeout
         while (
-            (self._running_agents or self._active_cron_job_count())
+            (
+                self._running_agents
+                or self._active_cron_job_count()
+                or self._active_api_run_count()
+            )
             and asyncio.get_running_loop().time() < deadline
         ):
             _maybe_update_status()
             await asyncio.sleep(0.1)
-        timed_out = bool(self._running_agents) or bool(self._active_cron_job_count())
+        timed_out = (
+            bool(self._running_agents)
+            or bool(self._active_cron_job_count())
+            or bool(self._active_api_run_count())
+        )
         _maybe_update_status(force=True)
         return snapshot, timed_out
 
@@ -8150,12 +8226,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     logger.debug("pre-drain mark_resume_pending failed for %s: %s", _sk, _e)
 
             _cron_at_start = self._active_cron_job_count()
+            _api_at_start = self._active_api_run_count()
             _drain_started_at = time.monotonic()
             active_agents, timed_out = await self._drain_active_agents(timeout)
             logger.info(
                 "Shutdown phase: drain done at +%.2fs (drain took %.2fs, "
                 "timed_out=%s, active_at_start=%d, active_now=%d, "
-                "cron_at_start=%d, cron_now=%d)",
+                "cron_at_start=%d, cron_now=%d, "
+                "api_at_start=%d, api_now=%d)",
                 _phase_elapsed(),
                 time.monotonic() - _drain_started_at,
                 timed_out,
@@ -8163,6 +8241,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 self._running_agent_count(),
                 _cron_at_start,
                 self._active_cron_job_count(),
+                _api_at_start,
+                self._active_api_run_count(),
             )
 
             if not timed_out:
@@ -8181,11 +8261,13 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
             if timed_out:
                 logger.warning(
-                    "Gateway drain timed out after %.1fs with %d active agent(s) "
-                    "and %d in-flight cron job(s); interrupting remaining work.",
+                    "Gateway drain timed out after %.1fs with %d active agent(s), "
+                    "%d in-flight cron job(s), and %d api_server run(s); "
+                    "interrupting remaining work.",
                     timeout,
                     self._running_agent_count(),
                     self._active_cron_job_count(),
+                    self._active_api_run_count(),
                 )
                 # Mark forcibly-interrupted sessions as resume_pending BEFORE
                 # interrupting the agents.  This preserves each session's
@@ -15426,11 +15508,17 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
             user_name=str(evt.get("user_name") or "").strip() or None,
         )
 
-    async def _inject_watch_notification(self, synth_text: str, evt: dict) -> None:
-        """Inject a watch-pattern notification as a synthetic message event.
+    async def _inject_watch_notification(
+        self, synth_text: str, evt: dict,
+    ) -> Optional[bool]:
+        """Inject a watch/completion notification as a synthetic message event.
 
-        Routing must come from the queued watch event itself, not from whatever
+        Routing must come from the queued event itself, not from whatever
         foreground message happened to be active when the queue was drained.
+        Returns ``True`` after adapter acceptance, ``False`` after a retryable
+        adapter failure, and ``None`` when the event has no gateway route. This
+        is not a transactional boundary: a process crash after adapter
+        acceptance can still cause durable at-least-once replay.
         """
         source = self._build_process_event_source(evt)
         if not source:
@@ -15438,7 +15526,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 "Dropping watch notification with no routing metadata for process %s",
                 evt.get("session_id", "unknown"),
             )
-            return
+            return None
         platform_name = source.platform.value if hasattr(source.platform, "value") else str(source.platform)
         adapter = None
         for p, a in self.adapters.items():
@@ -15446,7 +15534,7 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 adapter = a
                 break
         if not adapter:
-            return
+            return None
         try:
             metadata = {}
             parent_session_id = str(evt.get("parent_session_id") or "").strip()
@@ -15467,8 +15555,118 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 source.thread_id,
             )
             await adapter.handle_message(synth_event)
+            return True
         except Exception as e:
             logger.error("Watch notification injection error: %s", e)
+            return False
+
+    @staticmethod
+    def _completion_delivery_identity(evt: dict) -> Optional[tuple[str, str, object]]:
+        """Return a producer-stable identity when one is available.
+
+        Delegation UUIDs identify one producer completion. Process session IDs
+        are normally unique too, but include the persisted spawn epoch so an
+        explicitly reused ID represents a distinct process incarnation. Legacy
+        process events without ``started_at`` are delivered without deduplication
+        rather than risking suppression of a real completion.
+        """
+        evt_type = str(evt.get("type") or "")
+        if evt_type == "async_delegation":
+            producer_id = str(evt.get("delegation_id") or "")
+            return (evt_type, producer_id, "") if producer_id else None
+        if evt_type == "completion":
+            producer_id = str(evt.get("session_id") or "")
+            started_at = evt.get("started_at")
+            if producer_id and started_at is not None:
+                return (evt_type, producer_id, started_at)
+        return None
+
+    async def _deliver_completion_notification(
+        self, synth_text: str, evt: dict,
+    ) -> Optional[bool]:
+        """Deliver once per live gateway, or return False for a retry.
+
+        ``True`` means this caller reached adapter acceptance, ``False`` means
+        injection failed and the claim was released for retry, and ``None``
+        means either another same-lifecycle caller owns/delivered the producer
+        event or the event has no gateway route. No cross-process exactly-once
+        guarantee is claimed.
+        """
+        identity = self._completion_delivery_identity(evt)
+        durable_claim_id = ""
+        durable_delegation_id = ""
+        if evt.get("type") == "async_delegation":
+            durable_delegation_id = str(evt.get("delegation_id") or "")
+            if durable_delegation_id:
+                try:
+                    from tools.async_delegation import claim_completion_delivery
+
+                    durable_claim_id = f"gateway:{id(self)}:{__import__('uuid').uuid4().hex}"
+                    if not claim_completion_delivery(
+                        durable_delegation_id, durable_claim_id,
+                    ):
+                        return None
+                except Exception as exc:
+                    logger.warning(
+                        "Could not claim durable async completion %s: %s",
+                        durable_delegation_id, exc,
+                    )
+                    return False
+        if identity is not None:
+            with self._completion_delivery_lock:
+                if (
+                    identity in self._completion_deliveries_inflight
+                    or identity in self._completion_deliveries_delivered
+                ):
+                    return None
+                self._completion_deliveries_inflight.add(identity)
+
+        accepted = False
+        try:
+            injection_result = await self._inject_watch_notification(synth_text, evt)
+            if injection_result is not True:
+                return injection_result
+            accepted = True
+
+            if identity is not None:
+                with self._completion_delivery_lock:
+                    self._completion_deliveries_inflight.discard(identity)
+                    self._completion_deliveries_delivered[identity] = None
+                    while (
+                        len(self._completion_deliveries_delivered)
+                        > self._completion_delivery_retention
+                    ):
+                        self._completion_deliveries_delivered.popitem(last=False)
+
+            # If the durable async-delegation producer branch is present, its
+            # SQLite row remains the authoritative replay state. Acknowledge it
+            # after adapter acceptance; this gateway keeps no parallel ledger.
+            if durable_claim_id:
+                try:
+                    from tools.async_delegation import complete_completion_delivery
+
+                    complete_completion_delivery(
+                        durable_delegation_id, durable_claim_id,
+                    )
+                except Exception as exc:
+                    logger.warning(
+                        "Could not acknowledge durable async completion %s: %s",
+                        durable_delegation_id, exc,
+                    )
+            return True
+        finally:
+            if identity is not None and not accepted:
+                with self._completion_delivery_lock:
+                    self._completion_deliveries_inflight.discard(identity)
+            if durable_claim_id and not accepted:
+                try:
+                    from tools.async_delegation import release_completion_delivery
+
+                    release_completion_delivery(
+                        durable_delegation_id, durable_claim_id,
+                    )
+                except Exception:
+                    logger.debug("Could not release durable completion claim", exc_info=True)
 
     def _enrich_async_delegation_routing(self, evt: dict) -> None:
         """Fill platform/chat_id/thread_id/chat_type on an async-delegation event.
@@ -15531,8 +15729,11 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                     if not synth_text:
                         continue
                     try:
-                        await self._inject_watch_notification(synth_text, evt)
+                        delivered = await self._deliver_completion_notification(synth_text, evt)
+                        if delivered is False:
+                            _pr.completion_queue.put(evt)
                     except Exception as e:
+                        _pr.completion_queue.put(evt)
                         logger.error("Async delegation injection error: %s", e)
             except Exception as e:
                 logger.debug("Async delegation watcher error: %s", e)
@@ -15598,8 +15799,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 # (#10156) — a status check must not suppress this delivery turn.
                 from tools.process_registry import format_process_notification, process_registry as _pr_check
                 if agent_notify and not _pr_check.is_completion_consumed(session_id):
+                    from agent.redact import redact_terminal_output
                     from tools.ansi_strip import strip_ansi
+                    _command = getattr(session, "command", "") or ""
                     _raw = strip_ansi(session.output_buffer) if session.output_buffer else ""
+                    _raw = redact_terminal_output(_raw, _command)
+                    _command = _redact_gateway_user_facing_secrets(_command)
                     # Truncate at line boundaries so notifications never start
                     # mid-line (fixes #23284). Keep the last ~2000 chars but
                     # snap to the nearest preceding newline, then prepend a
@@ -15612,57 +15817,34 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                         _out = f"[… output truncated — showing last {len(_tail)} chars]\n{_tail}"
                     else:
                         _out = _raw
-                    synth_text = format_process_notification({
+                    completion_evt = {
                         "type": "completion",
-                        "session_id": session_id,
-                        "command": session.command,
-                        "exit_code": session.exit_code,
-                        "completion_reason": getattr(session, "completion_reason", "exited"),
-                        "termination_source": getattr(session, "termination_source", ""),
-                        "output": _out,
-                    })
-                    if not synth_text:
-                        break
-                    source = self._build_process_event_source({
                         "session_id": session_id,
                         "session_key": session_key,
                         "platform": platform_name,
+                        "chat_type": watcher.get("chat_type", ""),
                         "chat_id": chat_id,
                         "thread_id": thread_id,
                         "user_id": user_id,
                         "user_name": user_name,
-                    })
-                    if not source:
-                        logger.warning(
-                            "Dropping completion notification with no routing metadata for process %s",
-                            session_id,
-                        )
+                        "message_id": message_id,
+                        "started_at": getattr(session, "started_at", None),
+                        "command": _command,
+                        "exit_code": session.exit_code,
+                        "completion_reason": getattr(session, "completion_reason", "exited"),
+                        "termination_source": getattr(session, "termination_source", ""),
+                        "output": _out,
+                    }
+                    synth_text = format_process_notification(completion_evt)
+                    if not synth_text:
                         break
-
-                    adapter = None
-                    for p, a in self.adapters.items():
-                        if p == source.platform:
-                            adapter = a
-                            break
-                    if adapter and source.chat_id:
-                        try:
-                            synth_event = MessageEvent(
-                                text=synth_text,
-                                message_type=MessageType.TEXT,
-                                source=source,
-                                internal=True,
-                                message_id=message_id,
-                            )
-                            logger.info(
-                                "Process %s finished — injecting agent notification for session %s chat=%s thread=%s",
-                                session_id,
-                                session_key,
-                                source.chat_id,
-                                source.thread_id,
-                            )
-                            await adapter.handle_message(synth_event)
-                        except Exception as e:
-                            logger.error("Agent notify injection error: %s", e)
+                    delivered = await self._deliver_completion_notification(
+                        synth_text, completion_evt,
+                    )
+                    if delivered is False:
+                        # The process remains terminal; retry after failed
+                        # adapter injection instead of suppressing the result.
+                        continue
                     break
 
                 # --- Normal text-only notification ---
@@ -18641,6 +18823,8 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                                 session_key=_approval_session_key,
                                 description=desc,
                                 metadata=_status_thread_metadata,
+                                allow_permanent=approval_data.get("allow_permanent", True),
+                                smart_denied=approval_data.get("smart_denied", False),
                             ),
                             _loop_for_step,
                             logger=logger,
@@ -18665,13 +18849,12 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
                 # can actually type (`!approve`) — typed "/" is blocked in
                 # Slack threads and reserved by Matrix clients.
                 _p = getattr(_status_adapter, "typed_command_prefix", "/")
-                cmd_preview = cmd[:200] + "..." if len(cmd) > 200 else cmd
-                msg = (
-                    f"⚠️ **Dangerous command requires approval:**\n"
-                    f"```\n{cmd_preview}\n```\n"
-                    f"Reason: {desc}\n\n"
-                    f"Reply `{_p}approve` to execute, `{_p}approve session` to approve this pattern "
-                    f"for the session, `{_p}approve always` to approve permanently, or `{_p}deny` to cancel."
+                msg = _format_exec_approval_fallback(
+                    cmd,
+                    desc,
+                    _p,
+                    allow_permanent=approval_data.get("allow_permanent", True),
+                    smart_denied=approval_data.get("smart_denied", False),
                 )
                 try:
                     _approval_send_fut = safe_schedule_threadsafe(
@@ -20833,13 +21016,21 @@ async def start_gateway(config: Optional[GatewayConfig] = None, replace: bool =
     # historical in-process 60s ticker; an external provider (e.g. chronos)
     # may arm a schedule and return. Pass the event loop so cron delivery can
     # use live adapters (E2EE support).
-    from cron.scheduler_provider import resolve_cron_scheduler
+    from cron.scheduler_provider import InProcessCronScheduler, resolve_cron_scheduler
     cron_stop = threading.Event()
     cron_provider = resolve_cron_scheduler()
+    cron_start_kwargs = {"adapters": runner.adapters, "loop": asyncio.get_running_loop()}
+    # External cron providers own their remote scheduling contract. Only the
+    # in-process ticker polls local due jobs, so only it receives the local
+    # external-drain dispatch gate.
+    if isinstance(cron_provider, InProcessCronScheduler):
+        cron_start_kwargs["can_dispatch"] = lambda: not (
+            runner._draining or runner._external_drain_active
+        )
     cron_thread = threading.Thread(
         target=cron_provider.start,
         args=(cron_stop,),
-        kwargs={"adapters": runner.adapters, "loop": asyncio.get_running_loop()},
+        kwargs=cron_start_kwargs,
         daemon=True,
         name="cron-scheduler",
     )

hermes_cli/config.py

@@ -8097,6 +8097,20 @@ def edit_config():
     subprocess.run([editor, str(config_path)])
 
 
+def _default_value_for_key(dotted_key: str):
+    """Return the leaf value declared for *dotted_key* in ``DEFAULT_CONFIG``.
+
+    Unknown keys and non-leaf paths return ``None`` so they retain the legacy
+    best-effort coercion used by ``config set``.
+    """
+    node = DEFAULT_CONFIG
+    for part in dotted_key.split("."):
+        if not isinstance(node, dict) or part not in node:
+            return None
+        node = node[part]
+    return node if not isinstance(node, dict) else None
+
+
 def set_config_value(key: str, value: str):
     """Set a configuration value."""
     if is_managed():
@@ -8154,16 +8168,21 @@ def set_config_value(key: str, value: str):
     # _set_nested which preserves list-typed nodes; before #17876 the
     # inline navigation here silently overwrote lists with dicts.
 
-    # Convert value to appropriate type
-    if value.lower() in {'true', 'yes', 'on'}:
-        value = True
-    elif value.lower() in {'false', 'no', 'off'}:
-        value = False
-    elif value.isdigit():
-        value = int(value)
-    elif value.replace('.', '', 1).isdigit():
-        value = float(value)
+    # Preserve values for string-typed settings.  In particular, enum members
+    # such as approvals.mode="off" must not become YAML booleans.  Unknown keys
+    # retain the historical best-effort coercion behavior.
+    coerced_value: Any = value
+    if not isinstance(_default_value_for_key(key), str):
+        if value.lower() in {'true', 'yes', 'on'}:
+            coerced_value = True
+        elif value.lower() in {'false', 'no', 'off'}:
+            coerced_value = False
+        elif value.isdigit():
+            coerced_value = int(value)
+        elif value.replace('.', '', 1).isdigit():
+            coerced_value = float(value)
 
+    value = coerced_value
     _set_nested(user_config, key, value)
     # Normalize the api_base → base_url alias at set-time too (issue #8919),
     # so a fresh `hermes config set model.api_base ...` lands on the canonical

hermes_cli/kanban_db.py

@@ -135,6 +135,7 @@ BLOCK_RECURRENCE_LIMIT = 2
 VALID_WORKSPACE_KINDS = {"scratch", "worktree", "dir"}
 KNOWN_TOOLSET_NAMES = frozenset(name.casefold() for name in get_toolset_names())
 _IS_WINDOWS = sys.platform == "win32"
+KANBAN_ATTACHMENT_MAX_BYTES = 25 * 1024 * 1024
 
 
 def _fire_kanban_lifecycle_hook(event: str, task_id: str, **fields: Any) -> None:
@@ -3975,6 +3976,10 @@ class HallucinatedCardsError(ValueError):
         )
 
 
+class ArtifactPreservationError(RuntimeError):
+    """Raised when a declared scratch deliverable cannot be preserved."""
+
+
 def complete_task(
     conn: sqlite3.Connection,
     task_id: str,
@@ -4042,6 +4047,9 @@ def complete_task(
     else:
         verified_cards = []
 
+    metadata = _merge_completion_prose_artifacts(
+        conn, task_id, metadata, summary=summary, result=result,
+    )
     with write_txn(conn):
         if expected_run_id is None:
             cur = conn.execute(
@@ -4080,6 +4088,18 @@ def complete_task(
             )
         if cur.rowcount != 1:
             return False
+        if isinstance(metadata, dict):
+            _persist_scratch_completion_artifacts(conn, task_id, metadata)
+            for stored_path in metadata.pop("_staged_artifacts", []):
+                path = Path(stored_path)
+                _insert_completion_attachment(
+                    conn,
+                    task_id,
+                    filename=path.name,
+                    stored_path=str(path),
+                    size=path.stat().st_size,
+                    created_at=now,
+                )
         run_id = _end_run(
             conn, task_id,
             outcome="completed", status="done",
@@ -4174,6 +4194,256 @@ def complete_task(
 # Workspace / tmux cleanup
 # ---------------------------------------------------------------------------
 
+
+def _merge_completion_prose_artifacts(
+    conn: sqlite3.Connection,
+    task_id: str,
+    metadata: Optional[dict],
+    *,
+    summary: Optional[str],
+    result: Optional[str],
+) -> Optional[dict]:
+    """Promote existing scratch files named in legacy completion prose.
+
+    ``artifacts=[...]`` is preferred. Older workers only wrote an absolute
+    deliverable path in ``summary``/``result``; discover it while scratch still
+    exists so cleanup cannot erase the file the user was promised.
+    """
+    row = conn.execute(
+        "SELECT workspace_kind, workspace_path FROM tasks WHERE id = ?",
+        (task_id,),
+    ).fetchone()
+    if not row or row["workspace_kind"] != "scratch" or not row["workspace_path"]:
+        return metadata
+    workspace = Path(row["workspace_path"]).expanduser()
+    if not _is_managed_scratch_path(workspace):
+        return metadata
+    text = "\n".join(part for part in (summary, result) if part)
+    if not text:
+        return metadata
+    prefix = re.escape(str(workspace))
+    discovered: list[str] = []
+    for match in re.finditer(prefix + r"(?:[/\\][^\s`\"'<>]+)", text):
+        raw = match.group(0).rstrip(".,;:!?)]}")
+        candidate = Path(raw)
+        if candidate.is_file():
+            discovered.append(str(candidate))
+    if not discovered:
+        return metadata
+    updated = dict(metadata) if isinstance(metadata, dict) else {}
+    existing = updated.get("artifacts")
+    merged = list(existing) if isinstance(existing, (list, tuple)) else []
+    seen = {str(path) for path in merged}
+    for path in discovered:
+        if path not in seen:
+            merged.append(path)
+            seen.add(path)
+    updated["artifacts"] = merged
+    return updated
+
+
+def _persist_scratch_completion_artifacts(
+    conn: sqlite3.Connection,
+    task_id: str,
+    metadata: dict,
+) -> None:
+    """Copy scratch-workspace completion artifacts before cleanup removes them."""
+    raw_artifacts = metadata.get("artifacts")
+    if not isinstance(raw_artifacts, (list, tuple)):
+        return
+
+    row = conn.execute(
+        "SELECT workspace_kind, workspace_path FROM tasks WHERE id = ?",
+        (task_id,),
+    ).fetchone()
+    if not row or row["workspace_kind"] != "scratch" or not row["workspace_path"]:
+        return
+
+    workspace = Path(row["workspace_path"]).expanduser()
+    is_managed, board = _managed_scratch_path_info(workspace)
+    if not is_managed:
+        return
+
+    try:
+        workspace_root = workspace.resolve()
+    except OSError:
+        return
+
+    attachment_dir = task_attachments_dir(task_id, board=board)
+    persisted: list[str] = []
+    used_destinations: set[Path] = set()
+    changed = False
+
+    def _discard_copies() -> None:
+        for copied in used_destinations:
+            try:
+                copied.unlink(missing_ok=True)
+            except OSError:
+                pass
+        try:
+            attachment_dir.rmdir()
+        except OSError:
+            pass
+
+    for item in raw_artifacts:
+        artifact = str(item).strip() if isinstance(item, str) else ""
+        if not artifact:
+            continue
+        src = Path(artifact).expanduser()
+        try:
+            resolved_src = src.resolve()
+        except OSError:
+            persisted.append(artifact)
+            continue
+
+        if not resolved_src.is_relative_to(workspace_root):
+            persisted.append(artifact)
+            continue
+
+        if not src.is_file():
+            _discard_copies()
+            raise ArtifactPreservationError(
+                f"declared scratch artifact is unavailable or not a regular file: {artifact}"
+            )
+
+        size = resolved_src.stat().st_size
+        if size > KANBAN_ATTACHMENT_MAX_BYTES:
+            _discard_copies()
+            raise ArtifactPreservationError(
+                f"declared scratch artifact exceeds the "
+                f"{KANBAN_ATTACHMENT_MAX_BYTES}-byte limit: {artifact}"
+            )
+
+        dest: Optional[Path] = None
+        try:
+            attachment_dir.mkdir(parents=True, exist_ok=True)
+            dest = _unique_attachment_path(attachment_dir, resolved_src.name, used_destinations)
+            with resolved_src.open("rb") as source_file, dest.open("xb") as destination_file:
+                copied = 0
+                while chunk := source_file.read(1024 * 1024):
+                    copied += len(chunk)
+                    if copied > KANBAN_ATTACHMENT_MAX_BYTES:
+                        raise ArtifactPreservationError(
+                            f"declared scratch artifact grew beyond the size limit: {artifact}"
+                        )
+                    destination_file.write(chunk)
+        except Exception as exc:
+            if dest is not None:
+                try:
+                    dest.unlink(missing_ok=True)
+                except OSError:
+                    pass
+            _discard_copies()
+            if isinstance(exc, ArtifactPreservationError):
+                raise
+            raise ArtifactPreservationError(
+                f"could not preserve declared scratch artifact {artifact}: {exc}"
+            ) from exc
+
+        used_destinations.add(dest)
+        persisted.append(str(dest.resolve()))
+        changed = True
+
+    if changed:
+        metadata["artifacts"] = persisted
+        metadata["_staged_artifacts"] = [
+            path for path in persisted if path.startswith(str(attachment_dir.resolve()))
+        ]
+
+
+def _insert_completion_attachment(
+    conn: sqlite3.Connection,
+    task_id: str,
+    *,
+    filename: str,
+    stored_path: str,
+    size: int,
+    created_at: int,
+) -> None:
+    """Record a worker-produced artifact in the existing attachment table."""
+    conn.execute(
+        "INSERT INTO task_attachments "
+        "(task_id, filename, stored_path, content_type, size, uploaded_by, created_at) "
+        "VALUES (?, ?, ?, NULL, ?, 'kanban_complete', ?)",
+        (task_id, filename, stored_path, size, created_at),
+    )
+    _append_event(
+        conn,
+        task_id,
+        "attached",
+        {"filename": filename, "size": size, "by": "kanban_complete"},
+    )
+
+
+def _unique_attachment_path(directory: Path, filename: str, used: set[Path]) -> Path:
+    """Return a non-conflicting path under ``directory`` for ``filename``."""
+    safe_name = Path(filename).name or "artifact"
+    candidate = directory / safe_name
+    if candidate not in used and not candidate.exists():
+        return candidate
+
+    stem = Path(safe_name).stem or "artifact"
+    suffix = Path(safe_name).suffix
+    idx = 1
+    while True:
+        candidate = directory / f"{stem}_{idx}{suffix}"
+        if candidate not in used and not candidate.exists():
+            return candidate
+        idx += 1
+
+
+def _managed_scratch_path_info(p: Path) -> tuple[bool, Optional[str]]:
+    """Return whether *p* is managed scratch storage and the matching board."""
+    try:
+        p_abs = p.resolve(strict=False)
+    except OSError:
+        return False, None
+    roots: list[tuple[Path, Optional[str]]] = []
+    override = os.environ.get("HERMES_KANBAN_WORKSPACES_ROOT", "").strip()
+    if override:
+        try:
+            roots.append((Path(override).expanduser().resolve(strict=False), None))
+        except OSError:
+            pass
+    try:
+        home = kanban_home()
+    except OSError:
+        home = None
+    if home is not None:
+        try:
+            roots.append(((home / "kanban" / "workspaces").resolve(strict=False), DEFAULT_BOARD))
+        except OSError:
+            pass
+        try:
+            boards_parent = (home / "kanban" / "boards").resolve(strict=False)
+        except OSError:
+            boards_parent = None
+        if boards_parent is not None:
+            try:
+                entries = list(boards_parent.iterdir())
+            except OSError:
+                entries = []
+            for entry in entries:
+                try:
+                    if not entry.is_dir():
+                        continue
+                except OSError:
+                    continue
+                try:
+                    roots.append(((entry / "workspaces").resolve(strict=False), entry.name))
+                except OSError:
+                    continue
+    for root, board in roots:
+        if p_abs == root:
+            continue
+        try:
+            if p_abs.is_relative_to(root):
+                return True, board
+        except ValueError:
+            continue
+    return False, None
+
+
 def _is_managed_scratch_path(p: Path) -> bool:
     """Return True iff *p* is a strict descendant of a kanban-managed scratch root.
 
@@ -4199,54 +4469,8 @@ def _is_managed_scratch_path(p: Path) -> bool:
     real source tree can otherwise pair with ``workspace_kind='scratch'`` and
     cause task completion to delete user data (#28818).
     """
-    try:
-        p_abs = p.resolve(strict=False)
-    except OSError:
-        return False
-    roots: list[Path] = []
-    override = os.environ.get("HERMES_KANBAN_WORKSPACES_ROOT", "").strip()
-    if override:
-        try:
-            roots.append(Path(override).expanduser().resolve(strict=False))
-        except OSError:
-            pass
-    try:
-        home = kanban_home()
-    except OSError:
-        home = None
-    if home is not None:
-        try:
-            roots.append((home / "kanban" / "workspaces").resolve(strict=False))
-        except OSError:
-            pass
-        try:
-            boards_parent = (home / "kanban" / "boards").resolve(strict=False)
-        except OSError:
-            boards_parent = None
-        if boards_parent is not None:
-            try:
-                entries = list(boards_parent.iterdir())
-            except OSError:
-                entries = []
-            for entry in entries:
-                try:
-                    if not entry.is_dir():
-                        continue
-                except OSError:
-                    continue
-                try:
-                    roots.append((entry / "workspaces").resolve(strict=False))
-                except OSError:
-                    continue
-    for root in roots:
-        if p_abs == root:
-            continue
-        try:
-            if p_abs.is_relative_to(root):
-                return True
-        except ValueError:
-            continue
-    return False
+    is_managed, _board = _managed_scratch_path_info(p)
+    return is_managed
 
 
 def _cleanup_workspace(conn: sqlite3.Connection, task_id: str) -> None:
@@ -7822,6 +8046,13 @@ def _default_spawn(
         "chat",
         "-q", prompt,
     ])
+    if task.goal_mode:
+        # Goal-mode workers must take the fully-quiet single-query path:
+        # the kanban goal-loop hook (_run_kanban_goal_loop_q) only runs in
+        # cli.py's quiet branch. Without -Q the worker gets exactly one
+        # turn, prints text, exits rc=0, and the dispatcher records a
+        # protocol violation (incident 2026-06-09 t_d9cbe312).
+        cmd.append("-Q")
     # Redirect output to a per-task log under <board-root>/logs/.
     # Anchored at the board root (not the shared kanban root), so
     # `hermes kanban log` on a specific board reads its own file and

hermes_cli/oneshot.py

@@ -150,6 +150,12 @@ def _write_usage_file(path: Optional[str], result: dict, failure: Optional[str]
             "session_id": result.get("session_id"),
             "completed": result.get("completed"),
             "failed": bool(result.get("failed")) or failure is not None,
+            # Billing-audit field: the service tier this run REQUESTED via
+            # request_overrides.extra_body (e.g. OpenAI "flex"). None when
+            # unset. Lets batch pipelines verify the tier they think they're
+            # paying for actually went out on the wire (July 2026 incident:
+            # a config-matching bug silently dropped flex -> 2.3x billing).
+            "service_tier": result.get("service_tier"),
         }
         if failure is not None:
             report["failure"] = failure

hermes_cli/plugins.py

@@ -172,17 +172,19 @@ VALID_HOOKS: Set[str] = {
     # Kwargs: event: MessageEvent, gateway: GatewayRunner, session_store.
     "pre_gateway_dispatch",
     # Approval lifecycle hooks. Fired by tools/approval.py when a dangerous
-    # command needs user approval -- fires BOTH for CLI-interactive prompts
-    # and for gateway/ACP approvals (Telegram, Discord, Slack, TUI, etc.).
+    # command needs an approval decision -- fires for CLI-interactive prompts,
+    # gateway/ACP approvals, and smart-mode auxiliary-LLM decisions.
     # Observers only: return values are ignored. Plugins cannot veto or
     # pre-answer an approval from these hooks (use pre_tool_call to block
     # a tool before it reaches approval).
     #
     # Kwargs for pre_approval_request:
     #   command: str, description: str, pattern_key: str, pattern_keys: list[str],
-    #   session_key: str, surface: "cli" | "gateway"
+    #   session_key: str, surface: "cli" | "gateway" | "smart"
     # Kwargs for post_approval_response: same as above plus
     #   choice: "once" | "session" | "always" | "deny" | "timeout"
+    #           | "smart_approve" | "smart_deny"
+    #   decided_by: "aux_llm"  -- only on surface="smart"
     "pre_approval_request",
     "post_approval_response",
     # Kanban task lifecycle hooks. Fired by hermes_cli.kanban_db when a task

hermes_cli/tips.py

@@ -433,7 +433,7 @@ TIPS = [
     "hermes chat --source tool tags programmatic chats so they don't clutter hermes sessions list.",
     'hermes dump --show-keys includes redacted API key fingerprints for deeper support debugging.',
     'hermes sessions rename <ID> "new title" renames any past session; hermes sessions delete <ID> removes one.',
-    'hermes import restores a session export or profile archive produced by sessions export or profile export.',
+    'hermes import restores a full Hermes backup zip; session JSON/JSONL exports import from the dashboard Sessions page.',
     'hermes fallback manages the fallback_model chain interactively — no hand-editing config.yaml.',
     'hermes pairing rotates the DM pairing token — the first messager after rotation claims access to the bot.',
     'hermes setup walks first-time users through provider, keys, and platform wiring in one interactive flow.',

hermes_cli/web_server.py

@@ -9569,6 +9569,34 @@ class BulkDeleteSessions(BaseModel):
     profile: Optional[str] = None
 
 
+class SessionImport(BaseModel):
+    sessions: List[Dict[str, Any]]
+    profile: Optional[str] = None
+
+
+# Keep the dashboard import endpoint stream-safe: FastAPI otherwise parses and
+# buffers an arbitrarily large JSON body before SessionDB can enforce its own
+# per-session and transaction-work limits.
+_SESSION_IMPORT_MAX_BYTES = 25 * 1024 * 1024
+
+
+async def _read_session_import_body(request: Request) -> bytes:
+    body = bytearray()
+    async for chunk in request.stream():
+        if len(body) + len(chunk) > _SESSION_IMPORT_MAX_BYTES:
+            raise HTTPException(status_code=413, detail="Session import payload is too large")
+        body.extend(chunk)
+    return bytes(body)
+
+
+def _import_sessions_for_profile(profile: Optional[str], sessions: List[Dict[str, Any]]) -> Dict[str, Any]:
+    db = _open_session_db_for_profile(profile)
+    try:
+        return db.import_sessions(sessions)
+    finally:
+        db.close()
+
+
 @app.post("/api/sessions/bulk-delete")
 async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions):
     """Delete every session in ``body.ids`` in a single DB transaction.
@@ -9619,6 +9647,32 @@ async def bulk_delete_sessions_endpoint(body: BulkDeleteSessions):
         db.close()
 
 
+@app.post("/api/sessions/import")
+async def import_sessions_endpoint(request: Request):
+    """Import one or more sessions exported from the dashboard or CLI.
+
+    This is intentionally separate from ``/api/ops/import``: that endpoint
+    restores a whole Hermes backup archive, while this endpoint is scoped to
+    session rows/messages and is safe to use from the Sessions page.
+    """
+    try:
+        raw_body = await _read_session_import_body(request)
+        body = SessionImport.model_validate_json(raw_body)
+    except HTTPException:
+        raise
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail="Invalid session import payload") from exc
+
+    try:
+        result = await asyncio.to_thread(_import_sessions_for_profile, body.profile, body.sessions)
+    except ValueError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    if not result.get("ok", False):
+        raise HTTPException(status_code=400, detail=result)
+    return result
+
+
 @app.get("/api/sessions/empty/count")
 async def count_empty_sessions_endpoint(profile: Optional[str] = None):
     """Return the number of empty, ended, non-archived sessions.

hermes_state.py

@@ -140,7 +140,7 @@ T = TypeVar("T")
 
 DEFAULT_DB_PATH = get_hermes_home() / "state.db"
 
-SCHEMA_VERSION = 20
+SCHEMA_VERSION = 21
 
 # Cap on user-controlled FTS5 query input before regex/sanitizer processing.
 # Search queries do not need to be arbitrarily large, and bounding them keeps
@@ -827,6 +827,27 @@ CREATE TABLE IF NOT EXISTS compression_locks (
     expires_at REAL NOT NULL
 );
 
+CREATE TABLE IF NOT EXISTS async_delegations (
+    delegation_id TEXT PRIMARY KEY,
+    origin_session TEXT NOT NULL,
+    origin_ui_session_id TEXT NOT NULL DEFAULT '',
+    parent_session_id TEXT,
+    state TEXT NOT NULL,
+    dispatched_at REAL NOT NULL,
+    completed_at REAL,
+    updated_at REAL NOT NULL,
+    event_json TEXT,
+    result_json TEXT,
+    delivery_state TEXT NOT NULL DEFAULT 'pending',
+    delivery_attempts INTEGER NOT NULL DEFAULT 0,
+    delivered_at REAL,
+    owner_pid INTEGER,
+    owner_started_at INTEGER,
+    task_json TEXT,
+    delivery_claim TEXT,
+    delivery_claimed_at REAL
+);
+
 CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
 CREATE INDEX IF NOT EXISTS idx_sessions_source_id ON sessions(source, id);
 CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
@@ -835,6 +856,8 @@ CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestam
 CREATE INDEX IF NOT EXISTS idx_compression_locks_expires ON compression_locks(expires_at);
 CREATE INDEX IF NOT EXISTS idx_session_model_usage_session ON session_model_usage(session_id);
 CREATE INDEX IF NOT EXISTS idx_session_model_usage_model ON session_model_usage(model);
+CREATE INDEX IF NOT EXISTS idx_async_delegations_delivery
+    ON async_delegations(delivery_state, completed_at);
 """
 
 # Indexes that reference columns added in later schema versions must be
@@ -942,6 +965,15 @@ class SessionDB:
     # pays almost nothing; the cadence is deliberately coarse so the one-off
     # merge cost is amortised far below the checkpoint cadence.
     _OPTIMIZE_EVERY_N_WRITES = 1000
+    # Session imports intentionally use a lower cap than exports: import holds
+    # one BEGIN IMMEDIATE transaction, so bounded batches avoid starving live
+    # gateway/CLI writers. The dashboard accepts one exported JSON/JSONL file
+    # at a time, so these still cover normal history restores.
+    _IMPORT_MAX_SESSIONS = 500
+    _IMPORT_MAX_MESSAGES_PER_SESSION = 10_000
+    _IMPORT_MAX_TOTAL_MESSAGES = 50_000
+    _IMPORT_MAX_SESSION_BYTES = 5 * 1024 * 1024
+    _IMPORT_MAX_TOTAL_BYTES = 25 * 1024 * 1024
 
     def __init__(self, db_path: Path = None, read_only: bool = False):
         self.db_path = db_path or DEFAULT_DB_PATH
@@ -5278,13 +5310,16 @@ class SessionDB:
             return [session_id] if session else []
 
         root = session
+        ancestors = {root["id"]}
         while self._is_compression_child_row(root):
             parent = self.get_session(root["parent_session_id"])
-            if not parent:
+            if not parent or parent["id"] in ancestors:
                 break
             root = parent
+            ancestors.add(root["id"])
 
         lineage = [root["id"]]
+        seen = {root["id"]}
         current = root
         while current.get("end_reason") == "compression":
             with self._lock:
@@ -5302,9 +5337,10 @@ class SessionDB:
                 if not self._is_branch_child_row(candidate):
                     next_child = candidate
                     break
-            if not next_child:
+            if not next_child or next_child["id"] in seen:
                 break
             lineage.append(next_child["id"])
+            seen.add(next_child["id"])
             current = next_child
             if current["id"] == session_id:
                 # Continue to include later compression tips only when the
@@ -5352,6 +5388,403 @@ class SessionDB:
             results.append({**session, "messages": messages})
         return results
 
+    @staticmethod
+    def _import_text_or_none(value: Any, field: str) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return value
+        raise ValueError(f"{field} must be a string")
+
+    @staticmethod
+    def _import_json_object_or_none(value: Any, field: str) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            try:
+                parsed = json.loads(value)
+            except json.JSONDecodeError as exc:
+                raise ValueError(f"{field} must be valid JSON") from exc
+            if not isinstance(parsed, dict):
+                raise ValueError(f"{field} must be a JSON object")
+            return value
+        if not isinstance(value, dict):
+            raise ValueError(f"{field} must be a JSON object")
+        try:
+            return json.dumps(value)
+        except (TypeError, ValueError) as exc:
+            raise ValueError(f"{field} must be JSON serializable") from exc
+
+    @staticmethod
+    def _float_or_none(value: Any) -> Optional[float]:
+        if value is None:
+            return None
+        try:
+            return float(value)
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _import_int_or_none(value: Any, field: str) -> Optional[int]:
+        if value is None:
+            return None
+        try:
+            return int(value)
+        except (TypeError, ValueError) as exc:
+            raise ValueError(f"{field} must be an integer") from exc
+
+    @staticmethod
+    def _int_or_default(value: Any, default: int = 0) -> int:
+        if value is None:
+            return default
+        try:
+            return int(value)
+        except (TypeError, ValueError):
+            return default
+
+    @staticmethod
+    def _reasoning_json_value(value: Any) -> Any:
+        if not isinstance(value, str):
+            return value
+        try:
+            return json.loads(value)
+        except (json.JSONDecodeError, TypeError):
+            return value
+
+    @staticmethod
+    def _import_error(index: int, session_id: str, error: str) -> Dict[str, Any]:
+        item: Dict[str, Any] = {"index": index, "error": error}
+        if session_id:
+            item["session_id"] = session_id
+        return item
+
+    def import_sessions(self, sessions: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Import sessions exported by :meth:`export_session` or ``export_all``.
+
+        Existing session IDs are skipped. Imported child sessions keep their
+        parent only when that parent already exists or is included in the same
+        import payload; otherwise the child is detached so partial imports don't
+        fail foreign-key validation. Gateway routing, handoff, rewind, and other
+        live runtime state are intentionally reset: this restores conversation
+        history, not ownership of a live channel or process.
+        """
+        if not isinstance(sessions, list):
+            raise ValueError("sessions must be a list")
+        if len(sessions) > self._IMPORT_MAX_SESSIONS:
+            raise ValueError(
+                f"sessions must contain at most {self._IMPORT_MAX_SESSIONS} entries"
+            )
+
+        normalized: List[Dict[str, Any]] = []
+        errors: List[Dict[str, Any]] = []
+        seen_ids: set[str] = set()
+        total_messages = 0
+        total_bytes = 0
+        session_text_fields = (
+            "source",
+            "user_id",
+            "model",
+            "system_prompt",
+            "end_reason",
+            "cwd",
+            "git_branch",
+            "git_repo_root",
+            "billing_provider",
+            "billing_base_url",
+            "billing_mode",
+            "cost_status",
+            "cost_source",
+            "pricing_version",
+            "title",
+        )
+        message_text_fields = (
+            "role",
+            "tool_call_id",
+            "tool_name",
+            "effect_disposition",
+            "finish_reason",
+            "reasoning",
+            "reasoning_content",
+            "platform_message_id",
+            "message_id",
+        )
+
+        for index, raw in enumerate(sessions):
+            if not isinstance(raw, dict):
+                errors.append(self._import_error(index, "", "session must be an object"))
+                continue
+            session_id = str(raw.get("id") or "").strip()
+            if not session_id:
+                errors.append(self._import_error(index, "", "session id is required"))
+                continue
+            if session_id in seen_ids:
+                errors.append(self._import_error(index, session_id, "duplicate session id"))
+                continue
+            messages = raw.get("messages") or []
+            if not isinstance(messages, list):
+                errors.append(self._import_error(index, session_id, "messages must be a list"))
+                continue
+            if len(messages) > self._IMPORT_MAX_MESSAGES_PER_SESSION:
+                errors.append(
+                    self._import_error(
+                        index,
+                        session_id,
+                        "messages exceeds the per-session import limit",
+                    )
+                )
+                continue
+            if any(not isinstance(msg, dict) for msg in messages):
+                errors.append(
+                    self._import_error(
+                        index,
+                        session_id,
+                        "messages must contain only objects",
+                    )
+                )
+                continue
+
+            try:
+                session_bytes = len(
+                    json.dumps(raw, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+                )
+            except (TypeError, ValueError):
+                errors.append(
+                    self._import_error(index, session_id, "session must be JSON serializable")
+                )
+                continue
+            if session_bytes > self._IMPORT_MAX_SESSION_BYTES:
+                errors.append(
+                    self._import_error(index, session_id, "session exceeds the import size limit")
+                )
+                continue
+            total_bytes += session_bytes
+            if total_bytes > self._IMPORT_MAX_TOTAL_BYTES:
+                errors.append(
+                    self._import_error(index, session_id, "import exceeds the total size limit")
+                )
+                continue
+
+            try:
+                clean_session = dict(raw)
+                clean_session["id"] = session_id
+                clean_session["model_config"] = self._import_json_object_or_none(
+                    clean_session.get("model_config"), "model_config"
+                )
+                clean_session["parent_session_id"] = self._import_text_or_none(
+                    clean_session.get("parent_session_id"), "parent_session_id"
+                )
+                for field in session_text_fields:
+                    clean_session[field] = self._import_text_or_none(
+                        clean_session.get(field), field
+                    )
+
+                clean_messages: List[Dict[str, Any]] = []
+                for message_index, message in enumerate(messages):
+                    clean_message = dict(message)
+                    role = clean_message.get("role")
+                    if not isinstance(role, str) or not role:
+                        raise ValueError(f"messages[{message_index}].role must be a non-empty string")
+                    for field in message_text_fields:
+                        if field == "role":
+                            continue
+                        clean_message[field] = self._import_text_or_none(
+                            clean_message.get(field), field
+                        )
+                    clean_message["token_count"] = self._import_int_or_none(
+                        clean_message.get("token_count"), "token_count"
+                    )
+                    clean_messages.append(clean_message)
+            except ValueError as exc:
+                errors.append(self._import_error(index, session_id, str(exc)))
+                continue
+
+            total_messages += len(clean_messages)
+            if total_messages > self._IMPORT_MAX_TOTAL_MESSAGES:
+                errors.append(
+                    self._import_error(
+                        index,
+                        session_id,
+                        "messages exceeds the total import limit",
+                    )
+                )
+                continue
+            seen_ids.add(session_id)
+            normalized.append(
+                {"index": index, "session": clean_session, "messages": clean_messages}
+            )
+
+        if errors:
+            return {
+                "ok": False,
+                "imported": 0,
+                "skipped": 0,
+                "detached": 0,
+                "errors": errors,
+            }
+
+        def _do(conn):
+            imported_ids: List[str] = []
+            skipped_ids: List[str] = []
+            parent_updates: List[tuple[str, str]] = []
+            detached = 0
+
+            for item in normalized:
+                raw = item["session"]
+                messages = item["messages"]
+                session_id = str(raw.get("id") or "").strip()
+                exists = conn.execute(
+                    "SELECT 1 FROM sessions WHERE id = ? LIMIT 1",
+                    (session_id,),
+                ).fetchone()
+                if exists:
+                    skipped_ids.append(session_id)
+                    continue
+
+                started_at = self._float_or_none(raw.get("started_at"))
+                if started_at is None:
+                    started_at = time.time()
+                archived = 1 if raw.get("archived") else 0
+
+                conn.execute(
+                    """INSERT INTO sessions (
+                           id, source, user_id, model, model_config, system_prompt,
+                           parent_session_id, started_at, ended_at, end_reason,
+                           message_count, tool_call_count, input_tokens, output_tokens,
+                           cache_read_tokens, cache_write_tokens, reasoning_tokens,
+                           cwd, git_branch, git_repo_root,
+                           billing_provider, billing_base_url, billing_mode,
+                           estimated_cost_usd, actual_cost_usd, cost_status, cost_source,
+                           pricing_version, title, api_call_count, archived
+                       )
+                       VALUES (
+                           :id, :source, :user_id, :model, :model_config,
+                           :system_prompt, NULL, :started_at, :ended_at,
+                           :end_reason, 0, 0, :input_tokens, :output_tokens,
+                           :cache_read_tokens, :cache_write_tokens,
+                           :reasoning_tokens, :cwd, :git_branch, :git_repo_root,
+                           :billing_provider, :billing_base_url, :billing_mode,
+                           :estimated_cost_usd, :actual_cost_usd, :cost_status,
+                           :cost_source, :pricing_version, :title,
+                           :api_call_count, :archived
+                       )""",
+                    {
+                        "id": session_id,
+                        "source": str(raw.get("source") or "import"),
+                        "user_id": raw.get("user_id"),
+                        "model": raw.get("model"),
+                        "model_config": raw.get("model_config"),
+                        "system_prompt": raw.get("system_prompt"),
+                        "started_at": started_at,
+                        "ended_at": self._float_or_none(raw.get("ended_at")),
+                        "end_reason": raw.get("end_reason"),
+                        "input_tokens": self._int_or_default(raw.get("input_tokens")),
+                        "output_tokens": self._int_or_default(raw.get("output_tokens")),
+                        "cache_read_tokens": self._int_or_default(
+                            raw.get("cache_read_tokens")
+                        ),
+                        "cache_write_tokens": self._int_or_default(
+                            raw.get("cache_write_tokens")
+                        ),
+                        "reasoning_tokens": self._int_or_default(
+                            raw.get("reasoning_tokens")
+                        ),
+                        "cwd": raw.get("cwd"),
+                        "git_branch": raw.get("git_branch"),
+                        "git_repo_root": raw.get("git_repo_root"),
+                        "billing_provider": raw.get("billing_provider"),
+                        "billing_base_url": raw.get("billing_base_url"),
+                        "billing_mode": raw.get("billing_mode"),
+                        "estimated_cost_usd": self._float_or_none(
+                            raw.get("estimated_cost_usd")
+                        ),
+                        "actual_cost_usd": self._float_or_none(
+                            raw.get("actual_cost_usd")
+                        ),
+                        "cost_status": raw.get("cost_status"),
+                        "cost_source": raw.get("cost_source"),
+                        "pricing_version": raw.get("pricing_version"),
+                        "title": raw.get("title"),
+                        "api_call_count": self._int_or_default(raw.get("api_call_count")),
+                        "archived": archived,
+                    },
+                )
+
+                sanitized_messages: List[Dict[str, Any]] = []
+                for msg in messages:
+                    clean = dict(msg)
+                    for key in (
+                        "reasoning_details",
+                        "codex_reasoning_items",
+                        "codex_message_items",
+                    ):
+                        clean[key] = self._reasoning_json_value(clean.get(key))
+                    sanitized_messages.append(clean)
+
+                total_messages, total_tool_calls = self._insert_message_rows(
+                    conn,
+                    session_id,
+                    sanitized_messages,
+                )
+                conn.execute(
+                    "UPDATE sessions SET message_count = ?, tool_call_count = ? WHERE id = ?",
+                    (total_messages, total_tool_calls, session_id),
+                )
+
+                parent_id = str(raw.get("parent_session_id") or "").strip()
+                if parent_id:
+                    parent_updates.append((session_id, parent_id))
+                imported_ids.append(session_id)
+
+            parent_by_child = dict(parent_updates)
+
+            def _would_create_cycle(session_id: str, parent_id: str) -> bool:
+                seen = {session_id}
+                current = parent_id
+                while current:
+                    if current in seen:
+                        return True
+                    seen.add(current)
+                    if current in parent_by_child:
+                        current = parent_by_child[current]
+                        continue
+                    row = conn.execute(
+                        "SELECT parent_session_id FROM sessions WHERE id = ? LIMIT 1",
+                        (current,),
+                    ).fetchone()
+                    if row is None:
+                        return False
+                    current = row["parent_session_id"]
+                return False
+
+            for session_id, parent_id in parent_updates:
+                parent_exists = conn.execute(
+                    "SELECT 1 FROM sessions WHERE id = ? LIMIT 1",
+                    (parent_id,),
+                ).fetchone()
+                if parent_exists and not _would_create_cycle(session_id, parent_id):
+                    conn.execute(
+                        "UPDATE sessions SET parent_session_id = ? WHERE id = ?",
+                        (parent_id, session_id),
+                    )
+                else:
+                    # Drop only the closing edge. Later entries can still attach
+                    # to this now-root session, preserving the acyclic portion
+                    # of a malformed imported lineage.
+                    parent_by_child.pop(session_id, None)
+                    detached += 1
+
+            return {
+                "ok": True,
+                "imported": len(imported_ids),
+                "skipped": len(skipped_ids),
+                "detached": detached,
+                "imported_ids": imported_ids,
+                "skipped_ids": skipped_ids,
+                "errors": [],
+            }
+
+        return self._execute_write(_do)
+
     def clear_messages(self, session_id: str) -> None:
         """Delete all messages for a session and reset its counters."""
         def _do(conn):

locales/af.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]eenmalig  |  [s]sessie  |  [d]weier"
   prompt_long:     "      Keuse [o/s/a/D]: "
   prompt_short:    "      Keuse [o/s/D]: "
+  choose_smart_deny: "      [o]eenmalig  |  [d]weier"
+  prompt_smart_deny: "      Keuse [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Tyd verstreke - opdrag word geweier"
   allowed_once:    "      ✓ Eenmalig toegelaat"
   allowed_session: "      ✓ Vir hierdie sessie toegelaat"

locales/de.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]einmal  |  [s]sitzung  |  [d]ablehnen"
   prompt_long:     "      Auswahl [o/s/a/D]: "
   prompt_short:    "      Auswahl [o/s/D]: "
+  choose_smart_deny: "      [o]einmal  |  [d]ablehnen"
+  prompt_smart_deny: "      Auswahl [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Zeitüberschreitung – Befehl wird abgelehnt"
   allowed_once:    "      ✓ Einmalig erlaubt"
   allowed_session: "      ✓ Für diese Sitzung erlaubt"

locales/en.yaml

@@ -17,6 +17,10 @@ approval:
   choose_short:    "      [o]nce  |  [s]ession  |  [d]eny"
   prompt_long:     "      Choice [o/s/a/D]: "
   prompt_short:    "      Choice [o/s/D]: "
+  choose_smart_deny: "      [o]nce  |  [d]eny"
+  prompt_smart_deny: "      Choice [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Timeout - denying command"
   allowed_once:    "      ✓ Allowed once"
   allowed_session: "      ✓ Allowed for this session"

locales/es.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]una vez  |  [s]sesión  |  [d]denegar"
   prompt_long:     "      Opción [o/s/a/D]: "
   prompt_short:    "      Opción [o/s/D]: "
+  choose_smart_deny: "      [o]una vez  |  [d]denegar"
+  prompt_smart_deny: "      Opción [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Tiempo agotado — comando denegado"
   allowed_once:    "      ✓ Permitido una vez"
   allowed_session: "      ✓ Permitido en esta sesión"

locales/fr.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]ne fois  |  [s]ession  |  [r]efuser"
   prompt_long:     "      Choix [o/s/t/R] : "
   prompt_short:    "      Choix [o/s/R] : "
+  choose_smart_deny: "      [o]ne fois  |  [r]efuser"
+  prompt_smart_deny: "      Choix [o/R] : "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "r,refuser"
   timeout:         "      ⏱ Délai dépassé — commande refusée"
   allowed_once:    "      ✓ Autorisé une fois"
   allowed_session: "      ✓ Autorisé pour cette session"

locales/ga.yaml

@@ -11,6 +11,10 @@ approval:
   choose_short:    "      [o]uair amháin  |  [s]eisiún  |  [d]iúltaigh"
   prompt_long:     "      Rogha [o/s/a/D]: "
   prompt_short:    "      Rogha [o/s/D]: "
+  choose_smart_deny: "      [o]uair amháin  |  [d]iúltaigh"
+  prompt_smart_deny: "      Rogha [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Am istigh — ag diúltú don ordú"
   allowed_once:    "      ✓ Ceadaithe uair amháin"
   allowed_session: "      ✓ Ceadaithe don seisiún seo"

locales/hu.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]egyszer  |  [s]munkamenet  |  [d]elutasít"
   prompt_long:     "      Választás [o/s/a/D]: "
   prompt_short:    "      Választás [o/s/D]: "
+  choose_smart_deny: "      [o]egyszer  |  [d]elutasít"
+  prompt_smart_deny: "      Választás [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Időtúllépés - parancs elutasítva"
   allowed_once:    "      ✓ Egyszer engedélyezve"
   allowed_session: "      ✓ Engedélyezve ehhez a munkamenethez"

locales/it.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]una volta  |  [s]essione  |  [d]nega"
   prompt_long:     "      Scelta [o/s/a/D]: "
   prompt_short:    "      Scelta [o/s/D]: "
+  choose_smart_deny: "      [o]una volta  |  [d]nega"
+  prompt_smart_deny: "      Scelta [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Tempo scaduto — comando negato"
   allowed_once:    "      ✓ Consentito una volta"
   allowed_session: "      ✓ Consentito per questa sessione"

locales/ja.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]今回のみ  |  [s]セッション中  |  [d]拒否"
   prompt_long:     "      選択 [o/s/a/D]: "
   prompt_short:    "      選択 [o/s/D]: "
+  choose_smart_deny: "      [o]今回のみ  |  [d]拒否"
+  prompt_smart_deny: "      選択 [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ タイムアウト — コマンドを拒否しました"
   allowed_once:    "      ✓ 今回のみ許可"
   allowed_session: "      ✓ このセッション中は許可"

locales/ko.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]한 번  |  [s]세션  |  [d]거부"
   prompt_long:     "      선택 [o/s/a/D]: "
   prompt_short:    "      선택 [o/s/D]: "
+  choose_smart_deny: "      [o]한 번  |  [d]거부"
+  prompt_smart_deny: "      선택 [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ 시간 초과 - 명령을 거부합니다"
   allowed_once:    "      ✓ 한 번 허용됨"
   allowed_session: "      ✓ 이 세션에서 허용됨"

locales/pt.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]uma vez  |  [s]sessão  |  [d]negar"
   prompt_long:     "      Escolha [o/s/a/D]: "
   prompt_short:    "      Escolha [o/s/D]: "
+  choose_smart_deny: "      [o]uma vez  |  [d]negar"
+  prompt_smart_deny: "      Escolha [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Tempo esgotado — comando negado"
   allowed_once:    "      ✓ Permitido uma vez"
   allowed_session: "      ✓ Permitido nesta sessão"

locales/ru.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]один раз  |  [s]сеанс  |  [d]отклонить"
   prompt_long:     "      Выбор [o/s/a/D]: "
   prompt_short:    "      Выбор [o/s/D]: "
+  choose_smart_deny: "      [o]один раз  |  [d]отклонить"
+  prompt_smart_deny: "      Выбор [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Время ожидания истекло — команда отклонена"
   allowed_once:    "      ✓ Разрешено один раз"
   allowed_session: "      ✓ Разрешено для этого сеанса"

locales/tr.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [b]ir kez  |  [o]turum  |  [r]eddet"
   prompt_long:     "      Seçim [b/o/h/R]: "
   prompt_short:    "      Seçim [b/o/R]: "
+  choose_smart_deny: "      [b]ir kez  |  [r]eddet"
+  prompt_smart_deny: "      Seçim [b/R]: "
+  smart_deny_once_inputs: "b,bir kez"
+  smart_deny_deny_inputs: "r,reddet"
   timeout:         "      ⏱ Zaman aşımı — komut reddedildi"
   allowed_once:    "      ✓ Bir kez izin verildi"
   allowed_session: "      ✓ Bu oturum için izin verildi"

locales/uk.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]один раз  |  [s]сеанс  |  [d]відхилити"
   prompt_long:     "      Вибір [o/s/a/D]: "
   prompt_short:    "      Вибір [o/s/D]: "
+  choose_smart_deny: "      [o]один раз  |  [d]відхилити"
+  prompt_smart_deny: "      Вибір [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ Час очікування вичерпано — команду відхилено"
   allowed_once:    "      ✓ Дозволено один раз"
   allowed_session: "      ✓ Дозволено для цього сеансу"

locales/zh-hant.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]僅此一次  |  [s]本次工作階段  |  [d]拒絕"
   prompt_long:     "      選擇 [o/s/a/D]: "
   prompt_short:    "      選擇 [o/s/D]: "
+  choose_smart_deny: "      [o]僅此一次  |  [d]拒絕"
+  prompt_smart_deny: "      選擇 [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ 逾時 — 已拒絕指令"
   allowed_once:    "      ✓ 本次允許"
   allowed_session: "      ✓ 本次工作階段內允許"

locales/zh.yaml

@@ -7,6 +7,10 @@ approval:
   choose_short:    "      [o]仅此一次  |  [s]本次会话  |  [d]拒绝"
   prompt_long:     "      选择 [o/s/a/D]: "
   prompt_short:    "      选择 [o/s/D]: "
+  choose_smart_deny: "      [o]仅此一次  |  [d]拒绝"
+  prompt_smart_deny: "      选择 [o/D]: "
+  smart_deny_once_inputs: "o,once"
+  smart_deny_deny_inputs: "d,deny"
   timeout:         "      ⏱ 超时 — 已拒绝命令"
   allowed_once:    "      ✓ 本次允许"
   allowed_session: "      ✓ 本次会话内允许"

package-lock.json

@@ -1903,7 +1903,6 @@
       "os": [
         "aix"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1920,7 +1919,6 @@
       "os": [
         "android"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1937,7 +1935,6 @@
       "os": [
         "android"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1954,7 +1951,6 @@
       "os": [
         "android"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1971,7 +1967,6 @@
       "os": [
         "darwin"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -1988,7 +1983,6 @@
       "os": [
         "darwin"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2005,7 +1999,6 @@
       "os": [
         "freebsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2022,7 +2015,6 @@
       "os": [
         "freebsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2039,7 +2031,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2056,7 +2047,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2073,7 +2063,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2090,7 +2079,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2107,7 +2095,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2124,7 +2111,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2141,7 +2127,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2158,7 +2143,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2175,7 +2159,6 @@
       "os": [
         "linux"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2192,7 +2175,6 @@
       "os": [
         "netbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2209,7 +2191,6 @@
       "os": [
         "netbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2226,7 +2207,6 @@
       "os": [
         "openbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2243,7 +2223,6 @@
       "os": [
         "openbsd"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2260,7 +2239,6 @@
       "os": [
         "openharmony"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2277,7 +2255,6 @@
       "os": [
         "sunos"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2294,7 +2271,6 @@
       "os": [
         "win32"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2311,7 +2287,6 @@
       "os": [
         "win32"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }
@@ -2328,7 +2303,6 @@
       "os": [
         "win32"
       ],
-      "peer": true,
       "engines": {
         "node": ">=18"
       }

plugins/image_gen/openai-codex/__init__.py

@@ -40,6 +40,32 @@ from agent.image_gen_provider import (
 logger = logging.getLogger(__name__)
 
 
+class CodexImageGenerationUnsupportedError(RuntimeError):
+    """The active Codex account cannot use the hosted image tool."""
+
+
+_IMAGE_GENERATION_UNAVAILABLE_MESSAGE = (
+    "Image generation is not enabled for the current Codex account. "
+    "Switch the image provider to OpenAI API key, FAL, or xAI."
+)
+_IMAGE_GENERATION_UNSUPPORTED_ERROR = (
+    "Tool choice 'image_generation' not found in 'tools' parameter."
+)
+
+
+def _is_image_generation_unsupported_error(status_code: int, body: str) -> bool:
+    """Match only Codex's account-capability rejection for the image tool."""
+    if status_code != 400:
+        return False
+    try:
+        payload = json.loads(body)
+        error = payload.get("error") if isinstance(payload, dict) else None
+        message = error.get("message") if isinstance(error, dict) else None
+    except (TypeError, ValueError):
+        message = body
+    return isinstance(message, str) and message.strip() == _IMAGE_GENERATION_UNSUPPORTED_ERROR
+
+
 # ---------------------------------------------------------------------------
 # Model catalog — mirrors the ``openai`` plugin so the picker UX is identical.
 # ---------------------------------------------------------------------------
@@ -393,7 +419,13 @@ def _collect_image_b64(
                 response.raise_for_status()
             except httpx.HTTPStatusError as exc:
                 exc.response.read()
-                body = exc.response.text[:500]
+                full_body = exc.response.text
+                if _is_image_generation_unsupported_error(
+                    exc.response.status_code,
+                    full_body,
+                ):
+                    raise CodexImageGenerationUnsupportedError(full_body) from exc
+                body = full_body[:500]
                 raise RuntimeError(
                     f"Codex Responses API returned HTTP {exc.response.status_code}: {body}"
                 ) from exc
@@ -542,6 +574,19 @@ class OpenAICodexImageGenProvider(ImageGenProvider):
                 quality=meta["quality"],
                 input_images=input_images or None,
             )
+        except CodexImageGenerationUnsupportedError:
+            logger.debug(
+                "Codex account does not expose image generation",
+                exc_info=True,
+            )
+            return error_response(
+                error=_IMAGE_GENERATION_UNAVAILABLE_MESSAGE,
+                error_type="capability_unsupported",
+                provider="openai-codex",
+                model=tier_id,
+                prompt=prompt,
+                aspect_ratio=aspect,
+            )
         except Exception as exc:
             logger.debug("Codex image generation failed", exc_info=True)
             return error_response(

plugins/kanban/dashboard/dist/index.js

@@ -1093,6 +1093,7 @@
           taskId: selectedTaskId,
           boardSlug: board,
           onClose: function () { setSelectedTaskId(null); },
+          onOpenTask: setSelectedTaskId,
           onRefresh: loadBoard,
           renderMarkdown: renderMd,
           allTasks: boardData.columns.reduce(function (acc, c) { return acc.concat(c.tasks); }, []),
@@ -3222,6 +3223,10 @@
           onDeleteAttachment: handleDeleteAttachment,
           uploadBusy: uploadBusy,
           uploadErr: uploadErr,
+          onOpenTask: function (taskId) {
+            props.onClose();
+            if (props.onOpenTask) props.onOpenTask(taskId);
+          },
         }) : null,
         data ? h("div", { className: "hermes-kanban-drawer-comment-row" },
           h(Input, {
@@ -3358,6 +3363,7 @@
     const events = props.data.events || [];
     const attachments = props.data.attachments || [];
     const links = props.data.links || { parents: [], children: [] };
+    const childResults = props.data.child_results || [];
 
     return h("div", { className: "hermes-kanban-drawer-body" },
       h("div", { className: "hermes-kanban-drawer-title" },
@@ -3428,9 +3434,61 @@
         onAddChild: props.onAddChild,
         onRemoveChild: props.onRemoveChild,
       }),
-      t.result ? h("div", { className: "hermes-kanban-section" },
-        h("div", { className: "hermes-kanban-section-head" }, tx(i18n, "result", "Result")),
-        h(MarkdownBlock, { source: t.result, enabled: props.renderMarkdown }),
+      (function () {
+        var finalResult = t.result || t.latest_summary || null;
+        var isDone = t.status === "done";
+        var isParent = links.children.length > 0;
+        if (finalResult) {
+          var label = t.result
+            ? tx(i18n, "result", "Result")
+            : tx(i18n, "finalResult", "Final Result (run summary)");
+          return h("div", { className: "hermes-kanban-section" },
+            h("div", { className: "hermes-kanban-section-head" }, label),
+            h(MarkdownBlock, { source: finalResult, enabled: props.renderMarkdown }),
+          );
+        }
+        if (isDone && isParent) {
+          return h("div", { className: "hermes-kanban-section" },
+            h("div", { className: "hermes-kanban-section-head" }, tx(i18n, "result", "Result")),
+            h("div", { className: "hermes-kanban-done-no-result hermes-kanban-done-parent-note" },
+              tx(i18n, "doneParentNote",
+                "This card is an orchestrator / parent task. Review the child results section for the substantive work."),
+            ),
+          );
+        }
+        if (isDone) {
+          return h("div", { className: "hermes-kanban-section" },
+            h("div", { className: "hermes-kanban-section-head" }, tx(i18n, "result", "Result")),
+            h("div", { className: "hermes-kanban-done-no-result" },
+              tx(i18n, "doneNoResult",
+                "No final result was recorded. Check Run History, Logs, or Child Tasks for the worker output."),
+            ),
+          );
+        }
+        return null;
+      })(),
+      childResults.length > 0 ? h("div", { className: "hermes-kanban-section" },
+        h("div", { className: "hermes-kanban-section-head" },
+          `${tx(i18n, "childResults", "Child Results")} (${childResults.length})`),
+        childResults.map(function (child) {
+          var childResult = child.result || child.latest_summary || null;
+          return h("div", { key: child.id, className: "hermes-kanban-comment" },
+            h("div", { className: "hermes-kanban-comment-head" },
+              h("span", { className: "hermes-kanban-comment-author" },
+                `${child.id} · ${child.title || tx(i18n, "untitled", "(untitled)")}`),
+              h(Badge, { variant: "outline" }, child.status),
+              h("button", {
+                type: "button",
+                className: "hermes-kanban-diag-action-btn",
+                onClick: function () { if (props.onOpenTask) props.onOpenTask(child.id); },
+              }, tx(i18n, "open", "Open")),
+            ),
+            childResult
+              ? h(MarkdownBlock, { source: childResult, enabled: props.renderMarkdown })
+              : h("div", { className: "text-xs text-muted-foreground" },
+                  tx(i18n, "noChildResult", "No result recorded yet.")),
+          );
+        }),
       ) : null,
       h(AttachmentsSection, {
         attachments: attachments,

plugins/kanban/dashboard/plugin_api.py

@@ -546,6 +546,21 @@ def get_task(
         # a second round-trip. Cards on /board carry a 200-char preview.
         full_summary = kanban_db.latest_summary(conn, task_id)
         task_d = _task_dict(task, latest_summary=full_summary)
+        links = _links_for(conn, task_id)
+        child_ids = links["children"]
+        child_summaries = kanban_db.latest_summaries(conn, child_ids)
+        child_results = []
+        for child_id in child_ids:
+            child = kanban_db.get_task(conn, child_id)
+            if child is None:
+                continue
+            child_results.append({
+                "id": child.id,
+                "title": child.title,
+                "status": child.status,
+                "latest_summary": child_summaries.get(child.id),
+                "result": child.result,
+            })
         # Attach diagnostics so the drawer's Diagnostics section can
         # render recovery actions without a second round-trip.
         diags = _compute_task_diagnostics(conn, task_ids=[task_id])
@@ -558,7 +573,8 @@ def get_task(
             "comments": [_comment_dict(c) for c in kanban_db.list_comments(conn, task_id)],
             "events": [_event_dict(e) for e in kanban_db.list_events(conn, task_id)],
             "attachments": [_attachment_dict(a) for a in kanban_db.list_attachments(conn, task_id)],
-            "links": _links_for(conn, task_id),
+            "links": links,
+            "child_results": child_results,
             "runs": [
                 _run_dict(r)
                 for r in kanban_db.list_runs(
@@ -646,7 +662,7 @@ def create_task(payload: CreateTaskBody, board: Optional[str] = Query(None)):
 
 # Cap a single upload so a runaway request can't fill the disk. 25 MB
 # comfortably covers PDFs, images, and source docs — the kanban use case.
-_MAX_ATTACHMENT_BYTES = 25 * 1024 * 1024
+_MAX_ATTACHMENT_BYTES = kanban_db.KANBAN_ATTACHMENT_MAX_BYTES
 
 
 def _safe_attachment_name(raw: str) -> str:

plugins/platforms/discord/adapter.py

@@ -5571,6 +5571,8 @@ class DiscordAdapter(BasePlatformAdapter):
         self, chat_id: str, command: str, session_key: str,
         description: str = "dangerous command",
         metadata: Optional[dict] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """
         Send a button-based exec approval prompt for a dangerous command.
@@ -5606,6 +5608,8 @@ class DiscordAdapter(BasePlatformAdapter):
                 "Do you want Hermes to run this command?\n\n"
                 "**Requested command:**\n```bash\n"
             )
+            if smart_denied:
+                prompt_prefix += "**Smart DENY:** owner override applies to this one operation only.\n\n"
             mention_content = self._approval_mention_content()
             if mention_content:
                 prompt_prefix = f"{mention_content}\n{prompt_prefix}"
@@ -5642,6 +5646,8 @@ class DiscordAdapter(BasePlatformAdapter):
                 allowed_role_ids=self._allowed_role_ids,
                 require_admin=require_admin,
                 admin_user_ids=admin_user_ids,
+                allow_permanent=allow_permanent,
+                smart_denied=smart_denied,
             )
 
             send_kwargs: Dict[str, Any] = {"content": content, "embed": embed, "view": view}
@@ -6836,6 +6842,8 @@ def _define_discord_view_classes() -> None:
             allowed_role_ids: Optional[set] = None,
             require_admin: bool = False,
             admin_user_ids: Optional[set] = None,
+            allow_permanent: bool = True,
+            smart_denied: bool = False,
         ):
             super().__init__(timeout=_read_discord_prompt_timeout())
             self.session_key = session_key
@@ -6849,6 +6857,11 @@ def _define_discord_view_classes() -> None:
                 str(a).strip() for a in (admin_user_ids or set()) if str(a).strip()
             }
             self.resolved = False
+            if smart_denied:
+                self.remove_item(self.allow_session)
+                self.remove_item(self.allow_always)
+            elif not allow_permanent:
+                self.remove_item(self.allow_always)
 
         def _check_auth(self, interaction: discord.Interaction) -> bool:
             """Verify the user clicking is authorized.

plugins/platforms/feishu/adapter.py

@@ -1980,6 +1980,8 @@ class FeishuAdapter(BasePlatformAdapter):
         self, chat_id: str, command: str, session_key: str,
         description: str = "dangerous command",
         metadata: Optional[Dict[str, Any]] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """Send an interactive card with approval buttons.
 
@@ -2002,6 +2004,13 @@ class FeishuAdapter(BasePlatformAdapter):
                     "value": {"hermes_action": action_name, "approval_id": approval_id},
                 }
 
+            actions = [_btn("✅ Allow Once", "approve_once", "primary")]
+            if not smart_denied:
+                actions.append(_btn("✅ Session", "approve_session"))
+                if allow_permanent:
+                    actions.append(_btn("✅ Always", "approve_always"))
+            actions.append(_btn("❌ Deny", "deny", "danger"))
+            scope_note = "\n\n**Smart DENY:** owner override applies to this one operation only." if smart_denied else ""
             card = {
                 "config": {"wide_screen_mode": True},
                 "header": {
@@ -2011,16 +2020,11 @@ class FeishuAdapter(BasePlatformAdapter):
                 "elements": [
                     {
                         "tag": "markdown",
-                        "content": f"```\n{cmd_preview}\n```\n**Reason:** {description}",
+                        "content": f"```\n{cmd_preview}\n```\n**Reason:** {description}{scope_note}",
                     },
                     {
                         "tag": "action",
-                        "actions": [
-                            _btn("✅ Allow Once", "approve_once", "primary"),
-                            _btn("✅ Session", "approve_session"),
-                            _btn("✅ Always", "approve_always"),
-                            _btn("❌ Deny", "deny", "danger"),
-                        ],
+                        "actions": actions,
                     },
                 ],
             }

plugins/platforms/matrix/adapter.py

@@ -2000,6 +2000,8 @@ class MatrixAdapter(BasePlatformAdapter):
         session_key: str,
         description: str = "dangerous command",
         metadata: Optional[dict] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """Send a reaction-based exec approval prompt for Matrix."""
         if not self._client:
@@ -2007,12 +2009,18 @@ class MatrixAdapter(BasePlatformAdapter):
 
         requester_user_id = str((metadata or {}).get("requester_user_id") or "") or None
         cmd_preview = command[:2000] + "..." if len(command) > 2000 else command
+        scope_choices = ""
+        if smart_denied:
+            scope_choices = "Smart DENY: owner override applies to this one operation only.\n"
+        else:
+            scope_choices = "Reply `!approve session` to approve this pattern for the session, "
+            if allow_permanent:
+                scope_choices += "`!approve always` to approve permanently, "
         text = (
             "⚠️ **Dangerous command requires approval**\n"
             f"```\n{cmd_preview}\n```\n"
             f"Reason: {description}\n\n"
-            "Reply `!approve` to execute, `!approve session` to approve this pattern for the session, "
-            "`!approve always` to approve permanently, or `!deny` to cancel.\n\n"
+            f"{scope_choices}Reply `!approve` to execute once, or `!deny` to cancel.\n\n"
             "You can also click the reaction to approve:\n"
             "✅ = approve\n"
             "❎ = deny"
@@ -2035,7 +2043,8 @@ class MatrixAdapter(BasePlatformAdapter):
         self._approval_prompts_by_event[result.message_id] = prompt
         self._approval_prompt_by_session[session_key] = result.message_id
 
-        for emoji in ("✅", "♾️", "❌"):
+        reactions = ("✅", "❌") if smart_denied or not allow_permanent else ("✅", "♾️", "❌")
+        for emoji in reactions:
             try:
                 reaction_result = await self._send_reaction(chat_id, result.message_id, emoji)
                 # Save the bot's reaction event_id for later cleanup

plugins/platforms/slack/adapter.py

@@ -3244,6 +3244,8 @@ class SlackAdapter(BasePlatformAdapter):
         session_key: str,
         description: str = "dangerous command",
         metadata: Optional[Dict[str, Any]] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """Send a Block Kit approval prompt with interactive buttons.
 
@@ -3264,10 +3266,42 @@ class SlackAdapter(BasePlatformAdapter):
             # instead of a flat truncation that overflows once the header +
             # reason are added.
             header = ":warning: *Command Approval Required*\n"
+            if smart_denied:
+                header += "*Smart DENY:* owner override applies to this one operation only.\n"
             reason = f"Reason: {description[:500]}"
             budget = 3000 - len(header) - len(reason) - len("``````\n") - len("...")
             cmd_preview = command[:budget] + "..." if len(command) > budget else command
 
+            actions = [
+                {
+                    "type": "button",
+                    "text": {"type": "plain_text", "text": "Allow Once"},
+                    "style": "primary",
+                    "action_id": "hermes_approve_once",
+                    "value": session_key,
+                },
+            ]
+            if not smart_denied:
+                actions.append({
+                    "type": "button",
+                    "text": {"type": "plain_text", "text": "Allow Session"},
+                    "action_id": "hermes_approve_session",
+                    "value": session_key,
+                })
+                if allow_permanent:
+                    actions.append({
+                        "type": "button",
+                        "text": {"type": "plain_text", "text": "Always Allow"},
+                        "action_id": "hermes_approve_always",
+                        "value": session_key,
+                    })
+            actions.append({
+                "type": "button",
+                "text": {"type": "plain_text", "text": "Deny"},
+                "style": "danger",
+                "action_id": "hermes_deny",
+                "value": session_key,
+            })
             blocks = [
                 {
                     "type": "section",
@@ -3276,37 +3310,7 @@ class SlackAdapter(BasePlatformAdapter):
                         "text": f"{header}```{cmd_preview}```\n{reason}",
                     },
                 },
-                {
-                    "type": "actions",
-                    "elements": [
-                        {
-                            "type": "button",
-                            "text": {"type": "plain_text", "text": "Allow Once"},
-                            "style": "primary",
-                            "action_id": "hermes_approve_once",
-                            "value": session_key,
-                        },
-                        {
-                            "type": "button",
-                            "text": {"type": "plain_text", "text": "Allow Session"},
-                            "action_id": "hermes_approve_session",
-                            "value": session_key,
-                        },
-                        {
-                            "type": "button",
-                            "text": {"type": "plain_text", "text": "Always Allow"},
-                            "action_id": "hermes_approve_always",
-                            "value": session_key,
-                        },
-                        {
-                            "type": "button",
-                            "text": {"type": "plain_text", "text": "Deny"},
-                            "style": "danger",
-                            "action_id": "hermes_deny",
-                            "value": session_key,
-                        },
-                    ],
-                },
+                {"type": "actions", "elements": actions},
             ]
 
             kwargs: Dict[str, Any] = {

plugins/platforms/teams/adapter.py

@@ -1098,6 +1098,8 @@ class TeamsAdapter(BasePlatformAdapter):
         session_key: str,
         description: str = "dangerous command",
         metadata: Optional[Dict[str, Any]] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """Send an Adaptive Card approval prompt with Allow/Deny buttons."""
         if not self._app:
@@ -1111,39 +1113,34 @@ class TeamsAdapter(BasePlatformAdapter):
             "desc": description,
         }
 
-        card = (
-            AdaptiveCard()
-            .with_version("1.4")
-            .with_body([
-                TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"),
-                TextBlock(text=f"```\n{cmd_preview}\n```", wrap=True),
-                TextBlock(text=f"Reason: {description}", wrap=True, isSubtle=True),
-            ])
-            .with_actions([
-                ExecuteAction(
-                    title="Allow Once",
-                    verb="hermes_approve",
-                    data={**btn_data_base, "hermes_action": "approve_once"},
-                    style="positive",
-                ),
-                ExecuteAction(
-                    title="Allow Session",
-                    verb="hermes_approve",
-                    data={**btn_data_base, "hermes_action": "approve_session"},
-                ),
-                ExecuteAction(
-                    title="Always Allow",
-                    verb="hermes_approve",
+        actions = [ExecuteAction(
+            title="Allow Once", verb="hermes_approve",
+            data={**btn_data_base, "hermes_action": "approve_once"}, style="positive",
+        )]
+        if not smart_denied:
+            actions.append(ExecuteAction(
+                title="Allow Session", verb="hermes_approve",
+                data={**btn_data_base, "hermes_action": "approve_session"},
+            ))
+            if allow_permanent:
+                actions.append(ExecuteAction(
+                    title="Always Allow", verb="hermes_approve",
                     data={**btn_data_base, "hermes_action": "approve_always"},
-                ),
-                ExecuteAction(
-                    title="Deny",
-                    verb="hermes_approve",
-                    data={**btn_data_base, "hermes_action": "deny"},
-                    style="destructive",
-                ),
-            ])
-        )
+                ))
+        actions.append(ExecuteAction(
+            title="Deny", verb="hermes_approve",
+            data={**btn_data_base, "hermes_action": "deny"}, style="destructive",
+        ))
+        body = [
+            TextBlock(text="⚠️ Command Approval Required", wrap=True, weight="Bolder"),
+            TextBlock(text=f"```\n{cmd_preview}\n```", wrap=True),
+            TextBlock(text=f"Reason: {description}", wrap=True, isSubtle=True),
+        ]
+        if smart_denied:
+            body.append(TextBlock(
+                text="Smart DENY: owner override applies to this one operation only.", wrap=True
+            ))
+        card = AdaptiveCard().with_version("1.4").with_body(body).with_actions(actions)
 
         try:
             result = await self._send_card(chat_id, card)

plugins/platforms/telegram/adapter.py

@@ -4568,6 +4568,8 @@ class TelegramAdapter(BasePlatformAdapter):
         self, chat_id: str, command: str, session_key: str,
         description: str = "dangerous command",
         metadata: Optional[Dict[str, Any]] = None,
+        allow_permanent: bool = True,
+        smart_denied: bool = False,
     ) -> SendResult:
         """Send an inline-keyboard approval prompt with interactive buttons.
 
@@ -4584,6 +4586,8 @@ class TelegramAdapter(BasePlatformAdapter):
                 f"<pre>{_html.escape(cmd_preview)}</pre>\n\n"
                 f"Reason: {_html.escape(description)}"
             )
+            if smart_denied:
+                text += "\n\n<b>Smart DENY:</b> owner override applies to this one operation only."
 
             # Resolve thread context for thread replies
             thread_id = self._metadata_thread_id(metadata)
@@ -4596,16 +4600,19 @@ class TelegramAdapter(BasePlatformAdapter):
                 self._approval_counter = itertools.count(1)
             approval_id = next(self._approval_counter)
 
-            keyboard = InlineKeyboardMarkup([
-                [
-                    InlineKeyboardButton("✅ Allow Once", callback_data=f"ea:once:{approval_id}"),
-                    InlineKeyboardButton("✅ Session", callback_data=f"ea:session:{approval_id}"),
-                ],
-                [
-                    InlineKeyboardButton("✅ Always", callback_data=f"ea:always:{approval_id}"),
-                    InlineKeyboardButton("❌ Deny", callback_data=f"ea:deny:{approval_id}"),
-                ],
-            ])
+            buttons = [
+                InlineKeyboardButton("✅ Allow Once", callback_data=f"ea:once:{approval_id}")
+            ]
+            if not smart_denied:
+                buttons.append(
+                    InlineKeyboardButton("✅ Session", callback_data=f"ea:session:{approval_id}")
+                )
+                if allow_permanent:
+                    buttons.append(
+                        InlineKeyboardButton("✅ Always", callback_data=f"ea:always:{approval_id}")
+                    )
+            buttons.append(InlineKeyboardButton("❌ Deny", callback_data=f"ea:deny:{approval_id}"))
+            keyboard = InlineKeyboardMarkup([buttons])
 
             kwargs: Dict[str, Any] = {
                 "chat_id": normalize_telegram_chat_id(chat_id),

scripts/release.py

@@ -45,7 +45,11 @@ ACP_REGISTRY_MANIFEST = REPO_ROOT / "acp_registry" / "agent.json"
 
 # Auto-extracted from noreply emails + manual overrides
 AUTHOR_MAP = {
+    "m.guttmann@journaway.com": "mguttmann",  # PR #63738 salvage (Anthropic setup-token pool auth normalization)
     "VrtxOmega@pm.me": "VrtxOmega",  # PR #43809 salvage (desktop: WSL folder-picker path bridge)
+    "jake.long.vu@vucar.net": "jakelongvu-bot",  # PR #36683 partial salvage (approval: honor canonical approvals.timeout in gateway waits)
+    "luigi@users.noreply.github.com": "Tortugasaur",  # PR #43205 salvage (desktop: profile-aware three-way approval mode statusbar control)
+    "kavi@local.hermes": "kavioavio",  # Issue #46544 / PR #47705 evolution (smart DENY exact-operation owner override)
     "135129512+ansel-f@users.noreply.github.com": "ansel-f",  # PR #62388 salvage (approval: allow exact verifier temp cleanup without broadening rm safety boundary)
     "robert@modern-minds.ai": "Hopfensaft",  # PR #31933 salvage (dashboard: align approvals.mode dropdown with canonical engine values)
     "palmer@dugoutfantasy.com": "professorpalmer",  # PR #48591 salvage (sessions: CLI workspace filter + restore-cwd-on-resume)
@@ -238,6 +242,7 @@ AUTHOR_MAP = {
     "w31rdm4ch1n3z@protonmail.com": "w31rdm4ch1nZ",
     "xtpeeps@gmail.com": "x7peeps",
     "ahmad@madsgency.com": "ahmadashfq",
+    "182213728+yinkev@users.noreply.github.com": "yinkev",  # scratch artifact preservation salvage
     "rratmansky@gmail.com": "rratmansky",
     "lkz-de@users.noreply.github.com": "lkz-de",
     "charles@salesondemand.io": "salesondemandio",
@@ -318,6 +323,7 @@ AUTHOR_MAP = {
     "290859878+synapsesx@users.noreply.github.com": "synapsesx",
     "157689911+itsflownium@users.noreply.github.com": "itsflownium",
     "dirtyren@users.noreply.github.com": "dirtyren",
+    "johnmlussier@gmail.com": "John-Lussier",
     "chenkun_lws@126.com": "bytesnail",  # PR #60360 salvage (--yolo startup ordering; #60328)
     "iamgexin@qq.com": "nullptr0807",  # PR #60956 salvage (gateway hygiene in-place compaction; #60947)
     "embwl0x@users.noreply.github.com": "embwl0x",  # PR #60810 salvage (channel directory offload)

tests/acp/test_permissions.py

@@ -23,6 +23,7 @@ def _invoke_callback(
     outcome,
     *,
     allow_permanent=True,
+    smart_denied=False,
     timeout=60.0,
     use_prompt_path=False,
 ):
@@ -45,6 +46,7 @@ def _invoke_callback(
                 "rm -rf /",
                 "dangerous command",
                 allow_permanent=allow_permanent,
+                smart_denied=smart_denied,
                 approval_callback=cb,
             )
         else:
@@ -52,6 +54,7 @@ def _invoke_callback(
                 "rm -rf /",
                 "dangerous command",
                 allow_permanent=allow_permanent,
+                smart_denied=smart_denied,
             )
 
     scheduled["coro"].close()
@@ -115,6 +118,30 @@ class TestApprovalBridge:
         assert result == "session"
         assert option_ids == ["allow_once", "allow_session", "deny", "deny_always"]
 
+    def test_smart_deny_prompt_only_offers_once_and_deny(self):
+        result, kwargs, _, _, _ = _invoke_callback(
+            AllowedOutcome(option_id="allow_once", outcome="selected"),
+            allow_permanent=False,
+            smart_denied=True,
+            use_prompt_path=True,
+        )
+
+        assert result == "once"
+        assert [option.option_id for option in kwargs["options"]] == [
+            "allow_once", "deny",
+        ]
+
+    def test_smart_deny_rejects_disallowed_session_outcome(self):
+        result, kwargs, _, _, _ = _invoke_callback(
+            AllowedOutcome(option_id="allow_session", outcome="selected"),
+            smart_denied=True,
+        )
+
+        assert result == "deny"
+        assert [option.option_id for option in kwargs["options"]] == [
+            "allow_once", "deny",
+        ]
+
     def test_reject_always_outcome_denies_without_changing_policy(self):
         result, kwargs, _, _, _ = _invoke_callback(
             AllowedOutcome(option_id="deny_always", outcome="selected"),

tests/agent/test_credential_pool.py

@@ -831,6 +831,70 @@ def test_load_pool_does_not_persist_env_seeded_secret_value(tmp_path, monkeypatc
     assert persisted["secret_fingerprint"].startswith("sha256:")
 
 
+def test_load_pool_collapses_duplicate_env_rows_to_active_key(tmp_path, monkeypatch):
+    """One env source is one credential, even if auth.json contains stale duplicates."""
+    key = "sk-or-active-main-key"
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes"))
+    monkeypatch.setenv("OPENROUTER_API_KEY", key)
+    _write_auth_store(
+        tmp_path,
+        {
+            "version": 1,
+            "credential_pool": {
+                "openrouter": [
+                    {
+                        "id": "current-row",
+                        "label": "OPENROUTER_API_KEY",
+                        "auth_type": "api_key",
+                        "priority": 0,
+                        "source": "env:OPENROUTER_API_KEY",
+                    },
+                    {
+                        "id": "stale-duplicate",
+                        "label": "OPENROUTER_API_KEY",
+                        "auth_type": "api_key",
+                        "priority": 1,
+                        "source": "env:OPENROUTER_API_KEY",
+                    },
+                ]
+            },
+        },
+    )
+
+    from agent.credential_pool import load_pool
+
+    pool = load_pool("openrouter")
+
+    assert [(entry.id, entry.runtime_api_key) for entry in pool.entries()] == [
+        ("current-row", key)
+    ]
+    persisted = json.loads((tmp_path / "hermes" / "auth.json").read_text())
+    assert [entry["id"] for entry in persisted["credential_pool"]["openrouter"]] == [
+        "current-row"
+    ]
+
+
+def test_credential_pool_never_selects_empty_borrowed_entry():
+    from agent.credential_pool import CredentialPool, PooledCredential
+
+    pool = CredentialPool(
+        "openrouter",
+        [
+            PooledCredential(
+                provider="openrouter",
+                id="metadata-only",
+                label="OPENROUTER_API_KEY",
+                auth_type="api_key",
+                priority=0,
+                source="env:OPENROUTER_API_KEY",
+                access_token="",
+            )
+        ],
+    )
+
+    assert pool.select() is None
+    assert pool.acquire_lease() is None
+
 
 def test_load_pool_persists_bitwarden_origin_metadata_without_secret(tmp_path, monkeypatch):
     """Bitwarden-injected env vars retain source metadata but not raw values."""

tests/agent/test_credential_pool_oat_authtype.py

@@ -0,0 +1,138 @@
+"""Regression tests for #63737: sk-ant-oat pool entries are OAuth."""
+
+import json
+from pathlib import Path
+
+from agent.credential_pool import (
+    AUTH_TYPE_API_KEY,
+    AUTH_TYPE_OAUTH,
+    CredentialPool,
+    PooledCredential,
+)
+
+
+def test_manual_anthropic_oat_normalized_to_oauth():
+    # Pool auth_type gates OAuth-only resolver and refresh paths.
+    entry = PooledCredential.from_dict(
+        "anthropic",
+        {
+            "label": "MainKey",
+            "source": "manual",
+            "auth_type": "api_key",
+            "access_token": "sk-ant-oat-EXAMPLE",
+        },
+    )
+    assert entry.auth_type == AUTH_TYPE_OAUTH
+
+
+def test_anthropic_real_api_key_unchanged():
+    entry = PooledCredential.from_dict(
+        "anthropic",
+        {"auth_type": "api_key", "access_token": "sk-ant-api-EXAMPLE"},
+    )
+    assert entry.auth_type == AUTH_TYPE_API_KEY
+
+
+def test_anthropic_admin_key_unchanged():
+    entry = PooledCredential.from_dict(
+        "anthropic",
+        {"auth_type": "api_key", "access_token": "sk-ant-admin-EXAMPLE"},
+    )
+    assert entry.auth_type == AUTH_TYPE_API_KEY
+
+
+def test_non_anthropic_provider_unchanged():
+    entry = PooledCredential.from_dict(
+        "openrouter",
+        {"auth_type": "api_key", "access_token": "sk-ant-oat-WHATEVER"},
+    )
+    assert entry.auth_type == AUTH_TYPE_API_KEY
+
+
+def test_add_entry_normalizes_before_persisting(tmp_path, monkeypatch):
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+
+    pool = CredentialPool("anthropic", [])
+    entry = pool.add_entry(PooledCredential(
+        provider="anthropic",
+        id="manual-oat",
+        label="Manual setup token",
+        auth_type=AUTH_TYPE_API_KEY,
+        priority=0,
+        source="manual",
+        access_token="sk-ant-oat-manual-entry",
+    ))
+
+    persisted = json.loads((hermes_home / "auth.json").read_text())
+    assert entry.auth_type == AUTH_TYPE_OAUTH
+    assert persisted["credential_pool"]["anthropic"][0]["auth_type"] == AUTH_TYPE_OAUTH
+
+
+def test_load_heals_legacy_row_and_exposes_it_to_resolver(tmp_path, monkeypatch):
+    hermes_home = tmp_path / "hermes"
+    hermes_home.mkdir()
+    monkeypatch.setenv("HERMES_HOME", str(hermes_home))
+    for key in ("ANTHROPIC_API_KEY", "ANTHROPIC_TOKEN", "CLAUDE_CODE_OAUTH_TOKEN"):
+        monkeypatch.delenv(key, raising=False)
+    monkeypatch.setattr(
+        "agent.anthropic_adapter.read_claude_code_credentials",
+        lambda: None,
+    )
+    token = "sk-ant-oat-legacy-manual"
+    auth_file = hermes_home / "auth.json"
+    auth_file.write_text(json.dumps({
+        "version": 1,
+        "credential_pool": {
+            "anthropic": [{
+                "id": "legacy-oat",
+                "label": "Legacy setup token",
+                "auth_type": AUTH_TYPE_API_KEY,
+                "priority": 0,
+                "source": "manual",
+                "access_token": token,
+            }],
+        },
+    }))
+
+    from agent.anthropic_adapter import resolve_anthropic_token
+    from agent.credential_pool import load_pool
+
+    entry = load_pool("anthropic").entries()[0]
+    persisted = json.loads(auth_file.read_text())
+    assert entry.auth_type == AUTH_TYPE_OAUTH
+    assert persisted["credential_pool"]["anthropic"][0]["auth_type"] == AUTH_TYPE_OAUTH
+    assert resolve_anthropic_token() == token
+
+
+def test_profile_global_fallback_normalizes_in_memory_without_writing(tmp_path, monkeypatch):
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    global_root = tmp_path / ".hermes"
+    global_root.mkdir()
+    profile_home = global_root / "profiles" / "coder"
+    profile_home.mkdir(parents=True)
+    monkeypatch.setenv("HERMES_HOME", str(profile_home))
+    token = "sk-ant-oat-global-fallback"
+    global_auth = global_root / "auth.json"
+    global_auth.write_text(json.dumps({
+        "version": 1,
+        "credential_pool": {
+            "anthropic": [{
+                "id": "global-oat",
+                "label": "Global setup token",
+                "auth_type": AUTH_TYPE_API_KEY,
+                "priority": 0,
+                "source": "manual",
+                "access_token": token,
+            }],
+        },
+    }))
+
+    from agent.credential_pool import load_pool
+
+    entry = load_pool("anthropic").entries()[0]
+    persisted = json.loads(global_auth.read_text())
+    assert entry.auth_type == AUTH_TYPE_OAUTH
+    assert persisted["credential_pool"]["anthropic"][0]["auth_type"] == AUTH_TYPE_API_KEY
+    assert not (profile_home / "auth.json").exists()

tests/agent/test_custom_provider_extra_body_matching.py

@@ -0,0 +1,78 @@
+"""Tests for custom-provider model matching (extra_body / service_tier drop bug).
+
+July 2026 incident: a custom provider with a multi-model catalog
+(``models: {gpt-5.5: {}, gpt-5.6-terra: {}, ...}``) and a ``model``/
+``default_model`` differing from the session model failed
+``_custom_provider_model_matches``, so ``extra_body: {service_tier: flex}``
+was silently dropped — every request billed at standard tier (~2.3x).
+"""
+
+from agent.agent_init import (
+    _custom_provider_extra_body_for_agent,
+    _custom_provider_model_matches,
+)
+
+BASE = "https://api.openai.com/v1"
+
+
+def _entry(**over):
+    e = {
+        "name": "openai",
+        "base_url": BASE,
+        "extra_body": {"service_tier": "flex"},
+    }
+    e.update(over)
+    return e
+
+
+class TestModelMatches:
+    def test_models_dict_catalog_matches_session_model(self):
+        e = _entry(model="gpt-5.5", models={"gpt-5.5": {}, "gpt-5.6-terra": {}})
+        assert _custom_provider_model_matches("gpt-5.6-terra", e)
+
+    def test_models_list_catalog_matches(self):
+        e = _entry(model="gpt-5.5", models=["gpt-5.5", "gpt-5.6-sol"])
+        assert _custom_provider_model_matches("gpt-5.6-sol", e)
+
+    def test_catalog_miss_falls_back_to_model_field(self):
+        e = _entry(model="gpt-5.5", models={"gpt-5.5": {}})
+        assert _custom_provider_model_matches("gpt-5.5", e)
+        assert not _custom_provider_model_matches("gpt-4o", e)
+
+    def test_no_model_no_catalog_matches_everything(self):
+        e = _entry()
+        assert _custom_provider_model_matches("anything", e)
+
+    def test_catalog_case_insensitive(self):
+        e = _entry(models={"GPT-5.6-Terra": {}})
+        assert _custom_provider_model_matches("gpt-5.6-terra", e)
+
+
+class TestExtraBodyResolution:
+    def test_multi_model_provider_yields_extra_body(self):
+        # The exact sweeper-profile shape that failed in production.
+        entry = _entry(
+            model="gpt-5.5",
+            models={"gpt-5.5": {}, "gpt-5.6-sol": {}, "gpt-5.6-terra": {}},
+        )
+        got = _custom_provider_extra_body_for_agent(
+            provider="custom", model="gpt-5.6-terra",
+            base_url=BASE, custom_providers=[entry],
+        )
+        assert got == {"service_tier": "flex"}
+
+    def test_non_catalog_model_gets_no_override(self):
+        entry = _entry(model="gpt-5.5", models={"gpt-5.5": {}})
+        got = _custom_provider_extra_body_for_agent(
+            provider="custom", model="gpt-4o",
+            base_url=BASE, custom_providers=[entry],
+        )
+        assert got is None
+
+    def test_non_custom_provider_unaffected(self):
+        entry = _entry(models={"gpt-5.6-terra": {}})
+        got = _custom_provider_extra_body_for_agent(
+            provider="openrouter", model="gpt-5.6-terra",
+            base_url=BASE, custom_providers=[entry],
+        )
+        assert got is None

tests/cli/test_cli_approval_ui.py

@@ -67,6 +67,38 @@ def _make_background_cli_stub():
 
 
 class TestCliApprovalUi:
+    def test_smart_denied_callback_offers_only_once_and_deny(self):
+        cli = _make_cli_stub()
+        result = {}
+
+        def _run_callback():
+            result["value"] = cli._approval_callback(
+                "rm -rf /tmp/example",
+                "recursive delete",
+                allow_permanent=False,
+                smart_denied=True,
+            )
+
+        thread = threading.Thread(target=_run_callback, daemon=True)
+        thread.start()
+
+        deadline = time.time() + 2
+        while cli._approval_state is None and time.time() < deadline:
+            time.sleep(0.01)
+
+        assert cli._approval_state is not None
+        assert cli._approval_state["choices"] == ["once", "deny"]
+
+        cli._approval_state["response_queue"].put("deny")
+        thread.join(timeout=2)
+        assert result["value"] == "deny"
+
+    def test_non_smart_non_permanent_callback_preserves_session_choice(self):
+        cli = _make_cli_stub()
+        assert cli._approval_choices(
+            "rm -rf /tmp/example", allow_permanent=False, smart_denied=False
+        ) == ["once", "session", "deny"]
+
     def test_sudo_prompt_restores_existing_draft_after_response(self):
         cli = _make_cli_stub()
         cli._app.current_buffer = _FakeBuffer("draft command", cursor_position=5)

tests/cron/test_scheduler.py

@@ -1533,6 +1533,25 @@ class TestRunJobSessionPersistence:
         assert "final fallback report" in output
         assert "(FAILED)" not in output
 
+    def test_tick_skips_due_jobs_while_dispatch_is_paused(self, tmp_path):
+        """The drain gate runs before advancing a due job's schedule."""
+        from cron.scheduler import tick
+
+        job = {
+            "id": "paused-due-job",
+            "name": "paused due job",
+            "schedule": {"kind": "interval", "seconds": 60},
+            "next_run_at": "2020-01-01T00:00:00+00:00",
+            "enabled": True,
+        }
+        with patch("cron.scheduler.get_due_jobs", return_value=[job]), patch(
+            "cron.scheduler.advance_next_run"
+        ) as advance, patch("cron.scheduler.run_one_job") as run_one:
+            assert tick(verbose=False, sync=True, can_dispatch=lambda: False) == 0
+
+        advance.assert_not_called()
+        run_one.assert_not_called()
+
     def test_tick_marks_empty_response_as_error(self, tmp_path):
         """When run_job returns success=True but final_response is empty,
         tick() should mark the job as error so last_status != 'ok'.

tests/cron/test_scheduler_provider.py

@@ -129,13 +129,11 @@ def test_cronscheduler_default_is_available_true():
 
 
 def test_abc_growth_stays_additive():
-    """Forward-compat guard: the ABC's REQUIRED surface is exactly name+start.
+    """The provider interface stays source-compatible with existing plugins.
 
-    Any optional hook added later for the external provider
-    (on_jobs_changed/fire_due/reconcile) must be NON-abstract (carry a default),
-    so the built-in keeps satisfying the ABC without overriding them. This test
-    fails loudly if someone makes a future hook abstract (a breaking change that
-    would force every provider — including the built-in — to implement it).
+    ``start`` must be the only required implementation hook: future optional
+    behavior belongs in non-abstract default methods so custom plugins do not
+    break on import after an upgrade.
     """
     from cron.scheduler_provider import CronScheduler
 
@@ -174,6 +172,33 @@ def test_inprocess_provider_ticks_and_stops():
     assert calls[0].get("sync") is False
 
 
+def test_inprocess_provider_skips_dispatch_while_draining():
+    """A drain pause keeps due work pending until dispatch is re-enabled."""
+    from cron.scheduler_provider import InProcessCronScheduler
+
+    calls = []
+    stop = threading.Event()
+    allow_dispatch = threading.Event()
+    provider = InProcessCronScheduler()
+
+    with patch("cron.scheduler.tick", side_effect=lambda *a, **k: calls.append(k) or 0):
+        thread = threading.Thread(
+            target=provider.start,
+            args=(stop,),
+            kwargs={"interval": 0.01, "can_dispatch": allow_dispatch.is_set},
+            daemon=True,
+        )
+        thread.start()
+        time.sleep(0.05)
+        assert calls == []
+        allow_dispatch.set()
+        assert _wait_until(lambda: len(calls) >= 1), "provider never resumed dispatch"
+        stop.set()
+        thread.join(timeout=5)
+
+    assert not thread.is_alive()
+
+
 def test_inprocess_provider_stop_is_noop():
     """The default stop() hook is a safe no-op (the stop_event is the real
     stop signal for the built-in)."""

tests/gateway/restart_test_helpers.py

@@ -115,6 +115,18 @@ def make_restart_runner(
     runner._running_agent_count = GatewayRunner._running_agent_count.__get__(
         runner, GatewayRunner
     )
+    runner._active_cron_job_count = GatewayRunner._active_cron_job_count.__get__(
+        runner, GatewayRunner
+    )
+    runner._active_api_run_count = GatewayRunner._active_api_run_count.__get__(
+        runner, GatewayRunner
+    )
+    runner._active_work_count = GatewayRunner._active_work_count.__get__(
+        runner, GatewayRunner
+    )
+    runner._persist_active_agents = GatewayRunner._persist_active_agents.__get__(
+        runner, GatewayRunner
+    )
     runner._snapshot_running_agents = GatewayRunner._snapshot_running_agents.__get__(
         runner, GatewayRunner
     )

tests/gateway/test_api_server_active_work_drain.py

@@ -0,0 +1,326 @@
+"""Regression coverage for #63529 API-server shutdown draining.
+
+API-server work is adapter-owned rather than tracked by
+``GatewayRunner._running_agents``. The shutdown drain must account for the
+same live state as the API concurrency limiter, including a ``/v1/runs`` task
+that exists before its agent has been constructed, and it must refuse new API
+turns once the gateway starts draining.
+"""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import TestClient, TestServer
+
+from gateway.config import Platform, PlatformConfig
+from gateway.platforms.api_server import APIServerAdapter
+from tests.gateway.restart_test_helpers import make_restart_runner
+
+
+class _RunTask:
+    def __init__(self, done: bool = False):
+        self._done = done
+
+    def done(self) -> bool:
+        return self._done
+
+
+def _make_api_adapter(*, inflight: int = 0, queued_ids=()):
+    tasks = {run_id: _RunTask() for run_id in queued_ids}
+    adapter = SimpleNamespace(
+        platform=Platform.API_SERVER,
+        _inflight_agent_runs=inflight,
+        _active_run_tasks=tasks,
+    )
+
+    def active_agent_work_count() -> int:
+        return int(getattr(adapter, "_pending_agent_requests", 0)) + int(
+            adapter._inflight_agent_runs
+        ) + sum(not task.done() for task in adapter._active_run_tasks.values())
+
+    adapter.active_agent_work_count = active_agent_work_count
+    return adapter
+
+
+def _make_admission_app(adapter: APIServerAdapter) -> web.Application:
+    app = web.Application()
+    app.router.add_post("/api/sessions/{session_id}/chat", adapter._handle_session_chat)
+    app.router.add_post(
+        "/api/sessions/{session_id}/chat/stream", adapter._handle_session_chat_stream
+    )
+    app.router.add_post("/v1/chat/completions", adapter._handle_chat_completions)
+    app.router.add_post("/v1/responses", adapter._handle_responses)
+    app.router.add_post("/v1/runs", adapter._handle_runs)
+    return app
+
+
+class TestActiveApiRunCount:
+    def test_zero_when_no_api_adapter(self):
+        runner, _adapter = make_restart_runner()
+        runner.adapters = {}
+        assert runner._active_api_run_count() == 0
+
+    def test_delegates_to_primary_api_adapter(self):
+        runner, _adapter = make_restart_runner()
+        runner.adapters = {
+            Platform.API_SERVER: _make_api_adapter(inflight=2, queued_ids=["r1"])
+        }
+        assert runner._active_api_run_count() == 3
+
+    def test_ignores_non_api_platforms(self):
+        runner, _adapter = make_restart_runner()
+        other = SimpleNamespace(
+            platform=Platform.DISCORD,
+            active_agent_work_count=lambda: 99,
+        )
+        runner.adapters = {Platform.DISCORD: other}
+        assert runner._active_api_run_count() == 0
+
+    def test_never_raises_on_broken_adapter(self):
+        runner, _adapter = make_restart_runner()
+
+        class Bad:
+            platform = Platform.API_SERVER
+
+            @staticmethod
+            def active_agent_work_count() -> int:
+                raise RuntimeError("boom")
+
+        runner.adapters = {Platform.API_SERVER: Bad()}
+        assert runner._active_api_run_count() == 0
+
+
+class TestAPIServerAdapterWorkCount:
+    def test_concurrency_limit_counts_other_pending_admissions(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        adapter._max_concurrent_runs = 1
+        adapter._pending_agent_requests = 1
+
+        response = adapter._concurrency_limited_response()
+
+        assert response is not None
+        assert response.status == 429
+
+    @pytest.mark.asyncio
+    async def test_concurrency_limit_excludes_current_pending_admission(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        adapter._max_concurrent_runs = 1
+        app = _make_admission_app(adapter)
+
+        async with TestClient(TestServer(app)) as client:
+            with patch.object(adapter, "_run_agent", new=AsyncMock(return_value=({}, {}))):
+                response = await client.post(
+                    "/api/sessions/s/chat",
+                    json={"message": "hello"},
+                )
+
+        assert response.status == 404
+
+    def test_counts_pending_admission_before_agent_bookkeeping(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        adapter._pending_agent_requests = 1
+
+        assert adapter.active_agent_work_count() == 1
+
+    def test_counts_live_run_task_before_agent_creation(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        adapter._inflight_agent_runs = 2
+        adapter._active_run_tasks = {
+            "queued": _RunTask(),
+            "finished": _RunTask(done=True),
+        }
+        adapter._active_run_agents = {}
+
+        assert adapter.active_agent_work_count() == 3
+
+    def test_does_not_double_count_started_run_agent(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        adapter._inflight_agent_runs = 0
+        adapter._active_run_tasks = {"run-1": _RunTask()}
+        adapter._active_run_agents = {"run-1": object()}
+
+        assert adapter.active_agent_work_count() == 1
+
+
+class TestDrainWaitsForApiWork:
+    @pytest.mark.asyncio
+    async def test_drain_returns_immediately_when_nothing_active(self):
+        runner, _adapter = make_restart_runner()
+        runner.adapters = {}
+
+        _snapshot, timed_out = await runner._drain_active_agents(5.0)
+
+        assert timed_out is False
+
+    @pytest.mark.asyncio
+    async def test_drain_waits_for_real_queued_run_before_agent_creation(self):
+        """A live /v1/runs task must block drain before it has an agent."""
+        runner, _adapter = make_restart_runner()
+        api = APIServerAdapter(PlatformConfig(enabled=True))
+        runner.adapters = {Platform.API_SERVER: api}
+        app = _make_admission_app(api)
+        original_create_task = asyncio.create_task
+        task_started = asyncio.Event()
+        allow_task = asyncio.Event()
+
+        def delayed_create_task(coro):
+            async def delayed():
+                task_started.set()
+                await allow_task.wait()
+                return await coro
+
+            return original_create_task(delayed())
+
+        mock_agent = MagicMock()
+        mock_agent.run_conversation.return_value = {"final_response": "done"}
+        mock_agent.session_prompt_tokens = 0
+        mock_agent.session_completion_tokens = 0
+        mock_agent.session_total_tokens = 0
+
+        with patch(
+            "gateway.platforms.api_server.asyncio.create_task",
+            side_effect=delayed_create_task,
+        ), patch.object(api, "_create_agent", return_value=mock_agent):
+            async with TestClient(TestServer(app)) as client:
+                response = await client.post("/v1/runs", json={"input": "hello"})
+                assert response.status == 202
+                await task_started.wait()
+
+                assert api._active_run_agents == {}
+                assert runner._active_api_run_count() == 1
+                drain_task = original_create_task(runner._drain_active_agents(2.0))
+                await asyncio.sleep(0.1)
+                assert not drain_task.done()
+
+                allow_task.set()
+                _snapshot, timed_out = await drain_task
+
+        assert timed_out is False
+
+    @pytest.mark.asyncio
+    async def test_drain_times_out_if_api_run_outlives_the_window(self):
+        runner, _adapter = make_restart_runner()
+        runner.adapters = {Platform.API_SERVER: _make_api_adapter(queued_ids=["run-1"])}
+
+        _snapshot, timed_out = await runner._drain_active_agents(0.1)
+
+        assert timed_out is True
+
+    @pytest.mark.asyncio
+    async def test_drain_still_waits_for_chat_cron_and_api_work(self):
+        import cron.scheduler as sched
+
+        runner, _adapter = make_restart_runner()
+        runner._running_agents = {"session-1": MagicMock()}
+        sched._running_job_ids.add("job-1")
+        runner.adapters = {Platform.API_SERVER: _make_api_adapter(queued_ids=["run-1"])}
+
+        async def finish_all():
+            await asyncio.sleep(0.12)
+            runner._running_agents.clear()
+            sched._running_job_ids.discard("job-1")
+            runner.adapters[Platform.API_SERVER]._active_run_tasks.clear()
+
+        task = asyncio.create_task(finish_all())
+        try:
+            _snapshot, timed_out = await runner._drain_active_agents(2.0)
+        finally:
+            await task
+            sched._running_job_ids.discard("job-1")
+
+        assert timed_out is False
+
+
+class TestDrainAdmission:
+    @pytest.mark.asyncio
+    async def test_drain_refuses_every_agent_start_endpoint(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        runner = SimpleNamespace(_draining=True, _external_drain_active=False)
+        app = _make_admission_app(adapter)
+        paths = (
+            "/api/sessions/missing/chat",
+            "/api/sessions/missing/chat/stream",
+            "/v1/chat/completions",
+            "/v1/responses",
+            "/v1/runs",
+        )
+
+        with patch("gateway.run._gateway_runner_ref", lambda: runner):
+            async with TestClient(TestServer(app)) as client:
+                for path in paths:
+                    response = await client.post(path, json={})
+                    payload = await response.json()
+
+                    assert response.status == 503
+                    assert response.headers["Retry-After"] == "1"
+                    assert payload["error"]["code"] == "gateway_draining"
+
+    @pytest.mark.asyncio
+    async def test_external_drain_refuses_every_agent_start_endpoint(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        runner = SimpleNamespace(_draining=False, _external_drain_active=True)
+        app = _make_admission_app(adapter)
+        paths = (
+            "/api/sessions/missing/chat",
+            "/api/sessions/missing/chat/stream",
+            "/v1/chat/completions",
+            "/v1/responses",
+            "/v1/runs",
+        )
+
+        with patch("gateway.run._gateway_runner_ref", lambda: runner):
+            async with TestClient(TestServer(app)) as client:
+                for path in paths:
+                    response = await client.post(path, json={})
+                    payload = await response.json()
+
+                    assert response.status == 503
+                    assert payload["error"]["code"] == "gateway_draining"
+
+    @pytest.mark.asyncio
+    async def test_admitted_request_blocks_drain_before_agent_bookkeeping(self):
+        adapter = APIServerAdapter(PlatformConfig(enabled=True))
+        runner, _adapter = make_restart_runner()
+        runner.adapters = {Platform.API_SERVER: adapter}
+        app = _make_admission_app(adapter)
+        body_read_started = asyncio.Event()
+        allow_body_read = asyncio.Event()
+
+        async def delayed_read_json(_request):
+            body_read_started.set()
+            await allow_body_read.wait()
+            return {"message": "hello"}, None
+
+        with patch.object(
+            adapter,
+            "_get_existing_session_or_404",
+            return_value=({}, None),
+        ), patch.object(
+            adapter,
+            "_read_json_body",
+            side_effect=delayed_read_json,
+        ), patch.object(
+            adapter,
+            "_run_agent",
+            new=AsyncMock(return_value=({"final_response": "done"}, {})),
+        ):
+            async with TestClient(TestServer(app)) as client:
+                request_task = asyncio.create_task(
+                    client.post("/api/sessions/missing/chat", json={})
+                )
+                await body_read_started.wait()
+
+                assert adapter._pending_agent_requests == 1
+                drain_task = asyncio.create_task(runner._drain_active_agents(2.0))
+                await asyncio.sleep(0.1)
+                assert not drain_task.done()
+
+                allow_body_read.set()
+                response = await request_task
+                assert response.status == 200
+                _snapshot, timed_out = await drain_task
+
+        assert timed_out is False

tests/gateway/test_api_server_jobs.py

@@ -11,6 +11,7 @@ Covers:
 """
 
 import logging
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -494,6 +495,19 @@ class TestRunJob:
                 assert data["job"] == triggered_job
                 mock_trigger.assert_called_once_with(VALID_JOB_ID)
 
+    @pytest.mark.asyncio
+    async def test_run_job_refuses_during_gateway_drain(self, adapter):
+        app = _create_app(adapter)
+        runner = SimpleNamespace(_draining=False, _external_drain_active=True)
+
+        with patch("gateway.run._gateway_runner_ref", lambda: runner):
+            async with TestClient(TestServer(app)) as cli:
+                resp = await cli.post(f"/api/jobs/{VALID_JOB_ID}/run")
+                payload = await resp.json()
+
+        assert resp.status == 503
+        assert payload["error"]["code"] == "gateway_draining"
+
 
 # ---------------------------------------------------------------------------
 # 17. test_auth_required

tests/gateway/test_api_server_runs.py

@@ -20,6 +20,7 @@ from aiohttp.test_utils import TestClient, TestServer
 from gateway.config import PlatformConfig
 from gateway.platforms.api_server import (
     APIServerAdapter,
+    _approval_event_choices,
     cors_middleware,
     security_headers_middleware,
 )
@@ -31,6 +32,24 @@ from tools import approval as approval_mod
 # ---------------------------------------------------------------------------
 
 
+@pytest.mark.parametrize(
+    ("smart_denied", "allow_permanent", "expected"),
+    [
+        (False, True, ["once", "session", "always", "deny"]),
+        (False, False, ["once", "session", "deny"]),
+        (True, True, ["once", "deny"]),
+        (True, False, ["once", "deny"]),
+    ],
+)
+def test_approval_event_choices_follow_backend_capabilities(
+    smart_denied, allow_permanent, expected
+):
+    assert _approval_event_choices(
+        smart_denied=smart_denied,
+        allow_permanent=allow_permanent,
+    ) == expected
+
+
 def _make_adapter(api_key: str = "") -> APIServerAdapter:
     """Create an adapter with optional API key."""
     extra = {}

tests/gateway/test_approval_prompt_redaction.py

@@ -126,3 +126,63 @@ class TestApprovalCommandWiring:
         from gateway.platforms import api_server
 
         self._assert_redacts_then_uses(api_server, "_approval_notify", "put_nowait")
+
+    def test_chat_platform_threads_approval_capabilities_to_adapter(self):
+        """The gateway must not drop the backend's one-operation UI contract."""
+        import ast
+        import inspect
+        import gateway.run as run
+
+        tree = ast.parse(inspect.getsource(run))
+        notify = next(
+            node for node in ast.walk(tree)
+            if isinstance(node, ast.FunctionDef) and node.name == "_approval_notify_sync"
+        )
+        call = next(
+            node for node in ast.walk(notify)
+            if isinstance(node, ast.Call)
+            and isinstance(node.func, ast.Attribute)
+            and node.func.attr == "send_exec_approval"
+        )
+        keywords = {kw.arg: kw.value for kw in call.keywords}
+        for name, default in (("allow_permanent", True), ("smart_denied", False)):
+            value = keywords[name]
+            assert isinstance(value, ast.Call)
+            assert isinstance(value.func, ast.Attribute) and value.func.attr == "get"
+            assert isinstance(value.args[0], ast.Constant) and value.args[0].value == name
+            assert isinstance(value.args[1], ast.Constant) and value.args[1].value is default
+
+
+class TestApprovalTextFallbackContract:
+    def test_smart_deny_only_advertises_one_operation(self):
+        from gateway.run import _format_exec_approval_fallback
+
+        text = _format_exec_approval_fallback(
+            "rm -rf /", "dangerous deletion", "/",
+            allow_permanent=False, smart_denied=True,
+        )
+        assert "owner override" in text.lower()
+        assert "one operation" in text.lower()
+        assert "`/approve`" in text
+        assert "approve session" not in text
+        assert "approve always" not in text
+
+    def test_non_smart_restriction_preserves_session_choice(self):
+        from gateway.run import _format_exec_approval_fallback
+
+        text = _format_exec_approval_fallback(
+            "curl https://example.test", "content warning", "!",
+            allow_permanent=False, smart_denied=False,
+        )
+        assert "`!approve session`" in text
+        assert "approve always" not in text
+
+    def test_manual_prompt_preserves_all_choices(self):
+        from gateway.run import _format_exec_approval_fallback
+
+        text = _format_exec_approval_fallback(
+            "rm -rf /", "dangerous deletion", "/",
+            allow_permanent=True, smart_denied=False,
+        )
+        assert "`/approve session`" in text
+        assert "`/approve always`" in text

tests/gateway/test_approve_deny_commands.py

@@ -515,28 +515,42 @@ class TestBlockingApprovalE2E:
         assert "BLOCKED" in result_holder[0]["message"]
         unregister_gateway_notify(session_key)
 
-    def test_blocking_approval_timeout(self):
-        """check_all_command_guards returns BLOCKED on timeout."""
+    @pytest.mark.parametrize(
+        "approval_config",
+        [
+            {"mode": "manual", "timeout": 0},
+            {"mode": "manual", "timeout": 0, "gateway_timeout": 300},
+        ],
+        ids=["shared-timeout-only", "shared-timeout-is-canonical"],
+    )
+    def test_blocking_approval_uses_canonical_timeout(self, approval_config, monkeypatch):
+        """Gateway waits use approvals.timeout, without a second timeout knob."""
+        from tools import approval as approval_module
         from tools.approval import (
-            register_gateway_notify, unregister_gateway_notify,
             check_all_command_guards,
+            register_gateway_notify,
+            reset_current_session_key,
+            resolve_gateway_approval,
+            set_current_session_key,
+            unregister_gateway_notify,
         )
 
+        monkeypatch.setattr(approval_module, "_YOLO_MODE_FROZEN", False)
         session_key = "e2e-timeout"
         register_gateway_notify(session_key, lambda d: None)
 
         result_holder = [None]
 
         def agent_thread():
-            from tools.approval import reset_current_session_key, set_current_session_key
-
             token = set_current_session_key(session_key)
             os.environ["HERMES_GATEWAY_SESSION"] = "1"
             os.environ["HERMES_EXEC_ASK"] = "1"
             os.environ["HERMES_SESSION_KEY"] = session_key
             try:
-                with patch("tools.approval._get_approval_config",
-                           return_value={"gateway_timeout": 1}):
+                with patch(
+                    "tools.approval._get_approval_config",
+                    return_value=approval_config,
+                ):
                     result_holder[0] = check_all_command_guards(
                         "rm -rf /important", "local"
                     )
@@ -548,9 +562,13 @@ class TestBlockingApprovalE2E:
 
         t = threading.Thread(target=agent_thread)
         t.start()
-        t.join(timeout=10)
+        t.join(timeout=1)
+        if t.is_alive():
+            resolve_gateway_approval(session_key, "deny")
+            t.join(timeout=5)
 
         assert result_holder[0]["approved"] is False
+        assert result_holder[0]["outcome"] == "timeout"
         assert "timed out" in result_holder[0]["message"]
         unregister_gateway_notify(session_key)
 

tests/gateway/test_completion_delivery.py

@@ -0,0 +1,494 @@
+"""Lifecycle-scoped gateway delivery regressions for terminal completions.
+
+The gateway contract here is deliberately narrower than exactly-once: one live
+GatewayRunner suppresses concurrent/replayed copies after successful adapter
+injection, failed injection remains retryable, and durable async-delegation
+state (when available) is acknowledged through its authoritative SQLite API.
+"""
+
+import asyncio
+import json
+import queue
+from collections import OrderedDict
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from gateway.config import Platform
+from gateway.run import GatewayRunner
+from gateway.session import SessionSource
+from tools.process_registry import ProcessRegistry, ProcessSession
+
+
+@pytest.fixture(autouse=True)
+def isolated_registry(tmp_path, monkeypatch):
+    """Any current/future durable compatibility path must stay in tmp state."""
+    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
+    import tools.process_registry as pr_module
+
+    monkeypatch.setattr(pr_module, "CHECKPOINT_PATH", tmp_path / "processes.json")
+    registry = pr_module.ProcessRegistry()
+    monkeypatch.setattr(pr_module, "process_registry", registry)
+    return registry
+
+
+def _runner(adapter, *, origins=None):
+    runner = object.__new__(GatewayRunner)
+    runner._running = True
+    runner.adapters = {Platform.TELEGRAM: adapter}
+    runner.session_store = SimpleNamespace(
+        _ensure_loaded=lambda: None,
+        _entries=origins or {},
+    )
+    runner._session_source_cache = {}
+    runner._completion_delivery_lock = __import__("threading").Lock()
+    runner._completion_deliveries_inflight = set()
+    runner._completion_deliveries_delivered = OrderedDict()
+    runner._completion_delivery_retention = 2048
+    return runner
+
+
+def _async_event(delegation_id="deleg_duplicate"):
+    return {
+        "type": "async_delegation",
+        "delegation_id": delegation_id,
+        "session_key": "agent:main:telegram:dm:12345:678",
+        "goal": "Investigate flaky test",
+        "status": "completed",
+        "summary": "Found it",
+        "api_calls": 1,
+        "duration_seconds": 12.0,
+        "dispatched_at": 1000.0,
+        "completed_at": 1012.0,
+        # PR #62479 stamps these on gateway-owned events. They must not
+        # change the producer identity used for queue replay.
+        "origin_profile": "default",
+        "origin_hermes_home": "/tmp/hermes-default",
+    }
+
+
+def _completion_event(*, started_at, session_id="proc_reused"):
+    return {
+        "type": "completion",
+        "session_id": session_id,
+        "session_key": "agent:main:telegram:dm:123",
+        "platform": "telegram",
+        "chat_type": "dm",
+        "chat_id": "123",
+        "started_at": started_at,
+        "command": "echo done",
+        "exit_code": 0,
+        "completion_reason": "exited",
+        "output": "done\n",
+    }
+
+
+def _stop_after_sleeps(monkeypatch, runner, count):
+    sleep_calls = 0
+
+    async def _bounded_sleep(_delay):
+        nonlocal sleep_calls
+        sleep_calls += 1
+        if sleep_calls >= count:
+            runner._running = False
+
+    monkeypatch.setattr(asyncio, "sleep", _bounded_sleep)
+
+
+def test_duplicate_async_queue_replay_injects_once(monkeypatch, isolated_registry):
+    """Byte-identical queue replays produce one turn in one gateway lifecycle."""
+    isolated = queue.Queue()
+    monkeypatch.setattr(isolated_registry, "completion_queue", isolated)
+    isolated.put(dict(_async_event()))
+    isolated.put(dict(_async_event()))
+
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+    _stop_after_sleeps(monkeypatch, runner, count=2)
+
+    asyncio.run(runner._async_delegation_watcher(interval=0))
+
+    adapter.handle_message.assert_awaited_once()
+
+
+def test_unroutable_async_event_is_not_requeued_forever(
+    monkeypatch, isolated_registry,
+):
+    isolated = queue.Queue()
+    monkeypatch.setattr(isolated_registry, "completion_queue", isolated)
+    event = _async_event("deleg_desktop_or_cli")
+    event["session_key"] = "20260711_unparseable_ui_session"
+    isolated.put(event)
+
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+    _stop_after_sleeps(monkeypatch, runner, count=2)
+
+    asyncio.run(runner._async_delegation_watcher(interval=0))
+
+    adapter.handle_message.assert_not_awaited()
+    assert isolated.empty()
+
+
+def test_concurrent_claims_share_the_same_narrow_delivery_seam():
+    """Concurrent consumers in one runner cannot both enter the adapter."""
+    entered = asyncio.Event()
+    release = asyncio.Event()
+
+    async def _blocked_injection(_event):
+        entered.set()
+        await release.wait()
+
+    adapter = SimpleNamespace(handle_message=AsyncMock(side_effect=_blocked_injection))
+    runner = _runner(adapter)
+    event = _async_event()
+    text = "completion"
+
+    async def _exercise():
+        first = asyncio.create_task(runner._deliver_completion_notification(text, dict(event)))
+        await entered.wait()
+        second = asyncio.create_task(runner._deliver_completion_notification(text, dict(event)))
+        await asyncio.sleep(0)
+        release.set()
+        return await asyncio.gather(first, second)
+
+    assert sorted(asyncio.run(_exercise()), key=str) == [None, True]
+    adapter.handle_message.assert_awaited_once()
+
+
+def test_failed_async_injection_is_retried_and_only_success_is_acked(
+    monkeypatch, isolated_registry,
+):
+    isolated = queue.Queue()
+    monkeypatch.setattr(isolated_registry, "completion_queue", isolated)
+    isolated.put(_async_event())
+
+    adapter = SimpleNamespace(
+        handle_message=AsyncMock(side_effect=[RuntimeError("temporary"), None])
+    )
+    runner = _runner(adapter)
+    _stop_after_sleeps(monkeypatch, runner, count=3)
+
+    from tools import async_delegation
+
+    acknowledgements = []
+    monkeypatch.setattr(
+        async_delegation,
+        "complete_completion_delivery",
+        lambda delegation_id, _claim_id: acknowledgements.append(delegation_id) or True,
+        raising=False,
+    )
+
+    asyncio.run(runner._async_delegation_watcher(interval=0))
+
+    assert adapter.handle_message.await_count == 2
+    assert acknowledgements == ["deleg_duplicate"]
+
+
+def test_distinct_process_incarnations_are_not_deduplicated():
+    """Producer spawn time distinguishes a reused process session ID."""
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+
+    async def _exercise():
+        first = await runner._deliver_completion_notification(
+            "first", _completion_event(started_at=10.0)
+        )
+        second = await runner._deliver_completion_notification(
+            "second", _completion_event(started_at=20.0)
+        )
+        return first, second
+
+    assert asyncio.run(_exercise()) == (True, True)
+
+    assert adapter.handle_message.await_count == 2
+
+
+def test_delivered_identity_retention_is_bounded():
+    """Lifecycle dedupe cannot grow without bound in a long-running gateway."""
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+    runner._completion_delivery_retention = 2
+    runner._completion_deliveries_delivered = OrderedDict()
+
+    async def _exercise():
+        for index in range(3):
+            await runner._deliver_completion_notification(
+                f"completion {index}",
+                _async_event(f"deleg_retention_{index}"),
+            )
+
+    asyncio.run(_exercise())
+
+    assert len(runner._completion_deliveries_delivered) == 2
+    assert ("async_delegation", "deleg_retention_0", "") not in (
+        runner._completion_deliveries_delivered
+    )
+    assert ("async_delegation", "deleg_retention_2", "") in (
+        runner._completion_deliveries_delivered
+    )
+
+
+def test_delivery_state_is_isolated_per_gateway_profile_lifecycle():
+    """A process-local claim in one profile never suppresses another runner."""
+    default_adapter = SimpleNamespace(handle_message=AsyncMock())
+    profile_adapter = SimpleNamespace(handle_message=AsyncMock())
+    default_runner = _runner(default_adapter)
+    profile_runner = _runner(profile_adapter)
+    event = _async_event("deleg_same_producer_id")
+
+    async def _exercise():
+        first = await default_runner._deliver_completion_notification(
+            "default", dict(event),
+        )
+        second = await profile_runner._deliver_completion_notification(
+            "profile", dict(event),
+        )
+        return first, second
+
+    assert asyncio.run(_exercise()) == (True, True)
+    default_adapter.handle_message.assert_awaited_once()
+    profile_adapter.handle_message.assert_awaited_once()
+
+
+def test_async_completion_uses_canonical_origin_routing(monkeypatch, isolated_registry):
+    isolated = queue.Queue()
+    monkeypatch.setattr(isolated_registry, "completion_queue", isolated)
+    event = _async_event("deleg_routing")
+    isolated.put(event)
+
+    canonical = SessionSource(
+        platform=Platform.TELEGRAM,
+        chat_id="canonical-chat",
+        chat_type="group",
+        thread_id="canonical-topic",
+    )
+    entry = SimpleNamespace(origin=canonical)
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter, origins={event["session_key"]: entry})
+    _stop_after_sleeps(monkeypatch, runner, count=2)
+
+    asyncio.run(runner._async_delegation_watcher(interval=0))
+
+    delivered = adapter.handle_message.await_args.args[0]
+    assert delivered.source == canonical
+
+
+def test_explicit_kill_returns_output_before_consuming_notification(monkeypatch):
+    import tools.process_registry as pr_module
+
+    registry = ProcessRegistry()
+    session = ProcessSession(
+        id="proc_kill_consumed",
+        command="sleep 999",
+        task_id="task",
+        started_at=1.0,
+        output_buffer="important terminal output\n",
+        notify_on_complete=True,
+    )
+    session.process = MagicMock()
+    session.process.pid = 4242
+    registry._running[session.id] = session
+    monkeypatch.setattr(registry, "_terminate_host_pid", lambda *_a, **_kw: None)
+    monkeypatch.setattr(registry, "_write_checkpoint", lambda: None)
+    monkeypatch.setattr(pr_module, "process_registry", registry)
+
+    result = registry.kill_process(session.id)
+    assert result["status"] == "killed"
+    assert result["output"] == "important terminal output\n"
+    assert registry.is_completion_consumed(session.id)
+
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+
+    async def _instant_sleep(*_a, **_kw):
+        pass
+
+    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
+    asyncio.run(runner._run_process_watcher({
+        "session_id": session.id,
+        "check_interval": 0,
+        "session_key": "agent:main:telegram:dm:123",
+        "platform": "telegram",
+        "chat_type": "dm",
+        "chat_id": "123",
+        "notify_on_complete": True,
+    }))
+
+    adapter.handle_message.assert_not_awaited()
+
+
+def test_process_tool_redacts_explicit_kill_output(monkeypatch):
+    from tools import process_registry as pr_module
+
+    registry = ProcessRegistry()
+    session = ProcessSession(
+        id="proc_kill_redacted",
+        command="printenv",
+        task_id="task",
+        started_at=1.0,
+        output_buffer="PRIVATE_TOKEN=opaque-value\n",
+        exited=True,
+        exit_code=0,
+    )
+    registry._finished[session.id] = session
+    monkeypatch.setattr(pr_module, "process_registry", registry)
+
+    def _redact(result):
+        assert result["output"] == "PRIVATE_TOKEN=opaque-value\n"
+        result["output"] = "PRIVATE_TOKEN=<redacted>\n"
+        return result
+
+    monkeypatch.setattr(pr_module, "_redact_process_result", _redact)
+
+    result = json.loads(pr_module._handle_process({
+        "action": "kill",
+        "session_id": session.id,
+    }))
+    assert result["output"] == "PRIVATE_TOKEN=<redacted>\n"
+
+
+def test_kill_of_already_exited_process_returns_output_before_consuming():
+    registry = ProcessRegistry()
+    session = ProcessSession(
+        id="proc_already_exited",
+        command="echo complete",
+        task_id="task",
+        started_at=1.0,
+        output_buffer="complete\n",
+        exited=True,
+        exit_code=0,
+    )
+    registry._finished[session.id] = session
+
+    result = registry.kill_process(session.id)
+
+    assert result["status"] == "already_exited"
+    assert result["output"] == "complete\n"
+    assert registry.is_completion_consumed(session.id)
+
+
+def test_read_log_only_consumes_when_terminal_output_page_is_observed():
+    registry = ProcessRegistry()
+    session = ProcessSession(
+        id="proc_paged_log",
+        command="printf lines",
+        task_id="task",
+        started_at=1.0,
+        output_buffer="first\nsecond\nfinal\n",
+        exited=True,
+        exit_code=0,
+    )
+    registry._finished[session.id] = session
+
+    middle_page = registry.read_log(session.id, offset=1, limit=1)
+    assert middle_page["output"] == "second"
+    assert not registry.is_completion_consumed(session.id)
+
+    final_page = registry.read_log(session.id, offset=2, limit=1)
+    assert final_page["output"] == "final"
+    assert registry.is_completion_consumed(session.id)
+
+
+def test_bulk_kill_does_not_consume_discarded_completion_output(monkeypatch):
+    registry = ProcessRegistry()
+    session = ProcessSession(
+        id="proc_bulk_kill",
+        command="sleep 999",
+        task_id="task",
+        started_at=1.0,
+        output_buffer="output bulk cleanup does not return\n",
+        notify_on_complete=True,
+    )
+    session.process = MagicMock()
+    session.process.pid = 4243
+    registry._running[session.id] = session
+    monkeypatch.setattr(registry, "_terminate_host_pid", lambda *_a, **_kw: None)
+    monkeypatch.setattr(registry, "_write_checkpoint", lambda: None)
+
+    assert registry.kill_all() == 1
+    assert not registry.is_completion_consumed(session.id)
+    queued = registry.completion_queue.get_nowait()
+    assert queued["session_id"] == session.id
+    assert queued["started_at"] == session.started_at
+    assert queued["output"] == "output bulk cleanup does not return\n"
+
+
+def test_unobserved_normal_completion_still_notifies(monkeypatch):
+    import tools.process_registry as pr_module
+
+    class _Registry:
+        def get(self, _session_id):
+            return SimpleNamespace(
+                output_buffer="done\n",
+                exited=True,
+                exit_code=0,
+                command="echo done",
+                started_at=1234.5,
+            )
+
+        def is_completion_consumed(self, _session_id):
+            return False
+
+    monkeypatch.setattr(pr_module, "process_registry", _Registry())
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+
+    async def _instant_sleep(*_a, **_kw):
+        pass
+
+    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
+    asyncio.run(runner._run_process_watcher({
+        "session_id": "proc_unobserved",
+        "check_interval": 0,
+        "session_key": "agent:main:telegram:dm:123",
+        "platform": "telegram",
+        "chat_type": "dm",
+        "chat_id": "123",
+        "notify_on_complete": True,
+    }))
+
+    adapter.handle_message.assert_awaited_once()
+
+
+def test_autonomous_completion_redacts_real_command_and_output_secrets(monkeypatch):
+    import agent.redact as redact_module
+    import tools.process_registry as pr_module
+
+    secret = "abc123randomopaquetokenvalue999"
+    registry = ProcessRegistry()
+    session = ProcessSession(
+        id="proc_autonomous_redaction",
+        command=f"printenv MY_SERVICE_TOKEN={secret}",
+        task_id="task",
+        started_at=1234.5,
+        output_buffer=f"MY_SERVICE_TOKEN={secret}\nHOME=/home/user\n",
+        exited=True,
+        exit_code=0,
+        notify_on_complete=True,
+    )
+    registry._finished[session.id] = session
+    monkeypatch.setattr(pr_module, "process_registry", registry)
+    monkeypatch.setattr(redact_module, "_REDACT_ENABLED", True)
+
+    adapter = SimpleNamespace(handle_message=AsyncMock())
+    runner = _runner(adapter)
+
+    async def _instant_sleep(*_a, **_kw):
+        pass
+
+    monkeypatch.setattr(asyncio, "sleep", _instant_sleep)
+    asyncio.run(runner._run_process_watcher({
+        "session_id": session.id,
+        "check_interval": 0,
+        "session_key": "agent:main:telegram:dm:123",
+        "platform": "telegram",
+        "chat_type": "dm",
+        "chat_id": "123",
+        "notify_on_complete": True,
+    }))
+
+    delivered = adapter.handle_message.await_args.args[0]
+    assert secret not in delivered.text
+    assert "HOME=/home/user" in delivered.text

tests/gateway/test_cron_fire_webhook.py

@@ -8,6 +8,9 @@ test_chronos_verify.py.
 """
 
 import asyncio
+import threading
+from types import SimpleNamespace
+from unittest.mock import patch
 
 import pytest
 from aiohttp import web
@@ -108,6 +111,86 @@ async def test_missing_token_401(adapter, monkeypatch):
     assert spy.fired == []
 
 
+@pytest.mark.asyncio
+async def test_valid_token_refuses_during_gateway_drain(adapter, monkeypatch):
+    spy = _SpyProvider()
+    runner = SimpleNamespace(_draining=False, _external_drain_active=True)
+    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", lambda: spy)
+    monkeypatch.setattr(
+        "plugins.cron_providers.chronos.verify.get_fire_verifier",
+        lambda: (lambda **kw: {"purpose": "cron_fire"}),
+    )
+
+    app = _create_app(adapter)
+    with patch("gateway.run._gateway_runner_ref", lambda: runner):
+        async with TestClient(TestServer(app)) as cli:
+            response = await cli.post(
+                "/api/cron/fire",
+                headers={"Authorization": "Bearer good"},
+                json={"job_id": "abc123"},
+            )
+            payload = await response.json()
+
+    assert response.status == 503
+    assert payload["error"]["code"] == "gateway_draining"
+    assert spy.fired == []
+
+
+@pytest.mark.asyncio
+async def test_valid_fire_reservation_blocks_drain_before_body_and_task(adapter, monkeypatch):
+    runner = SimpleNamespace(_draining=False, _external_drain_active=False)
+    body_started = asyncio.Event()
+    release_body = asyncio.Event()
+    fired = threading.Event()
+    release_fire = threading.Event()
+
+    class BlockingProvider:
+        def fire_due(self, job_id, *, adapters=None, loop=None):
+            fired.set()
+            release_fire.wait(timeout=2)
+            return True
+
+    original_json = web.Request.json
+
+    async def delayed_json(request):
+        body_started.set()
+        await release_body.wait()
+        return await original_json(request)
+
+    monkeypatch.setattr("cron.scheduler_provider.resolve_cron_scheduler", BlockingProvider)
+    monkeypatch.setattr(
+        "plugins.cron_providers.chronos.verify.get_fire_verifier",
+        lambda: (lambda **kw: {"purpose": "cron_fire"}),
+    )
+    app = _create_app(adapter)
+    with patch("gateway.run._gateway_runner_ref", lambda: runner), patch.object(
+        web.Request, "json", delayed_json
+    ):
+        async with TestClient(TestServer(app)) as cli:
+            request_task = asyncio.create_task(
+                cli.post(
+                    "/api/cron/fire",
+                    headers={"Authorization": "Bearer good"},
+                    json={"job_id": "abc123"},
+                )
+            )
+            await body_started.wait()
+            assert adapter.active_agent_work_count() == 1
+
+            release_body.set()
+            response = await request_task
+            assert response.status == 202
+            await asyncio.to_thread(fired.wait, 2)
+            assert adapter.active_agent_work_count() == 1
+            release_fire.set()
+            for _ in range(50):
+                if adapter.active_agent_work_count() == 0:
+                    break
+                await asyncio.sleep(0.01)
+
+    assert adapter.active_agent_work_count() == 0
+
+
 @pytest.mark.asyncio
 async def test_missing_job_id_400(adapter, monkeypatch):
     """Valid token but no job_id → 400, no fire."""

tests/gateway/test_external_drain_control.py

@@ -19,6 +19,7 @@ import pytest
 
 import gateway.drain_control as dc
 from gateway.run import GatewayRunner
+from gateway.config import Platform
 from gateway.platforms.base import MessageEvent, MessageType
 from tests.gateway.restart_test_helpers import make_restart_runner, make_restart_source
 
@@ -238,6 +239,16 @@ def _drain_runner():
 
 
 class TestDrainStateMachine:
+    def test_active_work_count_includes_api_and_cron_work(self, monkeypatch):
+        runner, _ = _drain_runner()
+        runner.adapters = {
+            Platform.API_SERVER: MagicMock(active_agent_work_count=MagicMock(return_value=2))
+        }
+        runner._running_agents = {"session": MagicMock()}
+        monkeypatch.setattr("cron.scheduler.get_running_job_ids", lambda: {"job-1"})
+
+        assert runner._active_work_count() == 4
+
     def test_enter_sets_flag_and_flips_state(self):
         runner, _ = _drain_runner()
         runner._enter_external_drain()
@@ -289,6 +300,26 @@ class TestDrainStateMachine:
 
 
 class TestDrainWatcher:
+    @pytest.mark.asyncio
+    async def test_watcher_persists_aggregate_work_during_external_drain(self, home, monkeypatch):
+        runner, _ = _drain_runner()
+        runner._drain_control_watcher = GatewayRunner._drain_control_watcher.__get__(
+            runner, GatewayRunner
+        )
+        runner._persist_active_agents = MagicMock()
+        dc.write_drain_request()
+        task = asyncio.create_task(runner._drain_control_watcher(interval=0.01))
+        await asyncio.sleep(0.03)
+        runner._running = False
+        await asyncio.sleep(0.02)
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+        runner._persist_active_agents.assert_called()
+
     @pytest.mark.asyncio
     async def test_watcher_enters_then_exits_with_marker(self, home):
         runner, _ = _drain_runner()

tests/gateway/test_platform_base.py

@@ -786,6 +786,28 @@ class TestMediaDeliveryPathValidation:
 
         assert BasePlatformAdapter.validate_media_delivery_path(str(media_file)) == str(media_file.resolve())
 
+    def test_allows_stale_kanban_attachment_but_not_neighboring_workspace(
+        self, tmp_path, monkeypatch,
+    ):
+        """Strict mode trusts durable attachments without trusting scratch."""
+        self._patch_roots(monkeypatch)
+        monkeypatch.setenv("HERMES_KANBAN_HOME", str(tmp_path / "hermes"))
+        monkeypatch.setenv("HERMES_MEDIA_TRUST_RECENT_FILES", "0")
+        board_root = tmp_path / "hermes" / "kanban" / "boards" / "research"
+        board_root.mkdir(parents=True)
+        (board_root / "kanban.db").touch()
+        attachment = board_root / "attachments" / "t_12345678" / "report.pdf"
+        scratch = board_root / "workspaces" / "t_12345678" / "notes.txt"
+        attachment.parent.mkdir(parents=True)
+        scratch.parent.mkdir(parents=True)
+        attachment.write_bytes(b"%PDF")
+        scratch.write_text("private", encoding="utf-8")
+
+        assert BasePlatformAdapter.validate_media_delivery_path(str(attachment)) == str(
+            attachment.resolve()
+        )
+        assert BasePlatformAdapter.validate_media_delivery_path(str(scratch)) is None
+
     def test_recency_trust_allows_freshly_produced_file(self, tmp_path, monkeypatch):
         """A PDF the agent just wrote to /tmp should be deliverable.
 

tests/gateway/test_shutdown_cache_cleanup.py

@@ -60,6 +60,11 @@ class _FakeGateway:
         # there's never in-flight cron work to report.
         return 0
 
+    def _active_api_run_count(self):
+        # The shutdown log also reports adapter-owned API work (#63529).
+        # This fake has no API server adapter, so it is always idle.
+        return 0
+
     def _update_runtime_status(self, *_a, **_kw):
         pass