imccyu
9ff8f26450
ProjectionValueStore rows are either cached (viewed from the persisted checkpoint by the session list, no comparable seq) or sequenced (computed by the connected Host, seq comparable within the connection). Cached values fill empty keys only; every baseline discards cached rows before seeding, and every frame replaces them regardless of seq, so a connected Session's values always supersede list hints. The list response and api-session/added blocks write cached rows.
1128 lines
53 KiB
TypeScript
1128 lines
53 KiB
TypeScript
/**
|
|
* SessionManager orchestration: lazy resident instances, list lifecycle, host
|
|
* frame routing, and control baselines for uninstantiated sessions.
|
|
*/
|
|
|
|
import { describe, expect, vi } from 'vitest'
|
|
import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
|
|
import { SessionSeq } from '@deepseek-ai/dsh-session/types'
|
|
import { RemoteError } from '@deepseek-ai/dsh-typert-protocol'
|
|
import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller/types'
|
|
import { ok, type RemoteMock } from '@deepseek-ai/dsh-remote-mock'
|
|
import {
|
|
createClientTest, type ClientTestFixtures, webApp,
|
|
} from '@deepseek-ai/dsh-client-test-runtime/src/assembly/index.ts'
|
|
import type {} from '@deepseek-ai/dsh-session-title/client'
|
|
import { SessionManager } from '../src/client/sessions/manager.ts'
|
|
import type { SessionRemotes } from '../src/client/sessions/remotes.ts'
|
|
import { entries, plainTurn } from './event-script.client.ts'
|
|
import { FOLLOW, err, followScript, sessionWorld } from './remote/session.client.ts'
|
|
|
|
const S1 = 'fk-m1' as SessionId
|
|
const S2 = 'fk-m2' as SessionId
|
|
/** Gateway Client cone used by the subagent-catalog and connected-generation cases. */
|
|
const API_ROSTER = webApp.closure(['@deepseek-ai/dsh-api-gateway'])
|
|
const it = createClientTest({ roster: API_ROSTER })
|
|
|
|
type SummaryOver = Partial<{
|
|
updatedAt: number
|
|
running: boolean
|
|
agentAvailable: boolean
|
|
blank: boolean
|
|
cwd: string
|
|
parentSessionId: SessionId
|
|
origin: 'subagent'
|
|
}>
|
|
|
|
function summary(sessionId: SessionId, over: SummaryOver = {}) {
|
|
return { agentAvailable: true, sessionId, updatedAt: 100, running: false, blank: false, ...over }
|
|
}
|
|
|
|
function makeManager(
|
|
mock: RemoteMock,
|
|
remote: ClientTestFixtures['remote'],
|
|
): SessionManager {
|
|
mock.load(sessionWorld)
|
|
// Cases using this helper never open a Session, so they do not need the broader Client Remote's $stream member.
|
|
return new SessionManager(remote as unknown as SessionRemotes)
|
|
}
|
|
|
|
describe('SessionManager instances', () => {
|
|
it('lazily builds one resident instance per id and syncs the running bit from the list', async ({ mock, remote }) => {
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1, { running: true })] as never[] }))
|
|
const manager = makeManager(mock, remote)
|
|
await manager.refreshList()
|
|
const session = manager.get(S1)
|
|
expect(manager.get(S1)).toBe(session) // resident: same instance forever
|
|
expect(session.getSnapshot().running).toBe(true) // list preceded instantiation
|
|
})
|
|
|
|
it('initializes blankness from metadata published before the Session and list exist', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
const published = Promise.withResolvers<undefined>()
|
|
const stopObserving = manager.subscribe(() => { published.resolve(undefined) })
|
|
try {
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1,
|
|
key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 1200 }, seq: 8 })
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S2,
|
|
key: 'sessionListMetadata', value: { blank: true, lastPromptAt: null }, seq: 2 })
|
|
// The initial notification must finish while neither Session is resident.
|
|
await published.promise
|
|
|
|
expect(manager.getListSnapshot().items).toEqual([])
|
|
expect(manager.get(S1).getSnapshot()).toMatchObject({ blank: false, running: false })
|
|
expect(manager.get(S2).getSnapshot()).toMatchObject({ blank: true, running: false })
|
|
expect(manager.get('fk-unobserved' as SessionId).getSnapshot().blank).toBe(true)
|
|
} finally {
|
|
stopObserving()
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
})
|
|
|
|
describe('SessionManager query lifetime', () => {
|
|
it.for(['list', 'projection'] as const)('forwards an unexpected %s failure and still completes teardown', async (target, { mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
const failure = new Error('query implementation failed')
|
|
if (target === 'list') remote.session.list.mockRejectedValueOnce(failure)
|
|
else remote.session.projections.mockRejectedValueOnce(failure)
|
|
try {
|
|
await expect(target === 'list' ? manager.refreshList() : manager.refreshProjections(S1)).rejects.toBe(failure)
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it('keeps a rejected Remote list failure in the observable state', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
const failure = new RemoteError('gateway/internal', 'list unavailable', {})
|
|
remote.session.list.mockRejectedValueOnce(failure)
|
|
try {
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: failure })
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it.for([false, true])('preserves a rejected Remote projection failure with prior values %s', async (warm, { mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
const failure = new RemoteError('gateway/internal', 'projection unavailable', {})
|
|
const entries = [{ id: S2, createdAt: 1, mode: 'one-shot' as const }]
|
|
try {
|
|
if (warm) {
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 0, value: entries })
|
|
}
|
|
remote.session.projections.mockRejectedValueOnce(failure)
|
|
await manager.refreshProjections(S1)
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]).toMatchObject({
|
|
state: 'error', error: failure, values: warm ? { subagentCatalog: entries } : {},
|
|
})
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
})
|
|
|
|
describe('list lifecycle', () => {
|
|
it('uses metadata received before the list without reblanking started rows', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
try {
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1,
|
|
key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 1200 }, seq: 8 })
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S2,
|
|
key: 'sessionListMetadata', value: { blank: true, lastPromptAt: null }, seq: 2 })
|
|
remote.session.list.mockResolvedValue(ok({ items: [
|
|
summary(S1, { blank: true }), summary(S2),
|
|
] as never[] }))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items.map(item => item.blank)).toEqual([false, false])
|
|
expect(manager.get(S1).getSnapshot().blank).toBe(false)
|
|
expect(manager.get(S2).getSnapshot().blank).toBe(false)
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1,
|
|
key: 'sessionListMetadata', value: { blank: true, lastPromptAt: null }, seq: 7 })
|
|
expect(manager.getListSnapshot().items[0]).toMatchObject({ blank: false, updatedAt: 1200 })
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it('fills missing durable links without overwriting established rows and projects first-send engagement', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
remote.session.list.mockResolvedValueOnce(ok({ items: [
|
|
summary(S1, { blank: true }), summary(S2, { cwd: '/existing', blank: true }),
|
|
] }))
|
|
try {
|
|
await manager.refreshList()
|
|
manager.handleSessionAdded(summary(S1, { cwd: '/filled', parentSessionId: S2, origin: 'subagent', blank: false }))
|
|
manager.handleSessionAdded(summary(S2, { cwd: '/ignored', blank: true }))
|
|
manager.handleSessionActivity(S1, 50)
|
|
manager.handleSessionActivity(S2, 200)
|
|
await manager.get(S2).prompt([{ type: 'text', text: 'first message' }], 'queue')
|
|
expect(manager.getListSnapshot().items).toEqual(expect.arrayContaining([
|
|
expect.objectContaining({ sessionId: S1, cwd: '/filled', parentSessionId: S2, origin: 'subagent', blank: false }),
|
|
expect.objectContaining({ sessionId: S2, cwd: '/existing', updatedAt: 200, blank: false }),
|
|
]))
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it('keeps an opened conversation out of blank-session reuse after a stale list refresh', async ({ mock, remote }) => {
|
|
const stale = {
|
|
...summary(S1, { blank: true, cwd: '/workspace' }),
|
|
projections: { asOfSeq: 2, values: {
|
|
sessionListMetadata: { blank: true, lastPromptAt: null },
|
|
} },
|
|
}
|
|
remote.session.list.mockResolvedValue(ok({ items: [stale] as never[] }))
|
|
const manager = makeManager(mock, remote)
|
|
try {
|
|
await manager.refreshList()
|
|
const session = manager.get(S1)
|
|
expect(session.getSnapshot().blank).toBe(true)
|
|
|
|
// History opening can be newer than the metadata-only list cache.
|
|
session.projections.seed({ asOfSeq: SessionSeq(61), values: {
|
|
sessionListMetadata: { blank: false, lastPromptAt: 1200 },
|
|
} })
|
|
await vi.waitFor(() => {
|
|
expect(manager.getListSnapshot().items[0]).toMatchObject({ blank: false, updatedAt: 1200 })
|
|
expect(session.getSnapshot().blank).toBe(false)
|
|
})
|
|
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items[0]).toMatchObject({ blank: false, updatedAt: 1200 })
|
|
expect(session.getSnapshot().blank).toBe(false)
|
|
manager.handleSessionAdded(stale)
|
|
expect(session.getSnapshot().blank).toBe(false)
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it('projects a cold Session addition as cached until an empty-cut control baseline replaces it', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
try {
|
|
manager.handleSessionAdded({
|
|
...summary(S1), projections: { asOfSeq: -1, values: { title: 'before history' } },
|
|
})
|
|
expect(manager.getListSnapshot().items[0]?.title).toBe('before history')
|
|
// The control baseline is the connected Session's own value: it replaces
|
|
// the list-surface title even at the same cursor.
|
|
manager.handleControlFrame({
|
|
type: 'baseline', value: { jobs: { [S1]: [] }, projections: { [S1]: { asOfSeq: -1, values: { title: 'cold baseline' } } } },
|
|
})
|
|
expect(manager.getListSnapshot().items[0]?.title).toBe('cold baseline')
|
|
expect(manager.getListSnapshot().jobsBySession).toEqual({})
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it('single-flights refreshList and preserves the Host baseline order', async ({ mock, remote }) => {
|
|
const gate = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.list>>>()
|
|
remote.session.list.mockReturnValue(gate.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const first = manager.refreshList()
|
|
const second = manager.refreshList()
|
|
expect(manager.getListSnapshot().state).toBe('loading')
|
|
gate.resolve(ok({ items: [summary(S2, { updatedAt: 200 }), summary(S1)] as never[] }))
|
|
await Promise.all([first, second])
|
|
expect(remote.session.list).toHaveBeenCalledOnce()
|
|
const snapshot = manager.getListSnapshot()
|
|
expect(snapshot.state).toBe('idle')
|
|
expect(snapshot.items.map(i => i.sessionId)).toEqual([S2, S1])
|
|
})
|
|
|
|
it('replays incremental frames over hydration and never batch-reorders established ids', async ({ mock, remote }) => {
|
|
const first = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.list>>>()
|
|
remote.session.list.mockReturnValue(first.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const hydration = manager.refreshList()
|
|
manager.handleSessionAdded(summary(S2, { blank: true }))
|
|
first.resolve(ok({ items: [summary(S1)] as never[] }))
|
|
await hydration
|
|
expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
|
|
|
|
remote.session.list.mockResolvedValue(ok({
|
|
items: [summary(S1, { updatedAt: 900 }), summary(S2, { updatedAt: 800 })] as never[],
|
|
}))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
|
|
})
|
|
|
|
it('advances list activity from the filtered Host notification', async ({ mock, remote }) => {
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1)] as never[] }))
|
|
const manager = makeManager(mock, remote)
|
|
await manager.refreshList()
|
|
|
|
manager.handleSessionActivity(S1, 500)
|
|
expect(manager.getListSnapshot().items[0]?.updatedAt).toBe(500)
|
|
})
|
|
|
|
it('keeps the error in the list snapshot on failure', async ({ mock, remote }) => {
|
|
remote.session.list.mockResolvedValue(err(new RemoteError('gateway/internal', 'boom', {})))
|
|
const manager = makeManager(mock, remote)
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'gateway/internal' } })
|
|
// A failed pull does not step the arrival phase: still pending.
|
|
expect(manager.getListSnapshot().phase).toBe('pending')
|
|
})
|
|
|
|
it('phase steps pending → ready on the first successful pull and never returns', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
expect(manager.getListSnapshot().phase).toBe('pending')
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().phase).toBe('ready')
|
|
// Sticky across later failures: the pull-activity axis reports the error,
|
|
// the arrival phase holds.
|
|
remote.session.list.mockResolvedValue(err(new RemoteError('gateway/internal', 'down', {})))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot()).toMatchObject({ state: 'error', phase: 'ready' })
|
|
// And across an empty re-pull (empty-with-ready = truly no sessions).
|
|
remote.session.list.mockResolvedValue(ok({ items: [] as never[] }))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', phase: 'ready' })
|
|
expect(manager.getListSnapshot().items).toEqual([])
|
|
})
|
|
|
|
it('merges create into the list immediately without waiting for a refresh', async ({ mock, remote }) => {
|
|
remote.session.create.mockResolvedValue(ok({ sessionId: S2 }))
|
|
const manager = makeManager(mock, remote)
|
|
const result = await manager.create()
|
|
expect(result).toMatchObject({ ok: true, value: { sessionId: S2 } })
|
|
expect(manager.getListSnapshot().items.map(i => i.sessionId)).toEqual([S2])
|
|
})
|
|
|
|
it('retains title projections before list arrival, keeps last-wins by seq, and clears them on removal', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
const titleFrame = (title: string, seq: number) => {
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'title', value: title, seq })
|
|
}
|
|
titleFrame('Newest', 4)
|
|
titleFrame('Stale', 3)
|
|
titleFrame('Equal', 4)
|
|
remote.session.list.mockResolvedValue(ok({
|
|
items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[],
|
|
}))
|
|
await manager.refreshList()
|
|
|
|
const titled = manager.getListSnapshot()
|
|
expect(titled.items.map(item => item.sessionId)).toEqual([S1, S2])
|
|
expect(titled.items[0]?.title).toBe('Newest')
|
|
expect(titled.items[1]?.title).toBeUndefined()
|
|
|
|
manager.handleSessionRemoved(S1)
|
|
manager.handleSessionAdded(summary(S1, { blank: true }))
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined()
|
|
})
|
|
|
|
it('fills cold titles from the list rows\' projections block as cached values', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
// A push frame landed before the list: S2's title is a sequenced value.
|
|
manager.handleControlFrame({
|
|
type: 'projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9,
|
|
})
|
|
remote.session.list.mockResolvedValue(ok({
|
|
items: [
|
|
{ ...summary(S1), projections: { asOfSeq: 4, values: { title: 'Cold cached' } } },
|
|
{ ...summary(S2, { updatedAt: 200 }), projections: { asOfSeq: 50, values: { title: 'List stale' } } },
|
|
] as never[],
|
|
}))
|
|
await manager.refreshList()
|
|
const items = manager.getListSnapshot().items
|
|
// Cold row: title surfaces straight from the list block — no open, no history.
|
|
expect(items.find(item => item.sessionId === S1)?.title).toBe('Cold cached')
|
|
// A list block never displaces a sequenced value, whatever watermark it claims.
|
|
expect(items.find(item => item.sessionId === S2)?.title).toBe('Pushed')
|
|
})
|
|
|
|
it('discards the previous generation title before accepting its lower-seq replay', async ({ mock, remote }) => {
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1)] as never[] }))
|
|
const manager = makeManager(mock, remote)
|
|
try {
|
|
await manager.refreshList()
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'title', value: 'Unflushed', seq: 4 })
|
|
|
|
manager.handleConnected()
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items[0]?.title).toBeUndefined()
|
|
manager.handleControlFrame({
|
|
type: 'baseline',
|
|
value: {
|
|
jobs: {},
|
|
projections: { [S1]: { asOfSeq: 2, values: { title: 'Durable' } } },
|
|
},
|
|
})
|
|
expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
})
|
|
|
|
describe('search', () => {
|
|
it('returns bounded Host results and forwards the caller signal', async ({ mock, remote }) => {
|
|
remote.session.search.mockResolvedValue(ok({
|
|
items: [{ sessionId: S1, snippet: 'matching excerpt' }],
|
|
hasMore: true,
|
|
}))
|
|
const manager = makeManager(mock, remote)
|
|
const signal = new AbortController().signal
|
|
|
|
await expect(manager.search('exact phrase', signal)).resolves.toEqual({
|
|
ok: true,
|
|
value: {
|
|
items: [{ sessionId: S1, snippet: 'matching excerpt' }],
|
|
hasMore: true,
|
|
},
|
|
})
|
|
expect(remote.session.search).toHaveBeenCalledExactlyOnceWith({ query: 'exact phrase' }, signal)
|
|
})
|
|
|
|
it('preserves business errors and propagates a non-Remote throw', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
remote.session.search.mockResolvedValue(err(new RemoteError('gateway/internal', 'index unavailable', {})))
|
|
const signal = new AbortController().signal
|
|
await expect(manager.search('first', signal)).resolves.toMatchObject({
|
|
ok: false,
|
|
error: { code: 'gateway/internal', message: 'index unavailable' },
|
|
})
|
|
|
|
remote.session.search.mockRejectedValue(new Error('wire down'))
|
|
await expect(manager.search('second', signal)).rejects.toThrow('wire down')
|
|
})
|
|
})
|
|
|
|
describe('Host Remote event routing', () => {
|
|
it('adds/removes/flips sessions and keeps removed instances resident', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleSessionAdded(summary(S1, { blank: true }))
|
|
manager.handleSessionAdded(summary(S1, { blank: true })) // dup: ignored
|
|
expect(manager.getListSnapshot().items).toHaveLength(1)
|
|
|
|
const session = manager.get(S1)
|
|
manager.handleSessionStatus(S1, true)
|
|
expect(session.getSnapshot().running).toBe(true)
|
|
expect(manager.getListSnapshot().items[0]?.running).toBe(true)
|
|
|
|
manager.handleSessionError(S1, '炸了')
|
|
expect(session.getSnapshot().lastAgentError).toBe('炸了')
|
|
|
|
manager.handleSessionRemoved(S1)
|
|
expect(manager.getListSnapshot().items).toHaveLength(0)
|
|
expect(session.getSnapshot().removed).toBe(true)
|
|
expect(manager.get(S1)).toBe(session) // resident-instance rule survives removal
|
|
})
|
|
})
|
|
|
|
describe('subagent catalogs', () => {
|
|
it('keeps a catalog-discovered child address across identity resolution and status frames', async ({ mock, remote, start }) => {
|
|
remote.session.list.mockImplementation(() => Promise.resolve(ok({ items: [
|
|
summary(S1),
|
|
summary(S2, { parentSessionId: S1, origin: 'subagent' }),
|
|
] as never[] })))
|
|
remote.session.projections.mockImplementation(() => Promise.resolve(ok({ asOfSeq: 0, values: { subagentCatalog: [{ createdAt: 1,
|
|
id: S2, mode: 'continuable', label: 'worker',
|
|
}] } })))
|
|
mock.load(sessionWorld)
|
|
const client = await start()
|
|
const manager = new SessionManager(client.ctx.remote)
|
|
await manager.refreshList()
|
|
await manager.refreshProjections(S1)
|
|
manager.resolveTarget({ parentSessionId: S1, childSessionId: S2, mode: 'continuable' })
|
|
|
|
expect(manager.subagentAddress(S2)).toEqual({
|
|
parentSessionId: S1, childSessionId: S2, mode: 'continuable',
|
|
})
|
|
expect(manager.get(S2).getSnapshot().subagent).toEqual({
|
|
address: {
|
|
parentSessionId: S1, childSessionId: S2, mode: 'continuable',
|
|
},
|
|
parentAvailable: true,
|
|
})
|
|
// Clicking the same child through an ordinary list-selection path must not
|
|
// erase the catalog-derived address and fall back to session.* transport.
|
|
manager.resolveTarget(S2)
|
|
expect(manager.subagentAddress(S2)).toEqual({
|
|
parentSessionId: S1, childSessionId: S2, mode: 'continuable',
|
|
})
|
|
expect(manager.get(S2).getSnapshot().subagent).toEqual({
|
|
address: {
|
|
parentSessionId: S1, childSessionId: S2, mode: 'continuable',
|
|
},
|
|
parentAvailable: true,
|
|
})
|
|
await manager.get(S2).open()
|
|
await manager.get(S2).prompt([{ type: 'text', text: 'continue' }], 'queue')
|
|
expect(mock.log.requests(FOLLOW)).toEqual([
|
|
{
|
|
address: {
|
|
kind: 'subagent',
|
|
parentSessionId: S1,
|
|
childSessionId: S2,
|
|
mode: 'continuable',
|
|
},
|
|
assistantStream: true,
|
|
maxMessages: 50,
|
|
},
|
|
])
|
|
expect(mock.log.requests('session/page')).toEqual([])
|
|
expect(remote.subagents.prompt.mock.calls.map(([request]) => request)).toEqual([
|
|
{
|
|
requestId: expect.any(String) as unknown as string,
|
|
parentSessionId: S1, childSessionId: S2,
|
|
mode: 'continuable',
|
|
delivery: 'queue',
|
|
content: [{ type: 'text', text: 'continue' }],
|
|
clientTimeZone: new Intl.DateTimeFormat().resolvedOptions().timeZone,
|
|
},
|
|
])
|
|
expect(mock.log.requests('session/page')).toEqual([])
|
|
expect(remote.session.prompt.mock.calls.map(([request]) => request)).toEqual([])
|
|
const listCalls = remote.session.projections.mock.calls.map(([request]) => request).length
|
|
manager.handleSessionStatus(S2, false)
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog?.[0]).toMatchObject({
|
|
id: S2,
|
|
})
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(listCalls)
|
|
|
|
manager.handleSessionRemoved(S2)
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)).toMatchObject({
|
|
origin: 'subagent', parentSessionId: S1, running: false,
|
|
})
|
|
expect(manager.get(S2).getSnapshot()).toMatchObject({
|
|
removed: false,
|
|
subagent: {
|
|
address: {
|
|
parentSessionId: S1, childSessionId: S2, mode: 'continuable',
|
|
},
|
|
},
|
|
})
|
|
})
|
|
|
|
it('publishes pushed membership without a catalog request or an instantiated parent', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleSessionAdded(summary(S2, { origin: 'subagent', parentSessionId: S1, running: true }))
|
|
manager.handleControlFrame({
|
|
type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 4,
|
|
value: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'worker' }],
|
|
})
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog).toEqual([
|
|
{ id: S2, createdAt: 1, mode: 'continuable', label: 'worker' },
|
|
])
|
|
expect(manager.subagentAddress(S2)).toEqual({ parentSessionId: S1, childSessionId: S2, mode: 'continuable' })
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toEqual([])
|
|
})
|
|
|
|
it('keeps newer pushed membership when the initial read returns an older baseline', async ({ mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections.mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const read = manager.refreshProjections(S1)
|
|
manager.handleControlFrame({
|
|
type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 5,
|
|
value: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'pushed' }],
|
|
})
|
|
response.resolve(ok({ asOfSeq: 3, values: { subagentCatalog: [] } }))
|
|
await read
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog?.[0]?.label).toBe('pushed')
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(1)
|
|
})
|
|
|
|
it.for([false, true])('keeps live parent availability when an earlier initial read settles (failure: %s)', async (failure, { mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections.mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
manager.resolveTarget({ parentSessionId: S1, childSessionId: S2, mode: 'continuable' })
|
|
const child = manager.get(S2)
|
|
const read = manager.refreshProjections(S1)
|
|
manager.handleSessionAdded(summary(S1))
|
|
manager.handleSessionStatus(S1, false)
|
|
response.resolve(failure
|
|
? err(new RemoteError('gateway/internal', 'offline', {}))
|
|
: ok({ asOfSeq: 0, values: { subagentCatalog: [] } }))
|
|
await read
|
|
expect(child.getSnapshot().subagent?.parentAvailable).toBe(true)
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(1)
|
|
})
|
|
|
|
it('keeps parent availability unknown after list failure and accepts a later Host summary', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.resolveTarget({ parentSessionId: S1, childSessionId: S2, mode: 'continuable' })
|
|
const child = manager.get(S2)
|
|
remote.session.list.mockResolvedValue(err(new RemoteError('gateway/internal', 'offline', {})))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().state).toBe('error')
|
|
expect(child.getSnapshot().subagent?.parentAvailable).toBeUndefined()
|
|
manager.handleSessionAdded(summary(S1))
|
|
expect(child.getSnapshot().subagent?.parentAvailable).toBe(true)
|
|
await manager.dispose()
|
|
})
|
|
|
|
it('keeps a pushed catalog when an earlier initial read reports the parent missing', async ({ mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections.mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const read = manager.refreshProjections(S1)
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 1,
|
|
value: [{ id: S2, createdAt: 1, mode: 'one-shot' }] })
|
|
response.resolve(ok(null))
|
|
await read
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]?.values.subagentCatalog).toEqual([
|
|
{ id: S2, createdAt: 1, mode: 'one-shot' },
|
|
])
|
|
})
|
|
|
|
it('shares cold initial loading and reuses the loaded catalog when reopened', async ({ mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections.mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const read = manager.refreshProjections(S1)
|
|
expect(manager.refreshProjections(S1)).toBe(read)
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]?.state).toBe('loading')
|
|
response.resolve(ok({ asOfSeq: 0, values: {
|
|
subagentCatalog: [{ id: S2, createdAt: 1, mode: 'one-shot' }],
|
|
title: 'cold parent',
|
|
} }))
|
|
await read
|
|
await manager.refreshProjections(S1)
|
|
manager.handleSessionAdded(summary(S1))
|
|
expect(manager.getListSnapshot().items[0]?.title).toBe('cold parent')
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]).toMatchObject({
|
|
values: { subagentCatalog: [{ id: S2, createdAt: 1, mode: 'one-shot' }] }, state: 'ready',
|
|
})
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(1)
|
|
})
|
|
|
|
it('uses current Session status rather than the time an initial catalog read started', async ({ mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections.mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const read = manager.refreshProjections(S1)
|
|
manager.handleSessionAdded(summary(S2, { origin: 'subagent', parentSessionId: S1 }))
|
|
manager.handleSessionStatus(S2, true)
|
|
response.resolve(ok({ asOfSeq: 0, values: {
|
|
subagentCatalog: [{ id: S2, createdAt: 1, mode: 'continuable', label: 'worker' }],
|
|
} }))
|
|
await read
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.running).toBe(true)
|
|
manager.handleSessionStatus(S2, false)
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.running).toBe(false)
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(1)
|
|
})
|
|
|
|
it('keeps completed unselected child metadata with its durable catalog row', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleSessionAdded(summary(S2, { origin: 'subagent', parentSessionId: S1, running: true }))
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S2, key: 'title', value: 'finished child', seq: 2 })
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 1,
|
|
value: [{ id: S2, createdAt: 1, mode: 'one-shot' }] })
|
|
manager.handleSessionRemoved(S2)
|
|
expect(manager.getListSnapshot().items[0]).toMatchObject({ title: 'finished child', running: false })
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.running).toBe(false)
|
|
})
|
|
|
|
it('does not invent membership from Session summaries', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleSessionAdded(summary(S2, { origin: 'subagent', parentSessionId: S1 }))
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]).toBeUndefined()
|
|
})
|
|
|
|
it('retries a failed initial read without discarding pushed entries', async ({ mock, remote }) => {
|
|
remote.session.projections.mockImplementation(() => Promise.resolve(err(new RemoteError('gateway/internal', 'offline', {}))))
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'subagentCatalog', seq: 2,
|
|
value: [{ id: S2, createdAt: 1, mode: 'one-shot' }] })
|
|
await manager.refreshProjections(S1)
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]).toMatchObject({ state: 'error', values: { subagentCatalog: [{ id: S2 }] } })
|
|
remote.session.projections.mockImplementation(() => Promise.resolve(ok({ asOfSeq: 2, values: { subagentCatalog: [] } })))
|
|
await manager.refreshProjections(S1)
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]).toMatchObject({ state: 'ready', values: { subagentCatalog: [{ id: S2 }] } })
|
|
})
|
|
|
|
it('ignores a late response after parent removal without a trailing request', async ({ mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections.mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const read = manager.refreshProjections(S1)
|
|
manager.handleSessionRemoved(S1)
|
|
response.resolve(ok({ asOfSeq: 0, values: { subagentCatalog: [] } }))
|
|
await read
|
|
expect(manager.getListSnapshot().items.some(item => item.sessionId === S1)).toBe(false)
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(1)
|
|
})
|
|
|
|
it('represents a missing parent as an empty unavailable catalog', async ({ mock, remote }) => {
|
|
remote.session.projections.mockImplementation(() => Promise.resolve(ok(null)))
|
|
const manager = makeManager(mock, remote)
|
|
await manager.refreshProjections(S1)
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]).toEqual({
|
|
values: {}, state: 'ready', error: null,
|
|
})
|
|
})
|
|
})
|
|
|
|
describe('remaining branches', () => {
|
|
it('refreshList propagates a non-Remote throw', async ({ mock, remote }) => {
|
|
remote.session.list.mockRejectedValue(new Error('list wire down'))
|
|
const manager = makeManager(mock, remote)
|
|
await expect(manager.refreshList()).rejects.toThrow('list wire down')
|
|
})
|
|
|
|
it('refreshList pushes running bits down to already-instantiated sessions', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
const session = manager.get(S1)
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1, { running: true })] as never[] }))
|
|
await manager.refreshList()
|
|
expect(session.getSnapshot().running).toBe(true)
|
|
})
|
|
|
|
it('create passes cwd and a preallocated id, folds transport throws, and deduplicates the echo', async ({ mock, remote }) => {
|
|
remote.session.create.mockResolvedValue(ok({ sessionId: S1 }))
|
|
const manager = makeManager(mock, remote)
|
|
await manager.create({ cwd: '/tmp/w', sessionId: S1 })
|
|
expect(remote.session.create).toHaveBeenCalledExactlyOnceWith({ cwd: '/tmp/w', sessionId: S1 })
|
|
expect(manager.getListSnapshot().items[0]).toMatchObject({ sessionId: S1, cwd: '/tmp/w' })
|
|
await manager.create({ cwd: '/tmp/w' }) // same id returned: no duplicate row
|
|
expect(manager.getListSnapshot().items).toHaveLength(1)
|
|
remote.session.create.mockRejectedValue(new Error('create wire down'))
|
|
await expect(manager.create()).rejects.toThrow('create wire down')
|
|
// Business error passes through untouched.
|
|
remote.session.create.mockResolvedValue(err(new RemoteError('gateway/internal', 'no', {})))
|
|
expect(await manager.create()).toMatchObject({ ok: false })
|
|
})
|
|
|
|
it.for(['create', 'fork'] as const)('preserves pushed running state when a %s response arrives later', async (operation, { mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.create>>>()
|
|
remote.session[operation].mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const request = operation === 'create' ? manager.create() : manager.fork({ sessionId: S1 })
|
|
manager.handleSessionAdded(summary(S2, { running: true }))
|
|
response.resolve(ok({ sessionId: S2 }))
|
|
await request
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2))
|
|
.toMatchObject({ running: true })
|
|
manager.handleSessionAdded(summary(S2, { running: false, agentAvailable: false }))
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2))
|
|
.toMatchObject({ running: false })
|
|
await manager.dispose()
|
|
})
|
|
|
|
it('preserves Host availability when a local create echo follows Agent disposal', async ({ mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.create>>>()
|
|
remote.session.create.mockImplementation(() => response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
manager.resolveTarget({
|
|
parentSessionId: S1, childSessionId: S2, mode: 'continuable',
|
|
})
|
|
const child = manager.get(S2)
|
|
const request = manager.create()
|
|
manager.handleSessionAdded(summary(S1, { agentAvailable: false }))
|
|
response.resolve(ok({ sessionId: S1 }))
|
|
await request
|
|
manager.resolveTarget(S2)
|
|
expect(child.getSnapshot().subagent?.parentAvailable).toBe(false)
|
|
await manager.dispose()
|
|
})
|
|
|
|
it('publishes a real Ungrouped summary from workspace-attach-failed', async ({ mock, remote }) => {
|
|
remote.session.create.mockResolvedValue(err(new RemoteError('session/workspace-attach-failed', 'published but unattached', {
|
|
sessionId: S1, workspaceId: 'w1',
|
|
})))
|
|
const manager = makeManager(mock, remote)
|
|
const result = await manager.create({ workspaceId: 'w1' as never, sessionId: S1 })
|
|
expect(result).toMatchObject({ ok: false, error: { code: 'session/workspace-attach-failed' } })
|
|
expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({ sessionId: S1 })])
|
|
expect(manager.getListSnapshot().items[0]).not.toHaveProperty('cwd')
|
|
})
|
|
|
|
it('reconciles a fork child published before workspace attachment fails', async ({ mock, remote }) => {
|
|
remote.session.fork.mockResolvedValue(err(new RemoteError('session/workspace-attach-failed', 'forked but unattached', {
|
|
sessionId: S2, workspaceId: 'w1',
|
|
})))
|
|
const manager = makeManager(mock, remote)
|
|
const result = await manager.fork({ sessionId: S1 })
|
|
expect(result).toMatchObject({ ok: false, error: { code: 'session/workspace-attach-failed' } })
|
|
expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({
|
|
sessionId: S2,
|
|
parentSessionId: S1,
|
|
blank: true,
|
|
})])
|
|
|
|
manager.handleSessionAdded(summary(S2, { blank: true, parentSessionId: S1 }))
|
|
expect(manager.getListSnapshot().items[0]?.blank).toBe(true)
|
|
manager.handleSessionAdded(summary(S2, { blank: false, parentSessionId: S1 }))
|
|
expect(manager.getListSnapshot().items[0]?.blank).toBe(false)
|
|
})
|
|
|
|
it('reconciles a preallocated id after an ordinary transport failure', async ({ mock, remote }) => {
|
|
remote.session.create.mockRejectedValue(new Error('response lost'))
|
|
const manager = makeManager(mock, remote)
|
|
await expect(manager.create({ workspaceId: 'w1' as never, sessionId: S1 }))
|
|
.rejects.toThrow('response lost')
|
|
expect(manager.getListSnapshot().items).toEqual([])
|
|
|
|
manager.handleSessionAdded(summary(S1, { blank: true, cwd: '/w/one' }))
|
|
expect(manager.getListSnapshot().items).toEqual([
|
|
expect.objectContaining({ sessionId: S1, cwd: '/w/one' }),
|
|
])
|
|
manager.handleSessionAdded(summary(S1, { blank: true, cwd: '/w/one' }))
|
|
expect(manager.getListSnapshot().items).toHaveLength(1)
|
|
})
|
|
|
|
it('subscribe notifies on list changes and stops after unsubscribe', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
let notified = 0
|
|
const unsubscribe = manager.subscribe(() => { notified++ })
|
|
await manager.refreshList()
|
|
await new Promise(resolve => setTimeout(resolve, 0))
|
|
expect(notified).toBeGreaterThan(0)
|
|
const seen = notified
|
|
unsubscribe()
|
|
manager.handleSessionAdded(summary(S1, { blank: true }))
|
|
await new Promise(resolve => setTimeout(resolve, 0))
|
|
expect(notified).toBe(seen)
|
|
})
|
|
|
|
it('ignores Host status and error events for sessions without an instance', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleSessionStatus(S2, true)
|
|
manager.handleSessionError(S2, '无实例')
|
|
})
|
|
|
|
it('keeps list-entry identity for unchanged rows across an unrelated list change', async ({ mock, remote }) => {
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
|
|
const manager = makeManager(mock, remote)
|
|
await manager.refreshList()
|
|
const before = manager.getListSnapshot()
|
|
manager.handleSessionStatus(S2, true)
|
|
const after = manager.getListSnapshot()
|
|
expect(after.items).not.toBe(before.items)
|
|
const beforeS1 = before.items.find(e => e.sessionId === S1)
|
|
const afterS1 = after.items.find(e => e.sessionId === S1)
|
|
expect(afterS1).toBe(beforeS1) // untouched entry keeps identity (entryCache)
|
|
// Same-order same-entries snapshot reuses the items array.
|
|
manager.handleSessionError(S1, 'x')
|
|
expect(manager.getListSnapshot().items).toBe(after.items)
|
|
})
|
|
|
|
it('reuses refreshed rows and evicts missing rows independently of Client instances', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1), summary(S2)] as never[] }))
|
|
await manager.refreshList()
|
|
const first = manager.getListSnapshot()
|
|
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1), summary(S2)] as never[] }))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items).toBe(first.items)
|
|
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1)] as never[] }))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items).toEqual([first.items[0]])
|
|
expect(manager.getListSnapshot().items[0]).toBe(first.items[0])
|
|
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1), summary(S2)] as never[] }))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items[0]).toBe(first.items[0])
|
|
expect(manager.getListSnapshot().items[1]).not.toBe(first.items[1])
|
|
|
|
remote.session.list.mockResolvedValue(ok({ items: [] as never[] }))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items).toEqual([])
|
|
|
|
remote.session.list.mockResolvedValue(ok({ items: [summary(S1)] as never[] }))
|
|
await manager.refreshList()
|
|
expect(manager.getListSnapshot().items[0]).not.toBe(first.items[0])
|
|
})
|
|
|
|
it('bounds cached-row ID reads linearly during repeated list refreshes', async ({ mock, remote }) => {
|
|
const count = 1_000
|
|
const summaries = Array.from({ length: count }, (_, i) => summary(`list-${i}` as SessionId))
|
|
const manager = makeManager(mock, remote)
|
|
remote.session.list.mockResolvedValue(ok({ items: summaries as never[] }))
|
|
await manager.refreshList()
|
|
const first = manager.getListSnapshot()
|
|
let reads = 0
|
|
// Instance-local accessors count membership work without a machine-dependent timing budget.
|
|
for (const entry of first.items) {
|
|
const id = entry.sessionId
|
|
Object.defineProperty(entry, 'sessionId', { get: () => { reads++; return id }, configurable: true })
|
|
}
|
|
for (let refresh = 0; refresh < 2; refresh++) {
|
|
reads = 0
|
|
remote.session.list.mockResolvedValue(ok({ items: summaries.map(item => ({ ...item })) as never[] }))
|
|
await manager.refreshList()
|
|
const snapshot = manager.getListSnapshot()
|
|
expect(snapshot.items).toBe(first.items)
|
|
expect(reads).toBeLessThanOrEqual(count * 3)
|
|
}
|
|
})
|
|
|
|
it('carries parentSessionId from the added event into the lineage row', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleSessionAdded(summary(S1, { blank: true }))
|
|
manager.handleSessionAdded(summary(S2, {
|
|
blank: true, parentSessionId: S1, origin: 'subagent',
|
|
}))
|
|
const items = manager.getListSnapshot().items
|
|
expect(items.find(e => e.sessionId === S2)).toMatchObject({
|
|
parentSessionId: S1, origin: 'subagent', depth: 1,
|
|
})
|
|
})
|
|
})
|
|
|
|
describe('connected generation', () => {
|
|
it.for(['old-first', 'new-first'] as const)(
|
|
'ignores a previous generation list response (%s)',
|
|
async (order, { mock, remote }) => {
|
|
const oldList = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.list>>>()
|
|
const newList = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.list>>>()
|
|
let calls = 0
|
|
remote.session.list.mockImplementation(() => calls++ === 0 ? oldList.promise : newList.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const oldResult = ok({ items: [{ ...summary(S1), projections: {
|
|
asOfSeq: 20, values: { title: 'Unpersisted title' },
|
|
} }] as never[] })
|
|
const newResult = ok({ items: [{ ...summary(S1), projections: {
|
|
asOfSeq: 1, values: { title: 'Durable title' },
|
|
} }] as never[] })
|
|
const oldPull = manager.refreshList()
|
|
let newPull: Promise<void> | undefined
|
|
try {
|
|
manager.handleConnected()
|
|
newPull = manager.refreshList()
|
|
expect(remote.session.list.mock.calls).toHaveLength(2)
|
|
if (order === 'old-first') {
|
|
oldList.resolve(oldResult)
|
|
await oldPull
|
|
expect(manager.getListSnapshot().state).toBe('loading')
|
|
expect(manager.refreshList()).toBe(newPull)
|
|
}
|
|
newList.resolve(newResult)
|
|
await newPull
|
|
oldList.resolve(oldResult)
|
|
await oldPull
|
|
|
|
expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', error: null })
|
|
expect(manager.getListSnapshot().items[0]?.title).toBe('Durable title')
|
|
} finally {
|
|
oldList.resolve(oldResult)
|
|
newList.resolve(newResult)
|
|
await Promise.all([oldPull, newPull])
|
|
await manager.dispose()
|
|
}
|
|
},
|
|
)
|
|
|
|
it('ignores a previous generation request failure while the new list is loading', async ({ mock, remote }) => {
|
|
const oldList = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.list>>>()
|
|
const newList = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.list>>>()
|
|
let calls = 0
|
|
remote.session.list.mockImplementation(() => calls++ === 0 ? oldList.promise : newList.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const oldPull = manager.refreshList()
|
|
let newPull: Promise<void> | undefined
|
|
try {
|
|
manager.handleConnected()
|
|
newPull = manager.refreshList()
|
|
oldList.reject(new RemoteError('gateway/internal', 'old Host disconnected', {}))
|
|
await oldPull
|
|
expect(manager.getListSnapshot()).toMatchObject({ state: 'loading', error: null })
|
|
expect(manager.refreshList()).toBe(newPull)
|
|
newList.resolve(ok({ items: [summary(S1)] as never[] }))
|
|
await newPull
|
|
expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', error: null })
|
|
} finally {
|
|
oldList.resolve(ok({ items: [] }))
|
|
newList.resolve(ok({ items: [] }))
|
|
await Promise.all([oldPull, newPull])
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it.for([false, true])('refreshes an unselected cached catalog after offline activity changes: %s', async (running, { mock, remote }) => {
|
|
let hostRunning = !running
|
|
remote.session.list.mockImplementation(() => Promise.resolve(ok({ items: [
|
|
summary(S1),
|
|
summary(S2, { parentSessionId: S1, origin: 'subagent', running: hostRunning }),
|
|
] as never[] })))
|
|
remote.session.projections.mockImplementation(() => Promise.resolve(ok({ asOfSeq: 0, values: { subagentCatalog: [{ createdAt: 1, id: S2, mode: 'continuable', label: 'worker' }] } })))
|
|
const manager = makeManager(mock, remote)
|
|
try {
|
|
await manager.refreshList()
|
|
await manager.refreshProjections(S1)
|
|
const reads = remote.session.projections.mock.calls.map(([request]) => request).length
|
|
hostRunning = running
|
|
manager.handleConnected()
|
|
await manager.refreshList()
|
|
await vi.waitFor(() => {
|
|
expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.running)
|
|
.toBe(running)
|
|
})
|
|
expect(manager.getListSnapshot()).not.toHaveProperty('current')
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(reads + 1)
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it('starts a fresh catalog read on reconnect and ignores the previous response', async ({ mock, remote }) => {
|
|
const previous = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections
|
|
.mockReturnValueOnce(previous.promise)
|
|
.mockResolvedValue(ok({ asOfSeq: 0, values: { subagentCatalog: [] } }))
|
|
const manager = makeManager(mock, remote)
|
|
try {
|
|
const read = manager.refreshProjections(S1)
|
|
manager.handleConnected()
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(2)
|
|
previous.resolve(ok({ asOfSeq: 0, values: { subagentCatalog: [{ createdAt: 1, id: S2, mode: 'continuable', label: 'worker' }] } }))
|
|
await read
|
|
await vi.waitFor(() => {
|
|
expect(manager.getListSnapshot().projectionsBySession[S1]).toMatchObject({
|
|
values: {}, state: 'ready',
|
|
})
|
|
})
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toHaveLength(2)
|
|
} finally {
|
|
await manager.dispose()
|
|
}
|
|
})
|
|
|
|
it('refreshes query baselines without rebuilding independently resumed Session sources', async ({ mock, remote, start }) => {
|
|
mock.stream(FOLLOW, followScript(ok({
|
|
records: entries(plainTurn(SessionSeq(0), 0, 'a', 'b')) as never[],
|
|
hasMore: false,
|
|
modelSelection: { provider: 'deepseek-official', model: 'deepseek-chat' },
|
|
})))
|
|
const client = await start()
|
|
const manager = new SessionManager(client.ctx.remote)
|
|
const openedSession = manager.get(S1)
|
|
await openedSession.open()
|
|
manager.get(S2) // instantiated but never opened
|
|
const historyCallsBefore = remote.session.page.mock.calls.length
|
|
manager.handleConnected()
|
|
await vi.waitFor(() => {
|
|
expect(remote.session.list).toHaveBeenCalledOnce()
|
|
})
|
|
expect(remote.session.follow).toHaveBeenCalledOnce()
|
|
expect(remote.session.page).toHaveBeenCalledTimes(historyCallsBefore)
|
|
})
|
|
|
|
it('retains the durable parent address and refreshes that parent across reconnect', async ({ mock, remote }) => {
|
|
const address = {
|
|
parentSessionId: S1,
|
|
childSessionId: S2,
|
|
mode: 'continuable' as const,
|
|
}
|
|
remote.session.list.mockImplementation(() => Promise.resolve(ok({ items: [summary(S1)] as never[] })))
|
|
const parent = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.projections>>>()
|
|
remote.session.projections.mockImplementation(() => parent.promise)
|
|
const manager = makeManager(mock, remote)
|
|
manager.resolveTarget(address)
|
|
manager.get(S2)
|
|
|
|
manager.handleConnected()
|
|
expect(manager.get(S2).getSnapshot().subagent).toEqual({ address })
|
|
parent.resolve(ok({ asOfSeq: 0, values: { subagentCatalog: [{ createdAt: 1,
|
|
id: S2, mode: 'continuable', label: 'worker',
|
|
}] } }))
|
|
|
|
await vi.waitFor(() => {
|
|
expect(remote.session.list.mock.calls.map(([request]) => request)).toHaveLength(1)
|
|
})
|
|
await vi.waitFor(() => {
|
|
expect(remote.session.projections.mock.calls.map(([request]) => request)).toEqual([
|
|
{ sessionId: S1 },
|
|
])
|
|
})
|
|
expect(manager.get(S2).getSnapshot().subagent).toEqual({
|
|
address,
|
|
parentAvailable: true,
|
|
})
|
|
expect(manager.subagentAddress(S2)).toEqual(address)
|
|
})
|
|
})
|
|
|
|
describe('running facts without UI reminders', () => {
|
|
it('replays running status during hydration without publishing a completion marker', async ({ mock, remote }) => {
|
|
const response = Promise.withResolvers<Awaited<ReturnType<typeof remote.session.list>>>()
|
|
remote.session.list.mockReturnValueOnce(response.promise)
|
|
const manager = makeManager(mock, remote)
|
|
const refreshing = manager.refreshList()
|
|
manager.handleSessionStatus(S1, true)
|
|
manager.handleSessionStatus(S1, false)
|
|
response.resolve(ok({ items: [summary(S1)] }))
|
|
await refreshing
|
|
const entry = manager.getListSnapshot().items.find(item => item.sessionId === S1)
|
|
expect(entry).toMatchObject({ running: false })
|
|
expect(entry).not.toHaveProperty('completed')
|
|
expect(manager.getListSnapshot()).not.toHaveProperty('current')
|
|
})
|
|
})
|
|
|
|
describe('background-job mirror', () => {
|
|
const view = (over: Partial<{ id: string; status: string; label: string }> = {}) => ({
|
|
id: 'bash-1', kind: 'bash', label: 'pnpm run build', status: 'running', startedAt: 5, ...over,
|
|
})
|
|
const tasksFrame = (
|
|
sessionId: SessionId,
|
|
jobs: unknown[],
|
|
): Extract<SessionControlFrame, { type: 'jobs' }> => ({
|
|
type: 'jobs', sessionId, jobs: jobs as never,
|
|
})
|
|
|
|
it('mirrors the whole set last-wins, keyed per session, with no Session instance needed', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleControlFrame(tasksFrame(S1, [view()]))
|
|
manager.handleControlFrame(tasksFrame(S2, [view({ id: 'pwsh-1', label: 'other' })]))
|
|
const first = manager.getListSnapshot().jobsBySession
|
|
expect(first[S1]).toEqual([view()])
|
|
expect(first[S2]?.[0]?.label).toBe('other')
|
|
|
|
// Last-wins: the newer whole set replaces, it does not merge.
|
|
manager.handleControlFrame(tasksFrame(S1, [view({ status: 'completed' })]))
|
|
expect(manager.getListSnapshot().jobsBySession[S1]).toEqual([view({ status: 'completed' })])
|
|
})
|
|
|
|
it('stores an emptied set as an absent key so absence and [] read alike', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleControlFrame(tasksFrame(S1, [view()]))
|
|
expect(S1 in manager.getListSnapshot().jobsBySession).toBe(true)
|
|
manager.handleControlFrame(tasksFrame(S1, []))
|
|
expect(S1 in manager.getListSnapshot().jobsBySession).toBe(false)
|
|
})
|
|
|
|
it('clears the mirror when the next control baseline has no jobs', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleControlFrame(tasksFrame(S1, [view()]))
|
|
manager.handleControlFrame({
|
|
type: 'baseline',
|
|
value: { jobs: {}, projections: {} },
|
|
})
|
|
expect(S1 in manager.getListSnapshot().jobsBySession).toBe(false)
|
|
})
|
|
|
|
it('drops the rows when the session is removed, whichever stream lands first', ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
manager.handleSessionAdded(summary(S1, { blank: true }))
|
|
manager.handleControlFrame(tasksFrame(S1, [view()]))
|
|
manager.handleSessionRemoved(S1)
|
|
expect(S1 in manager.getListSnapshot().jobsBySession).toBe(false)
|
|
})
|
|
|
|
it('notifies list subscribers so an open header re-renders without a poll', async ({ mock, remote }) => {
|
|
const manager = makeManager(mock, remote)
|
|
const seen = vi.fn()
|
|
manager.subscribe(seen)
|
|
manager.handleControlFrame(tasksFrame(S1, [view()]))
|
|
// The notifier batches on a microtask; the frame itself is already applied.
|
|
await Promise.resolve()
|
|
expect(seen).toHaveBeenCalled()
|
|
})
|
|
})
|