fix(native-chat): retract status by authoritative visible catalog

This commit is contained in:
Neil
2026-09-07 17:53:41 -07:00
parent 588240043e
commit dc27f660d0
11 changed files with 296 additions and 36 deletions
@@ -54,7 +54,8 @@ import type {
StructuredAgentSessionHostSession,
StructuredAgentSessionReveal
} from './structured-agent-session-host-types'
import { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed'
import { createHostStatusFeed } from './structured-agent-session-status-catalog'
import type { StructuredAgentSessionStatusFeed } from './structured-agent-session-status-feed'
import { StructuredAgentSessionEventRecovery } from './structured-agent-session-event-recovery'
import { StructuredAgentSessionBackgroundTaskChannel } from './structured-agent-session-background-task-channel'
export type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types'
@@ -65,12 +66,7 @@ export class StructuredAgentSessionHost {
this
)
private readonly sessions = new Map<string, StructuredAgentSessionHostSession>()
private readonly statusFeed = new StructuredAgentSessionStatusFeed({
sessions: this.sessions,
getRecord: (sessionId) => this.deps.store.getRecord(sessionId),
now: () => this.now(),
onStatusChanged: (summary, options) => this.deps.onSessionStatusChanged?.(summary, options)
})
private readonly statusFeed: StructuredAgentSessionStatusFeed
private readonly subscribers = new AgentSessionSubscribers({
readCommands: (sessionId) => this.deps.adapter.readCommands?.(sessionId),
onJournalPublished: (sessionId, journal) => this.statusFeed.publish(sessionId, journal)
@@ -87,6 +83,7 @@ export class StructuredAgentSessionHost {
private readonly backgroundTasks: StructuredAgentSessionBackgroundTaskChannel
constructor(readonly deps: StructuredAgentSessionHostDeps) {
this.statusFeed = createHostStatusFeed(deps, this.sessions, () => this.now())
this.backgroundTasks = new StructuredAgentSessionBackgroundTaskChannel(
deps,
this.sessions,
@@ -216,8 +213,10 @@ export class StructuredAgentSessionHost {
getPersistedVisibleSessionTabIndex = (): { present: boolean; sessionIds: string[] } =>
this.deps.store.getVisibleSessionTabIndex()
setSessionTabVisibility = (sessionId: string, visible: boolean): Promise<void> =>
this.deps.store.setSessionTabVisibility(sessionId, visible)
setSessionTabVisibility = async (sessionId: string, visible: boolean): Promise<void> => {
await this.deps.store.setSessionTabVisibility(sessionId, visible)
this.statusFeed.visibilityChanged(sessionId, visible)
}
reconcileRestartLeases = async (): Promise<void> => {
const refusal = await this.reconcileLeases('startup')
@@ -0,0 +1,41 @@
import { describe, expect, it } from 'vitest'
import type { AgentSessionRecordStore } from '../../runtime/agent-session-record-store'
import type { AgentSessionStatusEvent } from '../../../shared/agent-session-wire'
import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types'
import { createHostStatusFeed } from './structured-agent-session-status-catalog'
describe('status catalog completeness', () => {
it.each([
{ present: true, readOnly: false, recoveredFromBackup: false, complete: true },
{ present: false, readOnly: false, recoveredFromBackup: false, complete: false },
{ present: true, readOnly: true, recoveredFromBackup: false, complete: false },
{ present: true, readOnly: false, recoveredFromBackup: true, complete: false }
])('keeps startup authority conservative: %o', (input) => {
let present = input.present
const store = {
readOnly: input.readOnly,
recoveredFromBackup: input.recoveredFromBackup,
getVisibleSessionTabIndex: () => ({ present, sessionIds: ['still-restoring'] }),
listVisibleSessionIds: () => ['still-restoring']
} as unknown as AgentSessionRecordStore
const feed = createHostStatusFeed(
{ store } as StructuredAgentSessionHostDeps,
new Map(),
() => 1
)
present = true
const events: AgentSessionStatusEvent[] = []
feed.subscribe({ id: 'reader', emit: (event) => events.push(event) })
expect(events).toEqual([
{
type: 'snapshot',
sessions: [],
catalog: {
epoch: expect.any(String),
complete: input.complete,
sessionIds: ['still-restoring']
}
}
])
})
})
@@ -0,0 +1,25 @@
import type { StructuredAgentSessionHostDeps } from './structured-agent-session-host-types'
import {
StructuredAgentSessionStatusFeed,
type StructuredAgentSessionStatusFeedDeps
} from './structured-agent-session-status-feed'
export function createHostStatusFeed(
deps: StructuredAgentSessionHostDeps,
sessions: StructuredAgentSessionStatusFeedDeps['sessions'],
now: () => number
): StructuredAgentSessionStatusFeed {
// Legacy hydration writes visibility incrementally; only an already loaded index is complete.
const complete =
deps.store.getVisibleSessionTabIndex().present &&
!deps.store.readOnly &&
!deps.store.recoveredFromBackup
return new StructuredAgentSessionStatusFeed({
sessions,
getRecord: (sessionId) => deps.store.getRecord(sessionId),
now,
catalog: () => ({ complete, sessionIds: deps.store.listVisibleSessionIds() }),
isVisible: (sessionId) => deps.store.isSessionTabVisible(sessionId),
onStatusChanged: (summary, options) => deps.onSessionStatusChanged?.(summary, options)
})
}
@@ -522,4 +522,42 @@ describe('StructuredAgentSessionStatusFeed', () => {
session: expect.objectContaining({ status: 'idle', latestPrompt: 'hello' })
})
})
it('bounds retention by explicit visibility without suppressing hidden runtime observations', async () => {
const journal = await openJournal()
const sessions = new Map<string, ReturnType<typeof indexed>>()
const visible = new Set<string>()
const observed: string[] = []
const feed = new StructuredAgentSessionStatusFeed({
sessions,
getRecord: () => null,
now: () => 1,
isVisible: (id) => visible.has(id),
catalog: () => ({ complete: true, sessionIds: [...visible] }),
onStatusChanged: (summary) => observed.push(summary.sessionId)
})
for (let i = 0; i < 3000; i++) {
const id = String(i)
sessions.set(id, indexed({ journal }))
visible.add(id)
feed.visibilityChanged(id, true)
visible.delete(id)
feed.visibilityChanged(id, false)
feed.publish(id)
sessions.delete(id)
}
const events: AgentSessionStatusEvent[] = []
feed.subscribe({ id: 'late', emit: (event) => events.push(event) })
expect(events).toEqual([
{
type: 'snapshot',
sessions: [],
catalog: {
complete: true,
epoch: expect.any(String),
sessionIds: []
}
}
])
expect(observed.length).toBeGreaterThan(0)
})
})
@@ -8,8 +8,9 @@
// The last projection is kept after the session's provider child is evicted: an idle session is
// still idle without a process, and a renderer that reloads must not lose every settled row until
// each chat is reopened. Restart is the one boundary that forgets, and restoring readable sessions
// republishes them.
// republishes them. Explicit tab removal retracts only the catalog projection.
import { randomUUID } from 'node:crypto'
import { agentProviderSessionsEqual } from '../../../shared/agent-session-resume'
import type { AgentSessionRecord } from '../../../shared/agent-session-record'
import { normalizeOptionalField } from '../../../shared/agent-status-field-normalization'
@@ -36,6 +37,8 @@ export type StructuredAgentSessionStatusFeedDeps = {
sessions: ReadonlyMap<string, StatusFeedSession>
getRecord: (sessionId: string) => AgentSessionRecord | null
now: () => number
catalog?: () => { complete: boolean; sessionIds: string[] }
isVisible?: (sessionId: string) => boolean
/** Every projection change, whether or not anyone is subscribed. `replay` marks a re-projection
* of state the host already knew (restore, an arriving subscriber) rather than a journal edge. */
onStatusChanged?: (summary: AgentSessionStatusSummary, options: { replay: boolean }) => void
@@ -62,6 +65,9 @@ export class StructuredAgentSessionStatusFeed {
private readonly subscribers = new Map<string, StructuredAgentSessionStatusSubscriber>()
private readonly published = new Map<string, AgentSessionStatusSummary>()
private readonly epoch = randomUUID()
private readonly lastProjection = new WeakMap<AgentSessionJournal, AgentSessionStatusSummary>()
constructor(private readonly deps: StructuredAgentSessionStatusFeedDeps) {}
/** Opens with every session this host has projected, live ones re-read, then only changes. */
@@ -72,7 +78,7 @@ export class StructuredAgentSessionStatusFeed {
this.publish(sessionId, undefined, { replay: true })
}
this.subscribers.set(subscriber.id, subscriber)
this.emit(subscriber, { type: 'snapshot', sessions: [...this.published.values()] })
this.emit(subscriber, this.snapshot())
return () => this.unsubscribe(subscriber.id)
}
@@ -95,13 +101,20 @@ export class StructuredAgentSessionStatusFeed {
if (!session) {
return
}
const summary = this.summaryFor(sessionId, session, journal ?? session.journal)
const previous = this.published.get(sessionId)
const source = journal ?? session.journal
const summary = this.summaryFor(sessionId, session, source)
const previous = this.lastProjection.get(source) ?? this.published.get(sessionId)
const retained = this.published.get(sessionId)
if (this.deps.isVisible?.(sessionId) !== false) {
if (!retained || !summariesEqual(retained, summary)) {
this.published.set(sessionId, summary)
this.broadcast({ type: 'status', session: summary })
}
}
this.lastProjection.set(source, summary)
if (previous && summariesEqual(previous, summary)) {
return
}
this.published.set(sessionId, summary)
this.broadcast({ type: 'status', session: summary })
try {
this.deps.onStatusChanged?.(summary, { replay: options?.replay === true })
} catch (error) {
@@ -110,6 +123,32 @@ export class StructuredAgentSessionStatusFeed {
}
}
visibilityChanged(sessionId: string, visible: boolean): void {
if (visible) {
this.publish(sessionId, undefined, { replay: true })
} else {
this.published.delete(sessionId)
this.broadcast({ type: 'snapshot', sessions: [], removedSessionIds: [sessionId] })
}
}
private snapshot(): AgentSessionStatusEvent {
const catalog = this.deps.catalog?.()
if (catalog?.complete) {
const visible = new Set(catalog.sessionIds)
for (const id of this.published.keys()) {
if (!visible.has(id)) {
this.published.delete(id)
}
}
}
return {
type: 'snapshot',
sessions: [...this.published.values()],
...(catalog ? { catalog: { ...catalog, epoch: this.epoch } } : {})
}
}
private summaryFor(
sessionId: string,
session: StatusFeedSession,
@@ -1,4 +1,4 @@
import { setVisibleSessionId } from './agent-session-visible-tab-index'
import * as visibleTabs from './agent-session-visible-tab-index'
import { commitConversationCommandRecord } from './agent-session-conversation-command-record'
/** Durable single-writer session records and their operation ledger. */
@@ -117,17 +117,15 @@ export class AgentSessionRecordStore {
listRecords = (): AgentSessionRecord[] => [...this.state.records.values()]
listVisibleSessionIds = (): string[] =>
[...this.state.visibleSessionIds].filter((sessionId) => this.state.records.has(sessionId))
isSessionTabVisible = (id: string) => visibleTabs.isSessionTabVisible(this.state, id)
getVisibleSessionTabIndex = (): { present: boolean; sessionIds: string[] } => ({
present: this.state.visibleSessionIdsIndexPresent,
sessionIds: this.listVisibleSessionIds()
})
listVisibleSessionIds = () => visibleTabs.listVisibleSessionIds(this.state)
getVisibleSessionTabIndex = () => visibleTabs.visibleSessionTabIndex(this.state)
/** Persist the user-visible tab reference separately from the rollback-sensitive profile tabs. */
setSessionTabVisibility(sessionId: string, visible: boolean): Promise<void> {
return this.transact(() => setVisibleSessionId(this.state, sessionId, visible))
return this.transact(() => visibleTabs.setVisibleSessionId(this.state, sessionId, visible))
}
listByScope(location: AgentSessionExecutionLocation): AgentSessionRecord[] {
@@ -36,3 +36,21 @@ export function setVisibleSessionId(
}
state.visibleSessionIdsIndexPresent = true
}
export function isSessionTabVisible(state: AgentSessionStoreState, sessionId: string): boolean {
return !state.visibleSessionIdsIndexPresent || state.visibleSessionIds.has(sessionId)
}
export function visibleSessionTabIndex(state: AgentSessionStoreState): {
present: boolean
sessionIds: string[]
} {
return {
present: state.visibleSessionIdsIndexPresent,
sessionIds: listVisibleSessionIds(state)
}
}
export function listVisibleSessionIds(state: AgentSessionStoreState): string[] {
return [...state.visibleSessionIds].filter((id) => state.records.has(id))
}
@@ -21,6 +21,7 @@ vi.mock('./runtime-rpc-client', () => ({
import {
getStructuredAgentSessionStatusFeed,
invalidateStructuredAgentSessionStatusFeed,
resetStructuredAgentSessionStatusFeedsForTests
} from './structured-agent-session-status-feed'
@@ -137,4 +138,57 @@ describe('structured agent session status feed', () => {
await vi.advanceTimersByTimeAsync(10_000)
expect(mocks.subscribeStatus).toHaveBeenCalledOnce()
})
it('retracts only complete catalog absences and explicit removals', async () => {
const feed = getStructuredAgentSessionStatusFeed(LOCAL)
feed.activate()
await vi.advanceTimersByTimeAsync(0)
hostEmit()({ type: 'snapshot', sessions: [summary('a'), summary('b'), summary('c')] })
hostEmit()({
type: 'snapshot',
sessions: [],
catalog: { epoch: 'one', complete: false, sessionIds: [] }
})
expect(feed.getSnapshot().size).toBe(3)
hostEmit()({
type: 'snapshot',
sessions: [],
catalog: { epoch: 'two', complete: true, sessionIds: ['a', 'b'] }
})
expect([...feed.getSnapshot().keys()]).toEqual(['a', 'b'])
hostEmit()({ type: 'snapshot', sessions: [], removedSessionIds: ['a'] })
expect([...feed.getSnapshot().keys()]).toEqual(['b'])
})
it('fences closed streams immediately and keeps retained counts bounded under churn', async () => {
const feed = getStructuredAgentSessionStatusFeed(LOCAL)
feed.activate()
await vi.advanceTimersByTimeAsync(0)
const emit = hostEmit()
for (let i = 0; i < 3000; i++) {
emit({ type: 'status', session: summary(String(i)) })
emit({ type: 'snapshot', sessions: [], removedSessionIds: [String(i)] })
}
expect(feed.getSnapshot().size).toBe(0)
emit({ type: 'end' })
emit({ type: 'status', session: summary('late') })
expect(feed.getSnapshot().size).toBe(0)
})
it('disposes removed runtime owners and clears paired identity state before reopening', async () => {
const feed = getStructuredAgentSessionStatusFeed(REMOTE)
feed.activate()
await vi.advanceTimersByTimeAsync(0)
const emit = hostEmit()
emit({ type: 'status', session: summary('old-account') })
invalidateStructuredAgentSessionStatusFeed('env-1', false)
expect(feed.getSnapshot().size).toBe(0)
emit({ type: 'status', session: summary('stale-account') })
expect(feed.getSnapshot().size).toBe(0)
await vi.advanceTimersByTimeAsync(0)
expect(mocks.subscribeStatus).toHaveBeenCalledTimes(2)
invalidateStructuredAgentSessionStatusFeed('env-1', true)
hostEmit(1)({ type: 'status', session: summary('late') })
expect(feed.getSnapshot().size).toBe(0)
expect(getStructuredAgentSessionStatusFeed(REMOTE)).not.toBe(feed)
})
})
@@ -2,8 +2,7 @@
//
// The feed is a read-only mirror: the host projects each session's status from its journal and
// this owner keeps the latest summary per session while anyone is looking. Losing the stream
// keeps the cached summaries and reconnects; a fresh snapshot merges over them.
// Which sessions are listed is the tab map's decision, so the feed never retracts a summary.
// keeps cached summaries and reconnects; only complete catalogs or explicit removals retract.
import type {
AgentSessionStatusEvent,
@@ -27,7 +26,10 @@ export type StructuredAgentSessionStatusFeedOwner = {
const RECONNECT_MAX_DELAY_MS = 5_000
/** `stop` is the map's own teardown, not part of the owner contract callers hold. */
type OwnedStatusFeed = StructuredAgentSessionStatusFeedOwner & { stop: () => void }
type OwnedStatusFeed = StructuredAgentSessionStatusFeedOwner & {
stop: () => void
invalidate: (removed: boolean) => void
}
const owners = new Map<string, OwnedStatusFeed>()
@@ -40,6 +42,7 @@ function createOwner(target: RuntimeClientTarget): OwnedStatusFeed {
const listeners = new Set<() => void>()
const activations = new Set<symbol>()
let generation = 0
let disposed = false
let handle: { unsubscribe: () => void } | null = null
let reconnectTimer: ReturnType<typeof setTimeout> | null = null
let reconnectAttempt = 0
@@ -56,9 +59,12 @@ function createOwner(target: RuntimeClientTarget): OwnedStatusFeed {
const applyEvent = (event: AgentSessionStatusEvent): void => {
if (event.type === 'snapshot') {
reconnectAttempt = 0
// Merged, not replaced: a restarted host restores its readable sessions asynchronously, so
// the first snapshot can be empty and dropping those rows flickers every one to no-status.
const next = new Map(snapshot)
// An empty projection during restore is not an empty catalog.
const visible = event.catalog?.complete ? new Set(event.catalog.sessionIds) : null
const next = new Map([...snapshot].filter(([id]) => !visible || visible.has(id)))
for (const id of event.removedSessionIds ?? []) {
next.delete(id)
}
for (const session of event.sessions) {
next.set(session.sessionId, session)
}
@@ -104,22 +110,25 @@ function createOwner(target: RuntimeClientTarget): OwnedStatusFeed {
return
}
if (event.type === 'end') {
generation += 1
dropHandle()
scheduleReconnect(candidate)
scheduleReconnect(generation)
return
}
applyEvent(event)
},
() => {
if (active(candidate)) {
generation += 1
dropHandle()
scheduleReconnect(candidate)
scheduleReconnect(generation)
}
},
() => {
if (active(candidate)) {
generation += 1
dropHandle()
scheduleReconnect(candidate)
scheduleReconnect(generation)
}
}
)
@@ -168,6 +177,9 @@ function createOwner(target: RuntimeClientTarget): OwnedStatusFeed {
return {
activate: () => {
if (disposed) {
return () => {}
}
const token = Symbol('status-feed')
activations.add(token)
if (activations.size === 1) {
@@ -185,7 +197,15 @@ function createOwner(target: RuntimeClientTarget): OwnedStatusFeed {
listeners.add(listener)
return () => listeners.delete(listener)
},
stop
stop,
invalidate: (removed) => {
stop()
disposed = removed
setSnapshot(new Map())
if (!removed && activations.size > 0) {
open()
}
}
}
}
@@ -201,6 +221,21 @@ export function getStructuredAgentSessionStatusFeed(
return owner
}
export function invalidateStructuredAgentSessionStatusFeed(
environmentId: string,
removed: boolean
): void {
const key = structuredAgentSessionStatusFeedKey({ kind: 'environment', environmentId })
const owner = owners.get(key)
if (!owner) {
return
}
if (removed) {
owners.delete(key)
}
owner.invalidate(removed)
}
export function resetStructuredAgentSessionStatusFeedsForTests(): void {
// Dropping the map alone leaves a live subscription and its pending reconnect running
// into the next test, where they reopen a stream nothing is holding.
@@ -1,3 +1,4 @@
import { invalidateStructuredAgentSessionStatusFeed } from '@/runtime/structured-agent-session-status-feed'
import type { StateCreator } from 'zustand'
import type { AppState } from '../types'
import type { RuntimeStatusSlice } from './runtime-status-types'
@@ -76,6 +77,12 @@ export const createRuntimeStatusSlice: StateCreator<AppState, [], [], RuntimeSta
const removedIds = get()
.runtimeEnvironments.map((environment) => environment.id)
.filter((id) => !nextIds.has(id))
for (const id of removedIds) {
invalidateStructuredAgentSessionStatusFeed(id, true)
}
for (const id of replacedEnvironmentIds) {
invalidateStructuredAgentSessionStatusFeed(id, false)
}
runtimeStatusRecheck.cancelRuntimeStatusRechecks([...removedIds, ...replacedEnvironmentIds])
set((s) => {
const keep = new Set(environments.map((environment) => environment.id))
+9 -3
View File
@@ -208,10 +208,16 @@ export type AgentSessionStatusSummary = {
updatedAt: number
}
/** A summary outlives its provider child: an evicted idle session is still idle, so the host
* keeps the last projection and never retracts one. Tabs, not this feed, decide what is listed. */
/** A summary outlives its provider child; the durable visible-tab catalog owns retraction. */
export type AgentSessionStatusEvent =
| { type: 'snapshot'; sessions: AgentSessionStatusSummary[] }
| {
type: 'snapshot'
sessions: AgentSessionStatusSummary[]
/** Absent on older hosts; only complete inventories authorize retraction. */
catalog?: { epoch: string; complete: boolean; sessionIds: string[] }
/** Explicit tab removals remain authoritative during a partial restore. */
removedSessionIds?: string[]
}
| { type: 'status'; session: AgentSessionStatusSummary }
| { type: 'end' }