NousResearch/hermes-agent

Background process notifications routed to wrong TUI session

Cluster 35652 · agent. Data as of 2026-10-27 11:58 UTC

1 open item. Fix merged in #65419 on 16 Jul 2026. Check it covers your case.

open issues

open PR

items grouped

Known?: Yes. 14 related items grouped here, 4 issues already closed.
Fix event: A fix was merged to main in PR #65419. Check it covers your case. No release data is tracked.
Tracker: The maintainers' tracker marks this partially resolved, deciding PR #65419. That is separate from the fix event.
Still open: 0 issues and 1 PR.

Next step: read #65419, then check it covers your case.

Read #65419 Pick up a Lab task

Most members concern background process or completion notifications reaching a session other than the one that started them, mainly in the TUI, with some gateway and delegation variants. One member (desktop session hygiene, archive, media streaming) appears unrelated to this theme.

Open issues0

No open issues.

Open PRs1

PR#61719fix(tui): isolate background notifications by UI session

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

#37099 feat(desktop): session hygiene, archive, media streaming + connecting overlay

merged to main

files: apps/desktop/electron/main.cjs, apps/desktop/src/app/chat/index.tsx, apps/desktop/src/app/chat/sidebar/index.tsx, apps/desktop/src/app/chat/sidebar/session-actions-menu.tsx, apps/desktop/src/app/chat/sidebar/session-row.tsx, apps/desktop/src/app/chat/sidebar/virtual-session-list.tsx, apps/desktop/src/app/command-center/index.tsx, apps/desktop/src/app/desktop-controller.tsx, apps/desktop/src/app/session/hooks/use-session-actions.ts, apps/desktop/src/app/settings/constants.ts, apps/desktop/src/app/settings/index.tsx, apps/desktop/src/app/settings/sessions-settings.tsx, apps/desktop/src/app/settings/types.ts, apps/desktop/src/components/assistant-ui/markdown-text.tsx, apps/desktop/src/components/gateway-connecting-overlay.tsx, apps/desktop/src/hermes.ts, apps/desktop/src/lib/icons.ts, apps/desktop/src/lib/media.ts, apps/desktop/src/types/hermes.ts, hermes_cli/web_server.py (+6 more)

Show the diff

apps/desktop/electron/main.cjs

@@ -8,6 +8,8 @@ const {
   ipcMain,
   nativeImage,
   nativeTheme,
+  net: electronNet,
+  protocol,
   safeStorage,
   session,
   shell,
@@ -364,6 +366,66 @@ app.setAboutPanelOptions({
   copyright: 'Copyright © 2026 Nous Research'
 })
 
+// Custom scheme for streaming local media (video/audio) into the renderer.
+// Reading large media through `readFileDataUrl` failed: it base64-loads the
+// whole file into memory and is hard-capped at DATA_URL_READ_MAX_BYTES (16 MB),
+// so any non-trivial video silently refused to load. Streaming via a protocol
+// handler removes the size cap and gives the <video> element seekable,
+// range-aware playback. Must be registered before the app is ready.
+const MEDIA_PROTOCOL = 'hermes-media'
+// Only audio/video may be streamed. Without this the handler would read any
+// non-blocklisted local file (no size cap) for any `fetch(hermes-media://…)`.
+const STREAMABLE_MEDIA_EXTS = new Set([
+  '.avi',
+  '.flac',
+  '.m4a',
+  '.mkv',
+  '.mov',
+  '.mp3',
+  '.mp4',
+  '.ogg',
+  '.opus',
+  '.wav',
+  '.webm'
+])
+
+protocol.registerSchemesAsPrivileged([
+  {
+    scheme: MEDIA_PROTOCOL,
+    privileges: {
+      secure: true,
+      standard: true,
+      stream: true,
+      supportFetchAPI: true
+    }
+  }
+])
+
+function registerMediaProtocol() {
+  protocol.handle(MEDIA_PROTOCOL, async request => {
+    let resolvedPath
+    try {
+      const url = new URL(request.url)
+      const filePath = decodeURIComponent(url.pathname.replace(/^\/+/, ''))
+      ;({ resolvedPath } = await resolveReadableFileForIpc(filePath, { purpose: 'Media stream' }))
+    } catch {
+      return new Response('Media not found', { status: 404 })
+    }
+
+    if (!STREAMABLE_MEDIA_EXTS.has(path.extname(resolvedPath).toLowerCase())) {
+      return new Response('Unsupported media type', { status: 415 })
+    }
+
+    // Delegate to Electron's net stack on a file:// URL — it resolves the
+    // content-type and honors Range requests so seeking works. Forward the
+    // renderer's headers (notably Range) and skip custom-protocol re-entry.
+    return electronNet.fetch(pathToFileURL(resolvedPath).toString(), {
+      bypassCustomProtocolHandlers: true,
+      headers: request.headers
+    })
+  })
+}
+
 let mainWindow = null
 let hermesProcess = null
 let connectionPromise = null
@@ -3654,6 +3716,7 @@ app.whenReady().then(() => {
     Menu.setApplicationMenu(null)
   }
   installMediaPermissions()
+  registerMediaProtocol()
   ensureWslWindowsFonts()
   createWindow()

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

@@ -97,7 +97,7 @@ function ChatHeader({
   const sessions = useStore($sessions)
   const pinnedSessionIds = useStore($pinnedSessionIds)
   const activeStoredSession = sessions.find(session => session.id === selectedSessionId) || null
-  const title = activeStoredSession ? sessionTitle(activeStoredSession) : 'New agent'
+  const title = activeStoredSession ? sessionTitle(activeStoredSession) : 'New session'
   const selectedIsPinned = selectedSessionId ? pinnedSessionIds.includes(selectedSessionId) : false
 
   return (

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

@@ -67,7 +67,12 @@ import { VirtualSessionList } from './virtual-session-list'
 const VIRTUALIZE_THRESHOLD = 25
 
 const SIDEBAR_NAV: SidebarNavItem[] = [
-  { id: 'new-session', label: 'New agent', icon: props => <Codicon name="robot" {...props} />, action: 'new-session' },
+  {
+    id: 'new-session',
+    label: 'New session',
+    icon: props => <Codicon name="robot" {...props} />,
+    action: 'new-session'
+  },
   { id: 'skills', label: 'Skills', icon: props => <Codicon name="symbol-misc" {...props} />, route: SKILLS_ROUTE },
   { id: 'messaging', label: 'Messaging', icon: props => <Codicon name="comment" {...props} />, route: MESSAGING_ROUTE },
   { id: 'artifacts', label: 'Artifacts', icon: props => <Codicon name="files" {...props} />, route: ARTIFACTS_ROUTE }
@@ -149,6 +154,8 @@ interface ChatSidebarProps extends React.ComponentProps<typeof Sidebar> {
   onLoadMoreSessions: () => void
   onResumeSession: (sessionId: string) => void
   onDeleteSession: (sessionId: string) => void
+  onArchiveSession: (sessionId: string) => void
+  onNewSessionInWorkspace: (path: null | string) => void
 }
 
 export function ChatSidebar({
@@ -156,7 +163,9 @@ export function ChatSidebar({
   onNavigate,
   onLoadMoreSessions,
   onResumeSession,
-  onDeleteSession
+  onDeleteSession,
+  onArchiveSession,
+  onNewSessionInWorkspace
 }: ChatSidebarProps) {
   const sidebarOpen = useStore($sidebarOpen)
   const agentsGrouped = useStore($sidebarAgentsGrouped)
@@ -328,6 +337,7 @@ export function ChatSidebar({
             dndSensors={dndSensors}
             emptyState={<SidebarPinnedEmptyState />}
             label="Pinned"
+            onArchiveSession={onArchiveSession}
             onDeleteSession={onDeleteSession}
             onReorder={handlePinnedDragEnd}
             onResumeSession={onResumeSession}
@@ -361,9 +371,9 @@ export function ChatSidebar({
             groups={agentsGrouped ? agentGroups : undefined}
             headerAction={
               <Button
-                aria-label={agentsGrouped ? 'Show agents as a single list' : 'Group agents by workspace'}
+                aria-label={agentsGrouped ? 'Show sessions as a single list' : 'Group sessions by workspace'}
                 className={cn(
-                  'cursor-pointer text-(--ui-text-tertiary) opacity-0 hover:bg-(--ui-control-hover-background) hover:text-foreground hover:opacity-100 focus-visible:opacity-100 group-hover/section:opacity-100',
+                  'cursor-pointer text-(--ui-text-tertiary) opacity-70 hover:bg-(--ui-control-hover-background) hover:text-foreground hover:opacity-100 focus-visible:opacity-100',
                   agentsGrouped && 'bg-(--ui-control-active-background) text-foreground opacity-100'
                 )}
                 onClick={event => {
@@ -372,15 +382,17 @@ export function ChatSidebar({
                   setSidebarAgentsGrouped(!agentsGrouped)
                 }}
                 size="icon-xs"
-                title={agentsGrouped ? 'Ungroup agents' : 'Group by workspace'}
+                title={agentsGrouped ? 'Ungroup sessions' : 'Group by workspace'}
                 variant="ghost"
               >
                 <Codicon name={agentsGrouped ? 'list-unordered' : 'root-folder'} size="0.75rem" />
               </Button>
             }
-            label="Agents"
+            label="Sessions"
             labelMeta={countLabel(agentSessions.length, knownSessionTotal)}
+            onArchiveSession={onArchiveSession}
             onDeleteSession={onDeleteSession}
+            onNewSessionInWorkspace={onNewSessionInWorkspace}
             onReorder={handleAgentDragEnd}
             onResumeSession={onResumeSession}
             onToggle={() => setSidebarRecentsOpen(!agentsOpen)}
@@ -472,7 +484,9 @@ interface SidebarSessionsSectionProps {
   workingSessionIdSet: Set<string>
   onResumeSession: (sessionId: string) => void
   onDeleteSession: (sessionId: string) => void
+  onArchiveSession: (sessionId: string) => void
   onTogglePin: (sessionId: string) => void
+  onNewSessionInWorkspace?: (path: null | string) => void
   pinned: boolean
   rootClassName?: string
   contentClassName?: string
@@ -496,7 +510,9 @@ function SidebarSessionsSection({
   workingSessionIdSet,
   onResumeSession,
   onDeleteSession,
+  onArchiveSession,
   onTogglePin,
+  onNewSessionInWorkspace,
   pinned,
   rootClassName,
   contentClassName,
@@ -518,6 +534,7 @@ function SidebarSessionsSection({
       isPinned: pinned,
       isSelected: session.id === activeSessionId,
       isWorking: workingSessionIdSet.has(session.id),
+      onArchive: () => onArchiveSession(session.id),
       onDelete: () => onDeleteSession(session.id),
       onPin: () => onTogglePin(session.id),
       onResume: () => onResumeSession(session.id),
@@ -551,9 +568,19 @@ function SidebarSessionsSection({
   } else if (groups?.length) {
     const groupNodes = groups.map(group =>
       dndActive ? (
-        <SortableSidebarWorkspaceGroup group={group} key={group.id} renderRows={renderSessionList} />
+        <SortableSidebarWorkspaceGroup
+          group={group}
+          key={group.id}
+          onNewSession={onNewSessionInWorkspace}
+          renderRows={renderSessionList}
+        />
       ) : (
-        <SidebarWorkspaceGroup group={group} key={group.id} renderRows={renderSessionList} />
+        <SidebarWorkspaceGroup
+          group={group}
+          key={group.id}
+          onNewSession={onNewSessionInWorkspace}
+          renderRows={renderSessionList}
+        />
       )
     )
 
@@ -568,6 +595,7 @@ function SidebarSessionsSection({
     inner = (
       <VirtualSessionList
         activeSessionId={activeSessionId}
+        onArchiveSession={onArchiveSession}
         onDeleteSession={onDeleteSession}
         onResumeSession={onResumeSession}
         onTogglePin={onTogglePin}
@@ -610,6 +638,7 @@ function SidebarSessionsSection({
 interface SidebarWorkspaceGroupProps extends React.ComponentProps<'div'> {
   group: SidebarSessionGroup
   renderRows: (sessions: SessionInfo[]) => React.ReactNode
+  onNewSession?: (path: null | string) => void
   reorderable?: boolean
   dragging?: boolean
   dragHandleProps?: React.HTMLAttributes<HTMLElement>
@@ -618,6 +647,7 @@ interface SidebarWorkspaceGroupProps extends React.ComponentProps<'div'> {
 function SidebarWorkspaceGroup({
   group,
   renderRows,
+  onNewSession,
   reorderable = false,
   dragging = false,
   dragHandleProps,
@@ -634,18 +664,31 @@ function SidebarWorkspaceGroup({
 
   return (
     <div className={cn('grid gap-px', dragging && 'z-10 opacity-60', className)} ref={ref} style={style} {...rest}>
-      <button
-        className="group/workspace flex min-h-6 cursor-pointer items-center gap-1 px-2 pt-1 text-left text-[0.6875rem] font-medium text-(--ui-text-tertiary) hover:text-(--ui-text-secondary)"
-        onClick={() => setOpen(value => !value)}
-        title={group.path ?? undefined}
-        type="button"
-      >
-        <span className="truncate">{group.label}</span>
-        <SidebarCount>{group.sessions.length}</SidebarCount>
-        <DisclosureCaret
-          className="text-(--ui-text-tertiary) opacity-0 transition group-hover/workspace:opacity-100"
-          open={open}
-        />
+      <div className="group/workspace flex min-h-6 items-center gap-1 px-2 pt-1 text-[0.6875rem] font-medium text-(--ui-text-tertiary)">
+        <button
+          className="flex min-w-0 cursor-pointer items-center gap-1 bg-transparent text-left hover:text-(--ui-text-secondary)"
+          onClick={() => setOpen(value => !value)}
+          title={group.path ?? undefined}
+          type="button"
+        >
+          <span className="truncate">{group.label}</span>
+          <SidebarCount>{group.sessions.length}</SidebarCount>
+          <DisclosureCaret
+            className="text-(--ui-text-tertiary) opacity-0 transition group-hover/workspace:opacity-100"
+            open={open}
+          />
+        </button>
+        {onNewSession && (
+          <button
+            aria-label={`New session in ${group.label}`}
+            className="grid size-4 shrink-0 cursor-pointer place-items-center rounded-sm bg-transparent text-(--ui-text-quaternary) opacity-0 transition-opacity hover:bg-(--ui-control-hover-background) hover:text-foreground group-hover/workspace:opacity-100"
+            onClick={() => onNewSession(group.path)}
+            title={`New session in ${group.label}`}
+            type="button"
+          >
+            <Codicon name="add" size="0.75rem" />
+          </button>
+        )}
         {reorderable && (
           <span
             {...dragHandleProps}
@@ -663,7 +706,7 @@ function SidebarWorkspaceGroup({
             />
           </span>
         )}
-      </button>
+      </div>
       {open && (
         <>
           {renderRows(visibleSessions)}
@@ -687,6 +730,7 @@ function SidebarWorkspaceGroup({
 interface SortableWorkspaceProps {
   group: SidebarSessionGroup
   renderRows: (sessions: SessionInfo[]) => React.ReactNode
+  onNewSession?: (path: null | string) => void
 }
 
 function SortableSidebarWorkspaceGroup(props: SortableWorkspaceProps) {
@@ -702,6 +746,7 @@ interface SortableSessionRowProps {
   isPinned: boolean
   isSelected: boolean
   isWorking: boolean
+  onArchive: () => void
   onDelete: () => void
   onPin: () => void
   onResume: () => void

apps/desktop/src/app/chat/sidebar/session-actions-menu.tsx

@@ -26,6 +26,7 @@ interface SessionActions {
   title: string
   pinned?: boolean
   onPin?: () => void
+  onArchive?: () => void
   onDelete?: () => void
 }
 
@@ -40,7 +41,7 @@ interface ItemSpec {
   variant?: 'destructive'
 }
 
-function useSessionActions({ sessionId, title, pinned = false, onPin, onDelete }: SessionActions) {
+function useSessionActions({ sessionId, title, pinned = false, onPin, onArchive, onDelete }: SessionActions) {
   const [renameOpen, setRenameOpen] = useState(false)
 
   const items: ItemSpec[] = [
@@ -81,6 +82,15 @@ function useSessionActions({ sessionId, title, pinned = false, onPin, onDelete }
         setRenameOpen(true)
       }
     },
+    {
+      disabled: !onArchive,
+      icon: 'archive',
+      label: 'Archive',
+      onSelect: () => {
+        triggerHaptic('selection')
+        onArchive?.()
+      }
+    },
     {
       className: 'text-destructive focus:text-destructive',
       disabled: !onDelete,

apps/desktop/src/app/chat/sidebar/session-row.tsx

@@ -14,6 +14,7 @@ interface SidebarSessionRowProps extends React.ComponentProps<'div'> {
   isPinned: boolean
   isSelected: boolean
   isWorking: boolean
+  onArchive: () => void
   onDelete: () => void
   onPin: () => void
   onResume: () => void
@@ -45,6 +46,7 @@ export function SidebarSessionRow({
   isPinned,
   isSelected,
   isWorking,
+  onArchive,
   onDelete,
   onPin,
   onResume,
@@ -61,7 +63,14 @@ export function SidebarSessionRow({
   const handleLabel = `Reorder ${title}`
 
   return (
-    <SessionContextMenu onDelete={onDelete} onPin={onPin} pinned={isPinned} sessionId={session.id} title={title}>
+    <SessionContextMenu
+      onArchive={onArchive}
+      onDelete={onDelete}
+      onPin={onPin}
+      pinned={isPinned}
+      sessionId={session.id}
+      title={title}
+    >
       <div
         className={cn(
           'group relative grid min-h-[1.625rem] cursor-pointer grid-cols-[minmax(0,1fr)_1.375rem] items-center rounded-md transition-colors duration-100 ease-out hover:bg-(--ui-row-hover-background) hover:transition-none',
@@ -88,6 +97,15 @@ export function SidebarSessionRow({
               return
             }
 
+            if (event.metaKey || event.ctrlKey) {
+              event.preventDefault()
+              event.stopPropagation()
+              triggerHaptic('selection')
+              onArchive()
+
+              return
+            }
+
             onResume()
           }}
           type="button"
@@ -127,7 +145,14 @@ export function SidebarSessionRow({
               {age}
             </span>
           )}
-          <SessionActionsMenu onDelete={onDelete} onPin={onPin} pinned={isPinned} sessionId={session.id} title={title}>
+          <SessionActionsMenu
+            onArchive={onArchive}
+            onDelete={onDelete}
+            onPin={onPin}
+            pinned={isPinned}
+            sessionId={session.id}
+            title={title}
+          >
             <Button
               aria-label={`Actions for ${title}`}
               className="size-5 rounded-md bg-transparent text-transparent transition-colors duration-100 hover:bg-(--ui-control-active-background) hover:text-foreground focus-visible:bg-(--ui-control-active-background) focus-visible:text-foreground focus-visible:ring-0 data-[state=open]:bg-(--ui-control-active-background) data-[state=open]:text-foreground group-hover:text-(--ui-text-tertiary) [&_svg]:size-3.5!"

apps/desktop/src/app/chat/sidebar/virtual-session-list.tsx

@@ -12,6 +12,7 @@ interface SessionRowCommonProps {
   isPinned: boolean
   isSelected: boolean
   isWorking: boolean
+  onArchive: () => void
   onDelete: () => void
   onPin: () => void
   onResume: () => void
@@ -20,6 +21,7 @@ interface SessionRowCommonProps {
 interface VirtualSessionListProps {
   activeSessionId: null | string
   className?: string
+  onArchiveSession: (sessionId: string) => void
   onDeleteSession: (sessionId: string) => void
   onResumeSession: (sessionId: string) => void
   onTogglePin: (sessionId: string) => void
@@ -35,6 +37,7 @@ const OVERSCAN_ROWS = 12
 export const VirtualSessionList: FC<VirtualSessionListProps> = ({
   activeSessionId,
   className,
+  onArchiveSession,
   onDeleteSession,
   onResumeSession,
   onTogglePin,
@@ -72,6 +75,7 @@ export const VirtualSessionList: FC<VirtualSessionListProps> = ({
       isPinned: pinned,
       isSelected: session.id === activeSessionId,
       isWorking: workingSessionIdSet.has(session.id),
+      onArchive: () => onArchiveSession(session.id),
       onDelete: () => onDeleteSession(session.id),
       onPin: () => onTogglePin(session.id),
       onResume: () => onResumeSession(session.id)

apps/desktop/src/app/command-center/index.tsx

@@ -113,7 +113,7 @@ interface SectionSearchEntry {
 }
 
 const NAVIGATION_SEARCH_ENTRIES: readonly NavigationSearchEntry[] = [
-  { id: 'nav-new-chat', route: NEW_CHAT_ROUTE, title: 'New agent', detail: 'Start a fresh session' },
+  { id: 'nav-new-chat', route: NEW_CHAT_ROUTE, title: 'New session', detail: 'Start a fresh session' },
   { id: 'nav-settings', route: SETTINGS_ROUTE, title: 'Settings', detail: 'Configure Hermes desktop' },
   { id: 'nav-skills', route: SKILLS_ROUTE, title: 'Skills', detail: 'Enable and inspect skills' },
   {

apps/desktop/src/app/desktop-controller.tsx

@@ -6,6 +6,7 @@ import { Navigate, Route, Routes, useLocation, useNavigate, useParams } from 're
 import { BootFailureOverlay } from '@/components/boot-failure-overlay'
 import { DesktopInstallOverlay } from '@/components/desktop-install-overlay'
 import { DesktopOnboardingOverlay } from '@/components/desktop-onboarding-overlay'
+import { GatewayConnectingOverlay } from '@/components/gateway-connecting-overlay'
 import { Pane, PaneMain } from '@/components/pane-shell'
 import { useSkinCommand } from '@/themes/use-skin-command'
 
@@ -33,6 +34,8 @@ import {
   $selectedStoredSessionId,
   setAwaitingResponse,
   setBusy,
+  setCurrentBranch,
+  setCurrentCwd,
   setCurrentModel,
   setCurrentProvider,
   setMessages,
@@ -122,6 +125,7 @@ export function DesktopController() {
     settingsOpen,
     toggleCommandCenter
   } = useOverlayRouting()
+
   const terminalTakeoverActive = chatOpen && terminalTakeover
 
   const titlebarToolGroups = useGroupRegistry<TitlebarTool>()
@@ -192,7 +196,10 @@ export function DesktopController() {
 
     try {
       const limit = $sessionsLimit.get()
-      const result = await listSessions(limit)
+      // Require at least one message so abandoned/empty "Untitled" drafts (one
+      // was created per TUI/desktop launch before the lazy-create fix) don't
+      // clutter the sidebar.
+      const result = await listSessions(limit, 1)
 
       if (refreshSessionsRequestRef.current === requestId) {
         setSessions(result.sessions)
@@ -324,6 +331,7 @@ export function DesktopController() {
   })
 
   const {
+    archiveSession,
     branchCurrentSession,
     createBackendSessionForSend,
     openSettings,
@@ -392,6 +400,29 @@ export function DesktopController() {
     [branchCurrentSession, refreshSessions]
   )
 
+  const startSessionInWorkspace = useCallback(
+    (path: null | string) => {
+      startFreshSessionDraft()
+
+      const target = path?.trim()
+
+      if (!target) {
+        return
+      }
+
+      // The next message creates the backend session in $currentCwd, so seed
+      // it (and the branch) from the workspace the user clicked the + on.
+      setCurrentCwd(target)
+      void requestGateway<{ branch?: string; cwd?: string }>('config.get', { key: 'project', cwd: target })
+        .then(info => {
+          setCurrentCwd(info.cwd || target)
+          setCurrentBranch(info.branch || '')
+        })
+        .catch(() => undefined)
+    },
+    [requestGateway, startFreshSessionDraft]
+  )
+
   const handleSkinCommand = useSkinCommand()
 
   const { cancelRun, editMessage, handleThreadMessagesChange, reloadFromMessage, submitText, transcribeVoiceAudio } =
@@ -461,9 +492,11 @@ export function DesktopController() {
   const sidebar = (
     <ChatSidebar
       currentView={currentView}
+      onArchiveSession={sessionId => void archiveSession(sessionId)}
       onDeleteSession={sessionId => void removeSession(sessionId)}
       onLoadMoreSessions={loadMoreSessions}
       onNavigate={selectSidebarItem}
+      onNewSessionInWorkspace={startSessionInWorkspace}
       onResumeSession={sessionId => navigate(sessionRoute(sessionId))}
     />
   )
@@ -485,6 +518,7 @@ export function DesktopController() {
       />
       <ModelPickerOverlay gateway={gatewayRef.current || undefined} onSelect={selectModel} />
       <UpdatesOverlay />
+      <GatewayConnectingOverlay />
       <BootFailureOverlay />
 
       {settingsOpen && (
@@ -575,10 +609,10 @@ export function DesktopController() {
       titlebarTools={titlebarToolGroups.flat.right}
     >
       <Pane
+        disabled={terminalTakeoverActive}
         id="chat-sidebar"
         maxWidth={SIDEBAR_MAX_WIDTH}
         minWidth={SIDEBAR_DEFAULT_WIDTH}
-        disabled={terminalTakeoverActive}
         resizable
         side="left"
         width={`${SIDEBAR_DEFAULT_WIDTH}px`}

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

@@ -2,7 +2,7 @@ import type { MutableRefObject } from 'react'
 import { useCallback, useRef } from 'react'
 import type { NavigateFunction } from 'react-router-dom'
 
-import { deleteSession, getSessionMessages } from '@/hermes'
+import { deleteSession, getSessionMessages, setSessionArchived } from '@/hermes'
 import { type ChatMessage, chatMessageText, preserveLocalAssistantErrors, toChatMessages } from '@/lib/chat-messages'
 import { normalizePersonalityValue } from '@/lib/chat-runtime'
 import { embeddedImageUrls, textWithoutEmbeddedImages } from '@/lib/embedded-images'
@@ -751,7 +751,39 @@ export function useSessionActions({
     ]
   )
 
+  const archiveSession = useCallback(
+    async (storedSessionId: string) => {
+      clearNotifications()
+
+      const archived = $sessions.get().find(s => s.id === storedSessionId)
+      const wasSelected = selectedStoredSessionId === storedSessionId
+      const previousPinned = $pinnedSessionIds.get()
+
+      // Soft-hide: drop from the sidebar immediately, keep the data.
+      setSessions(prev => prev.filter(s => s.id !== storedSessionId))
+      $pinnedSessionIds.set(previousPinned.filter(id => id !== storedSessionId))
+
+      if (wasSelected) {
+        startFreshSessionDraft(true)
+      }
+
+      try {
+        await setSessionArchived(storedSessionId, true)
+        notify({ durationMs: 2_000, kind: 'success', message: 'Archived' })
+      } catch (err) {
+        if (archived) {
+          setSessions(prev => [archived, ...prev.filter(s => s.id !== storedSessionId)])
+        }
+
+        $pinnedSessionIds.set(previousPinned)
+        notifyError(err, 'Archive failed')
+      }
+    },
+    [selectedStoredSessionId, startFreshSessionDraft]
+  )
+
   return {
+    archiveSession,
     branchCurrentSession,
     closeSettings,
     createBackendSessionForSend,

apps/desktop/src/app/settings/constants.ts

@@ -311,11 +311,15 @@ export const MODE_OPTIONS: ModeOption[] = [
   { id: 'system', label: 'System', description: 'Follow OS appearance', icon: Monitor }
 ]
 
-export const SEARCH_PLACEHOLDER: Record<'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'tools', string> = {
+export const SEARCH_PLACEHOLDER: Record<
+  'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'sessions' | 'tools',
+  string
+> = {
   about: 'About Hermes Desktop',
   config: 'Search settings...',
   gateway: 'Gateway connection...',
   keys: 'Search API keys...',
   mcp: 'Search MCP servers...',
+  sessions: 'Search archived sessions...',
   tools: 'Search skills and tools...'
 }

apps/desktop/src/app/settings/index.tsx

@@ -3,7 +3,7 @@ import { useEffect, useRef, useState } from 'react'
 
 import { getHermesConfigDefaults, getHermesConfigRecord, saveHermesConfig } from '@/hermes'
 import { triggerHaptic } from '@/lib/haptics'
-import { Globe, Info, KeyRound, Package, Wrench } from '@/lib/icons'
+import { Archive, Globe, Info, KeyRound, Package, Wrench } from '@/lib/icons'
 import { notifyError } from '@/store/notifications'
 
 import { useRouteEnumParam } from '../hooks/use-route-enum-param'
@@ -19,6 +19,7 @@ import { SEARCH_PLACEHOLDER, SECTIONS } from './constants'
 import { GatewaySettings } from './gateway-settings'
 import { KeysSettings } from './keys-settings'
 import { McpSettings } from './mcp-settings'
+import { SessionsSettings } from './sessions-settings'
 import { ToolsSettings } from './tools-settings'
 import type { SettingsPageProps, SettingsQueryKey, SettingsView as SettingsViewId } from './types'
 
@@ -27,6 +28,7 @@ const SETTINGS_VIEWS: readonly SettingsViewId[] = [
   'gateway',
   'keys',
   'mcp',
+  'sessions',
   'tools',
   'about'
 ]
@@ -40,6 +42,7 @@ export function SettingsView({ gateway, onClose, onConfigSaved }: SettingsPagePr
     gateway: '',
     keys: '',
     mcp: '',
+    sessions: '',
     tools: ''
   })
 
@@ -149,6 +152,12 @@ export function SettingsView({ gateway, onClose, onConfigSaved }: SettingsPagePr
             label="MCP"
             onClick={() => setActiveView('mcp')}
           />
+          <OverlayNavItem
+            active={activeView === 'sessions'}
+            icon={Archive}
+            label="Archived Chats"
+            onClick={() => setActiveView('sessions')}
+          />
           <div className="my-2 h-px bg-border/30" />
           <OverlayNavItem
             active={activeView === 'about'}
@@ -200,6 +209,8 @@ export function SettingsView({ gateway, onClose, onConfigSaved }: SettingsPagePr
             <KeysSettings query={queries.keys} />
           ) : activeView === 'mcp' ? (
             <McpSettings gateway={gateway} onConfigSaved={onConfigSaved} query={queries.mcp} />
+          ) : activeView === 'sessions' ? (
+            <SessionsSettings query={queries.sessions} />
           ) : (
             <ToolsSettings query={queries.tools} />
           )}

apps/desktop/src/app/settings/sessions-settings.tsx

@@ -0,0 +1,168 @@
+import { useCallback, useEffect, useMemo, useState } from 'react'
+
+import { Button } from '@/components/ui/button'
+import { deleteSession, listSessions, setSessionArchived } from '@/hermes'
+import { sessionTitle } from '@/lib/chat-runtime'
+import { triggerHaptic } from '@/lib/haptics'
+import { Archive, ArchiveOff, Loader2, Trash2 } from '@/lib/icons'
+import { notify, notifyError } from '@/store/notifications'
+import { setSessions } from '@/store/session'
+import type { SessionInfo } from '@/types/hermes'
+
+import { EmptyState, ListRow, LoadingState, SectionHeading, SettingsContent } from './primitives'
+import type { SearchProps } from './types'
+
+const ARCHIVED_FETCH_LIMIT = 200
+
+function workspaceLabel(cwd: null | string | undefined): string {
+  const path = cwd?.trim()
+
+  if (!path) {
+    return ''
+  }
+
+  return (
+    path
+      .replace(/[/\\]+$/, '')
+      .split(/[/\\]/)
+      .filter(Boolean)
+      .pop() ?? path
+  )
+}
+
+export function SessionsSettings({ query }: SearchProps) {
+  const [sessions, setLocalSessions] = useState<SessionInfo[]>([])
+  const [loading, setLoading] = useState(true)
+  const [busyId, setBusyId] = useState<string | null>(null)
+
+  const load = useCallback(async () => {
+    setLoading(true)
+
+    try {
+      const result = await listSessions(ARCHIVED_FETCH_LIMIT, 0, 'only')
+      setLocalSessions(result.sessions)
+    } catch (err) {
+      notifyError(err, 'Could not load archived sessions')
+    } finally {
+      setLoading(false)
+    }
+  }, [])
+
+  useEffect(() => {
+    void load()
+  }, [load])
+
+  const unarchive = useCallback(async (session: SessionInfo) => {
+    setBusyId(session.id)
+
+    try {
+      await setSessionArchived(session.id, false)
+      setLocalSessions(prev => prev.filter(s => s.id !== session.id))
+      // Surface it again in the sidebar without waiting for a full refresh.
+      setSessions(prev => [{ ...session, archived: false }, ...prev.filter(s => s.id !== session.id)])
+      triggerHaptic('selection')
+      notify({ durationMs: 2_000, kind: 'success', message: 'Restored' })
+    } catch (err) {
+      notifyError(err, 'Unarchive failed')
+    } finally {
+      setBusyId(null)
+    }
+  }, [])
+
+  const remove = useCallback(async (session: SessionInfo) => {
+    if (!window.confirm(`Permanently delete "${sessionTitle(session)}"? This cannot be undone.`)) {
+      return
+    }
+
+    setBusyId(session.id)
+
+    try {
+      await deleteSession(session.id)
+      setLocalSessions(prev => prev.filter(s => s.id !== session.id))
+      triggerHaptic('warning')
+    } catch (err) {
+      notifyError(err, 'Delete failed')
+    } finally {
+      setBusyId(null)
+    }
+  }, [])
+
+  const filtered = useMemo(() => {
+    const needle = query.trim().toLowerCase()
+
+    if (!needle) {
+      return sessions
+    }
+
+    return sessions.filter(session =>
+      [sessionTitle(session), session.preview ?? '', session.cwd ?? ''].join(' ').toLowerCase().includes(needle)
+    )
+  }, [query, sessions])
+
+  if (loading) {
+    return <LoadingState label="Loading archived sessions…" />
+  }
+
+  return (
+    <SettingsContent>
+      <SectionHeading
+        icon={Archive}
+        meta={sessions.length ? String(sessions.length) : undefined}
+        title="Archived sessions"
+      />
+      <p className="mb-2 text-[length:var(--conversation-caption-font-size)] text-(--ui-text-tertiary)">
+        Archived chats are hidden from the sidebar but keep all their messages. Ctrl/⌘-click a chat in the sidebar to
+        archive it.
+      </p>
+
+      {filtered.length === 0 ? (
+        <EmptyState
+          description={query.trim() ? 'No archived chats match your search.' : 'Archive a chat to hide it here.'}
+          title="Nothing archived"
+        />
+      ) : (
+        <div className="divide-y divide-border/30">
+          {filtered.map(session => {
+            const label = workspaceLabel(session.cwd)
+            const busy = busyId === session.id
+
+            return (
+              <ListRow
+                action={
+                  <div className="flex items-center gap-1.5">
+                    <Button
+                      disabled={busy}
+                      onClick={() => void unarchive(session)}
+                      size="sm"
+                      type="button"
+                      variant="outline"
+                    >
+                      {busy ? <Loader2 className="size-3.5 animate-spin" /> : <ArchiveOff className="size-3.5" />}
+                      <span>Unarchive</span>
+                    </Button>
+                    <Button
+                      aria-label="Delete permanently"
+                      className="text-muted-foreground hover:text-destructive"
+                      disabled={busy}
+                      onClick={() => void remove(session)}
+                      size="icon"
+                      title="Delete permanently"
+                      type="button"
+                      variant="ghost"
+                    >
+                      <Trash2 className="size-3.5" />
+                    </Button>
+                  </div>
+                }
+                description={session.preview || undefined}
+                hint={label ? `${label} · ${session.message_count} messages` : `${session.message_count} messages`}
+                key={session.id}
+                title={sessionTitle(session)}
+              />
+            )
+          })}
+        </div>
+      )}
+    </SettingsContent>
+  )
+}

apps/desktop/src/app/settings/types.ts

@@ -4,8 +4,8 @@ import type { HermesGateway } from '@/hermes'
 import type { IconComponent } from '@/lib/icons'
 import type { EnvVarInfo } from '@/types/hermes'
 
-export type SettingsView = 'about' | 'gateway' | 'keys' | 'mcp' | 'tools' | `config:${string}`
-export type SettingsQueryKey = 'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'tools'
+export type SettingsView = 'about' | 'gateway' | 'keys' | 'mcp' | 'sessions' | 'tools' | `config:${string}`
+export type SettingsQueryKey = 'about' | 'config' | 'gateway' | 'keys' | 'mcp' | 'sessions' | 'tools'
 export type EnvPatch = Partial<Pick<EnvVarInfo, 'is_set' | 'redacted_value'>>
 
 export interface SettingsPageProps {

apps/desktop/src/components/assistant-ui/markdown-text.tsx

@@ -19,9 +19,9 @@ import {
   filePathFromMediaPath,
   mediaExternalUrl,
   mediaKind,
-  mediaMime,
   mediaName,
-  mediaPathFromMarkdownHref
+  mediaPathFromMarkdownHref,
+  mediaStreamUrl
 } from '@/lib/media'
 import { previewTargetFromMarkdownHref } from '@/lib/preview-targets'
 import { cn } from '@/lib/utils'
@@ -40,24 +40,22 @@ import { cn } from '@/lib/utils'
 // LLM convention). The default false-setting only accepts `$$...$$`.
 const mathPlugin = createMemoizedMathPlugin({ singleDollarTextMath: true })
 
-async function typedBlobUrl(dataUrl: string, mime: string): Promise<string> {
-  const blob = await fetch(dataUrl).then(response => response.blob())
-
-  return URL.createObjectURL(new Blob([await blob.arrayBuffer()], { type: mime }))
-}
-
 async function mediaSrc(path: string): Promise<string> {
   if (/^(?:https?|data):/i.test(path)) {
     return path
   }
 
+  // Stream audio/video through the custom protocol: data URLs are capped and
+  // load the whole file into memory, which broke playback for larger videos.
+  if (window.hermesDesktop && ['audio', 'video'].includes(mediaKind(path))) {
+    return mediaStreamUrl(path)
+  }
+
   if (!window.hermesDesktop?.readFileDataUrl) {
     return mediaExternalUrl(path)
   }
 
-  const dataUrl = await window.hermesDesktop.readFileDataUrl(filePathFromMediaPath(path))
-
-  return ['audio', 'video'].includes(mediaKind(path)) ? typedBlobUrl(dataUrl, mediaMime(path)) : dataUrl
+  return window.hermesDesktop.readFileDataUrl(filePathFromMediaPath(path))
 }
 
 function OpenMediaButton({ kind, path }: { kind: 'audio' | 'video'; path: string }) {
@@ -278,10 +276,7 @@ const MarkdownTextImpl = () => {
   // render, which churns Streamdown's outer memo + propagates new prop
   // identities into every Block. The plugin set really only varies on
   // `isStreaming`, so memoize on that.
-  const plugins = useMemo(
-    () => (isStreaming ? { math: mathPlugin } : { math: mathPlugin, code }),
-    [isStreaming]
-  )
+  const plugins = useMemo(() => (isStreaming ? { math: mathPlugin } : { math: mathPlugin, code }), [isStreaming])
 
   const components = useMemo(
     () =>

apps/desktop/src/components/gateway-connecting-overlay.tsx

@@ -0,0 +1,183 @@
+import { useStore } from '@nanostores/react'
+import { useEffect, useRef, useState } from 'react'
+
+import { cn } from '@/lib/utils'
+import { $desktopBoot } from '@/store/boot'
+import { $gatewayState } from '@/store/session'
+
+// Static, always-legible prefix; only TAIL ever scrambles. Splitting them at
+// the render level means no timer logic (even a stale HMR one) can ever
+// scramble "CONN".
+const PREFIX = 'CONN'
+const TAIL = 'ECTING'
+// Even-weight mono ascii so cycling glyphs don't jump width (matches the
+// nousnet-web download-button decode effect).
+const SCRAMBLE_CHARS = '/\\|-_=+<>~:*'
+const TICK_MS = 45
+
+// Exit choreography (ms): text fades down + out, hold, then the overlay fades.
+const TEXT_OUT_MS = 360
+const POST_TEXT_HOLD_MS = 300
+const OVERLAY_OUT_MS = 520
+// Preview-only: how long to "connect" for, and the pause before replaying.
+const PREVIEW_CONNECT_MS = 2600
+const PREVIEW_REPLAY_MS = 1100
+
+type Phase = 'live' | 'text-out' | 'overlay-out' | 'gone'
+
+// Dev affordance: a warm Cmd+R reconnects almost instantly, so the overlay
+// only flashes. Load with `?connecting=1` to force a looping preview.
+function forcedPreview(): boolean {
+  if (!import.meta.env.DEV || typeof window === 'undefined') {
+    return false
+  }
+
+  try {
+    return new URLSearchParams(window.location.search).get('connecting') === '1'
+  } catch {
+    return false
+  }
+}
+
+function scrambledTail(resolvedCount: number): string {
+  return Array.from(TAIL, (ch, i) =>
+    i < resolvedCount ? ch : SCRAMBLE_CHARS[(Math.random() * SCRAMBLE_CHARS.length) | 0]
+  ).join('')
+}
+
+export function GatewayConnectingOverlay() {
+  const gatewayState = useStore($gatewayState)
+  const boot = useStore($desktopBoot)
+  const [previewing] = useState(forcedPreview)
+  const [tail, setTail] = useState(TAIL)
+  const [phase, setPhase] = useState<Phase>('live')
+
+  const connecting = gatewayState !== 'open' && !boot.error
+  // Latches once we've actually shown the overlay, so the brief frame where
+  // gatewayState flips to "open" (connecting -> false) before the exit phase
+  // kicks in doesn't unmount us and cause a flash.
+  const shownRef = useRef(false)
+
+  if (previewing || connecting) {
+    shownRef.current = true
+  }
+
+  // Decode loop — only while live (freeze the resolved word during the exit).
+  useEffect(() => {
+    if (phase !== 'live' || (!previewing && !connecting)) {
+      return
+    }
+
+    let resolved = 0
+    let hold = 0
+
+    const id = window.setInterval(() => {
+      if (resolved >= TAIL.length) {
+        hold += 1
+
+        if (hold > 16) {
+          resolved = 0
+          hold = 0
+        }
+
+        setTail(TAIL)
+
+        return
+      }
+
+      resolved += 0.5
+      setTail(scrambledTail(Math.floor(resolved)))
+    }, TICK_MS)
+
+    return () => window.clearInterval(id)
+  }, [phase, previewing, connecting])
+
+  // Kick off the exit when connected: real connect, or a faked timer in preview.
+  useEffect(() => {
+    if (phase !== 'live') {
+      return
+    }
+
+    if (previewing) {
+      const id = window.setTimeout(() => {
+        setTail(TAIL)
+        setPhase('text-out')
+      }, PREVIEW_CONNECT_MS)
+
+      return () => window.clearTimeout(id)
+    }
+
+    if (gatewayState === 'open' && shownRef.current) {
+      setTail(TAIL)
+      setPhase('text-out')
+    }
+  }, [phase, previewing, gatewayState])
+
+  // Advance the exit choreography: text-out -> overlay-out -> gone.
+  useEffect(() => {
+    if (phase === 'text-out') {
+      const id = window.setTimeout(() => setPhase('overlay-out'), TEXT_OUT_MS + POST_TEXT_HOLD_MS)
+
+      return () => window.clearTimeout(id)
+    }
+
+    if (phase === 'overlay-out') {
+      const id = window.setTimeout(() => setPhase('gone'), OVERLAY_OUT_MS)
+
+      return () => window.clearTimeout(id)
+    }
+
+    // Preview replays so we can keep watching the transition.
+    if (phase === 'gone' && previewing) {
+      const id = window.setTimeout(() => {
+        setTail(TAIL)
+        setPhase('live')
+      }, PREVIEW_REPLAY_MS)
+
+      return () => window.clearTimeout(id)
+    }
+  }, [phase, previewing])
+
+  // Boot failed — BootFailureOverlay owns the screen; don't linger behind it.
+  if (boot.error && !previewing) {
+    return null
+  }
+
+  // Real connect: once the fade finishes, get out of the way for good.
+  if (phase === 'gone' && !previewing) {
+    return null
+  }
+
+  // Never showed (e.g. gateway already up on a warm reload) — stay out.
+  if (!previewing && !connecting && !shownRef.current) {
+    return null
+  }
+
+  const leaving = phase !== 'live'
+  const overlayHidden = phase === 'overlay-out' || phase === 'gone'
+
+  return (
+    <div
+      className={cn(
+        'fixed inset-0 z-[1200] grid place-items-center bg-(--ui-chat-surface-background) transition-opacity duration-500 ease-out',
+        overlayHidden ? 'pointer-events-none opacity-0' : 'opacity-100'
+      )}
+    >
+      <style>{'@keyframes gco-cursor { 0%, 49% { opacity: 1 } 50%, 100% { opacity: 0 } }'}</style>
+      <span
+        className={cn(
+          'inline-flex items-center pl-[0.4em] font-mono text-[0.64rem] font-semibold uppercase tracking-[0.4em] tabular-nums text-(--theme-primary) transition duration-300 ease-out',
+          leaving ? 'translate-y-2 opacity-0 saturate-0' : 'translate-y-0 opacity-100 saturate-100'
+        )}
+      >
+        {PREFIX}
+        {tail}
+        <span
+          aria-hidden="true"
+          className="dither ml-0.5 inline-block size-2 shrink-0 -translate-y-px rounded-[1px]"
+          style={{ animation: 'gco-cursor 1s step-end infinite' }}
+        />
+      </span>
+    </div>
+  )
+}

apps/desktop/src/hermes.ts

@@ -111,9 +111,13 @@ export class HermesGateway extends JsonRpcGatewayClient {
   }
 }
 
-export async function listSessions(limit = 40, minMessages = 0): Promise<PaginatedSessions> {
+export async function listSessions(
+  limit = 40,
+  minMessages = 0,
+  archived: 'exclude' | 'include' | 'only' = 'exclude'
+): Promise<PaginatedSessions> {
   const result = await window.hermesDesktop.api<PaginatedSessions>({
-    path: `/api/sessions?limit=${limit}&offset=0&min_messages=${Math.max(0, minMessages)}`
+    path: `/api/sessions?limit=${limit}&offset=0&min_messages=${Math.max(0, minMessages)}&archived=${archived}`
   })
 
   return {
@@ -123,6 +127,14 @@ export async function listSessions(limit = 40, minMessages = 0): Promise<Paginat
   }
 }
 
+export function setSessionArchived(id: string, archived: boolean): Promise<{ ok: boolean }> {
+  return window.hermesDesktop.api<{ ok: boolean }>({
+    path: `/api/sessions/${encodeURIComponent(id)}`,
+    method: 'PATCH',
+    body: { archived }
+  })
+}
+
 export function searchSessions(query: string): Promise<SessionSearchResponse> {
   return window.hermesDesktop.api<SessionSearchResponse>({
     path: `/api/sessions/search?q=${encodeURIComponent(query)}`

apps/desktop/src/lib/icons.ts

@@ -2,6 +2,8 @@ import {
   IconActivity as Activity,
   IconAlertCircle as AlertCircle,
   IconAlertTriangle as AlertTriangle,
+  IconArchive as Archive,
+  IconArchiveOff as ArchiveOff,
   IconArrowUp as ArrowUp,
   IconArrowUpRight as ArrowUpRight,
   IconAt as AtSign,
@@ -98,6 +100,8 @@ export {
   Activity,
   AlertCircle,
   AlertTriangle,
+  Archive,
+  ArchiveOff,
   ArrowUp,
   ArrowUpRight,
   AtSign,

apps/desktop/src/lib/media.ts

@@ -58,6 +58,13 @@ export function mediaExternalUrl(path: string): string {
   return /^(?:https?|file):/i.test(path) ? path : `file://${path}`
 }
 
+// Custom Electron scheme (registered in electron/main.cjs) that streams a local
+// file with Range support. Used for audio/video so playback bypasses the data
+// URL size cap and supports seeking. `path` may be a plain path or `file://…`.
+export function mediaStreamUrl(path: string): string {
+  return `hermes-media://stream/${encodeURIComponent(filePathFromMediaPath(path))}`
+}
+
 export function mediaPathFromMarkdownHref(href?: string): string | null {
   if (!href?.startsWith('#media:')) {
     return null

apps/desktop/src/types/hermes.ts

@@ -240,6 +240,7 @@ export interface SessionCreateResponse {
 }
 
 export interface SessionInfo {
+  archived?: boolean
   cwd?: null | string
   ended_at: null | number
   id: string

hermes_cli/web_server.py

@@ -1131,22 +1131,51 @@ async def get_action_status(name: str, lines: int = 200):
 
 
 @app.get("/api/sessions")
-async def get_sessions(limit: int = 20, offset: int = 0, min_messages: int = 0):
+async def get_sessions(
+    limit: int = 20,
+    offset: int = 0,
+    min_messages: int = 0,
+    archived: str = "exclude",
+):
+    """List sessions.
+
+    ``archived`` controls how soft-archived sessions are treated:
+    ``exclude`` (default) hides them, ``only`` returns just the archived ones
+    (used by the desktop "Archived sessions" settings panel), and ``include``
+    returns both.
+    """
+    if archived not in ("exclude", "only", "include"):
+        raise HTTPException(
+            status_code=400,
+            detail="archived must be one of: exclude, only, include",
+        )
     try:
         from hermes_state import SessionDB
         db = SessionDB()
         try:
             min_message_count = max(0, min_messages)
+            archived_only = archived == "only"
+            include_archived = archived == "include"
             sessions = db.list_sessions_rich(
-                limit=limit, offset=offset, min_message_count=min_message_count
+                limit=limit,
+                offset=offset,
+                min_message_count=min_message_count,
+                include_archived=include_archived,
+                archived_only=archived_only,
+            )
+            total = db.session_count(
+                min_message_count=min_message_count,
+                include_archived=include_archived,
+                archived_only=archived_only,
             )
-            total = db.session_count(min_message_count=min_message_count)
             now = time.time()
             for s in sessions:
                 s["is_active"] = (
                     s.get("ended_at") is None
                     and (now - s.get("last_active", s.get("started_at", 0))) < 300
                 )
+                # SQLite stores the flag as 0/1; expose a real JSON boolean.
+                s["archived"] = bool(s.get("archived"))
             return {"sessions": sessions, "total": total, "limit": limit, "offset": offset}
         finally:
             db.close()
@@ -3707,25 +3736,39 @@ async def delete_session_endpoint(session_id: str):
 
 class SessionRename(BaseModel):
     title: Optional[str] = None
+    archived: Optional[bool] = None
 
 
 @app.patch("/api/sessions/{session_id}")
 async def rename_session_endpoint(session_id: str, body: SessionRename):
-    """Rename a session (or clear its title when ``title`` is empty/null)."""
+    """Update a session: rename (or clear its title) and/or archive it.
+
+    ``title`` renames (empty/null clears the title); ``archived`` soft-hides or
+    restores the session. Either field may be omitted.
+    """
     from hermes_state import SessionDB
     db = SessionDB()
     try:
         sid = db.resolve_session_id(session_id)
         if not sid:
             raise HTTPException(status_code=404, detail="Session not found")
-        try:
-            updated = db.set_session_title(sid, body.title or "")
-        except ValueError as e:
-            # Title too long, invalid characters, or already in use.
-            raise HTTPException(status_code=400, detail=str(e))
-        if not updated:
-            raise HTTPException(status_code=404, detail="Session not found")
-        return {"ok": True, "title": db.get_session_title(sid) or ""}
+        if body.title is None and body.archived is None:
+            raise HTTPException(
+                status_code=400,
+                detail="Nothing to update; provide 'title' and/or 'archived'.",
+            )
+        if body.title is not None:
+            try:
+                db.set_session_title(sid, body.title or "")
+            except ValueError as e:
+                # Title too long, invalid characters, or already in use.
+                raise HTTPException(status_code=400, detail=str(e))
+        if body.archived is not None:
+            db.set_session_archived(sid, body.archived)
+        result = {"ok": True, "title": db.get_session_title(sid) or ""}
+        if body.archived is not None:
+            result["archived"] = bool(body.archived)
+        return result
     finally:
         db.close()

hermes_state.py

@@ -264,6 +264,7 @@ def _log_wal_fallback_once(db_label: str, exc: Exception) -> None:
     handoff_platform TEXT,
     handoff_error TEXT,
     rewind_count INTEGER NOT NULL DEFAULT 0,
+    archived INTEGER NOT NULL DEFAULT 0,
     FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
 );
 
@@ -1430,6 +1431,22 @@ def get_session_title(self, session_id: str) -> Optional[str]:
             row = cursor.fetchone()
         return row["title"] if row else None
 
+    def set_session_archived(self, session_id: str, archived: bool) -> bool:
+        """Archive or unarchive a session.
+
+        Archived sessions are hidden from the default session list but keep all
+        their messages — this is a soft hide, not a delete. Returns True when a
+        row was updated.
+        """
+        def _do(conn):
+            cursor = conn.execute(
+                "UPDATE sessions SET archived = ? WHERE id = ?",
+                (1 if archived else 0, session_id),
+            )
+            return cursor.rowcount
+        rowcount = self._execute_write(_do)
+        return rowcount > 0
+
     def get_session_by_title(self, title: str) -> Optional[Dict[str, Any]]:
         """Look up a session by exact title. Returns session dict or None."""
         with self._lock:
@@ -1549,6 +1566,8 @@ def list_sessions_rich(
         min_message_count: int = 0,
         project_compression_tips: bool = True,
         order_by_last_active: bool = False,
+        include_archived: bool = False,
+        archived_only: bool = False,
     ) -> List[Dict[str, Any]]:
         """List sessions with preview (first user message) and last active timestamp.
 
@@ -1604,6 +1623,10 @@ def list_sessions_rich(
         if min_message_count > 0:
             where_clauses.append("s.message_count >= ?")
             params.append(min_message_count)
+        if archived_only:
+            where_clauses.append("s.archived = 1")
+        elif not include_archived:
+            where_clauses.append("s.archived = 0")
 
         where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
         if order_by_last_active:
@@ -3027,7 +3050,13 @@ def search_sessions(
     # Utility
     # =========================================================================
 
-    def session_count(self, source: str = None, min_message_count: int = 0) -> int:
+    def session_count(
+        self,
+        source: str = None,
+        min_message_count: int = 0,
+        include_archived: bool = False,
+        archived_only: bool = False,
+    ) -> int:
         """Count sessions, optionally filtered by source."""
         where_clauses = []
         params = []
@@ -3038,6 +3067,10 @@ def session_count(self, source: str = None, min_message_count: int = 0) -> int:
         if min_message_count > 0:
             where_clauses.append("message_count >= ?")
             params.append(min_message_count)
+        if archived_only:
+            where_clauses.append("archived = 1")
+        elif not include_archived:
+            where_clauses.append("archived = 0")
 
         where_sql = f" WHERE {' AND '.join(where_clauses)}" if where_clauses else ""

tests/hermes_cli/test_web_server.py

@@ -187,11 +187,11 @@ class _FakeDB:
             def __init__(self, *args, **kwargs):
                 pass
 
-            def list_sessions_rich(self, limit, offset, min_message_count=0):
+            def list_sessions_rich(self, limit, offset, min_message_count=0, **kwargs):
                 captured["list"] = min_message_count
                 return []
 
-            def session_count(self, min_message_count=0):
+            def session_count(self, min_message_count=0, **kwargs):
                 captured["count"] = min_message_count
                 return 0
 
@@ -250,6 +250,76 @@ def test_rename_session_not_found(self):
         resp = self.client.patch("/api/sessions/does-not-exist", json={"title": "x"})
         assert resp.status_code == 404
 
+    def test_archive_session_via_patch(self):
+        """PATCH archived=true soft-hides a session; archived=false restores it."""
+        from hermes_state import SessionDB
+
+        db = SessionDB()
+        try:
+            db.create_session(session_id="arch-me", source="cli")
+            db.append_message(session_id="arch-me", role="user", content="hi")
+        finally:
+            db.close()
+
+        resp = self.client.patch("/api/sessions/arch-me", json={"archived": True})
+        assert resp.status_code == 200
+        assert resp.json()["archived"] is True
+
+        # Hidden from the default list, surfaced by archived=only.
+        listed = self.client.get("/api/sessions").json()
+        assert all(s["id"] != "arch-me" for s in listed["sessions"])
+        only = self.client.get("/api/sessions?archived=only").json()
+        assert any(s["id"] == "arch-me" for s in only["sessions"])
+
+        resp = self.client.patch("/api/sessions/arch-me", json={"archived": False})
+        assert resp.status_code == 200
+        restored = self.client.get("/api/sessions").json()
+        assert any(s["id"] == "arch-me" for s in restored["sessions"])
+
+    def test_patch_session_without_fields_is_400(self):
+        """An existing session + empty body is a bad request, not a 404."""
+        from hermes_state import SessionDB
+
+        db = SessionDB()
+        try:
+            db.create_session(session_id="no-fields", source="cli")
+        finally:
+            db.close()
+
+        resp = self.client.patch("/api/sessions/no-fields", json={})
+        assert resp.status_code == 400
+
+    def test_get_sessions_rejects_unknown_archived_value(self):
+        resp = self.client.get("/api/sessions?archived=bogus")
+        assert resp.status_code == 400
+
+    def test_get_sessions_archived_is_boolean(self):
+        from hermes_state import SessionDB
+
+        db = SessionDB()
+        try:
+            db.create_session(session_id="bool-arch", source="cli")
+            db.append_message(session_id="bool-arch", role="user", content="hi")
+        finally:
+            db.close()
+
+        row = next(s for s in self.client.get("/api/sessions").json()["sessions"] if s["id"] == "bool-arch")
+        assert row["archived"] is False
+
+    def test_rename_response_omits_archived_when_not_set(self):
+        """Title-only PATCH keeps its legacy {ok, title} response shape."""
+        from hermes_state import SessionDB
+
+        db = SessionDB()
+        try:
+            db.create_session(session_id="title-only", source="cli")
+        finally:
+            db.close()
+
+        resp = self.client.patch("/api/sessions/title-only", json={"title": "Hi"})
+        assert resp.status_code == 200
+        assert "archived" not in resp.json()
+
     def test_audio_transcription_endpoint(self, monkeypatch):
         import tools.transcription_tools as transcription_tools

tests/test_hermes_state.py

@@ -3509,3 +3509,43 @@ def execute(self, sql, params=()):
         assert any("journal_mode=WAL" in sql for sql in conn.executed), (
             "set-pragma must fire when probe returns 'delete'"
         )
+
+
+class TestSessionArchive:
+    """Soft-archiving hides a session from default listings without deleting it."""
+
+    def _seed(self, db, sid, *, archived=False):
+        db.create_session(session_id=sid, source="cli")
+        db.append_message(session_id=sid, role="user", content=f"hello from {sid}")
+        if archived:
+            db.set_session_archived(sid, True)
+
+    def test_set_session_archived_roundtrip(self, db):
+        self._seed(db, "s1")
+        assert db.set_session_archived("s1", True) is True
+        assert db.get_session("s1")["archived"] == 1
+        assert db.set_session_archived("s1", False) is True
+        assert db.get_session("s1")["archived"] == 0
+
+    def test_set_session_archived_missing_row(self, db):
+        assert db.set_session_archived("nope", True) is False
+
+    def test_archived_excluded_by_default(self, db):
+        self._seed(db, "live")
+        self._seed(db, "hidden", archived=True)
+
+        ids = [s["id"] for s in db.list_sessions_rich()]
+        assert ids == ["live"]
+        assert db.session_count() == 1
+
+    def test_archived_only_and_include(self, db):
+        self._seed(db, "live")
+        self._seed(db, "hidden", archived=True)
+
+        only = [s["id"] for s in db.list_sessions_rich(archived_only=True)]
+        assert only == ["hidden"]
+        assert db.session_count(archived_only=True) == 1
+
+        both = {s["id"] for s in db.list_sessions_rich(include_archived=True)}
+        assert both == {"live", "hidden"}
+        assert db.session_count(include_archived=True) == 2

tests/test_tui_gateway_server.py

@@ -884,6 +884,73 @@ def set_session_title(self, _key, _title):
         server._sessions.pop("sid", None)
 
 
+def test_notification_event_routing_by_session_key(monkeypatch):
+    """Background-process events surface only in the session that owns them."""
+    mine = _session(session_key="mine")
+    other = _session(session_key="other")
+    monkeypatch.setattr(server, "_sessions", {"a": mine, "b": other})
+
+    # My own event → handle it.
+    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "mine"}) is False
+    # Global/system event with no owner → handle it.
+    assert server._notification_event_belongs_elsewhere(mine, {"session_key": ""}) is False
+    assert server._notification_event_belongs_elsewhere(mine, {}) is False
+    # Owned by another *live* session → defer to that session's poller.
+    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "other"}) is True
+    # Owner is gone (not in _sessions) → handle as fallback so it isn't lost.
+    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "ghost"}) is False
+
+
+def test_session_create_does_not_persist_empty_row(monkeypatch):
+    """session.create must NOT eagerly write a DB row.
+
+    Every TUI/desktop launch opens a session here just to paint the composer;
+    eagerly creating a row left an empty "Untitled" session behind for every
+    launch the user never typed into. The row is created lazily on first prompt.
+    """
+    created = []
+
+    class _FakeDB:
+        def create_session(self, *args, **kwargs):
+            created.append((args, kwargs))
+
+    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
+    monkeypatch.setattr(server, "_start_agent_build", lambda *a, **k: None)
+    monkeypatch.setattr(
+        server.threading,
+        "Timer",
+        lambda *a, **k: types.SimpleNamespace(daemon=False, start=lambda: None),
+    )
+
+    resp = server.handle_request(
+        {"id": "1", "method": "session.create", "params": {"cols": 80}}
+    )
+    sid = resp["result"]["session_id"]
+    try:
+        assert resp["result"]["stored_session_id"]
+        assert created == [], "session.create should not persist an empty DB row"
+    finally:
+        server._sessions.pop(sid, None)
+
+
+def test_ensure_session_db_row_persists_with_cwd(monkeypatch, tmp_path):
+    """First prompt persists the row (INSERT OR IGNORE) capturing cwd up front."""
+    created = []
+
+    class _FakeDB:
+        def create_session(self, key, source=None, model=None, cwd=None):
+            created.append({"key": key, "source": source, "model": model, "cwd": cwd})
+
+    monkeypatch.setattr(server, "_get_db", lambda: _FakeDB())
+    monkeypatch.setattr(server, "_resolve_model", lambda: "test-model")
+
+    server._ensure_session_db_row({"session_key": "k1", "cwd": str(tmp_path)})
+
+    assert created == [
+        {"key": "k1", "source": "tui", "model": "test-model", "cwd": str(tmp_path)}
+    ]
+
+
 def test_session_title_clears_pending_after_persist(monkeypatch):
     class _FakeDB:
         def __init__(self):

tools/process_registry.py

@@ -880,6 +880,7 @@ def _move_to_finished(self, session: ProcessSession):
             self.completion_queue.put({
                 "type": "completion",
                 "session_id": session.id,
+                "session_key": session.session_key,
                 "command": session.command,
                 "exit_code": session.exit_code,
                 "output": output_tail,

tui_gateway/server.py

@@ -702,6 +702,32 @@ def _register_session_cwd(session: dict | None) -> None:
         pass
 
 
+def _ensure_session_db_row(session: dict) -> None:
+    """Idempotently persist the session's DB row on first real activity.
+
+    Called from prompt.submit so a row only exists once the user actually sends
+    a message — abandoned drafts never leave an empty "Untitled" session behind.
+    Uses INSERT OR IGNORE under the hood, so re-calls (and the AIAgent's own
+    lazy create) are no-ops. Captures cwd up front so workspace grouping works
+    without waiting for a separate cwd update.
+    """
+    key = session.get("session_key")
+    if not key:
+        return
+    db = _get_db()
+    if db is None:
+        return
+    try:
+        db.create_session(
+            key,
+            source="tui",
+            model=_resolve_model(),
+            cwd=_session_cwd(session),
+        )
+    except Exception:
+        logger.debug("failed to persist desktop session row", exc_info=True)
+
+
 def _set_session_cwd(session: dict, cwd: str) -> str:
     resolved = os.path.abspath(os.path.expanduser(str(cwd)))
     if not os.path.isdir(resolved):
@@ -2750,17 +2776,12 @@ def _(rid, params: dict) -> dict:
         "transport": current_transport() or _stdio_transport,
     }
     _register_session_cwd(_sessions[sid])
-    db = _get_db()
-    if db is not None:
-        try:
-            db.create_session(
-                key,
-                source="tui",
-                model=_resolve_model(),
-                cwd=_sessions[sid]["cwd"],
-            )
-        except Exception:
-            logger.debug("failed to pre-create desktop session row", exc_info=True)
+    # NOTE: we intentionally do NOT persist a DB row here. Every TUI/desktop
+    # launch (and every "New agent" / draft) opens a session here just to paint
+    # the composer, so eagerly creating a row left an "Untitled" empty session
+    # behind for every launch the user never typed into. The row is now created
+    # lazily on the first prompt (see _ensure_session_db_row + prompt.submit),
+    # and the AIAgent's own INSERT-OR-IGNORE persists it on the first turn too.
 
     # Return the lightweight session immediately so Ink can paint the composer
     # + skeleton panel, then build the real AIAgent just after this response is
@@ -3841,6 +3862,8 @@ def _(rid, params: dict) -> dict:
         session["last_active"] = time.time()
         _start_inflight_turn(session, text)
 
+    # Persist the DB row lazily, now that the user has actually sent a message.
+    _ensure_session_db_row(session)
     _start_agent_build(sid, session)
 
     def run_after_agent_ready() -> None:
@@ -3865,6 +3888,35 @@ def run_after_agent_ready() -> None:
     return _ok(rid, {"status": "streaming"})
 
 
+def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool:
+    """True if ``evt`` is owned by a *different* live session.
+
+    Background-process events carry the ``session_key`` of the session that
+    started the process. Since all desktop sessions share one process-wide
+    completion queue, each poller must skip events it doesn't own so a
+    background job's completion surfaces in the session that launched it — not
+    whichever poller happened to dequeue first. Orphaned events (owner gone)
+    and global/system events (empty ``session_key``) return False so the
+    current poller still handles them rather than losing them.
+    """
+    evt_key = str(evt.get("session_key") or "")
+    if not evt_key:
+        return False
+    if evt_key == str(session.get("session_key") or ""):
+        return False
+    try:
+        snapshot = list(_sessions.values())
+    except Exception:
+        # If we can't safely enumerate live sessions, fail open so we don't
+        # crash the poller thread or drop the event.
+        return False
+
+    return any(
+        s is not session and str(s.get("session_key") or "") == evt_key
+        for s in snapshot
+    )
+
+
 def _notification_poller_loop(
     stop_event: threading.Event, sid: str, session: dict
 ) -> None:
@@ -3887,6 +3939,16 @@ def _notification_poller_loop(
         except Exception:
             continue
 
+        # Multiple desktop sessions share this one process-wide queue. Only
+        # consume events that belong to *this* session — otherwise a background
+        # process started in session A would surface its completion in whichever
+        # session's poller happened to wake first (Ben's "reported in a
+        # different session" bug). Leave foreign events for their owner.
+        if _notification_event_belongs_elsewhere(session, evt):
+            process_registry.completion_queue.put(evt)
+            time.sleep(0.1)
+            continue
+
         _evt_sid = evt.get("session_id", "")
         if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
             continue
@@ -3917,12 +3979,17 @@ def _notification_poller_loop(
                 session["running"] = False
 
     # Drain any remaining events after stop signal (process all pending
-    # before exiting so nothing is lost on shutdown).
+    # before exiting so nothing is lost on shutdown). Events owned by other
+    # live sessions are set aside and re-queued so their poller still sees them.
+    deferred: list = []
     while not process_registry.completion_queue.empty():
         try:
             evt = process_registry.completion_queue.get_nowait()
         except Exception:
             break
+        if _notification_event_belongs_elsewhere(session, evt):
+            deferred.append(evt)
+            continue
         _evt_sid = evt.get("session_id", "")
         if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
             continue
@@ -3951,6 +4018,10 @@ def _notification_poller_loop(
             with session["history_lock"]:
                 session["running"] = False
 
+    # Hand any other sessions' events back to the shared queue.
+    for evt in deferred:
+        process_registry.completion_queue.put(evt)
+
 
 def _start_notification_poller(sid: str, session: dict) -> threading.Event:
     """Start the background notification poller for a TUI session."""

#60863 fix(delegation): route async results to their origin session across compression (#55578)

merged to main

files: gateway/session_context.py, tests/test_tui_gateway_server.py, tests/tools/test_async_delegation.py, tests/tui_gateway/test_delegation_session_lifecycle.py, tools/async_delegation.py, tools/delegate_tool.py, tui_gateway/server.py

Show the diff

gateway/session_context.py

@@ -79,6 +79,13 @@ def session_context_engaged() -> bool:
 _SESSION_USER_NAME: ContextVar = ContextVar("HERMES_SESSION_USER_NAME", default=_UNSET)
 _SESSION_KEY: ContextVar = ContextVar("HERMES_SESSION_KEY", default=_UNSET)
 _SESSION_ID: ContextVar = ContextVar("HERMES_SESSION_ID", default=_UNSET)
+# In-process UI session/window id for multi-session desktop/TUI hosts. This is
+# intentionally separate from HERMES_SESSION_ID: the latter is the durable
+# conversation/session-db id, while the UI id is the live frontend tab/window
+# that commissioned a detached completion. Background completions use it as a
+# precise return address so a stale/rotated durable session key cannot be
+# consumed by whichever desktop poller wakes first.
+_SESSION_UI_SESSION_ID: ContextVar = ContextVar("HERMES_UI_SESSION_ID", default=_UNSET)
 # ID of the message that triggered the current turn. Used as a reply anchor
 # so background-process notifications stay inside the originating Telegram
 # private-chat topic (those lanes route only with thread id + reply anchor).
@@ -123,6 +130,7 @@ def session_context_engaged() -> bool:
     "HERMES_SESSION_USER_NAME": _SESSION_USER_NAME,
     "HERMES_SESSION_KEY": _SESSION_KEY,
     "HERMES_SESSION_ID": _SESSION_ID,
+    "HERMES_UI_SESSION_ID": _SESSION_UI_SESSION_ID,
     "HERMES_SESSION_MESSAGE_ID": _SESSION_MESSAGE_ID,
     "HERMES_SESSION_PROFILE": _SESSION_PROFILE,
     "HERMES_CRON_AUTO_DELIVER_PLATFORM": _CRON_AUTO_DELIVER_PLATFORM,
@@ -160,6 +168,7 @@ def set_session_vars(
     profile: str = "",
     cwd: str = "",
     async_delivery: bool = True,
+    ui_session_id: str = "",
 ) -> list:
     """Set all session context variables and return reset tokens.
 
@@ -191,6 +200,7 @@ def set_session_vars(
         _SESSION_USER_NAME.set(user_name),
         _SESSION_KEY.set(session_key),
         _SESSION_ID.set(session_id),
+        _SESSION_UI_SESSION_ID.set(ui_session_id),
         _SESSION_MESSAGE_ID.set(message_id),
         _SESSION_PROFILE.set(profile),
         _SESSION_ASYNC_DELIVERY.set(bool(async_delivery)),
@@ -225,6 +235,7 @@ def clear_session_vars(tokens: list) -> None:
         _SESSION_USER_NAME,
         _SESSION_KEY,
         _SESSION_ID,
+        _SESSION_UI_SESSION_ID,
         _SESSION_MESSAGE_ID,
         _SESSION_PROFILE,
     ):

tests/test_tui_gateway_server.py

@@ -2112,14 +2112,85 @@ def test_notification_event_routing_by_session_key(monkeypatch):
     monkeypatch.setattr(server, "_sessions", {"a": mine, "b": other})
 
     # My own event → handle it.
-    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "mine"}) is False
+    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "mine"}) is False
     # Global/system event with no owner → handle it.
-    assert server._notification_event_belongs_elsewhere(mine, {"session_key": ""}) is False
-    assert server._notification_event_belongs_elsewhere(mine, {}) is False
+    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": ""}) is False
+    assert server._notification_event_belongs_elsewhere("a", mine, {}) is False
     # Owned by another *live* session → defer to that session's poller.
-    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "other"}) is True
+    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "other"}) is True
     # Owner is gone (not in _sessions) → handle as fallback so it isn't lost.
-    assert server._notification_event_belongs_elsewhere(mine, {"session_key": "ghost"}) is False
+    assert server._notification_event_belongs_elsewhere("a", mine, {"session_key": "ghost"}) is False
+
+
+def test_async_delegation_event_prefers_origin_ui_session(monkeypatch):
+    """Detached subagent completions return to the commissioning TUI tab.
+
+    Regression: when the durable session key was stale/orphaned, whichever
+    desktop poller woke first could consume the async result and inject it into
+    an unrelated session.
+    """
+    mine = _session(session_key="current-key")
+    other = _session(session_key="unrelated-key")
+    monkeypatch.setattr(server, "_sessions", {"origin-sid": mine, "other-sid": other})
+    monkeypatch.setattr(server, "_get_db", lambda: None)
+    evt = {
+        "type": "async_delegation",
+        "session_key": "stale-or-rotated-key",
+        "origin_ui_session_id": "origin-sid",
+    }
+
+    assert server._notification_event_belongs_elsewhere("other-sid", other, evt) is True
+    assert server._notification_event_belongs_elsewhere("origin-sid", mine, evt) is False
+
+
+def test_notification_event_follows_compression_continuation(monkeypatch):
+    """Events keyed to a compressed parent route to the live continuation."""
+    old_parent = _session(session_key="old-parent")
+    live_tip = _session(session_key="new-tip")
+    monkeypatch.setattr(server, "_sessions", {"old-sid": old_parent, "tip-sid": live_tip})
+
+    class _DB:
+        def resolve_resume_session_id(self, session_id):
+            return "new-tip" if session_id == "old-parent" else session_id
+
+    monkeypatch.setattr(server, "_get_db", lambda: _DB())
+    evt = {"type": "async_delegation", "session_key": "old-parent"}
+
+    assert server._notification_event_belongs_elsewhere("old-sid", old_parent, evt) is True
+    assert server._notification_event_belongs_elsewhere("tip-sid", live_tip, evt) is False
+    # A third session must leave it alone for the continuation's poller.
+    third = _session(session_key="third")
+    monkeypatch.setattr(
+        server,
+        "_sessions",
+        {"old-sid": old_parent, "tip-sid": live_tip, "third-sid": third},
+    )
+    assert server._notification_event_belongs_elsewhere("third-sid", third, evt) is True
+
+
+def test_finalized_origin_ui_session_falls_back_to_live_continuation(monkeypatch):
+    """A closed origin tab must not steal its resumed continuation's result."""
+    finalized_origin = _session(session_key="old-parent", _finalized=True)
+    live_tip = _session(session_key="new-tip")
+    monkeypatch.setattr(
+        server,
+        "_sessions",
+        {"origin-sid": finalized_origin, "tip-sid": live_tip},
+    )
+
+    class _DB:
+        def resolve_resume_session_id(self, session_id):
+            return "new-tip" if session_id == "old-parent" else session_id
+
+    monkeypatch.setattr(server, "_get_db", lambda: _DB())
+    evt = {
+        "type": "async_delegation",
+        "session_key": "old-parent",
+        "origin_ui_session_id": "origin-sid",
+    }
+
+    assert server._notification_event_belongs_elsewhere("origin-sid", finalized_origin, evt) is True
+    assert server._notification_event_belongs_elsewhere("tip-sid", live_tip, evt) is False
 
 
 def test_prompt_submit_rejects_negative_truncate_ordinal(monkeypatch):

tests/tools/test_async_delegation.py

@@ -289,6 +289,68 @@ def slow_child(task_index, goal, child=None, parent_agent=None, **kw):
     assert "the real task" in text
 
 
+def test_delegate_task_background_uses_live_tui_agent_session_id(monkeypatch):
+    """TUI async delegation must route to the live/compressed agent id.
+
+    Regression: delegate_task captured the stale approval/session context key
+    after compression rotated parent_agent.session_id. The resulting completion
+    was orphaned and could be consumed by an unrelated desktop session poller.
+    """
+    import json
+    from unittest.mock import MagicMock
+    import tools.delegate_tool as dt
+    from gateway.session_context import clear_session_vars, set_session_vars
+    from tools.approval import reset_current_session_key, set_current_session_key
+
+    parent = MagicMock()
+    parent._delegate_depth = 0
+    parent.session_id = "post-compress-tip"
+    parent._interrupt_requested = False
+    parent._active_children = []
+    parent._active_children_lock = None
+    fake_child = MagicMock()
+    fake_child._delegate_role = "leaf"
+
+    creds = {
+        "model": "m", "provider": None, "base_url": None, "api_key": None,
+        "api_mode": None, "command": None, "args": None,
+    }
+    monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
+    monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: creds)
+    monkeypatch.setattr(
+        dt,
+        "_run_single_child",
+        lambda *a, **k: {
+            "task_index": 0,
+            "status": "completed",
+            "summary": "done",
+            "api_calls": 1,
+            "duration_seconds": 0.1,
+            "model": "m",
+            "exit_reason": "completed",
+        },
+    )
+
+    approval_token = set_current_session_key("pre-compress-parent")
+    session_tokens = set_session_vars(
+        source="tui",
+        session_key="pre-compress-parent",
+        ui_session_id="origin-tab",
+    )
+    try:
+        out = dt.delegate_task(goal="bg task", background=True, parent_agent=parent)
+        assert json.loads(out)["status"] == "dispatched"
+        evt = _drain_one()
+    finally:
+        reset_current_session_key(approval_token)
+        clear_session_vars(session_tokens)
+
+    assert evt is not None
+    assert evt["type"] == "async_delegation"
+    assert evt["session_key"] == "post-compress-tip"
+    assert evt["origin_ui_session_id"] == "origin-tab"
+
+
 def test_delegate_task_background_batch_runs_as_one_unit(monkeypatch):
     """A multi-item batch with background=True dispatches the WHOLE fan-out as
     ONE background unit (one handle, one async slot). The children run in

tests/tui_gateway/test_delegation_session_lifecycle.py

@@ -0,0 +1,158 @@
+"""Fail-closed ownership + session-scoped delegation lifecycle (#55578).
+
+Covers the two hardening rules layered on top of the origin-routing salvage:
+
+1. ``_session_owns_notification_event`` — positive-proof ownership. An
+   async-delegation completion may only be injected into a session that
+   PROVABLY commissioned it (origin UI id, or session-key/lineage match).
+   Orphans are never adopted by a foreign chat.
+
+2. ``interrupt_for_session`` — a session's in-flight async delegations end
+   with the session. ``_finalize_session`` interrupts delegations owned by
+   the closing session (by origin UI id always; by durable key only when the
+   TUI owns the lifecycle).
+"""
+
+import threading
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+import tools.async_delegation as ad
+from tui_gateway.server import (
+    _finalize_session,
+    _session_owns_notification_event,
+)
+
+
[redacted:email](autouse=True)
+def _reset_async_delegation():
+    ad._reset_for_tests()
+    yield
+    ad._reset_for_tests()
+
+
+class TestSessionOwnsNotificationEvent:
+    def _session(self, key="sess_key_1"):
+        return {"session_key": key, "_finalized": False}
+
+    def test_origin_ui_match_owns(self):
+        evt = {"type": "async_delegation", "origin_ui_session_id": "tab1", "session_key": "other"}
+        assert _session_owns_notification_event("tab1", self._session(), evt) is True
+
+    def test_session_key_match_owns(self):
+        evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": "sess_key_1"}
+        assert _session_owns_notification_event("tabX", self._session("sess_key_1"), evt) is True
+
+    def test_orphan_is_not_owned(self):
+        """No origin match, no key match, owner gone → NOT ours (fail closed)."""
+        evt = {"type": "async_delegation", "origin_ui_session_id": "dead_tab", "session_key": "gone_key"}
+        assert _session_owns_notification_event("tab1", self._session(), evt) is False
+
+    def test_empty_key_and_origin_not_owned(self):
+        """A delegation event with no return address at all is never adopted."""
+        evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": ""}
+        assert _session_owns_notification_event("tab1", self._session(), evt) is False
+
+    def test_finalized_session_owns_nothing(self):
+        evt = {"type": "async_delegation", "origin_ui_session_id": "tab1", "session_key": "sess_key_1"}
+        sess = self._session()
+        sess["_finalized"] = True
+        assert _session_owns_notification_event("tab1", sess, evt) is False
+
+    def test_compression_chain_resolution_owns(self):
+        evt = {"type": "async_delegation", "origin_ui_session_id": "", "session_key": "parent_key"}
+        db = MagicMock()
+        db.resolve_resume_session_id.return_value = "child_key"
+        with patch("tui_gateway.server._get_db", return_value=db):
+            assert _session_owns_notification_event("tabX", self._session("child_key"), evt) is True
+
+
+class TestInterruptForSession:
+    def _seed_record(self, delegation_id, session_key="", origin_ui_session_id="", status="running"):
+        fn = MagicMock()
+        with ad._records_lock:
+            ad._records[delegation_id] = {
+                "delegation_id": delegation_id,
+                "status": status,
+                "session_key": session_key,
+                "origin_ui_session_id": origin_ui_session_id,
+                "interrupt_fn": fn,
+            }
+        return fn
+
+    def test_interrupts_only_matching_session(self):
+        mine = self._seed_record("d1", session_key="sess_A")
+        other = self._seed_record("d2", session_key="sess_B")
+        n = ad.interrupt_for_session(session_key="sess_A")
+        assert n == 1
+        mine.assert_called_once()
+        other.assert_not_called()
+
+    def test_matches_by_origin_ui_session_id(self):
+        mine = self._seed_record("d1", origin_ui_session_id="tab1")
+        other = self._seed_record("d2", origin_ui_session_id="tab2")
+        n = ad.interrupt_for_session(origin_ui_session_id="tab1")
+        assert n == 1
+        mine.assert_called_once()
+        other.assert_not_called()
+
+    def test_no_selector_is_noop(self):
+        fn = self._seed_record("d1", session_key="sess_A")
+        assert ad.interrupt_for_session() == 0
+        fn.assert_not_called()
+
+    def test_completed_records_untouched(self):
+        fn = self._seed_record("d1", session_key="sess_A", status="completed")
+        assert ad.interrupt_for_session(session_key="sess_A") == 0
+        fn.assert_not_called()
+
+
+class TestFinalizeInterruptsOwnDelegations:
+    def _make_session(self, session_key="sess_A", sid="tab1"):
+        agent = MagicMock()
+        agent.session_id = session_key
+        agent._session_messages = None
+        agent.model = "m"
+        agent.platform = "tui"
+        return {
+            "agent": agent,
+            "history": [{"role": "user", "content": "x"}],
+            "history_lock": threading.Lock(),
+            "session_key": session_key,
+            "_finalized": False,
+            "_sid": sid,
+        }
+
+    @patch("tui_gateway.server._get_db")
+    def test_finalize_interrupts_sessions_delegations(self, mock_get_db):
+        mock_db = MagicMock()
+        mock_db.get_session.return_value = {"source": "tui"}
+        mock_get_db.return_value = mock_db
+
+        with patch("tools.async_delegation.interrupt_for_session") as mock_int:
+            _finalize_session(self._make_session(), end_reason="tui_close")
+
+        mock_int.assert_called_once()
+        kwargs = mock_int.call_args.kwargs
+        assert kwargs["session_key"] == "sess_A"
+        assert kwargs["origin_ui_session_id"] == "tab1"
+
+    @patch("tui_gateway.server._get_db")
+    def test_viewer_of_gateway_session_only_interrupts_by_origin(self, mock_get_db):
+        """Closing a TUI viewer tab on a live gateway session must not kill
+        the gateway's own background work — key-based interrupt is skipped,
+        origin-id interrupt (this tab's own dispatches) still applies."""
+        mock_db = MagicMock()
+        mock_db.get_session.return_value = {"source": "telegram"}
+        mock_get_db.return_value = mock_db
+
+        with patch("tools.async_delegation.interrupt_for_session") as mock_int:
+            _finalize_session(
+                self._make_session(session_key="agent:main:telegram:dm:123", sid="tab9"),
+                end_reason="ws_orphan_reap",
+            )
+
+        kwargs = mock_int.call_args.kwargs
+        assert kwargs["session_key"] == ""
+        assert kwargs["origin_ui_session_id"] == "tab9"

tools/async_delegation.py

@@ -130,6 +130,7 @@ def dispatch_async_delegation(
     model: Optional[str],
     session_key: str,
     runner: Callable[[], Dict[str, Any]],
+    origin_ui_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
 ) -> Dict[str, Any]:
@@ -172,6 +173,7 @@ def dispatch_async_delegation(
         "role": role,
         "model": model,
         "session_key": session_key,
+        "origin_ui_session_id": origin_ui_session_id,
         "status": "running",
         "dispatched_at": dispatched_at,
         "completed_at": None,
@@ -282,6 +284,7 @@ def _push_completion_event(
         # session_key routes the completion back to the originating gateway
         # session; empty string => CLI (single-session) path.
         "session_key": record.get("session_key", ""),
+        "origin_ui_session_id": record.get("origin_ui_session_id", ""),
         "goal": record.get("goal", ""),
         "context": record.get("context"),
         "toolsets": record.get("toolsets"),
@@ -317,6 +320,7 @@ def dispatch_async_delegation_batch(
     model: Optional[str],
     session_key: str,
     runner: Callable[[], Dict[str, Any]],
+    origin_ui_session_id: str = "",
     interrupt_fn: Optional[Callable[[], None]] = None,
     max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
 ) -> Dict[str, Any]:
@@ -356,6 +360,7 @@ def dispatch_async_delegation_batch(
         "role": role,
         "model": model,
         "session_key": session_key,
+        "origin_ui_session_id": origin_ui_session_id,
         "status": "running",
         "dispatched_at": dispatched_at,
         "completed_at": None,
@@ -453,6 +458,7 @@ def _finalize_batch(
         "type": "async_delegation",
         "delegation_id": delegation_id,
         "session_key": event_record.get("session_key", ""),
+        "origin_ui_session_id": event_record.get("origin_ui_session_id", ""),
         "goal": event_record.get("goal", ""),
         "goals": event_record.get("goals"),
         "context": event_record.get("context"),
@@ -519,6 +525,54 @@ def interrupt_all(reason: str = "shutdown") -> int:
     return count
 
 
+def interrupt_for_session(
+    session_key: str = "",
+    origin_ui_session_id: str = "",
+    reason: str = "session_end",
+) -> int:
+    """Signal running async delegations owned by ONE session to stop.
+
+    A delegation's lifecycle is bound to the session that spawned it: when
+    that session ends, its in-flight background subagents must end with it —
+    a completed orphan would otherwise sit on the shared completion queue
+    with no live owner, either leaking into another chat or burning tokens
+    with no one listening (#55578).
+
+    Matches on ``origin_ui_session_id`` (the live UI session that
+    commissioned the work) and/or the durable ``session_key``; either
+    matching field claims the record. Returns how many were interrupted.
+    """
+    if not session_key and not origin_ui_session_id:
+        return 0
+    count = 0
+    with _records_lock:
+        targets = [
+            r for r in _records.values()
+            if r.get("status") == "running"
+            and (
+                (origin_ui_session_id and str(r.get("origin_ui_session_id") or "") == origin_ui_session_id)
+                or (session_key and str(r.get("session_key") or "") == session_key)
+            )
+        ]
+    for r in targets:
+        fn = r.get("interrupt_fn")
+        if callable(fn):
+            try:
+                fn()
+                count += 1
+            except Exception as exc:
+                logger.debug(
+                    "interrupt_for_session: %s interrupt failed: %s",
+                    r.get("delegation_id"), exc,
+                )
+    if count:
+        logger.info(
+            "Interrupted %d async delegation(s) for ending session (%s)",
+            count, reason,
+        )
+    return count
+
+
 def _reset_for_tests() -> None:
     """Test-only: clear all state and tear down the executor."""
     global _executor, _executor_max_workers

tools/delegate_tool.py

@@ -2796,6 +2796,25 @@ def _execute_and_aggregate() -> dict:
             return json.dumps(_sync_result, ensure_ascii=False)
 
         _session_key = get_current_session_key(default="")
+        _origin_ui_session_id = ""
+        try:
+            from gateway.session_context import get_session_env
+
+            _source = get_session_env("HERMES_SESSION_SOURCE", "")
+            _origin_ui_session_id = get_session_env("HERMES_UI_SESSION_ID", "")
+            # In desktop/TUI, the routable session key is the durable
+            # AIAgent.session_id. Context compression can rotate that id during
+            # the same turn before the TUI-side session dict is re-anchored;
+            # if we capture the stale approval/session context key here, the
+            # async completion becomes an orphan and any desktop poller may
+            # consume it. Gateway chats are different: their session_key is the
+            # platform conversation key (agent:main:...), so keep it there.
+            if _source == "tui":
+                _agent_session_id = str(getattr(parent_agent, "session_id", "") or "")
+                if _agent_session_id:
+                    _session_key = _agent_session_id
+        except Exception:
+            _origin_ui_session_id = ""
         _child_agents = [c for (_, _, c) in children]
 
         # Detach every child from the parent's interrupt-propagation list — the
@@ -2836,6 +2855,7 @@ def _batch_interrupt():
             role=top_role,
             model=creds["model"],
             session_key=_session_key,
+            origin_ui_session_id=_origin_ui_session_id,
             runner=_batch_runner,
             interrupt_fn=_batch_interrupt,
             max_async_children=_get_max_async_children(),

tui_gateway/server.py

@@ -625,6 +625,7 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
     # Use session_id (from agent.session_id) not session_key — after compression,
     # session_key may be stale (the ended parent) while session_id is the live
     # continuation. Fix for #20001.
+    _tui_owns_lifecycle = True
     if session_id:
         try:
             db = _get_db()
@@ -638,11 +639,40 @@ def _finalize_session(session: dict | None, end_reason: str = "tui_close") -> No
                 # repeats on every inbound message.  (#60609)
                 row = db.get_session(session_id)
                 source = (row or {}).get("source", "")
-                if not _is_gateway_owned_source(source):
+                _tui_owns_lifecycle = not _is_gateway_owned_source(source)
+                if _tui_owns_lifecycle:
                     db.end_session(session_id, end_reason)
         except Exception:
             pass
 
+    # A session's in-flight async delegations end WITH the session (#55578):
+    # once nobody owns the return address, a still-running background subagent
+    # can only burn tokens and park an orphaned completion on the shared
+    # queue. Always interrupt delegations commissioned by THIS live UI session
+    # (its sid); additionally interrupt by durable session_key, but only when
+    # the TUI owns the lifecycle — closing a viewer tab on a live gateway
+    # session must not kill the gateway's own background work.
+    try:
+        from tools.async_delegation import interrupt_for_session
+
+        _own_sid = str(session.get("_sid") or "")
+        if not _own_sid:
+            try:
+                with _sessions_lock:
+                    for _cand_sid, _cand in _sessions.items():
+                        if _cand is session:
+                            _own_sid = _cand_sid
+                            break
+            except Exception:
+                _own_sid = ""
+        interrupt_for_session(
+            session_key=str(session_key or "") if _tui_owns_lifecycle else "",
+            origin_ui_session_id=_own_sid,
+            reason=end_reason,
+        )
+    except Exception:
+        pass
+
     # Close the slash-worker subprocess as part of finalize itself, not just
     # in the callers. Defense-in-depth: every session-end path goes through
     # _finalize_session (it's the single ``_finalized``-guarded chokepoint), so
@@ -712,6 +742,10 @@ def _close_session_by_id(sid: str, *, end_reason: str = "tui_close") -> bool:
         session = _sessions.pop(sid, None)
     if session is None:
         return False
+    # The session is already out of _sessions here, so downstream teardown
+    # (e.g. _finalize_session's per-session async-delegation interrupt) can't
+    # recover its live id by scanning the dict — stamp it on the record.
+    session["_sid"] = sid
     _teardown_session(session, end_reason=end_reason)
     return True
 
@@ -1946,7 +1980,12 @@ def _cwd_for_session_key(session_key: str) -> str:
     return ""
 
 
-def _set_session_context(session_key: str, cwd: str | None = None) -> list:
+def _set_session_context(
+    session_key: str,
+    cwd: str | None = None,
+    *,
+    ui_session_id: str = "",
+) -> list:
     try:
         from gateway.session_context import set_session_vars
 
@@ -1961,7 +2000,12 @@ def _set_session_context(session_key: str, cwd: str | None = None) -> list:
                 if sess.get("session_key") == session_key:
                     source = _session_source(sess)
                     break
-        return set_session_vars(session_key=session_key, source=source, cwd=resolved)
+        return set_session_vars(
+            session_key=session_key,
+            source=source,
+            cwd=resolved,
+            ui_session_id=ui_session_id,
+        )
     except Exception:
         return []
 
@@ -8451,22 +8495,76 @@ def run_after_agent_ready() -> None:
     return _ok(rid, {"status": "streaming"})
 
 
-def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool:
+def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) -> bool:
     """True if ``evt`` is owned by a *different* live session.
 
-    Background-process events carry the ``session_key`` of the session that
-    started the process. Since all desktop sessions share one process-wide
-    completion queue, each poller must skip events it doesn't own so a
-    background job's completion surfaces in the session that launched it — not
-    whichever poller happened to dequeue first. Orphaned events (owner gone)
-    and global/system events (empty ``session_key``) return False so the
-    current poller still handles them rather than losing them.
+    Background completions carry the ``session_key`` of the session that started
+    the work. Async delegation completions from the desktop also carry
+    ``origin_ui_session_id``: the live TUI tab/window that commissioned them.
+    Since all desktop sessions share one process-wide completion queue, each
+    poller must skip events it doesn't own so a detached result surfaces in the
+    launching session, not whichever poller happened to dequeue first.
     """
+    evt_ui_sid = str(evt.get("origin_ui_session_id") or "")
+    if evt_ui_sid:
+        if evt_ui_sid == str(sid or "") and not session.get("_finalized"):
+            return False
+        try:
+            with _sessions_lock:
+                owner_live = evt_ui_sid in _sessions and not _sessions[evt_ui_sid].get("_finalized")
+        except Exception:
+            owner_live = False
+        if owner_live:
+            return True
+        # If the exact UI tab is gone, fall through to durable session_key
+        # routing. That avoids wrong-session delivery while still allowing a
+        # resumed continuation with the same durable key/lineage to claim it.
+
     evt_key = str(evt.get("session_key") or "")
     if not evt_key:
         return False
-    if evt_key == str(session.get("session_key") or ""):
+
+    current_keys = {
+        str(session.get("session_key") or ""),
+        _session_lookup_key(session, fallback=sid),
+    }
+
+    # Compression can rotate AIAgent.session_id while the detached child is
+    # still running. Resolve the event's original key to its continuation tip so
+    # an event captured before or after compression still maps to the same live
+    # desktop session instead of becoming an orphan that any poller may consume.
+    resolved_key = evt_key
+    try:
+        db = _get_db()
+        if db is not None:
+            resolved_key = db.resolve_resume_session_id(evt_key) or evt_key
+    except Exception:
+        resolved_key = evt_key
+
+    # If the key has a live continuation, prefer that continuation over the
+    # compressed parent. Otherwise a stale parent tab could consume the event
+    # before the real current conversation sees it.
+    if resolved_key != evt_key:
+        if resolved_key in current_keys:
+            return False
+        try:
+            with _sessions_lock:
+                continuation_live = any(
+                    not s.get("_finalized")
+                    and (
+                        str(s.get("session_key") or "") == resolved_key
+                        or _session_lookup_key(s, fallback="") == resolved_key
+                    )
+                    for s in _sessions.values()
+                )
+        except Exception:
+            continuation_live = False
+        if continuation_live:
+            return True
+
+    if evt_key in current_keys:
         return False
+
     try:
         with _sessions_lock:
             snapshot = list(_sessions.values())
@@ -8476,11 +8574,50 @@ def _notification_event_belongs_elsewhere(session: dict, evt: dict) -> bool:
         return False
 
     return any(
-        s is not session and str(s.get("session_key") or "") == evt_key
+        s is not session
+        and not s.get("_finalized")
+        and (
+            str(s.get("session_key") or "") in {evt_key, resolved_key}
+            or _session_lookup_key(s, fallback="") in {evt_key, resolved_key}
+        )
         for s in snapshot
     )
 
 
+def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool:
+    """True iff *this* session PROVABLY owns ``evt``.
+
+    Positive ownership — the mirror of ``_notification_event_belongs_elsewhere``
+    minus its orphan-adoption fallback. An event owns-matches when its
+    ``origin_ui_session_id`` is this live session, or its ``session_key``
+    (raw or resolved through the compression chain) matches this session's
+    key/lineage. Used as a fail-closed gate for async-delegation payloads:
+    "not provably elsewhere" is NOT good enough to inject a conversation
+    payload into this chat (#55578).
+    """
+    if session.get("_finalized"):
+        return False
+    if str(evt.get("origin_ui_session_id") or "") == str(sid or ""):
+        return True
+    evt_key = str(evt.get("session_key") or "")
+    if not evt_key:
+        return False
+    current_keys = {
+        str(session.get("session_key") or ""),
+        _session_lookup_key(session, fallback=sid),
+    }
+    if evt_key in current_keys:
+        return True
+    try:
+        db = _get_db()
+        resolved_key = (
+            db.resolve_resume_session_id(evt_key) if db is not None else evt_key
+        ) or evt_key
+    except Exception:
+        resolved_key = evt_key
+    return resolved_key in current_keys
+
+
 def _notification_event_dedup_key(evt: dict) -> tuple:
     """Return the UI-emission identity for a process notification event.
 
@@ -8546,11 +8683,37 @@ def _notification_poller_loop(
         # process started in session A would surface its completion in whichever
         # session's poller happened to wake first (Ben's "reported in a
         # different session" bug). Leave foreign events for their owner.
-        if _notification_event_belongs_elsewhere(session, evt):
+        if _notification_event_belongs_elsewhere(sid, session, evt):
             process_registry.completion_queue.put(evt)
             time.sleep(0.1)
             continue
 
+        # Fail closed for async-delegation results (#55578): these carry a
+        # conversation payload, and injecting one into any chat other than the
+        # one that commissioned it is a hard cross-session leak. The
+        # belongs-elsewhere check above already re-queued events owned by
+        # another LIVE session; what reaches here is either ours or an
+        # orphan whose owner is gone. Orphaned delegation payloads are
+        # DROPPED, not adopted — the subagent's summary is already persisted
+        # in the delegation records/output store, so nothing is lost, whereas
+        # a wrong-chat injection is unrecoverable. Non-delegation events
+        # (background process completions etc.) keep the historical
+        # adopt-orphans behavior.
+        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
+            sid, session, evt
+        ):
+            logger.warning(
+                "async-delegation completion %s has no live owner "
+                "(origin=%r key=%r); dropping from injection instead of "
+                "delivering to session %s (#55578 fail-closed; result "
+                "remains in the delegation records)",
+                evt.get("delegation_id", "?"),
+                str(evt.get("origin_ui_session_id") or ""),
+                str(evt.get("session_key") or ""),
+                sid,
+            )
+            continue
+
         _evt_sid = evt.get("session_id", "")
         if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
             continue
@@ -8604,7 +8767,16 @@ def _notification_poller_loop(
             evt = process_registry.completion_queue.get_nowait()
         except Exception:
             break
-        if _notification_event_belongs_elsewhere(session, evt):
+        if _notification_event_belongs_elsewhere(sid, session, evt):
+            deferred.append(evt)
+            continue
+        # Same fail-closed rule as the live loop: an orphaned async-delegation
+        # payload is never adopted by a foreign session — defer it (a later
+        # resume of the owner's lineage can still claim it) rather than
+        # injecting another chat's conversation here (#55578).
+        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
+            sid, session, evt
+        ):
             deferred.append(evt)
             continue
         _evt_sid = evt.get("session_id", "")
@@ -8729,7 +8901,10 @@ def run():
             )
 
             approval_token = [redacted:secret-assignment]"session_key"])
-            session_tokens = [redacted:secret-assignment]"session_key"])
+            session_tokens = _set_session_context(
+                session["session_key"],
+                ui_session_id=sid,
+            )
             _profile_home_str = session.get("profile_home")
             if _profile_home_str:
                 home_token = [redacted:secret-assignment]

#65419 fix(tui): route background-process notifications to their owning session (#35652)

merged to main

files: scripts/release.py, tests/test_tui_gateway_server.py, tests/tools/test_process_registry.py, tools/process_registry.py, tui_gateway/server.py

Show the diff

scripts/release.py

@@ -45,6 +45,7 @@
 
 # Auto-extracted from noreply emails + manual overrides
 AUTHOR_MAP = {
+    "[email protected]": "2751738943",  # PR #54785 salvage (tui: post-turn completion ownership routing)
     "[email protected]": "Burgunthy",  # PR #20096 salvage (gateway: profile-based routing for inbound messages)
     "[email protected]": "webtecnica",  # PR #63360 salvage (nous: restore inference-api base_url)
     "[redacted:email]": "Epoxidex",  # PR #29820 salvage (ollama: top-level reasoning_effort=none; #25758)

tests/test_tui_gateway_server.py

@@ -2284,7 +2284,611 @@ def replace_messages(self, key, messages):
         server._sessions.pop("trunc-sid", None)
 
 
-def test_session_create_does_not_persist_empty_row(monkeypatch):
+class _StopAfterOneNotificationPoll:
+    def __init__(self):
+        self._checks = 0
+
+    def is_set(self):
+        self._checks += 1
+        return self._checks > 1
+
+
+def test_notification_poller_live_loop_requeues_foreign_completion_for_owner(
+    monkeypatch,
+):
+    """A foreign live-loop dequeue is handed back to its proven owner."""
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    delivered = {"a": [], "b": []}
+    emitted = []
+    session_a = _session(session_key="session-a-live-handoff")
+    session_b = _session(session_key="session-b-live-handoff")
+    event = {
+        "type": "completion",
+        "session_id": "proc-live-handoff",
+        "session_key": "session-a-live-handoff",
+        "command": "echo owner",
+        "exit_code": 0,
+        "output": "owner",
+    }
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    monkeypatch.setattr(server, "_get_db", lambda: None)
+    monkeypatch.setattr(server, "_emit", lambda *args, **_kwargs: emitted.append(args))
+
+    def _deliver(_rid, sid, session, text):
+        delivered["a" if sid == "sid-a-live-handoff" else "b"].append(text)
+        session["running"] = False
+
+    monkeypatch.setattr(server, "_run_prompt_submit", _deliver)
+    server._sessions.update(
+        {
+            "sid-a-live-handoff": session_a,
+            "sid-b-live-handoff": session_b,
+        }
+    )
+    process_registry._completion_consumed.discard(event["session_id"])
+
+    try:
+        server._notification_poller_loop(
+            _StopAfterOneNotificationPoll(), "sid-b-live-handoff", session_b
+        )
+
+        assert delivered["b"] == []
+        assert emitted == []
+        assert isolated_queue.qsize() == 1
+        assert isolated_queue.queue[0] is event
+
+        server._notification_poller_loop(
+            _StopAfterOneNotificationPoll(), "sid-a-live-handoff", session_a
+        )
+
+        assert len(delivered["a"]) == 1
+        assert "proc-live-handoff completed normally" in delivered["a"][0]
+        assert delivered["b"] == []
+        assert isolated_queue.empty()
+    finally:
+        server._sessions.pop("sid-a-live-handoff", None)
+        server._sessions.pop("sid-b-live-handoff", None)
+        process_registry._completion_consumed.discard(event["session_id"])
+        while not isolated_queue.empty():
+            isolated_queue.get_nowait()
+
+
+def test_completion_ownership_lineage_lookup_failure_fails_closed(monkeypatch):
+    """A provenance lookup failure cannot turn an addressed event into ours."""
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    class _BrokenDB:
+        def resolve_resume_session_id(self, _session_key):
+            raise RuntimeError("lineage database unavailable")
+
+    session = _session(session_key="unrelated-live-session")
+    event = {
+        "type": "completion",
+        "session_id": "proc-unknown-lineage",
+        "session_key": "unknown-parent",
+        "command": "echo unknown",
+        "exit_code": 0,
+        "output": "unknown",
+    }
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    monkeypatch.setattr(server, "_get_db", lambda: _BrokenDB())
+
+    drained = process_registry.drain_notifications(
+        session_key="unrelated-live-session",
+        owns_event=lambda candidate: server._session_owns_notification_event(
+            "sid-unrelated-live", session, candidate
+        ),
+    )
+
+    assert drained == []
+    assert isolated_queue.qsize() == 1
+    assert isolated_queue.get_nowait() is event
+
+
[redacted:email](
+    "routing",
+    [
+        {"session_key": "missing-owner-key"},
+        {"origin_ui_session_id": "missing-owner-sid"},
+    ],
+)
+def test_notification_poller_live_loop_drops_addressed_orphan(
+    monkeypatch, routing
+):
+    """A live poll never injects an addressed event whose owner is gone."""
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    delivered = []
+    emitted = []
+    session = _session(session_key="unrelated-live-key")
+    event = {
+        "type": "completion",
+        "session_id": "proc-live-orphan",
+        "command": "echo orphan",
+        "exit_code": 0,
+        "output": "orphan",
+        **routing,
+    }
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    monkeypatch.setattr(server, "_get_db", lambda: None)
+    monkeypatch.setattr(server, "_emit", lambda *args, **_kwargs: emitted.append(args))
+    monkeypatch.setattr(
+        server,
+        "_run_prompt_submit",
+        lambda _rid, _sid, _session, text: delivered.append(text),
+    )
+    server._sessions["sid-live-orphan"] = session
+    process_registry._completion_consumed.discard(event["session_id"])
+
+    try:
+        server._notification_poller_loop(
+            _StopAfterOneNotificationPoll(), "sid-live-orphan", session
+        )
+
+        assert delivered == []
+        assert emitted == []
+        assert isolated_queue.empty()
+    finally:
+        server._sessions.pop("sid-live-orphan", None)
+        process_registry._completion_consumed.discard(event["session_id"])
+        while not isolated_queue.empty():
+            isolated_queue.get_nowait()
+
+
[redacted:email](
+    "routing",
+    [
+        {"session_key": "session-b"},
+        {"origin_ui_session_id": "sid_gone"},
+    ],
+)
+def test_notification_poller_drops_orphaned_events(monkeypatch, routing):
+    """Addressed completions whose owner is gone are dropped, not hijacked."""
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    emitted = []
+    delivered = []
+    sess = _session(session_key="session-a")
+    server._sessions["sid_a"] = sess
+    monkeypatch.setattr(server, "_emit", lambda *a, **kw: emitted.append(a))
+    monkeypatch.setattr(
+        server,
+        "_run_prompt_submit",
+        lambda _rid, _sid, _session, text: delivered.append(text),
+    )
+    monkeypatch.setattr(server, "_get_db", lambda: None)
+
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    process_registry._completion_consumed.discard("proc_ghost")
+    isolated_queue.put(
+        {
+            "type": "completion",
+            "session_id": "proc_ghost",
+            "command": "echo from ghost",
+            "exit_code": 0,
+            "output": "ghost output",
+            **routing,
+        }
+    )
+
+    stop = threading.Event()
+    stop.set()
+
+    try:
+        server._notification_poller_loop(stop, "sid_a", sess)
+
+        assert [a for a in emitted if a[0] == "status.update"] == []
+        assert delivered == []
+    finally:
+        server._sessions.pop("sid_a", None)
+        while not process_registry.completion_queue.empty():
+            process_registry.completion_queue.get_nowait()
+
+
[redacted:email](
+    ("routing", "resolved_key"),
+    [
+        ({"session_key": "session-a"}, None),
+        (
+            {
+                "session_key": "stale-durable-key",
+                "origin_ui_session_id": "sid_a",
+            },
+            None,
+        ),
+        ({"session_key": "old-parent-key"}, "session-a"),
+    ],
+)
+def test_notification_poller_delivers_owned_events(
+    monkeypatch, routing, resolved_key
+):
+    """Direct, UI-origin, and compression-lineage owners are delivered."""
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    class _CompressionDB:
+        def resolve_resume_session_id(self, key):
+            return resolved_key if key == "old-parent-key" and resolved_key else key
+
+    delivered = []
+    emitted = []
+    sess = _session(session_key="session-a")
+    server._sessions["sid_a"] = sess
+    monkeypatch.setattr(server, "_emit", lambda *a, **kw: emitted.append(a))
+    monkeypatch.setattr(
+        server,
+        "_run_prompt_submit",
+        lambda _rid, _sid, _session, text: delivered.append(text),
+    )
+    monkeypatch.setattr(server, "_get_db", lambda: _CompressionDB())
+
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    process_registry._completion_consumed.discard("proc_mine")
+    isolated_queue.put(
+        {
+            "type": "completion",
+            "session_id": "proc_mine",
+            "command": "echo mine",
+            "exit_code": 0,
+            "output": "mine",
+            **routing,
+        }
+    )
+
+    stop = threading.Event()
+    stop.set()
+
+    try:
+        server._notification_poller_loop(stop, "sid_a", sess)
+
+        status_calls = [a for a in emitted if a[0] == "status.update"]
+        assert len(status_calls) == 1
+        assert status_calls[0][2]["kind"] == "process"
+        assert len(delivered) == 1
+        assert "proc_mine" in delivered[0]
+    finally:
+        server._sessions.pop("sid_a", None)
+        while not process_registry.completion_queue.empty():
+            process_registry.completion_queue.get_nowait()
+
+
+def _configure_immediate_prompt_run(
+    monkeypatch, tmp_path, *, immediate_threads=True
+):
+    class _ImmediateThread:
+        def __init__(self, target=None, daemon=None, **_kwargs):
+            self._target = target
+
+        def start(self):
+            if self._target is not None:
+                self._target()
+
+        def is_alive(self):
+            return False
+
+    if immediate_threads:
+        monkeypatch.setattr(server.threading, "Thread", _ImmediateThread)
+    monkeypatch.setattr(server, "_emit", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(server, "make_stream_renderer", lambda _cols: None)
+    monkeypatch.setattr(server, "render_message", lambda _raw, _cols: None)
+    monkeypatch.setattr(server, "_wire_callbacks", lambda _sid: None)
+    monkeypatch.setattr(server, "_sync_agent_model_with_config", lambda *_args: None)
+    monkeypatch.setattr(server, "_session_cwd", lambda _session: str(tmp_path))
+    monkeypatch.setattr(server, "_register_session_cwd", lambda _session: None)
+    monkeypatch.setattr(server, "_set_session_context", lambda *_args, **_kwargs: [])
+    monkeypatch.setattr(server, "_clear_session_context", lambda _tokens: None)
+    monkeypatch.setattr(server, "_session_info", lambda *_args: {})
+    monkeypatch.setattr(server, "_get_usage", lambda _agent: {})
+    monkeypatch.setattr(
+        server, "_sync_session_key_after_compress", lambda *_args, **_kwargs: None
+    )
+    monkeypatch.setattr(server, "_drain_queued_prompt", lambda *_args: False)
+    monkeypatch.setattr(server, "_voice_tts_enabled", lambda: False)
+    monkeypatch.setattr(server, "_get_db", lambda: None)
+
+
+class _RecordingAgent:
+    model = "test-model"
+    provider = "test-provider"
+
+    def __init__(self, turns):
+        self._turns = turns
+
+    def clear_interrupt(self):
+        return None
+
+    def run_conversation(
+        self, prompt, conversation_history=None, stream_callback=None
+    ):
+        self._turns.append(prompt)
+        return {"final_response": "", "messages": []}
+
+
[redacted:email]("exit_code", [0, 7])
+def test_run_prompt_submit_requeues_foreign_completion(
+    monkeypatch, tmp_path, exit_code
+):
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    _configure_immediate_prompt_run(monkeypatch, tmp_path)
+    turns = []
+    session_a = _session(session_key="session-a")
+    session_b = _session(
+        session_key="session-b",
+        agent=_RecordingAgent(turns),
+        running=True,
+    )
+    event = {
+        "type": "completion",
+        "session_id": f"proc_foreign_{exit_code}",
+        "session_key": "session-a",
+        "command": "safe-test-command",
+        "exit_code": exit_code,
+        "output": "foreign",
+    }
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    server._sessions["sid_a"] = session_a
+    server._sessions["sid_b"] = session_b
+
+    try:
+        server._run_prompt_submit("rid-b", "sid_b", session_b, "session-b-turn")
+
+        assert turns == ["session-b-turn"]
+        assert isolated_queue.get_nowait() == event
+        assert isolated_queue.empty()
+    finally:
+        server._sessions.pop("sid_a", None)
+        server._sessions.pop("sid_b", None)
+        process_registry._completion_consumed.discard(event["session_id"])
+
+
+def test_run_prompt_submit_delivers_completion_observed_by_poll(monkeypatch, tmp_path):
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    _configure_immediate_prompt_run(monkeypatch, tmp_path)
+    turns = []
+    session = _session(
+        session_key="session-a",
+        agent=_RecordingAgent(turns),
+        running=True,
+    )
+    event = {
+        "type": "completion",
+        "session_id": "proc_polled",
+        "session_key": "session-a",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "observed but not consumed",
+    }
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    process_registry._completion_consumed.discard(event["session_id"])
+    process_registry._poll_observed.add(event["session_id"])
+    server._sessions["sid_a"] = session
+
+    try:
+        server._run_prompt_submit("rid-a", "sid_a", session, "session-a-turn")
+
+        assert turns[0] == "session-a-turn"
+        assert len(turns) == 2
+        assert "proc_polled" in turns[1]
+        assert isolated_queue.empty()
+    finally:
+        server._sessions.pop("sid_a", None)
+        process_registry._completion_consumed.discard(event["session_id"])
+        process_registry._poll_observed.discard(event["session_id"])
+
+
+def test_run_prompt_submit_requeues_all_unstarted_notifications_with_real_threading(
+    monkeypatch, tmp_path
+):
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    _configure_immediate_prompt_run(
+        monkeypatch, tmp_path, immediate_threads=False
+    )
+    real_thread_class = threading.Thread
+    threads = []
+    nested_started = threading.Event()
+    release_nested = threading.Event()
+    turns = []
+
+    def _recording_thread(*args, **kwargs):
+        thread = real_thread_class(*args, **kwargs)
+        threads.append(thread)
+        return thread
+
+    class _BlockingNotificationAgent(_RecordingAgent):
+        def run_conversation(
+            self, prompt, conversation_history=None, stream_callback=None
+        ):
+            turns.append(prompt)
+            if "proc_batch_1" in prompt:
+                nested_started.set()
+                if not release_nested.wait(timeout=5):
+                    raise TimeoutError("notification turn was not released")
+            return {"final_response": "", "messages": []}
+
+    monkeypatch.setattr(server.threading, "Thread", _recording_thread)
+    session = _session(
+        session_key="session-a",
+        agent=_BlockingNotificationAgent(turns),
+        running=True,
+    )
+    events = [
+        {
+            "type": "completion",
+            "session_id": f"proc_batch_{index}",
+            "session_key": "session-a",
+            "command": "safe-test-command",
+            "exit_code": 0,
+            "output": f"owned-{index}",
+        }
+        for index in range(1, 4)
+    ]
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    for event in events:
+        isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    server._sessions["sid_a"] = session
+
+    try:
+        server._run_prompt_submit("rid-a", "sid_a", session, "session-a-turn")
+
+        assert nested_started.wait(timeout=5)
+        threads[0].join(timeout=5)
+        assert not threads[0].is_alive()
+        queued = []
+        while not isolated_queue.empty():
+            queued.append(isolated_queue.get_nowait())
+        assert [event["session_id"] for event in queued] == [
+            "proc_batch_2",
+            "proc_batch_3",
+        ]
+    finally:
+        release_nested.set()
+        for thread in threads:
+            thread.join(timeout=5)
+        server._sessions.pop("sid_a", None)
+        while not isolated_queue.empty():
+            isolated_queue.get_nowait()
+        for event in events:
+            process_registry._completion_consumed.discard(event["session_id"])
+            process_registry._poll_observed.discard(event["session_id"])
+
+
+def test_run_prompt_submit_delivers_completion_owned_through_compression_lineage(
+    monkeypatch, tmp_path
+):
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    class _CompressionDB:
+        def resolve_resume_session_id(self, key):
+            return "new-child-key" if key == "old-parent-key" else key
+
+    _configure_immediate_prompt_run(monkeypatch, tmp_path)
+    monkeypatch.setattr(server, "_get_db", lambda: _CompressionDB())
+    ownership_checks = []
+    original_owns_event = server._session_owns_notification_event
+
+    def _record_ownership_check(sid, checked_session, checked_event):
+        ownership_checks.append(checked_event["session_id"])
+        return original_owns_event(sid, checked_session, checked_event)
+
+    monkeypatch.setattr(
+        server, "_session_owns_notification_event", _record_ownership_check
+    )
+    turns = []
+    session = _session(
+        session_key="new-child-key",
+        agent=_RecordingAgent(turns),
+        running=True,
+    )
+    event = {
+        "type": "completion",
+        "session_id": "proc_precompression",
+        "session_key": "old-parent-key",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "owned",
+    }
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    server._sessions["sid_b"] = session
+
+    try:
+        server._run_prompt_submit("rid-b", "sid_b", session, "session-b-turn")
+
+        assert turns[0] == "session-b-turn"
+        assert len(turns) == 2
+        assert "proc_precompression" in turns[1]
+        assert ownership_checks == ["proc_precompression"]
+        assert isolated_queue.empty()
+    finally:
+        server._sessions.pop("sid_b", None)
+        process_registry._completion_consumed.discard(event["session_id"])
+
+
+def test_run_prompt_submit_prefers_origin_ui_session_id(monkeypatch, tmp_path):
+    import queue as _queue_mod
+
+    from tools.process_registry import process_registry
+
+    _configure_immediate_prompt_run(monkeypatch, tmp_path)
+    ownership_checks = []
+    original_owns_event = server._session_owns_notification_event
+
+    def _record_ownership_check(sid, checked_session, checked_event):
+        ownership_checks.append(checked_event["session_id"])
+        return original_owns_event(sid, checked_session, checked_event)
+
+    monkeypatch.setattr(
+        server, "_session_owns_notification_event", _record_ownership_check
+    )
+    turns = []
+    session = _session(
+        session_key="current-key",
+        agent=_RecordingAgent(turns),
+        running=True,
+    )
+    event = {
+        "type": "completion",
+        "session_id": "proc_origin_owned",
+        "session_key": "stale-durable-key",
+        "origin_ui_session_id": "sid_b",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "owned",
+    }
+    isolated_queue: _queue_mod.Queue = _queue_mod.Queue()
+    isolated_queue.put(event)
+    monkeypatch.setattr(process_registry, "completion_queue", isolated_queue)
+    server._sessions["sid_b"] = session
+
+    try:
+        server._run_prompt_submit("rid-b", "sid_b", session, "session-b-turn")
+
+        assert turns[0] == "session-b-turn"
+        assert len(turns) == 2
+        assert "proc_origin_owned" in turns[1]
+        assert ownership_checks == ["proc_origin_owned"]
+        assert isolated_queue.empty()
+    finally:
+        server._sessions.pop("sid_b", None)
+        process_registry._completion_consumed.discard(event["session_id"])
+
+
+
     """session.create must NOT eagerly write a DB row.
 
     Every TUI/desktop launch opens a session here just to paint the composer;

tests/tools/test_process_registry.py

@@ -1336,6 +1336,66 @@ def test_drain_notifications_skips_consumed():
             process_registry.completion_queue.get_nowait()
 
 
+def test_drain_notifications_can_deliver_poll_observed_for_gateway(registry):
+    event = {
+        "type": "completion",
+        "session_id": "proc_polled",
+        "session_key": "session-a",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "observed but not consumed",
+    }
+    registry._poll_observed.add(event["session_id"])
+    registry.completion_queue.put(event)
+
+    try:
+        results = registry.drain_notifications(
+            session_key="session-a",
+            owns_event=lambda _event: True,
+            skip_poll_observed=False,
+        )
+
+        assert [raw for raw, _ in results] == [event]
+    finally:
+        registry._poll_observed.discard(event["session_id"])
+
+
[redacted:email](
+    "skip_state", ["_poll_observed", "_completion_consumed"]
+)
+def test_drain_notifications_routes_foreign_before_local_skip(
+    registry, skip_state
+):
+    event = {
+        "type": "completion",
+        "session_id": f"proc_foreign_{skip_state}",
+        "session_key": "session-a",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "foreign",
+    }
+    ownership_calls = []
+    getattr(registry, skip_state).add(event["session_id"])
+    registry.completion_queue.put(event)
+
+    def owns_event(checked_event):
+        ownership_calls.append(checked_event)
+        return False
+
+    try:
+        results = registry.drain_notifications(
+            session_key="session-b",
+            owns_event=owns_event,
+        )
+
+        assert results == []
+        assert ownership_calls == [event]
+        assert registry.completion_queue.get_nowait() == event
+        assert registry.completion_queue.empty()
+    finally:
+        getattr(registry, skip_state).discard(event["session_id"])
+
+
 def test_drain_notifications_empty_queue():
     from tools.process_registry import process_registry
 
@@ -1346,6 +1406,151 @@ def test_drain_notifications_empty_queue():
     assert results == []
 
 
[redacted:email]("exit_code", [0, 7])
+def test_drain_notifications_filters_addressed_completion_by_owns_event(
+    registry, exit_code
+):
+    owned = {
+        "type": "completion",
+        "session_id": f"proc_owned_{exit_code}",
+        "session_key": "session-a",
+        "command": "safe-test-command",
+        "exit_code": exit_code,
+        "output": "owned",
+    }
+    foreign = {
+        "type": "completion",
+        "session_id": f"proc_foreign_{exit_code}",
+        "session_key": "session-b",
+        "command": "safe-test-command",
+        "exit_code": exit_code,
+        "output": "foreign",
+    }
+    registry.completion_queue.put(owned)
+    registry.completion_queue.put(foreign)
+
+    results = registry.drain_notifications(
+        session_key="session-a",
+        owns_event=lambda event: event.get("session_key") == "session-a",
+    )
+
+    assert [event["session_id"] for event, _ in results] == [
+        f"proc_owned_{exit_code}"
+    ]
+    assert registry.completion_queue.get_nowait() == foreign
+    assert registry.completion_queue.empty()
+
+
+def test_drain_notifications_filters_addressed_completion_by_session_key(registry):
+    owned = {
+        "type": "completion",
+        "session_id": "proc_owned",
+        "session_key": "session-a",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "owned",
+    }
+    foreign = {
+        "type": "completion",
+        "session_id": "proc_foreign",
+        "session_key": "session-b",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "foreign",
+    }
+    registry.completion_queue.put(owned)
+    registry.completion_queue.put(foreign)
+
+    results = registry.drain_notifications(session_key="session-a")
+
+    assert [event["session_id"] for event, _ in results] == ["proc_owned"]
+    assert registry.completion_queue.get_nowait() == foreign
+    assert registry.completion_queue.empty()
+
+
+def test_drain_notifications_session_key_filter_requeues_origin_only_event(registry):
+    event = {
+        "type": "completion",
+        "session_id": "proc_origin_only",
+        "origin_ui_session_id": "ui-session-a",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "done",
+    }
+    registry.completion_queue.put(event)
+
+    results = registry.drain_notifications(session_key="session-a")
+
+    assert results == []
+    assert registry.completion_queue.get_nowait() == event
+    assert registry.completion_queue.empty()
+
+
+def test_drain_notifications_ownerless_completion_preserves_legacy_delivery(registry):
+    event = {
+        "type": "completion",
+        "session_id": "proc_ownerless",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "ownerless",
+    }
+    registry.completion_queue.put(event)
+
+    results = registry.drain_notifications(
+        session_key="session-a",
+        owns_event=lambda _event: False,
+    )
+
+    assert [raw for raw, _ in results] == [event]
+    assert registry.completion_queue.empty()
+
+
+def test_drain_notifications_ownerless_async_delegation_still_requires_proof(registry):
+    event = {
+        "type": "async_delegation",
+        "delegation_id": "deleg_ownerless",
+        "goal": "task",
+        "status": "completed",
+        "summary": "done",
+        "api_calls": 1,
+        "duration_seconds": 0.1,
+    }
+    registry.completion_queue.put(event)
+
+    results = registry.drain_notifications(
+        session_key="session-a",
+        owns_event=lambda _event: False,
+    )
+
+    assert results == []
+    assert registry.completion_queue.get_nowait() == event
+    assert registry.completion_queue.empty()
+
+
+def test_drain_notifications_completion_callback_exception_fails_closed(registry):
+    event = {
+        "type": "completion",
+        "session_id": "proc_callback_error",
+        "session_key": "session-a",
+        "command": "safe-test-command",
+        "exit_code": 0,
+        "output": "done",
+    }
+    registry.completion_queue.put(event)
+
+    def broken(_event):
+        raise RuntimeError("ownership check exploded")
+
+    results = registry.drain_notifications(
+        session_key="session-a",
+        owns_event=broken,
+    )
+
+    assert results == []
+    assert registry.completion_queue.get_nowait() == event
+    assert registry.completion_queue.empty()
+
+
 def test_drain_notifications_filters_async_delegation_by_session_key():
     """Async-delegation events should only be consumed by the matching session's drain.

tools/process_registry.py

@@ -1146,8 +1146,10 @@ def is_session_waiting(self, session_id: str) -> bool:
                 return False
         return True
 
-    def _drain_should_skip(self, session_id: str) -> bool:
-        """Whether the CLI drain should skip a completion event for this session.
+    def _drain_should_skip(
+        self, session_id: str, *, skip_poll_observed: bool = True
+    ) -> bool:
+        """Whether this drain should skip a completion event for this session.
 
         Skips when the agent has either truly consumed the output (wait/log →
         ``_completion_consumed``) or observed the exit inline via poll()
@@ -1157,32 +1159,45 @@ def _drain_should_skip(self, session_id: str) -> bool:
         check only ``is_completion_consumed`` so a read-only poll never
         suppresses their autonomous delivery turn (#10156).
         """
-        return session_id in self._completion_consumed or session_id in self._poll_observed
+        return session_id in self._completion_consumed or (
+            skip_poll_observed and session_id in self._poll_observed
+        )
 
     def drain_notifications(
-        self, session_key: str = "", owns_event=None,
+        self,
+        session_key: str = "",
+        owns_event=None,
+        *,
+        skip_poll_observed: bool = True,
     ) -> "list[tuple[dict, str]]":
         """Pop all pending notification events and return formatted pairs.
 
         Returns a list of (raw_event, formatted_text) tuples.
         Skips completion events the agent already consumed via wait/log or
-        observed inline via poll() (see ``_drain_should_skip``).
+        observed inline via poll() (see ``_drain_should_skip``). Gateway/TUI
+        callers pass ``skip_poll_observed=False`` because read-only polling must
+        not suppress autonomous delivery there.
 
-        Async-delegation events carry a conversation payload, so draining one
-        into the wrong session is a cross-chat leak (#58684, #55578). Two
-        filter modes, strongest wins:
+        When a routing filter is supplied, addressed notifications must not be
+        drained into the wrong session. Async-delegation events always require
+        conversation payload; ordinary notifications require routing when they
+        carry ``session_key`` or ``origin_ui_session_id`` metadata. Two filter
+        modes are supported, strongest first:
 
         - ``owns_event(evt) -> bool``: positive-proof ownership callback.
-          When provided, an async-delegation event is consumed ONLY if the
-          callback returns True; everything else is re-queued for its owner.
+          When provided, a routed event is consumed ONLY if the callback
+          returns True; everything else is re-queued for its owner.
           The TUI passes its compression-chain-aware ownership check here so
           a post-compression session still claims its own pre-compression
           dispatches.
         - ``session_key``: plain key equality (CLI and other single-session
-          callers). Non-matching async-delegation events are re-queued.
+          callers). Non-matching addressed events are re-queued.
 
         With neither set, all events are consumed (legacy single-session
-        behavior, backward compatible).
+        behavior, backward compatible). Ownerless ordinary notifications also
+        retain that legacy behavior even when a filter is provided. When a
+        filter is provided, ownerless async-delegation events remain
+        fail-closed and require positive proof.
         """
         results: "list[tuple[dict, str]]" = []
         requeue: "list[dict]" = []
@@ -1191,39 +1206,43 @@ def drain_notifications(
                 evt = self.completion_queue.get_nowait()
             except Exception:
                 break
-            _evt_sid = evt.get("session_id", "")
-            if evt.get("type") == "completion" and self._drain_should_skip(_evt_sid):
-                continue
-            # Filter async-delegation events so they are not delivered to the
-            # wrong session/thread (#58684). Positive-proof callback beats
-            # bare key equality when the caller can provide one.
-            if evt.get("type") == "async_delegation":
-                if owns_event is not None:
-                    try:
-                        owned = bool(owns_event(evt))
-                    except Exception:
-                        owned = False  # fail closed — never leak on a broken check
-                    if not owned:
-                        requeue.append(evt)
-                        continue
-                elif session_key:
-                    evt_session_key = evt.get("session_key", "") or ""
-                    if evt_session_key != session_key:
-                        requeue.append(evt)
-                        continue
-                elif evt.get("restored"):
-                    # Legacy unfiltered drain (no ownership callback, no
-                    # session key). That behavior was safe when the in-memory
-                    # queue could only hold events created by this very
-                    # process — but durable restore (#63494) re-enqueues
-                    # completions from PREVIOUS processes at startup, so an
-                    # unfiltered consumer here would adopt a dead, unrelated
-                    # session's conversation payload (#64484). Fail closed:
-                    # leave restored events queued (still 'pending' on disk)
-                    # for a consumer that can positively prove ownership,
-                    # e.g. the owning session's --resume.
+            # Positive-proof ownership beats bare key equality. Delegation
+            # payloads always require proof; ordinary events require it once
+            # they carry routing metadata. Ownerless ordinary events preserve
+            # legacy single-session delivery.
+            is_async_delegation = evt.get("type") == "async_delegation"
+            evt_session_key = str(evt.get("session_key") or "")
+            evt_origin_sid = str(evt.get("origin_ui_session_id") or "")
+            requires_positive_proof = is_async_delegation or bool(
+                evt_session_key or evt_origin_sid
+            )
+            if owns_event is not None and requires_positive_proof:
+                try:
+                    owned = bool(owns_event(evt))
+                except Exception:
+                    owned = False  # fail closed — never leak on a broken check
+                if not owned:
+                    requeue.append(evt)
+                    continue
+            elif session_key and requires_positive_proof:
+                if evt_session_key != session_key:
                     requeue.append(evt)
                     continue
+            elif is_async_delegation and evt.get("restored"):
+                # Durable restore can enqueue previous-process payloads into a
+                # fresh registry. An unfiltered legacy drain cannot prove
+                # ownership, so leave those events queued for the owner.
+                requeue.append(evt)
+                continue
+            # Local consumed/observed state may suppress only events this
+            # session owns (or legacy ownerless ordinary events). Routing must
+            # happen first so a foreign session cannot drop the owner's event.
+            _evt_sid = evt.get("session_id", "")
+            if evt.get("type") == "completion" and self._drain_should_skip(
+                _evt_sid, skip_poll_observed=skip_poll_observed
+            ):
+                continue
+
             text = format_process_notification(evt)
             if text:
                 results.append((evt, text))

tui_gateway/server.py

@@ -8652,9 +8652,9 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
     minus its orphan-adoption fallback. An event owns-matches when its
     ``origin_ui_session_id`` is this live session, or its ``session_key``
     (raw or resolved through the compression chain) matches this session's
-    key/lineage. Used as a fail-closed gate for async-delegation payloads:
-    "not provably elsewhere" is NOT good enough to inject a conversation
-    payload into this chat (#55578).
+    key/lineage. Used as the fail-closed gate for every addressed notification:
+    "not provably elsewhere" is NOT good enough to inject a payload into this
+    chat (#55578).
     """
     if session.get("_finalized"):
         return False
@@ -8679,6 +8679,14 @@ def _session_owns_notification_event(sid: str, session: dict, evt: dict) -> bool
     return resolved_key in current_keys
 
 
+def _notification_event_requires_owner(evt: dict) -> bool:
+    """Whether ``evt`` must be positively claimed before TUI delivery."""
+    return evt.get("type") == "async_delegation" or bool(
+        str(evt.get("origin_ui_session_id") or "")
+        or str(evt.get("session_key") or "")
+    )
+
+
 def _notification_event_dedup_key(evt: dict) -> tuple:
     """Return the UI-emission identity for a process notification event.
 
@@ -8725,10 +8733,9 @@ def _notification_poller_loop(
     status.update (kind=process) for user visibility, then chains an
     agent turn via _run_prompt_submit if the session is idle.
 
-    NOTE: The completion_queue is global (one per process). If multiple
-    TUI sessions coexist, whichever poller wakes first grabs the event,
-    even if the process was started by a different session. This matches
-    CLI/gateway behavior (single session per process).
+    The completion_queue is process-global. In multi-session Desktop each
+    poller requeues events owned by another live session and drops addressed
+    events whose owner is gone; ownerless legacy notifications remain global.
     """
     from tools.process_registry import process_registry, format_process_notification
 
@@ -8749,26 +8756,22 @@ def _notification_poller_loop(
             time.sleep(0.1)
             continue
 
-        # Fail closed for async-delegation results (#55578): these carry a
-        # conversation payload, and injecting one into any chat other than the
-        # one that commissioned it is a hard cross-session leak. The
-        # belongs-elsewhere check above already re-queued events owned by
-        # another LIVE session; what reaches here is either ours or an
-        # orphan whose owner is gone. Orphaned delegation payloads are
-        # DROPPED, not adopted — the subagent's summary is already persisted
-        # in the delegation records/output store, so nothing is lost, whereas
-        # a wrong-chat injection is unrecoverable. Non-delegation events
-        # (background process completions etc.) keep the historical
-        # adopt-orphans behavior.
-        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
-            sid, session, evt
-        ):
-            logger.warning(
-                "async-delegation completion %s has no live owner "
-                "(origin=%r key=%r); dropping from injection instead of "
-                "delivering to session %s (#55578 fail-closed; result "
-                "remains in the delegation records)",
-                evt.get("delegation_id", "?"),
+        # What reaches here is not owned by another LIVE session. Addressed
+        # events still require positive proof before injection: exact UI origin,
+        # direct durable key, or compression lineage. If none proves ownership,
+        # the event is orphaned and must not be adopted by this chat. Truly
+        # ownerless ordinary notifications retain legacy global delivery.
+        requires_owner = _notification_event_requires_owner(evt)
+        if requires_owner and not _session_owns_notification_event(sid, session, evt):
+            log = (
+                logger.warning
+                if evt.get("type") == "async_delegation"
+                else logger.debug
+            )
+            log(
+                "Dropping unowned %s notification (origin=%r key=%r) instead "
+                "of delivering to session %s",
+                evt.get("type", "completion"),
                 str(evt.get("origin_ui_session_id") or ""),
                 str(evt.get("session_key") or ""),
                 sid,
@@ -8830,6 +8833,7 @@ def _notification_poller_loop(
     # Drain any remaining events after stop signal (process all pending
     # before exiting so nothing is lost on shutdown). Events owned by other
     # live sessions are set aside and re-queued so their poller still sees them.
+    # Orphaned events (owner gone) are dropped — same guard as the main loop.
     deferred: list = []
     while not process_registry.completion_queue.empty():
         try:
@@ -8839,14 +8843,21 @@ def _notification_poller_loop(
         if _notification_event_belongs_elsewhere(sid, session, evt):
             deferred.append(evt)
             continue
-        # Same fail-closed rule as the live loop: an orphaned async-delegation
-        # payload is never adopted by a foreign session — defer it (a later
-        # resume of the owner's lineage can still claim it) rather than
-        # injecting another chat's conversation here (#55578).
-        if evt.get("type") == "async_delegation" and not _session_owns_notification_event(
-            sid, session, evt
-        ):
-            deferred.append(evt)
+        # Same positive-proof rule as the live loop. Preserve the existing
+        # shutdown behavior for orphaned delegation payloads by deferring them
+        # for a later resume; ordinary addressed orphans are dropped.
+        requires_owner = _notification_event_requires_owner(evt)
+        if requires_owner and not _session_owns_notification_event(sid, session, evt):
+            if evt.get("type") == "async_delegation":
+                deferred.append(evt)
+            else:
+                logger.debug(
+                    "Dropping unowned %s notification during shutdown drain "
+                    "(origin=%r key=%r)",
+                    evt.get("type", "completion"),
+                    str(evt.get("origin_ui_session_id") or ""),
+                    str(evt.get("session_key") or ""),
+                )
             continue
         _evt_sid = evt.get("session_id", "")
         if evt.get("type") == "completion" and process_registry.is_completion_consumed(_evt_sid):
@@ -9411,21 +9422,29 @@ def _stream(delta):
         # Drain completion notifications that arrived during this turn.
         # The background poller handles between-turn delivery; this is
         # the safety net for events that arrived mid-turn.
+        #
+        # Ownership filter (#42674, #35652): a turn finishing in session B
+        # must not consume an event that belongs to session A. The registry
+        # requeues every addressed event this session cannot positively claim;
+        # the poller then delivers it to a live owner or drops an orphan.
         try:
             from tools.process_registry import process_registry
 
             # Positive-proof ownership (compression-chain aware) — the same
             # fail-closed gate the poller uses, so the post-turn drain can't
-            # adopt another session's (or an orphan's) delegation payload,
-            # while a post-compression session still claims its own
-            # pre-compression dispatches (#55578).
-            for _evt, synth in process_registry.drain_notifications(
+            # adopt another session's addressed notification while a
+            # post-compression session still claims its own pre-compression
+            # dispatches (#55578).
+            drained = process_registry.drain_notifications(
                 session_key=session.get("session_key", ""),
                 owns_event=lambda e: _session_owns_notification_event(sid, session, e),
-            ):
+                skip_poll_observed=False,
+            )
+            for index, (_evt, synth) in enumerate(drained):
                 with session["history_lock"]:
                     if session.get("running"):
-                        process_registry.completion_queue.put(_evt)
+                        for pending_evt, _pending_synth in drained[index:]:
+                            process_registry.completion_queue.put(pending_evt)
                         break
                     session["running"] = True
                 from tools.async_delegation import (

6 PRs closed without merging

From the close comments: superseded by another PR 3 · done elsewhere 2 · partial fix 1

#35667 fix(tui): route completion notifications to owner live session · superseded by another PR → #60863
#35672 fix(tui): route process notifications to owner session · done elsewhere → 85b65e2, #37099, #35667
#42731 fix(tui): keep background notifications in owning session (#42674) · superseded by another PR → #65419
#54785 fix(tui): route bg process notifications to owning session, drop orphaned events · partial fix → 35803dd, 569b912, #54785, #63317, #42731, #35667, #57586, #60863, #64484
#60441 fix(gateway): prevent background process notifications from being injected into wrong sessions · done elsewhere → dee592a, #60426, #42674

Show 1 more

#63317 fix(tui): enforce notification session ownership across all drain paths · superseded by another PR → #54785

Closed issues · 4

Closed#35652TUI: background process completion notifications can route to the wrong live session
Closed#42674[Bug]: Background process notify_on_complete leaks into wrong TUI session (cross-session bleed)
Closed#52447Background process notifications route to active session instead of originating session
Closed#60426Background process notifications routed to wrong session

How this cluster was grouped

Link evidence (partly structural links): some members are connected by structural evidence (shared changed files, or a closing/superseded reference); the rest are joined by an automated relevance judgement over their text.
Members outside that structural core may belong to a different underlying problem.
Pairwise link density 0.16: the share of item pairs that carry a link, counting a structural link in full and a relevance-only link at 0.36. 1.00 means every pair is linked.

Pick up a Lab task

Check whether a fix covers the open issues here, with your own agent. Bring back evidence for maintainers.

Open the Lab

At a glance

Subsystem: agent
Merged PRs: #37099, #60863, #65419
Closed issues: 4
Attempts: 6 PRs closed without merging
Fix link checked: 16 Jul 2026

For your agent

The same facts, typed.

cluster/35652.json
cluster/35652.md

Background process notifications routed to wrong TUI session

The short answer

Open issues0

Open PRs1

Fixes on record

6 PRs closed without merging

Pick up a Lab task

At a glance

For your agent