Merge remote-tracking branch 'origin/master' into codex/pr-1150-conflict-fix
# Conflicts: # packages/client/ui-conversation/README.i18n.yaml
This commit is contained in:
@@ -59,6 +59,8 @@ interface CatalogInflight {
|
||||
readonly promise: Promise<void>
|
||||
readonly expandableRows: Set<SessionId>
|
||||
readonly activityRows: Map<SessionId, 'running' | 'inactive'>
|
||||
/** Removal-time invalidation replayed over the response this request predates. */
|
||||
parentAvailableOverride: false | undefined
|
||||
}
|
||||
|
||||
type SessionListMutation =
|
||||
@@ -101,6 +103,8 @@ export class SessionManager {
|
||||
private readonly addresses = new Map<SessionId, SubagentAddress>()
|
||||
private readonly catalogs = new Map<SessionId, SubagentCatalogSnapshot>()
|
||||
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 catalogDebounce = new Map<SessionId, ReturnType<typeof setTimeout>>()
|
||||
|
||||
@@ -301,22 +305,26 @@ export class SessionManager {
|
||||
try {
|
||||
const { result } = await this.api.subagents.list({ parentSessionId })
|
||||
if (result.ok) {
|
||||
const parentAvailable = this.catalogInflight.get(parentSessionId)?.parentAvailableOverride
|
||||
?? result.value.parentAvailable
|
||||
this.catalogs.set(parentSessionId, {
|
||||
...result.value,
|
||||
entries: this.withCatalogMutations(result.value.entries, expandableRows, activityRows),
|
||||
parentAvailable,
|
||||
state: 'ready',
|
||||
error: null,
|
||||
})
|
||||
for (const [childId, address] of this.addresses) {
|
||||
if (address.parentSessionId !== parentSessionId) continue
|
||||
this.sessions.get(childId)?.handleSubagentParentAvailable(result.value.parentAvailable)
|
||||
this.sessions.get(childId)?.handleSubagentParentAvailable(parentAvailable)
|
||||
}
|
||||
} else {
|
||||
this.catalogs.set(parentSessionId, {
|
||||
entries: this.withCatalogMutations(
|
||||
previous?.entries ?? [], expandableRows, activityRows,
|
||||
),
|
||||
parentAvailable: previous?.parentAvailable ?? false,
|
||||
parentAvailable: this.catalogInflight.get(parentSessionId)?.parentAvailableOverride
|
||||
?? previous?.parentAvailable ?? false,
|
||||
state: 'error',
|
||||
error: result.error,
|
||||
})
|
||||
@@ -327,16 +335,26 @@ export class SessionManager {
|
||||
entries: this.withCatalogMutations(
|
||||
previous?.entries ?? [], expandableRows, activityRows,
|
||||
),
|
||||
parentAvailable: previous?.parentAvailable ?? false,
|
||||
parentAvailable: this.catalogInflight.get(parentSessionId)?.parentAvailableOverride
|
||||
?? previous?.parentAvailable ?? false,
|
||||
state: 'error',
|
||||
error: folded.ok ? null : folded.error,
|
||||
})
|
||||
} finally {
|
||||
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.catalogInflight.set(parentSessionId, { promise: operation, expandableRows, activityRows })
|
||||
this.catalogInflight.set(parentSessionId, {
|
||||
promise: operation,
|
||||
expandableRows,
|
||||
activityRows,
|
||||
parentAvailableOverride: undefined,
|
||||
})
|
||||
return operation
|
||||
}
|
||||
|
||||
@@ -673,6 +691,29 @@ export class SessionManager {
|
||||
this.pendingBuffers.delete(frame.sessionId) // a removed session's buffered frames must not replay on a future instantiation
|
||||
this.waitingApprovals.delete(frame.sessionId) // a removed session cannot wait on anyone
|
||||
if (!durableSubagent) this.projectionStores.delete(frame.sessionId)
|
||||
// A pull already in flight was requested before this removal and can
|
||||
// carry the pre-removal parentAvailable:true, which would resurrect
|
||||
// the writable editor this invalidation just closed. Replay false over
|
||||
// that response and queue one trailing refresh so the post-removal
|
||||
// host truth converges.
|
||||
const inflightCatalog = this.catalogInflight.get(frame.sessionId)
|
||||
if (inflightCatalog !== undefined) {
|
||||
inflightCatalog.parentAvailableOverride = false
|
||||
this.catalogStale.add(frame.sessionId)
|
||||
}
|
||||
// The removed session can no longer be the delivery owner of its
|
||||
// catalog: invalidate availability immediately. Removal schedules no
|
||||
// catalog refresh, and without this an addressed child keeps a
|
||||
// writable editor against a dead continuation owner until an
|
||||
// unrelated refresh (or forever, for a closed menu).
|
||||
const ownedCatalog = this.catalogs.get(frame.sessionId)
|
||||
if (ownedCatalog !== undefined && ownedCatalog.parentAvailable) {
|
||||
this.catalogs.set(frame.sessionId, { ...ownedCatalog, parentAvailable: false })
|
||||
}
|
||||
for (const [childId, address] of this.addresses) {
|
||||
if (address.parentSessionId !== frame.sessionId) continue
|
||||
this.sessions.get(childId)?.handleSubagentParentAvailable(false)
|
||||
}
|
||||
return
|
||||
}
|
||||
case 'host/session-status': {
|
||||
@@ -724,11 +765,18 @@ export class SessionManager {
|
||||
for (const session of this.sessions.values()) void session.resync()
|
||||
}
|
||||
|
||||
/** Debounce membership refetches while one parent catalog is open. */
|
||||
/** Debounce membership refetches while one parent catalog is selected or open. */
|
||||
private scheduleCatalogRefresh(parentSessionId: SessionId): void {
|
||||
if (this.catalogDebounce.has(parentSessionId)) return
|
||||
const timer = setTimeout(() => {
|
||||
this.catalogDebounce.delete(parentSessionId)
|
||||
// The in-flight response predates the membership frame that scheduled
|
||||
// this callback. Queue one post-settlement pull instead of treating an
|
||||
// ordinary overlapping read as evidence that catalog membership changed.
|
||||
if (this.catalogInflight.has(parentSessionId)) {
|
||||
this.catalogStale.add(parentSessionId)
|
||||
return
|
||||
}
|
||||
void this.refreshSubagents(parentSessionId)
|
||||
}, 50)
|
||||
this.catalogDebounce.set(parentSessionId, timer)
|
||||
|
||||
@@ -529,6 +529,149 @@ describe('subagent catalogs', () => {
|
||||
{ kind: 'child', id: S2, activity: 'inactive' },
|
||||
])
|
||||
})
|
||||
|
||||
it('coalesces overlapping catalog reads without scheduling a trailing pull', async () => {
|
||||
const api = new FakeApiClient()
|
||||
const root = 'fk-root' as SessionId
|
||||
const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
|
||||
api.onSubagentList = () => first.promise
|
||||
const manager = new SessionManager(api)
|
||||
|
||||
const refresh = manager.refreshSubagents(root)
|
||||
expect(manager.refreshSubagents(root)).toBe(refresh)
|
||||
api.onSubagentList = () => Promise.resolve(ok({ entries: [], parentAvailable: true }))
|
||||
first.resolve(ok({ entries: [], parentAvailable: true }))
|
||||
await refresh
|
||||
|
||||
expect(api.callsOf('subagent.list')).toHaveLength(1)
|
||||
})
|
||||
|
||||
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, root)
|
||||
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()
|
||||
}
|
||||
})
|
||||
|
||||
it('keeps removal invalidation across a stale success and failed trailing pull', async () => {
|
||||
const api = new FakeApiClient()
|
||||
const root = 'fk-root' as SessionId
|
||||
const child = () => ({
|
||||
kind: 'child' as const, id: S2, mode: 'continuable' as const, label: 'worker',
|
||||
activity: 'inactive' as const, hasChildren: false,
|
||||
})
|
||||
const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
|
||||
api.onSubagentList = () => first.promise
|
||||
const manager = new SessionManager(api)
|
||||
const refresh = manager.refreshSubagents(root)
|
||||
first.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
|
||||
await refresh
|
||||
manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
|
||||
|
||||
// The removal lands while a second pull is in flight: the invalidation
|
||||
// must survive the pre-removal ok response, so one trailing pull runs.
|
||||
const mid = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
|
||||
api.onSubagentList = () => mid.promise
|
||||
const midRefresh = manager.refreshSubagents(root)
|
||||
manager.handleHostEnvelope({
|
||||
rpcId: 'parent-removed-mid-pull' as never,
|
||||
payload: { type: 'host/session-removed', sessionId: root },
|
||||
})
|
||||
const trailing = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
|
||||
api.onSubagentList = () => trailing.promise
|
||||
mid.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
|
||||
await midRefresh
|
||||
expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
|
||||
expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
|
||||
|
||||
trailing.resolve(err({ code: 'internal', message: 'trailing pull failed', details: {} }))
|
||||
await vi.waitFor(() => {
|
||||
expect(manager.getListSnapshot().subagentsByParent[root]).toMatchObject({
|
||||
state: 'error',
|
||||
parentAvailable: false,
|
||||
})
|
||||
})
|
||||
|
||||
const rootCalls = api.callsOf('subagent.list')
|
||||
.filter(call => (call as { parentSessionId: SessionId }).parentSessionId === root)
|
||||
expect(rootCalls).toHaveLength(3)
|
||||
expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
|
||||
expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
|
||||
})
|
||||
|
||||
it('invalidates catalog availability when the owning parent is removed', async () => {
|
||||
const api = new FakeApiClient()
|
||||
const root = 'fk-root' as SessionId
|
||||
api.onSubagentList = () => Promise.resolve(ok({
|
||||
entries: [{
|
||||
kind: 'child', id: S2, mode: 'continuable', label: 'worker',
|
||||
activity: 'inactive', hasChildren: false,
|
||||
}] as never[],
|
||||
parentAvailable: true,
|
||||
}))
|
||||
const manager = new SessionManager(api)
|
||||
await manager.refreshSubagents(root)
|
||||
manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
|
||||
expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: true })
|
||||
|
||||
manager.handleHostEnvelope({
|
||||
rpcId: 'parent-removed' as never,
|
||||
payload: { type: 'host/session-removed', sessionId: root },
|
||||
})
|
||||
|
||||
expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
|
||||
expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
|
||||
})
|
||||
})
|
||||
|
||||
describe('remaining branches', () => {
|
||||
|
||||
Reference in New Issue
Block a user