fix(desktop): sidebar rows not on screen show running subagents and background work

$subagentsBySession and $backgroundStatusBySession were fed only by
the visible pane (subagent.list / process.list polls on the mounted
tile, plus live events on that tile's stream), and both RPCs need the
session attached to this transport. A conversation that was not open,
and every row right after a renderer reload or socket reconnect, read
idle while its delegated children or background processes ran.

Backend: new sessionless subagent.roster RPC returns per-conversation
counts of live top-level children (keyed by owner_agent_session_id, the
durable spine completion delivery uses) and running background
processes (registry session_key). It reports only conversations stored
in the requested profile's state.db (the same _profile_db scope
session.list reads), so one process serving several profiles never
hands a caller another profile's session ids. Contract + regenerated
apps/shared outputs included.

Renderer: useLiveWorkRoster (mounted app-wide from useBackgroundSync) reads
the roster on connect/reconnect and every 10 s while the window is
visible, once per backend owner the sidebar shows rows for, routed
through requestForSessionProfile like the rows' own session RPCs. The
result lands in $liveWorkRoster and is unioned into
$delegatingSessionIds / $backgroundRunningSessionIds only for
conversations no event-fed entry covers, so an open pane's live events
stay authoritative. A source that fails three times in a row (an older
backend without the RPC) stops until the next reconnect or window
focus.

Refs #143077, #137847. Reconnect-hydration idea first proposed in
#87915 (@aleksclark).
This commit is contained in:
Teknium committed 2026-11-01 09:05:22 -08:00
1 parent c627ec8466
commit e032b8a264
12 files changed
+452 -10

No files matched your search

@@ -67,6 +67,8 @@ import {
import type { ClientSessionState } from '../../types'
import type { GatewayRequester } from '../types'
import { useLiveWorkRoster } from './use-live-work-roster'
interface ActiveTranscriptSession {
ownerRoute?: SessionOwnerRoute
profile?: string | null
@@ -1068,6 +1070,7 @@ export function useBackgroundSync({
requestGateway,
updateSessionState
}: BackgroundSyncParams): void {
useLiveWorkRoster()
const changeEventsAvailable = useStore($changeEventsAvailable)
const cronChangeTick = useStore($cronChangeTick)
const activeTranscriptRefreshPendingRef = useRef<string | null>(null)
@@ -0,0 +1,125 @@
import { useStore } from '@nanostores/react'
import { useEffect } from 'react'
import { $gateway } from '@/store/gateway'
import { $liveWorkRoster, setLiveWorkRosterSource } from '@/store/live-work-roster'
import { $gatewayState, $sessions } from '@/store/session'
import { ambientRequestFor } from '@/store/session-gone-latch'
import { requestForSessionProfile, type SessionOwnerScope } from '@/store/session-request-router'
import { knownOwnerForSession } from '@/store/session-states'
const POLL_MS = 10_000
const MAX_FAILURES = 3
const AMBIENT = 'ambient'
interface RosterResult {
sessions?: unknown[]
}
/** Every backend owner the sidebar shows rows for. The ambient socket answers
* for the launch profile; a row owned by another profile (or connection) is
* asked on the same route its own session RPCs take, so each backend only
* reports conversations stored in that profile. */
function rosterSources(): Map<string, SessionOwnerScope> {
const sources = new Map<string, SessionOwnerScope>([[AMBIENT, undefined]])
for (const row of $sessions.get()) {
const owner = knownOwnerForSession(row.id)
if (owner) {
sources.set(JSON.stringify(owner), owner)
}
}
return sources
}
/** App-level hydration of `$liveWorkRoster` (sidebar rows not on screen, and
* every row after a reload / socket reconnect): one `subagent.roster` read
* per backend owner on connect and every 10 s while the window is visible.
* A source that keeps failing (an older backend without the RPC) stops until
* the next reconnect or window focus, so it never hot-loops. */
export function useLiveWorkRoster() {
const gatewayState = useStore($gatewayState)
useEffect(() => {
if (gatewayState !== 'open') {
// Counts from a dropped socket are unverified; the reconnect re-reads them.
$liveWorkRoster.set({})
return
}
let cancelled = false
let pending = false
const failures = new Map<string, number>()
const refresh = async () => {
const gateway = $gateway.get()
if (cancelled || pending || !gateway) {
return
}
pending = true
const ambient = ambientRequestFor(gateway)
const sources = rosterSources()
try {
await Promise.all(
[...sources].map(async ([key, owner]) => {
if ((failures.get(key) ?? 0) >= MAX_FAILURES) {
return
}
try {
const result = await requestForSessionProfile<RosterResult>(owner, ambient, 'subagent.roster', {})
if (!cancelled) {
setLiveWorkRosterSource(key, Array.isArray(result?.sessions) ? result.sessions : [])
}
failures.delete(key)
} catch {
failures.set(key, (failures.get(key) ?? 0) + 1)
if (!cancelled) {
setLiveWorkRosterSource(key, null)
}
}
})
)
// Owners whose rows left the sidebar stop contributing.
for (const key of Object.keys($liveWorkRoster.get())) {
if (!sources.has(key)) {
setLiveWorkRosterSource(key, null)
}
}
} finally {
pending = false
}
}
void refresh()
const timer = window.setInterval(() => {
if (document.visibilityState === 'visible') {
void refresh()
}
}, POLL_MS)
const retry = () => {
failures.clear()
void refresh()
}
window.addEventListener('focus', retry)
return () => {
cancelled = true
window.clearInterval(timer)
window.removeEventListener('focus', retry)
}
}, [gatewayState])
}
+17 -4
View File
@@ -6,6 +6,7 @@ import { type TodoItem, type TodoStatus, todoTree } from '@/lib/todos'
import { $gateway } from './gateway'
import { $goalsBySession, type GoalStatus } from './goals'
import { $liveWorkRoster, rosterLiveIds } from './live-work-roster'
import { dispatchNativeNotification } from './native-notifications'
import { notifyError } from './notifications'
import { isSessionGone, isSessionGoneForBackgroundPolling, markSessionGone, noteRuntimeAlive } from './runtime-gone'
@@ -66,22 +67,34 @@ export const $backgroundStatusBySession = atom<Record<string, ComposerStatusItem
// reference when unchanged so rows reading this don't re-render per token.
let backgroundRunningIds: readonly string[] = []
export const $backgroundRunningSessionIds = computed(
[$backgroundStatusBySession, $sessionStates, $sessions],
(bg, states, sessions) => {
[$backgroundStatusBySession, $sessionStates, $sessions, $liveWorkRoster],
(bg, states, sessions, roster) => {
const ids = new Set<string>()
const covered = new Set<string>()
for (const [runtimeId, items] of Object.entries(bg)) {
if (!items.some(i => i.state === 'running')) {
if (items.length === 0) {
continue
}
const running = items.some(i => i.state === 'running')
// Same fresh-chat fallback as the working/attention projections: before a
// conversation is persisted its runtime id is the id surfaces key on.
for (const alias of lineageAliases(states[runtimeId]?.storedSessionId ?? runtimeId, sessions)) {
ids.add(alias)
covered.add(alias)
if (running) {
ids.add(alias)
}
}
}
// Rows no mounted tile feeds (off screen, or every row after a reload).
for (const alias of rosterLiveIds(roster, 'background', covered, sessions)) {
ids.add(alias)
}
return (backgroundRunningIds = stableArray(backgroundRunningIds, [...ids]))
}
)
@@ -0,0 +1,27 @@
import { afterEach, describe, expect, it } from 'vitest'
import { $liveWorkRoster, setLiveWorkRosterSource } from './live-work-roster'
import { $delegatingSessionIds } from './session-dot-state'
import { $subagentsBySession, type SubagentProgress } from './subagents'
const child = (status: SubagentProgress['status']) => ({ id: 'c1', index: 0, status }) as unknown as SubagentProgress
afterEach(() => {
$liveWorkRoster.set({})
$subagentsBySession.set({})
})
describe('backend roster feeds rows no mounted tile speaks for', () => {
it('lights a conversation the roster reports with running children', () => {
setLiveWorkRosterSource('ambient', [{ session_id: 'off-screen', subagents: 2, background: 0 }])
expect($delegatingSessionIds.get()).toContain('off-screen')
})
it('keeps a live event-fed entry authoritative over a stale roster count', () => {
setLiveWorkRosterSource('ambient', [{ session_id: 'open-pane', subagents: 1, background: 0 }])
$subagentsBySession.set({ 'open-pane': [child('completed')] })
expect($delegatingSessionIds.get()).not.toContain('open-pane')
})
})
@@ -0,0 +1,88 @@
import { atom } from 'nanostores'
import { lineageAliases } from './session'
// App-wide answer to "which conversations still have work running outside
// their turn?" from the backend's `subagent.roster` read. The event-fed stores
// ($subagentsBySession, $backgroundStatusBySession) only hear about sessions
// whose tile is mounted, and their `subagent.list` / `process.list` snapshots
// need the session attached to this transport — so a row that is not on
// screen, and every row right after a reload or socket reconnect, read idle
// while children ran. Keyed by STORED session id (the backend reports the
// durable conversation id), one bucket per backend source so a failing source
// never wipes another's rows.
export interface LiveWorkCounts {
background: number
subagents: number
}
export type LiveWorkRoster = Record<string, Record<string, LiveWorkCounts>>
export const $liveWorkRoster = atom<LiveWorkRoster>({})
export function setLiveWorkRosterSource(source: string, entries: readonly unknown[] | null) {
const current = $liveWorkRoster.get()
if (!entries) {
if (!(source in current)) {
return
}
const { [source]: _dropped, ...rest } = current
$liveWorkRoster.set(rest)
return
}
const bucket: Record<string, LiveWorkCounts> = {}
for (const entry of entries) {
if (!entry || typeof entry !== 'object') {
continue
}
const row = entry as Record<string, unknown>
const sid = typeof row.session_id === 'string' ? row.session_id : ''
if (!sid) {
continue
}
bucket[sid] = { background: Number(row.background) || 0, subagents: Number(row.subagents) || 0 }
}
$liveWorkRoster.set({ ...current, [source]: bucket })
}
/** Stored ids (all lineage aliases) the roster says have `kind` running,
* minus conversations a live, event-fed entry already speaks for: those
* entries carry richer per-child state and stay authoritative. */
export function rosterLiveIds(
roster: LiveWorkRoster,
kind: keyof LiveWorkCounts,
liveCovered: ReadonlySet<string>,
sessions: Parameters<typeof lineageAliases>[1]
): string[] {
const ids = new Set<string>()
for (const bucket of Object.values(roster)) {
for (const [storedId, counts] of Object.entries(bucket)) {
if (counts[kind] <= 0) {
continue
}
const aliases = lineageAliases(storedId, sessions)
if (aliases.some(alias => liveCovered.has(alias))) {
continue
}
for (const alias of aliases) {
ids.add(alias)
}
}
}
return [...ids]
}
+19 -5
View File
@@ -28,6 +28,7 @@ import { computed } from 'nanostores'
import { stableArray, stableRecord } from '@/lib/stable-array'
import { $backgroundRunningSessionIds } from './composer-status'
import { $liveWorkRoster, rosterLiveIds } from './live-work-roster'
import { $pausedQueueSessionIds } from './owner-queue-reconcile'
import { $messagingSessions, $sessions, $unreadFinishedSessionIds, lineageAliases } from './session'
import {
@@ -50,20 +51,33 @@ import { $subagentsBySession, activeSubagentCount } from './subagents'
// lineageAliases covers whichever tip of the conversation a surface holds.
let delegatingIds: readonly string[] = []
export const $delegatingSessionIds = computed(
[$subagentsBySession, $sessionStates, $sessions],
(bySession, states, sessions) => {
[$subagentsBySession, $sessionStates, $sessions, $liveWorkRoster],
(bySession, states, sessions, roster) => {
const ids = new Set<string>()
const covered = new Set<string>()
for (const [runtimeId, items] of Object.entries(bySession)) {
if (activeSubagentCount(items) === 0) {
if (items.length === 0) {
continue
}
for (const alias of lineageAliases(states[runtimeId]?.storedSessionId ?? runtimeId, sessions)) {
ids.add(alias)
const aliases = lineageAliases(states[runtimeId]?.storedSessionId ?? runtimeId, sessions)
const active = activeSubagentCount(items) > 0
for (const alias of aliases) {
covered.add(alias)
if (active) {
ids.add(alias)
}
}
}
// Rows no mounted tile feeds (off screen, or every row after a reload).
for (const alias of rosterLiveIds(roster, 'subagents', covered, sessions)) {
ids.add(alias)
}
return (delegatingIds = stableArray(delegatingIds, [...ids]))
}
)
@@ -2389,6 +2389,14 @@ export interface FailedDelegation {
dispatched_at?: number | null
completed_at?: number | null
}
export interface SubagentRosterResult {
sessions?: SubagentRosterEntry[]
}
export interface SubagentRosterEntry {
session_id: string
subagents?: number
background?: number
}
export interface SubagentIdParams {
session_id: string
profile?: string | null
@@ -6492,6 +6500,8 @@ export interface RpcMethods {
'subagent.interrupt': { params: SubagentIdParams; result: SubagentInterruptResult }
/** Live children owned by this session (other sessions' children never leak). */
'subagent.list': { params: SessionParams; result: SubagentListResult }
/** Per-conversation counts of live delegated children and running background processes, limited to conversations stored in the named profile (launch profile when omitted). Needs no attached session. */
'subagent.roster': { params: ProfileParams; result: SubagentRosterResult }
/** Queue steering text into a live delegated child owned by this session. */
'subagent.steer': { params: SubagentSteerParams; result: SubagentSteerResult }
/** Last 16KB of an owned child's live transcript. */
@@ -6783,6 +6793,7 @@ export const RPC_METHODS = [
'spawn_tree.save',
'subagent.interrupt',
'subagent.list',
'subagent.roster',
'subagent.steer',
'subagent.tail',
'subscription.change',
@@ -4074,6 +4074,24 @@
}
}
},
{
"name": "subagent.roster",
"summary": "Per-conversation counts of live delegated children and running background processes, limited to conversations stored in the named profile (launch profile when omitted). Needs no attached session.",
"params": [
{
"name": "params",
"schema": {
"$ref": "#/components/schemas/ProfileParams"
}
}
],
"result": {
"name": "result",
"schema": {
"$ref": "#/components/schemas/SubagentRosterResult"
}
}
},
{
"name": "subagent.steer",
"summary": "Queue steering text into a live delegated child owned by this session.",
@@ -42591,6 +42609,44 @@
"title": "SubagentOutputTailEntry",
"type": "object"
},
"SubagentRosterEntry": {
"additionalProperties": false,
"properties": {
"session_id": {
"title": "Session Id",
"type": "string"
},
"subagents": {
"default": 0,
"title": "Subagents",
"type": "integer"
},
"background": {
"default": 0,
"title": "Background",
"type": "integer"
}
},
"required": [
"session_id"
],
"title": "SubagentRosterEntry",
"type": "object"
},
"SubagentRosterResult": {
"additionalProperties": false,
"properties": {
"sessions": {
"items": {
"$ref": "#/components/schemas/SubagentRosterEntry"
},
"title": "Sessions",
"type": "array"
}
},
"title": "SubagentRosterResult",
"type": "object"
},
"SubagentSnapshot": {
"additionalProperties": false,
"description": "``methods_subagents._SUBAGENT_SNAPSHOT_FIELDS`` projection of one live child record.",
+46
View File
@@ -0,0 +1,46 @@
"""``subagent.roster``: the sessionless, profile-scoped feed the Desktop sidebar hydrates from."""
import json
import threading
import pytest
@pytest.fixture()
def roster(monkeypatch, tmp_path):
from hermes_state import SessionDB
from tools import delegate_tool_registry as reg
from tui_gateway import server
db = SessionDB(db_path=tmp_path / "state.db")
db.create_session("parent-a", source="tui")
monkeypatch.setattr(server, "_get_db", lambda: db)
monkeypatch.setattr(reg, "_active_subagents", {})
monkeypatch.setattr(reg, "_active_subagents_lock", threading.RLock())
def call():
resp = server.handle_request({"jsonrpc": "2.0", "id": 1, "method": "subagent.roster", "params": {}})
return {e["session_id"]: e for e in resp["result"]["sessions"]}
yield reg._active_subagents, call
db.close()
def test_roster_reports_live_children_without_an_attached_session(roster):
"""A conversation nobody has open (no live session record, no transport) still reports its running
top-level children, so its sidebar row can light up after a reload."""
active, call = roster
active["c1"] = {"subagent_id": "c1", "depth": 0, "owner_agent_session_id": "parent-a", "status": "running"}
active["c2"] = {"subagent_id": "c2", "depth": 0, "owner_agent_session_id": "parent-a", "status": "running"}
active["g1"] = {"subagent_id": "g1", "depth": 1, "owner_agent_session_id": "child-x", "status": "running"}
assert call() == {"parent-a": {"session_id": "parent-a", "subagents": 2, "background": 0}}
def test_roster_omits_conversations_stored_in_another_profile(roster):
"""One process serves several profiles: a child whose conversation is not in the requested
profile's state.db is never reported, so its session id does not leak across profiles."""
active, call = roster
active["c1"] = {"subagent_id": "c1", "depth": 0, "owner_agent_session_id": "other-profile-sid"}
assert json.dumps(call()) == "{}"
@@ -674,6 +674,21 @@ method("subagent.list", params=SessionParams, result=SubagentListResult,
doc="Live children owned by this session (other sessions' children never leak).")
class SubagentRosterEntry(Result):
session_id: str
subagents: int = 0
background: int = 0
class SubagentRosterResult(Result):
sessions: list[SubagentRosterEntry] = Field(default_factory=list)
method("subagent.roster", params=ProfileParams, result=SubagentRosterResult,
doc="Per-conversation counts of live delegated children and running background processes, limited to "
"conversations stored in the named profile (launch profile when omitted). Needs no attached session.")
class SubagentIdParams(SessionParams):
subagent_id: str
+44
View File
@@ -63,6 +63,50 @@ def _(rid, params):
})
def _live_work_by_conversation():
"""{durable conversation id: {"subagents": n, "background": m}} over the whole process: top-level live
children keyed by ``owner_agent_session_id`` (the same durable spine completion delivery routes by), running
background processes by their registry ``session_key``."""
from tools.delegate_tool_registry import _active_subagents, _active_subagents_lock
from tools.process_registry import process_registry
counts = {}
with _active_subagents_lock:
owners = [str(r.get("owner_agent_session_id") or "") for r in _active_subagents.values()
if not r.get("depth")]
for sid in filter(None, owners):
counts.setdefault(sid, {"subagents": 0, "background": 0})["subagents"] += 1
for entry in process_registry.list_sessions():
proc = process_registry.get(entry.get("session_id"))
key = str(getattr(proc, "session_key", "") or "") if proc is not None else ""
if key and not getattr(proc, "exited", True):
counts.setdefault(key, {"subagents": 0, "background": 0})["background"] += 1
return counts
@method("subagent.roster")
def _(rid, params):
"""App-wide sidebar feed: which conversations still have children or background processes running.
``subagent.list`` needs the conversation's live, transport-attached session record, so it cannot answer
for a row that is not open or for any row right after a renderer reload; this read needs neither. One
process hosts several profiles, so only conversations stored in the requested profile's state.db are
returned — a caller never learns another profile's session ids."""
counts = _live_work_by_conversation()
if not counts:
return _ok(rid, {"sessions": []})
scope = params if isinstance(params, dict) else {}
if not str(scope.get("profile") or "").strip():
# A legacy-fallback socket ticketed to a non-launch profile answers for that profile only.
from tui_gateway.ws_legacy_fallback import connection_profile_home
if home := connection_profile_home():
scope = {"profile": Path(home).name}
with _profile_db(scope) as db:
if db is None:
return _ok(rid, {"sessions": []})
owned = [sid for sid in counts if db.get_session(sid)]
return _ok(rid, {"sessions": [{"session_id": sid, **counts[sid]} for sid in owned]})
def _failed_delegations(session_id, owner):
"""Recently failed async delegation tasks for this session from the durable store (the live
roster forgets ended children and dies with a renderer reload, #97202). Read under the session's
+1 -1
View File
@@ -198,7 +198,7 @@ _LONG_HANDLERS = frozenset({
"setup.runtime_check", "setup.status", "free_tier.provision", "voice.toggle", "voice.record", "voice.tts", "wake.start",
"wake.status", "session.active_list", "session.branch", "session.compress", "session.list",
"session.resume", "session.save", "session.workspace.move", "shell.exec", "skills.manage", "slash.exec",
"onboarding.start", "onboarding.state",
"onboarding.start", "onboarding.state", "subagent.roster",
"command.dispatch", # /goal draft invokes the auxiliary model; never block the RPC reader
"shared_metrics.set", # consent reconcile waits on the metrics store's write lock
})