fix(web): run a trailing catalog refresh for coalesced membership changes
`refreshSubagents` single-flights per catalog owner: a request arriving while a pull is in flight returns the in-flight promise and is silently coalesced into it. The in-flight response was requested before the triggering change, so it can never contain that change — a debounced membership refresh (50ms after `host/session-added`) firing during a slow pull therefore lost the new child, and the catalog stayed stale until an unrelated trigger (reselection, menu reopen, reconnect). Mark the owner stale on coalescing and re-arm one trailing pull in the settlement `finally`, so every membership change observed during a pull is carried by a follow-up refresh exactly once. Bounded: the trailing pull only runs when a refresh request was actually coalesced, and a new coalescing during the trailing pull re-marks the same set. Adds a fake-timer regression test: a `host/session-added` debounce firing mid-pull yields exactly two `subagent.list` calls and the catalog eventually contains the new child.
This commit is contained in:
@@ -101,6 +101,8 @@ export class SessionManager {
|
|||||||
private readonly addresses = new Map<SessionId, SubagentAddress>()
|
private readonly addresses = new Map<SessionId, SubagentAddress>()
|
||||||
private readonly catalogs = new Map<SessionId, SubagentCatalogSnapshot>()
|
private readonly catalogs = new Map<SessionId, SubagentCatalogSnapshot>()
|
||||||
private readonly catalogInflight = new Map<SessionId, CatalogInflight>()
|
private readonly catalogInflight = new Map<SessionId, CatalogInflight>()
|
||||||
|
/** Catalog owners whose membership changed while a pull was in flight: one trailing refresh after it settles. */
|
||||||
|
private readonly catalogStale = new Set<SessionId>()
|
||||||
private readonly openCatalogs = new Set<SessionId>()
|
private readonly openCatalogs = new Set<SessionId>()
|
||||||
private readonly catalogDebounce = new Map<SessionId, ReturnType<typeof setTimeout>>()
|
private readonly catalogDebounce = new Map<SessionId, ReturnType<typeof setTimeout>>()
|
||||||
|
|
||||||
@@ -286,7 +288,16 @@ export class SessionManager {
|
|||||||
*/
|
*/
|
||||||
refreshSubagents(parentSessionId: SessionId): Promise<void> {
|
refreshSubagents(parentSessionId: SessionId): Promise<void> {
|
||||||
const existing = this.catalogInflight.get(parentSessionId)
|
const existing = this.catalogInflight.get(parentSessionId)
|
||||||
if (existing !== undefined) return existing.promise
|
if (existing !== undefined) {
|
||||||
|
// A refresh requested while a pull is in flight must not be silently
|
||||||
|
// coalesced into it: the in-flight response was requested before the
|
||||||
|
// triggering change (a membership frame or an opened menu), so it can
|
||||||
|
// never contain that change. Queue one trailing refresh that runs after
|
||||||
|
// the pull settles; without it the change stays invisible until an
|
||||||
|
// unrelated later trigger (reselection, menu reopen, reconnect).
|
||||||
|
this.catalogStale.add(parentSessionId)
|
||||||
|
return existing.promise
|
||||||
|
}
|
||||||
const previous = this.catalogs.get(parentSessionId)
|
const previous = this.catalogs.get(parentSessionId)
|
||||||
const expandableRows = new Set<SessionId>()
|
const expandableRows = new Set<SessionId>()
|
||||||
const activityRows = new Map<SessionId, 'running' | 'inactive'>()
|
const activityRows = new Map<SessionId, 'running' | 'inactive'>()
|
||||||
@@ -333,6 +344,10 @@ export class SessionManager {
|
|||||||
})
|
})
|
||||||
} finally {
|
} finally {
|
||||||
this.catalogInflight.delete(parentSessionId)
|
this.catalogInflight.delete(parentSessionId)
|
||||||
|
// Re-arm the trailing pull before the dirty notify: the response the
|
||||||
|
// caller observed predates the stale-marking change, so the follow-up
|
||||||
|
// refresh is the only carrier of that change.
|
||||||
|
if (this.catalogStale.delete(parentSessionId)) void this.refreshSubagents(parentSessionId)
|
||||||
this.notifier.markDirty()
|
this.notifier.markDirty()
|
||||||
}
|
}
|
||||||
})()
|
})()
|
||||||
|
|||||||
@@ -529,6 +529,64 @@ describe('subagent catalogs', () => {
|
|||||||
{ kind: 'child', id: S2, activity: 'inactive' },
|
{ kind: 'child', id: S2, activity: 'inactive' },
|
||||||
])
|
])
|
||||||
})
|
})
|
||||||
|
|
||||||
|
it('runs one trailing catalog refresh for a membership change coalesced into an in-flight pull', async () => {
|
||||||
|
vi.useFakeTimers()
|
||||||
|
try {
|
||||||
|
const api = new FakeApiClient()
|
||||||
|
const root = 'fk-root' as SessionId
|
||||||
|
const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
|
||||||
|
const second = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
|
||||||
|
api.onSubagentList = () => first.promise
|
||||||
|
const manager = new SessionManager(api)
|
||||||
|
manager.setSubagentCatalogOpen(root, true)
|
||||||
|
const refresh = manager.refreshSubagents(root)
|
||||||
|
|
||||||
|
// A membership frame arrives while the pull is in flight; the debounced
|
||||||
|
// refresh it schedules fires 50ms later and is coalesced into the pull —
|
||||||
|
// which was requested before the new child existed. The stale mark must
|
||||||
|
// queue one trailing pull carrying the change.
|
||||||
|
manager.handleHostEnvelope({
|
||||||
|
rpcId: 'child-added' as never,
|
||||||
|
payload: {
|
||||||
|
type: 'host/session-added', sessionId: S2, parentSessionId: root, blank: false,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
await vi.advanceTimersByTimeAsync(50)
|
||||||
|
api.onSubagentList = () => second.promise
|
||||||
|
first.resolve(ok({
|
||||||
|
entries: [{
|
||||||
|
kind: 'child', id: S1, mode: 'continuable', label: 'older',
|
||||||
|
activity: 'inactive', hasChildren: false,
|
||||||
|
}] as never[],
|
||||||
|
parentAvailable: true,
|
||||||
|
}))
|
||||||
|
await refresh
|
||||||
|
// The trailing pull is already in flight (kicked synchronously in finally).
|
||||||
|
second.resolve(ok({
|
||||||
|
entries: [
|
||||||
|
{
|
||||||
|
kind: 'child', id: S1, mode: 'continuable', label: 'older',
|
||||||
|
activity: 'inactive', hasChildren: false,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
kind: 'child', id: S2, mode: 'continuable', label: 'new child',
|
||||||
|
activity: 'inactive', hasChildren: false,
|
||||||
|
},
|
||||||
|
] as never[],
|
||||||
|
parentAvailable: true,
|
||||||
|
}))
|
||||||
|
await second.promise
|
||||||
|
|
||||||
|
expect(api.callsOf('subagent.list')).toHaveLength(2)
|
||||||
|
expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
|
||||||
|
{ kind: 'child', id: S1, label: 'older' },
|
||||||
|
{ kind: 'child', id: S2, label: 'new child' },
|
||||||
|
])
|
||||||
|
} finally {
|
||||||
|
vi.useRealTimers()
|
||||||
|
}
|
||||||
|
})
|
||||||
})
|
})
|
||||||
|
|
||||||
describe('remaining branches', () => {
|
describe('remaining branches', () => {
|
||||||
|
|||||||
Reference in New Issue
Block a user