mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-08-29 04:26:38 +00:00
1717 lines
72 KiB
TypeScript
1717 lines
72 KiB
TypeScript
import { afterEach, describe, expect, it, vi } from 'vitest'
|
|
import { mkdtempSync, rmSync } from 'node:fs'
|
|
import { tmpdir } from 'node:os'
|
|
import { join } from 'node:path'
|
|
import { Context } from '@deepseek-ai/cordis'
|
|
import type { Agent } from '@deepseek-ai/dsh-agent'
|
|
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
|
|
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
|
|
import { createUserMessage } from '@deepseek-ai/dsh-llm'
|
|
import { SessionId, type Session } from '@deepseek-ai/dsh-session'
|
|
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
|
|
import SubagentService from '@deepseek-ai/dsh-subagent'
|
|
import * as SubagentFork from '@deepseek-ai/dsh-subagent-fork-in-process'
|
|
import * as SubagentSpawn from '@deepseek-ai/dsh-subagent-spawn-in-process'
|
|
import { MockAdapter, textResponse } from '../../../core/agent-loop/tests/mock-adapter.ts'
|
|
import TeamService, { foldTeam, TeamError, TeamId, TeamMessageId, TeamTaskId } from '../src/index.ts'
|
|
import { TeamRuntimeLifecycle } from '../src/lifecycle.ts'
|
|
import type { TeamMemberSnapshot, TeamMessageSnapshot, TeamTaskSnapshot } from '../src/index.ts'
|
|
import { TestSessionQuery } from './test-session-query.ts'
|
|
|
|
const SIGNAL = new AbortController().signal
|
|
const roots: string[] = []
|
|
|
|
afterEach(() => {
|
|
vi.useRealTimers()
|
|
for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true })
|
|
})
|
|
|
|
/** Detached durable Team read: the service exposes views, so assertions fold the Lead log. */
|
|
function durable(agent: Agent): {
|
|
members: TeamMemberSnapshot[]
|
|
tasks: TeamTaskSnapshot[]
|
|
pendingMessages: TeamMessageSnapshot[]
|
|
} {
|
|
const state = foldTeam(agent.id, agent.session.events)
|
|
return {
|
|
members: [...state.members.values()],
|
|
tasks: [...state.tasks.values()],
|
|
pendingMessages: [...state.messages.values()].filter(message => !state.delivered.has(message.id)),
|
|
}
|
|
}
|
|
|
|
async function setup(
|
|
script: ConstructorParameters<typeof MockAdapter>[0],
|
|
config: ConstructorParameters<typeof TeamService>[1] = {},
|
|
) {
|
|
const ctx = new Context()
|
|
await mountAgentLoopTestDependencies(ctx)
|
|
const storageRoot = mkdtempSync(join(tmpdir(), 'dsh-team-'))
|
|
roots.push(storageRoot)
|
|
await ctx.plugin(JsonlSessionPersistence, { root: storageRoot })
|
|
await ctx.plugin(TestSessionQuery)
|
|
await ctx.plugin(AgentLoop, { agents: [] })
|
|
await ctx.plugin(SubagentService)
|
|
await ctx.plugin(SubagentSpawn, { providerName: 'spawn' })
|
|
await ctx.plugin(SubagentFork, { providerName: 'fork' })
|
|
const teamFiber = await ctx.plugin(TeamService, config)
|
|
const adapter = new MockAdapter(script)
|
|
ctx.llm.registerAdapter(['mock'], adapter)
|
|
const lead = ctx.agentLoop.create(SessionId('lead'), { provider: 'mock', model: 'mock' })
|
|
return { ctx, lead, adapter, storageRoot, teamFiber }
|
|
}
|
|
|
|
function content(text: string) {
|
|
return [{ type: 'text' as const, text }]
|
|
}
|
|
|
|
interface TeamServiceInternals {
|
|
readonly roster: {
|
|
readonly inFlightCreations: Set<Promise<unknown>>
|
|
checkpointInitialPrompt(childId: SessionId, messageId: string, signal: AbortSignal): Promise<void>
|
|
reconcileProvisioning(root: Agent, signal: AbortSignal): Promise<void>
|
|
liveChildrenByRoot(): Map<Agent, SessionId[]>
|
|
}
|
|
readonly mailbox: {
|
|
tryDispatch(root: Agent, message: TeamMessageSnapshot, signal: AbortSignal): Promise<boolean>
|
|
serializeDispatch(message: TeamMessageSnapshot, operation: () => Promise<boolean>): Promise<boolean>
|
|
markDelivered(root: Agent, messageId: ReturnType<typeof TeamMessageId>, targetId: SessionId): Promise<void>
|
|
}
|
|
readonly journal: {
|
|
state(root: Agent): unknown
|
|
}
|
|
disposeRuntime(): Promise<void>
|
|
recoverFor(agent: Agent): Promise<void>
|
|
scheduleRecovery(agent: Agent): void
|
|
}
|
|
|
|
/** White-box access follows the runtime owners so coverage does not widen the service API. */
|
|
function teamInternals(ctx: Context): TeamServiceInternals {
|
|
return ctx.agentTeams as unknown as TeamServiceInternals
|
|
}
|
|
|
|
function spawn(
|
|
ctx: Context,
|
|
lead: Agent,
|
|
name: string,
|
|
options: { context?: 'fresh' | 'fork'; provider?: string } = {},
|
|
) {
|
|
const context = options.context ?? 'fresh'
|
|
return ctx.agentTeams.spawnTeammate(lead, {
|
|
name,
|
|
description: `${name} responsibility`,
|
|
prompt: content(`${name} initial`),
|
|
context,
|
|
provider: options.provider ?? (context === 'fork' ? 'fork' : 'spawn'),
|
|
signal: SIGNAL,
|
|
})
|
|
}
|
|
|
|
async function waitNoAgent(ctx: Context, id: SessionId): Promise<void> {
|
|
await vi.waitFor(() => { expect(ctx.agents.get(id)).toBeUndefined() }, { timeout: 5_000 })
|
|
}
|
|
|
|
async function waitRunning(ctx: Context, id: SessionId): Promise<Agent> {
|
|
return vi.waitFor(() => {
|
|
const agent = ctx.agents.get(id)
|
|
expect(agent?.status).toBe('running')
|
|
return agent!
|
|
}, { timeout: 5_000 })
|
|
}
|
|
|
|
describe('Team identity and provisioning', () => {
|
|
it('rejects deployment limits that are not positive safe integers', async () => {
|
|
const fields = [
|
|
'maxMembers',
|
|
'maxTasks',
|
|
'maxPendingMessagesPerMember',
|
|
'maxMessageBytes',
|
|
'disposalTimeoutMs',
|
|
] as const
|
|
for (const field of fields) {
|
|
for (const value of [0, 1.5, Number.MAX_SAFE_INTEGER + 1]) {
|
|
await expect(setup([], { [field]: value })).rejects.toThrow()
|
|
}
|
|
}
|
|
})
|
|
|
|
it('supports direct-constructor defaults and recovers roots that already exist', async () => {
|
|
const ctx = new Context()
|
|
await mountAgentLoopTestDependencies(ctx)
|
|
const storageRoot = mkdtempSync(join(tmpdir(), 'dsh-team-direct-'))
|
|
roots.push(storageRoot)
|
|
await ctx.plugin(JsonlSessionPersistence, { root: storageRoot })
|
|
await ctx.plugin(AgentLoop, { agents: [] })
|
|
await ctx.plugin(SubagentService)
|
|
const lead = ctx.agentLoop.create(SessionId('preexisting-lead'), {})
|
|
const service = new TeamService(ctx)
|
|
|
|
expect(service.listMembers(lead)).toEqual([expect.objectContaining({
|
|
name: 'lead',
|
|
status: 'idle',
|
|
diagnostics: [],
|
|
})])
|
|
const provisioning = {
|
|
id: SessionId('preexisting-child'),
|
|
name: 'preexisting-worker',
|
|
description: 'preexisting responsibility',
|
|
provider: 'spawn',
|
|
context: 'fresh' as const,
|
|
phase: 'provisioning' as const,
|
|
}
|
|
lead.session.append('team/member', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
member: provisioning,
|
|
})
|
|
expect(service.listMembers(lead)[1]).toEqual(expect.objectContaining({
|
|
name: 'preexisting-worker',
|
|
status: 'provisioning',
|
|
diagnostics: [],
|
|
}))
|
|
expect(service.listMembers(lead)[1]).not.toHaveProperty('model')
|
|
await Promise.resolve()
|
|
})
|
|
|
|
it('creates fresh and fork teammates with immutable names and bounded roster size', async () => {
|
|
const { ctx, lead } = await setup([
|
|
textResponse('lead answer'),
|
|
textResponse('fork answer'),
|
|
textResponse('fresh answer'),
|
|
], { maxMembers: 2 })
|
|
lead.followup(createUserMessage({ content: content('lead turn'), source: { kind: 'user' } }))
|
|
await lead.whenIdle()
|
|
|
|
const forked = await spawn(ctx, lead, 'fork-worker', { context: 'fork' })
|
|
await waitNoAgent(ctx, forked.member.id)
|
|
const fresh = await spawn(ctx, lead, 'fresh-worker')
|
|
await waitNoAgent(ctx, fresh.member.id)
|
|
|
|
expect((await ctx.sessionPersistence.inspect(forked.member.id)).meta.seedLength).toBeGreaterThan(0)
|
|
expect((await ctx.sessionPersistence.inspect(fresh.member.id)).meta.seedLength ?? 0).toBe(0)
|
|
expect(ctx.agentTeams.listMembers(lead).map(row => [row.name, row.context, row.status])).toEqual([
|
|
['lead', undefined, 'idle'],
|
|
['fork-worker', 'fork', 'inactive'],
|
|
['fresh-worker', 'fresh', 'inactive'],
|
|
])
|
|
await expect(spawn(ctx, lead, 'third-worker')).rejects.toMatchObject({ code: 'TEAM_MEMBER_LIMIT' })
|
|
await expect(spawn(ctx, lead, 'fresh-worker')).rejects.toMatchObject({ code: 'TEAM_MEMBER_NAME_TAKEN' })
|
|
})
|
|
|
|
it('flushes the accepted child prompt before committing the active roster edge', async () => {
|
|
const { ctx, lead } = await setup([textResponse('checkpointed child answer')])
|
|
const flush = ctx.sessions.flush.bind(ctx.sessions)
|
|
const order: string[] = []
|
|
vi.spyOn(ctx.sessions, 'flush').mockImplementation(async (session) => {
|
|
if (session.id === lead.id && durable(lead).members[0]?.phase === 'active') {
|
|
order.push('lead-active')
|
|
} else if (session.id !== lead.id) {
|
|
order.push('child')
|
|
}
|
|
return flush(session)
|
|
})
|
|
|
|
const started = await spawn(ctx, lead, 'checkpoint-worker')
|
|
expect(order.indexOf('child')).toBeGreaterThanOrEqual(0)
|
|
expect(order.indexOf('child')).toBeLessThan(order.indexOf('lead-active'))
|
|
await waitNoAgent(ctx, started.member.id)
|
|
})
|
|
|
|
it('checkpoints live and detached inbox receipts and aborts an unresolved checkpoint', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
const internal = teamInternals(ctx).roster
|
|
let liveSession: Session | undefined
|
|
const liveFiber = await ctx.plugin(Object.assign(function checkpointFixture(childCtx: Context) {
|
|
liveSession = childCtx.sessions.create(SessionId('checkpoint-child'))
|
|
}, { inject: ['sessions'] }))
|
|
if (liveSession === undefined) throw new Error('checkpoint fixture did not create its Session')
|
|
const initial = createUserMessage({ content: content('checkpoint me'), source: { kind: 'user' } })
|
|
const checkpoint = internal.checkpointInitialPrompt(liveSession.id, initial.id, SIGNAL)
|
|
await Promise.resolve()
|
|
lead.inject(createUserMessage({ content: content('unrelated progress'), source: { kind: 'user' } }))
|
|
const unrelatedFiber = await ctx.plugin(Object.assign(function unrelatedCheckpointFixture(childCtx: Context) {
|
|
childCtx.sessions.create(SessionId('unrelated-checkpoint-child'))
|
|
}, { inject: ['sessions'] }))
|
|
await unrelatedFiber.dispose()
|
|
liveSession.append('agent/inbox/spliced', {
|
|
target: 'next-turn', start: 0, inserted: [initial],
|
|
})
|
|
await checkpoint
|
|
await liveFiber.dispose()
|
|
|
|
await expect(internal.checkpointInitialPrompt(liveSession.id, initial.id, SIGNAL)).resolves.toBeUndefined()
|
|
const missing = createUserMessage({ content: content('missing'), source: { kind: 'user' } })
|
|
await expect(internal.checkpointInitialPrompt(liveSession.id, missing.id, SIGNAL))
|
|
.rejects.toMatchObject({ code: 'TEAM_PROVISIONING_CONFLICT' })
|
|
|
|
let disposedSession: Session | undefined
|
|
const disposedFiber = await ctx.plugin(Object.assign(function disposedCheckpointFixture(childCtx: Context) {
|
|
disposedSession = childCtx.sessions.create(SessionId('disposed-checkpoint-child'))
|
|
}, { inject: ['sessions'] }))
|
|
if (disposedSession === undefined) throw new Error('disposed checkpoint fixture did not create its Session')
|
|
const disposed = internal.checkpointInitialPrompt(disposedSession.id, missing.id, SIGNAL)
|
|
const disposedResult = expect(disposed).rejects.toThrow('not found')
|
|
await Promise.resolve()
|
|
await disposedFiber.dispose()
|
|
await disposedResult
|
|
|
|
let abortedSession: Session | undefined
|
|
const abortedFiber = await ctx.plugin(Object.assign(function abortedCheckpointFixture(childCtx: Context) {
|
|
abortedSession = childCtx.sessions.create(SessionId('aborted-checkpoint-child'))
|
|
}, { inject: ['sessions'] }))
|
|
if (abortedSession === undefined) throw new Error('aborted checkpoint fixture did not create its Session')
|
|
const controller = new AbortController()
|
|
const aborted = internal.checkpointInitialPrompt(abortedSession.id, missing.id, controller.signal)
|
|
await Promise.resolve()
|
|
controller.abort({ kind: 'test' })
|
|
await expect(aborted).rejects.toMatchObject({ code: 'TEAM_DISPOSED' })
|
|
|
|
const errorController = new AbortController()
|
|
const errorAborted = internal.checkpointInitialPrompt(abortedSession.id, missing.id, errorController.signal)
|
|
const errorResult = expect(errorAborted).rejects.toThrow('checkpoint stopped')
|
|
await Promise.resolve()
|
|
errorController.abort(new Error('checkpoint stopped'))
|
|
await errorResult
|
|
await abortedFiber.dispose()
|
|
})
|
|
|
|
it('drains an accepted child when its initial durability checkpoint fails', async () => {
|
|
const { ctx, lead } = await setup(['hang'])
|
|
vi.spyOn(teamInternals(ctx).roster, 'checkpointInitialPrompt')
|
|
.mockRejectedValueOnce(new Error('checkpoint failed'))
|
|
|
|
await expect(spawn(ctx, lead, 'checkpoint-failure')).rejects.toThrow('checkpoint failed')
|
|
const member = durable(lead).members[0]
|
|
expect(member).toMatchObject({ phase: 'failed', error: 'checkpoint failed' })
|
|
if (member !== undefined) await waitNoAgent(ctx, member.id)
|
|
})
|
|
|
|
it('records failed provisioning durably, reserves its name, and counts it against the limit', async () => {
|
|
const { ctx, lead } = await setup([], { maxMembers: 1 })
|
|
await expect(spawn(ctx, lead, 'failed-worker', { provider: 'missing' })).rejects.toThrow()
|
|
|
|
expect(ctx.agentTeams.listMembers(lead)[1]).toMatchObject({
|
|
name: 'failed-worker',
|
|
status: 'failed',
|
|
provider: 'missing',
|
|
})
|
|
await expect(spawn(ctx, lead, 'failed-worker')).rejects.toMatchObject({ code: 'TEAM_MEMBER_NAME_TAKEN' })
|
|
await expect(spawn(ctx, lead, 'other-worker')).rejects.toMatchObject({ code: 'TEAM_MEMBER_LIMIT' })
|
|
})
|
|
|
|
it('records non-Error provider failures and contains a reversed provisioning settlement race', async () => {
|
|
const first = await setup([])
|
|
vi.spyOn(first.ctx.subagents, 'startContinuable').mockRejectedValueOnce('string provider failure')
|
|
await expect(spawn(first.ctx, first.lead, 'string-failure')).rejects.toBe('string provider failure')
|
|
expect(first.ctx.agentTeams.listMembers(first.lead)[1]).toMatchObject({
|
|
status: 'failed',
|
|
diagnostics: ['string provider failure'],
|
|
})
|
|
await expect(first.ctx.agentTeams.sendMessage(first.lead, {
|
|
target: 'string-failure', content: content('cannot deliver'), delivery: 'quiet', signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_MEMBER_NOT_FOUND' })
|
|
|
|
const second = await setup([])
|
|
vi.spyOn(second.ctx.subagents, 'startContinuable').mockImplementationOnce(async () => {
|
|
const provisioning = durable(second.lead).members[0]
|
|
if (provisioning === undefined) throw new Error('missing provisioning edge')
|
|
second.lead.session.append('team/member', {
|
|
version: 1,
|
|
teamId: TeamId(second.lead.id),
|
|
member: { ...provisioning, phase: 'active' },
|
|
})
|
|
await second.ctx.sessions.flush(second.lead.session)
|
|
throw new Error('creator failed after recovery settled active')
|
|
})
|
|
await expect(spawn(second.ctx, second.lead, 'reverse-race')).rejects.toBeInstanceOf(AggregateError)
|
|
expect(durable(second.lead).members[0]?.phase).toBe('active')
|
|
})
|
|
|
|
it('cleans up a child when recovery settles its provisioning record first', async () => {
|
|
const { ctx, lead } = await setup(['hang'])
|
|
const start = ctx.subagents.startContinuable.bind(ctx.subagents)
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
let childId: SessionId | undefined
|
|
vi.spyOn(ctx.subagents, 'startContinuable').mockImplementation(async (spec) => {
|
|
childId = spec.childId
|
|
entered.resolve(undefined)
|
|
await release.promise
|
|
return start(spec)
|
|
})
|
|
|
|
const spawning = spawn(ctx, lead, 'racing-worker')
|
|
const rejected = expect(spawning).rejects.toMatchObject({ code: 'TEAM_PROVISIONING_CONFLICT' })
|
|
await entered.promise
|
|
await teamInternals(ctx).roster.reconcileProvisioning(lead, SIGNAL)
|
|
expect(durable(lead).members[0]?.phase).toBe('failed')
|
|
|
|
release.resolve(undefined)
|
|
await rejected
|
|
if (childId === undefined) throw new Error('reserved child id was not observed')
|
|
await waitNoAgent(ctx, childId)
|
|
})
|
|
|
|
it('handles a continuation that settles before the active roster view or conflict cleanup lookup', async () => {
|
|
const first = await setup([])
|
|
vi.spyOn(teamInternals(first.ctx).roster, 'checkpointInitialPrompt').mockResolvedValueOnce()
|
|
vi.spyOn(first.ctx.subagents, 'startContinuable').mockImplementationOnce(async spec => ({
|
|
childId: spec.childId!,
|
|
messageId: createUserMessage({ content: content('accepted'), source: { kind: 'user' } }).id,
|
|
}))
|
|
const inactive = await spawn(first.ctx, first.lead, 'instant-worker')
|
|
expect(inactive.member).toMatchObject({ status: 'inactive', diagnostics: [] })
|
|
expect(inactive.member).not.toHaveProperty('model')
|
|
|
|
const second = await setup([])
|
|
vi.spyOn(teamInternals(second.ctx).roster, 'checkpointInitialPrompt').mockResolvedValueOnce()
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
vi.spyOn(second.ctx.subagents, 'startContinuable').mockImplementationOnce(async (spec) => {
|
|
entered.resolve(undefined)
|
|
await release.promise
|
|
return {
|
|
childId: spec.childId!,
|
|
messageId: createUserMessage({ content: content('accepted'), source: { kind: 'user' } }).id,
|
|
}
|
|
})
|
|
const spawning = spawn(second.ctx, second.lead, 'instant-conflict')
|
|
const rejected = expect(spawning).rejects.toMatchObject({ code: 'TEAM_PROVISIONING_CONFLICT' })
|
|
await entered.promise
|
|
await teamInternals(second.ctx).roster.reconcileProvisioning(second.lead, SIGNAL)
|
|
release.resolve(undefined)
|
|
await rejected
|
|
})
|
|
|
|
it('validates names and permits only the Lead to create or interrupt teammates', async () => {
|
|
const { ctx, lead } = await setup(['hang'])
|
|
for (const name of ['Lead', 'lead', '-bad', 'bad-', 'bad_name', 'x'.repeat(65)]) {
|
|
await expect(spawn(ctx, lead, name)).rejects.toMatchObject({ code: 'TEAM_INVALID_MEMBER_NAME' })
|
|
}
|
|
const started = await spawn(ctx, lead, 'worker')
|
|
const worker = await waitRunning(ctx, started.member.id)
|
|
await expect(spawn(ctx, worker, 'nested')).rejects.toMatchObject({ code: 'TEAM_LEAD_REQUIRED' })
|
|
expect(() => ctx.agentTeams.interrupt(worker, 'worker')).toThrow(expect.objectContaining({ code: 'TEAM_LEAD_REQUIRED' }))
|
|
expect(ctx.agentTeams.interrupt(lead, 'worker')).toEqual({ previousStatus: 'running' })
|
|
await waitNoAgent(ctx, worker.id)
|
|
expect(ctx.agentTeams.interrupt(lead, 'worker')).toEqual({ previousStatus: 'inactive' })
|
|
expect(() => ctx.agentTeams.interrupt(lead, 'lead')).toThrow(expect.objectContaining({ code: 'TEAM_INVALID_TARGET' }))
|
|
})
|
|
|
|
it('validates teammate text fields and pre-provisioning cancellation', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
await expect(ctx.agentTeams.spawnTeammate(lead, {
|
|
name: 'empty-description',
|
|
description: ' ',
|
|
prompt: content('unused'),
|
|
context: 'fresh',
|
|
provider: 'spawn',
|
|
signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
await expect(ctx.agentTeams.spawnTeammate(lead, {
|
|
name: 'empty-provider',
|
|
description: 'valid description',
|
|
prompt: content('unused'),
|
|
context: 'fresh',
|
|
provider: ' ',
|
|
signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
const controller = new AbortController()
|
|
controller.abort(new TeamError('cancelled before provisioning', 'TEST_CANCELLED'))
|
|
await expect(ctx.agentTeams.spawnTeammate(lead, {
|
|
name: 'cancelled-worker',
|
|
description: 'never provisioned',
|
|
prompt: content('unused'),
|
|
context: 'fresh',
|
|
provider: 'spawn',
|
|
signal: controller.signal,
|
|
})).rejects.toMatchObject({ code: 'TEST_CANCELLED' })
|
|
expect(durable(lead).members).toEqual([])
|
|
})
|
|
|
|
it('treats an ordinary fork as a new Root Team and filters inherited Team state', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
await ctx.agentTeams.createTask(lead, { subject: 'parent task', description: 'belongs to parent' })
|
|
const handle = await ctx.agents.create({
|
|
sessionId: SessionId('ordinary-fork'),
|
|
seed: lead.session.events,
|
|
meta: { parentSession: lead.id, seedLength: lead.session.seq },
|
|
agentOptions: { provider: 'mock', model: 'mock' },
|
|
})
|
|
|
|
expect(ctx.agentTeams.membership(handle.agent)).toMatchObject({
|
|
id: TeamId(handle.agent.id),
|
|
role: 'lead',
|
|
name: 'lead',
|
|
})
|
|
expect(durable(handle.agent)).toMatchObject({ members: [], tasks: [], pendingMessages: [] })
|
|
await handle.dispose()
|
|
})
|
|
|
|
it('rejects stale Agent identities and non-Team subagent children', async () => {
|
|
const { ctx, lead } = await setup([textResponse('done')])
|
|
const started = await ctx.subagents.startContinuable({
|
|
provider: 'spawn',
|
|
label: 'ordinary worker',
|
|
request: { prompt: content('ordinary'), parent: lead },
|
|
signal: SIGNAL,
|
|
})
|
|
const live = ctx.agents.get(started.childId)
|
|
if (live !== undefined) expect(ctx.agentTeams.tryMembership(live)).toBeUndefined()
|
|
await waitNoAgent(ctx, started.childId)
|
|
expect(() => ctx.agentTeams.membership(lead)).not.toThrow()
|
|
|
|
const impostor = { ...lead } as Agent
|
|
expect(ctx.agentTeams.tryMembership(impostor)).toBeUndefined()
|
|
expect(() => ctx.agentTeams.membership(impostor)).toThrow(expect.objectContaining({ code: 'TEAM_NOT_MEMBER' }))
|
|
|
|
const orphanRoot = await ctx.agents.create({
|
|
sessionId: SessionId('orphan-ordinary-root'),
|
|
meta: { parentSession: SessionId('absent-parent') },
|
|
agentOptions: { provider: 'mock', model: 'mock' },
|
|
})
|
|
expect(ctx.agentTeams.membership(orphanRoot.agent)).toMatchObject({ role: 'lead', name: 'lead' })
|
|
await orphanRoot.dispose()
|
|
})
|
|
|
|
it('does not reinterpret an orphaned provider child or malformed parent stream as a Team root', async () => {
|
|
const first = await setup([textResponse('ordinary child done')])
|
|
const parent = await first.ctx.agents.create({
|
|
sessionId: SessionId('temporary-parent'),
|
|
agentOptions: { provider: 'mock', model: 'mock' },
|
|
})
|
|
const started = await first.ctx.subagents.startContinuable({
|
|
provider: 'spawn',
|
|
label: 'ordinary child',
|
|
request: { prompt: content('finish'), parent: parent.agent },
|
|
signal: SIGNAL,
|
|
})
|
|
await waitNoAgent(first.ctx, started.childId)
|
|
await parent.dispose()
|
|
const orphan = await first.ctx.agents.resume({
|
|
resumeSessionId: started.childId,
|
|
agentOptions: { provider: 'mock', model: 'mock' },
|
|
})
|
|
expect(first.ctx.agentTeams.tryMembership(orphan.agent)).toBeUndefined()
|
|
expect(teamInternals(first.ctx).roster.liveChildrenByRoot()).toEqual(new Map())
|
|
await orphan.dispose()
|
|
|
|
const second = await setup([])
|
|
const child = await second.ctx.agents.create({
|
|
sessionId: SessionId('malformed-parent-child'),
|
|
meta: { parentSession: second.lead.id },
|
|
agentOptions: { provider: 'mock', model: 'mock' },
|
|
})
|
|
const journal = teamInternals(second.ctx).journal
|
|
const state = journal.state.bind(journal)
|
|
journal.state = () => { throw new Error('malformed Team stream') }
|
|
expect(second.ctx.agentTeams.tryMembership(child.agent)).toBeUndefined()
|
|
journal.state = state
|
|
await child.dispose()
|
|
})
|
|
})
|
|
|
|
describe('Team shared task DAG', () => {
|
|
it('fails loudly when the durable numeric task id space is exhausted', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
const id = TeamTaskId(`task-${Number.MAX_SAFE_INTEGER}`)
|
|
lead.session.append('team/task', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
task: {
|
|
id,
|
|
revision: 1,
|
|
subject: 'last numeric task',
|
|
description: 'occupies the final safe numeric task id',
|
|
status: 'pending',
|
|
blockedBy: [],
|
|
writeScopes: [],
|
|
},
|
|
})
|
|
await ctx.sessions.flush(lead.session)
|
|
|
|
await expect(ctx.agentTeams.createTask(lead, {
|
|
subject: 'cannot allocate',
|
|
description: 'no safe numeric task id remains',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_LIMIT' })
|
|
})
|
|
|
|
it('bounds non-deleted tasks while retaining deleted task ids as tombstones', async () => {
|
|
const { ctx, lead } = await setup([], { maxTasks: 1 })
|
|
const first = await ctx.agentTeams.createTask(lead, { subject: 'first', description: 'first task' })
|
|
await expect(ctx.agentTeams.createTask(lead, { subject: 'overflow', description: 'overflow task' }))
|
|
.rejects.toMatchObject({ code: 'TEAM_TASK_LIMIT' })
|
|
|
|
const deleted = await ctx.agentTeams.updateTask(lead, {
|
|
taskId: first.id,
|
|
expectedRevision: first.revision,
|
|
action: 'delete',
|
|
})
|
|
const second = await ctx.agentTeams.createTask(lead, { subject: 'second', description: 'second task' })
|
|
expect(deleted.status).toBe('deleted')
|
|
expect(second.id).toBe(TeamTaskId('task-2'))
|
|
expect(ctx.agentTeams.getTask(lead, first.id).status).toBe('deleted')
|
|
expect(ctx.agentTeams.listTasks(lead).map(task => task.id)).toEqual([second.id])
|
|
})
|
|
|
|
it('enforces CAS, ownership, dependencies, transitions, and write-scope warnings', async () => {
|
|
const { ctx, lead } = await setup(['hang', 'hang'])
|
|
const firstMember = await spawn(ctx, lead, 'alpha')
|
|
const alpha = await waitRunning(ctx, firstMember.member.id)
|
|
const secondMember = await spawn(ctx, lead, 'beta')
|
|
const beta = await waitRunning(ctx, secondMember.member.id)
|
|
|
|
const first = await ctx.agentTeams.createTask(alpha, {
|
|
subject: 'first',
|
|
description: 'first task',
|
|
writeScopes: ['src', './src/', 'src'],
|
|
})
|
|
const second = await ctx.agentTeams.createTask(beta, {
|
|
subject: 'second',
|
|
description: 'second task',
|
|
blockedBy: [first.id],
|
|
writeScopes: ['src/feature'],
|
|
})
|
|
expect(first.writeScopes).toEqual(['src'])
|
|
await expect(ctx.agentTeams.updateTask(beta, {
|
|
taskId: second.id,
|
|
expectedRevision: second.revision,
|
|
action: 'claim',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_BLOCKED' })
|
|
|
|
const claimed = await ctx.agentTeams.updateTask(alpha, {
|
|
taskId: first.id,
|
|
expectedRevision: first.revision,
|
|
action: 'claim',
|
|
})
|
|
await expect(ctx.agentTeams.updateTask(beta, {
|
|
taskId: first.id,
|
|
expectedRevision: claimed.revision,
|
|
action: 'claim',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_ALREADY_CLAIMED' })
|
|
expect(ctx.agentTeams.getTask(beta, second.id)).toMatchObject({
|
|
ready: false,
|
|
writeScopeWarnings: [`write scopes overlap with ${first.id}`],
|
|
})
|
|
await expect(ctx.agentTeams.updateTask(beta, {
|
|
taskId: first.id,
|
|
expectedRevision: claimed.revision,
|
|
action: 'edit',
|
|
subject: 'stolen',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_UNAUTHORIZED' })
|
|
await expect(ctx.agentTeams.updateTask(alpha, {
|
|
taskId: first.id,
|
|
expectedRevision: first.revision,
|
|
action: 'complete',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_STALE_REVISION' })
|
|
|
|
const completed = await ctx.agentTeams.updateTask(alpha, {
|
|
taskId: first.id,
|
|
expectedRevision: claimed.revision,
|
|
action: 'complete',
|
|
})
|
|
expect(completed.status).toBe('completed')
|
|
expect(ctx.agentTeams.getTask(beta, second.id).ready).toBe(true)
|
|
const secondClaim = await ctx.agentTeams.updateTask(beta, {
|
|
taskId: second.id,
|
|
expectedRevision: second.revision,
|
|
action: 'claim',
|
|
})
|
|
const released = await ctx.agentTeams.updateTask(beta, {
|
|
taskId: second.id,
|
|
expectedRevision: secondClaim.revision,
|
|
action: 'release',
|
|
})
|
|
expect(released).toMatchObject({ status: 'pending', ready: true })
|
|
expect('ownerId' in released).toBe(false)
|
|
|
|
ctx.agentTeams.interrupt(lead, 'alpha')
|
|
ctx.agentTeams.interrupt(lead, 'beta')
|
|
await Promise.all([waitNoAgent(ctx, alpha.id), waitNoAgent(ctx, beta.id)])
|
|
})
|
|
|
|
it('rejects malformed scopes and every invalid dependency relation', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
const first = await ctx.agentTeams.createTask(lead, { subject: 'one', description: 'one' })
|
|
const second = await ctx.agentTeams.createTask(lead, {
|
|
subject: 'two', description: 'two', blockedBy: [first.id],
|
|
})
|
|
await expect(ctx.agentTeams.createTask(lead, {
|
|
subject: 'bad', description: 'bad', blockedBy: [TeamTaskId('missing')],
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_NOT_FOUND' })
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: first.id,
|
|
expectedRevision: first.revision,
|
|
action: 'set_dependencies',
|
|
blockedBy: [second.id],
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_DEPENDENCY_CYCLE' })
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: first.id,
|
|
expectedRevision: first.revision,
|
|
action: 'set_dependencies',
|
|
blockedBy: [first.id],
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_DEPENDENCY_CYCLE' })
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: first.id,
|
|
expectedRevision: first.revision,
|
|
action: 'set_dependencies',
|
|
blockedBy: [second.id, second.id],
|
|
})).rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
for (const scope of ['', '.', '..', '/root', 'C:\\root', 'C:root', 'a//b', 'a/../b']) {
|
|
await expect(ctx.agentTeams.createTask(lead, {
|
|
subject: 'scope', description: 'scope', writeScopes: [scope],
|
|
})).rejects.toMatchObject({ code: 'TEAM_INVALID_WRITE_SCOPE' })
|
|
}
|
|
})
|
|
|
|
it('rejects incomplete mutations, invalid transitions, and deletion of a live blocker', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
await expect(ctx.agentTeams.createTask(lead, { subject: ' ', description: 'invalid' }))
|
|
.rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
await expect(ctx.agentTeams.createTask(lead, { subject: 'invalid', description: '' }))
|
|
.rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
await expect(ctx.agentTeams.createTask(lead, { subject: 'x'.repeat(201), description: 'too long' }))
|
|
.rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
const blocker = await ctx.agentTeams.createTask(lead, { subject: 'blocker', description: 'blocker' })
|
|
await ctx.agentTeams.createTask(lead, {
|
|
subject: 'dependent', description: 'dependent', blockedBy: [blocker.id],
|
|
})
|
|
expect(() => ctx.agentTeams.getTask(lead, TeamTaskId('missing')))
|
|
.toThrow(expect.objectContaining({ code: 'TEAM_TASK_NOT_FOUND' }))
|
|
for (const action of ['release', 'complete', 'reopen'] as const) {
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: blocker.id,
|
|
expectedRevision: blocker.revision,
|
|
action,
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_INVALID_TRANSITION' })
|
|
}
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: blocker.id,
|
|
expectedRevision: blocker.revision,
|
|
action: 'edit',
|
|
})).rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: blocker.id,
|
|
expectedRevision: blocker.revision,
|
|
action: 'set_dependencies',
|
|
})).rejects.toMatchObject({ code: 'TEAM_INVALID_ARGUMENT' })
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: blocker.id,
|
|
expectedRevision: blocker.revision,
|
|
action: 'delete',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_HAS_DEPENDENTS' })
|
|
})
|
|
|
|
it('supports Lead reassignment, completion, reopen, and deletion permissions', async () => {
|
|
const { ctx, lead } = await setup(['hang'])
|
|
const started = await spawn(ctx, lead, 'owner')
|
|
const owner = await waitRunning(ctx, started.member.id)
|
|
const task = await ctx.agentTeams.createTask(owner, { subject: 'lifecycle', description: 'lifecycle' })
|
|
const assigned = await ctx.agentTeams.updateTask(lead, {
|
|
taskId: task.id,
|
|
expectedRevision: task.revision,
|
|
action: 'reassign',
|
|
owner: 'owner',
|
|
})
|
|
await expect(ctx.agentTeams.updateTask(owner, {
|
|
taskId: task.id,
|
|
expectedRevision: assigned.revision,
|
|
action: 'reassign',
|
|
owner: 'lead',
|
|
})).rejects.toMatchObject({ code: 'TEAM_LEAD_REQUIRED' })
|
|
const complete = await ctx.agentTeams.updateTask(owner, {
|
|
taskId: task.id,
|
|
expectedRevision: assigned.revision,
|
|
action: 'complete',
|
|
})
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: task.id,
|
|
expectedRevision: complete.revision,
|
|
action: 'reassign',
|
|
owner: 'lead',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_INVALID_TRANSITION' })
|
|
const reopened = await ctx.agentTeams.updateTask(owner, {
|
|
taskId: task.id,
|
|
expectedRevision: complete.revision,
|
|
action: 'reopen',
|
|
})
|
|
const claimed = await ctx.agentTeams.updateTask(owner, {
|
|
taskId: task.id,
|
|
expectedRevision: reopened.revision,
|
|
action: 'claim',
|
|
})
|
|
const deleted = await ctx.agentTeams.updateTask(owner, {
|
|
taskId: task.id,
|
|
expectedRevision: claimed.revision,
|
|
action: 'delete',
|
|
})
|
|
expect(deleted.status).toBe('deleted')
|
|
expect(ctx.agentTeams.listTasks(lead)).toEqual([])
|
|
await expect(ctx.agentTeams.updateTask(owner, {
|
|
taskId: task.id,
|
|
expectedRevision: deleted.revision,
|
|
action: 'edit',
|
|
subject: 'late',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_DELETED' })
|
|
ctx.agentTeams.interrupt(lead, 'owner')
|
|
await waitNoAgent(ctx, owner.id)
|
|
})
|
|
|
|
it('covers partial edits, Lead ownership, unassignment, and blocked reassignment', async () => {
|
|
const { ctx, lead } = await setup(['hang'])
|
|
const started = await spawn(ctx, lead, 'editor')
|
|
const editor = await waitRunning(ctx, started.member.id)
|
|
const blocker = await ctx.agentTeams.createTask(lead, { subject: 'blocker', description: 'blocker' })
|
|
const task = await ctx.agentTeams.createTask(lead, {
|
|
subject: 'draft',
|
|
description: 'draft description',
|
|
blockedBy: [blocker.id],
|
|
})
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: TeamTaskId('missing-update'), expectedRevision: 1, action: 'delete',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_NOT_FOUND' })
|
|
await expect(ctx.agentTeams.updateTask(lead, {
|
|
taskId: task.id, expectedRevision: task.revision, action: 'reassign', owner: 'editor',
|
|
})).rejects.toMatchObject({ code: 'TEAM_TASK_BLOCKED' })
|
|
|
|
const leadClaim = await ctx.agentTeams.updateTask(lead, {
|
|
taskId: blocker.id, expectedRevision: blocker.revision, action: 'claim',
|
|
})
|
|
expect(leadClaim.ownerName).toBe('lead')
|
|
const completedBlocker = await ctx.agentTeams.updateTask(lead, {
|
|
taskId: blocker.id, expectedRevision: leadClaim.revision, action: 'complete',
|
|
})
|
|
expect(completedBlocker.status).toBe('completed')
|
|
const assigned = await ctx.agentTeams.updateTask(lead, {
|
|
taskId: task.id, expectedRevision: task.revision, action: 'reassign', owner: 'editor',
|
|
})
|
|
const subject = await ctx.agentTeams.updateTask(editor, {
|
|
taskId: task.id, expectedRevision: assigned.revision, action: 'edit', subject: 'edited subject',
|
|
})
|
|
const description = await ctx.agentTeams.updateTask(editor, {
|
|
taskId: task.id,
|
|
expectedRevision: subject.revision,
|
|
action: 'edit',
|
|
description: 'edited description',
|
|
})
|
|
const scopes = await ctx.agentTeams.updateTask(editor, {
|
|
taskId: task.id,
|
|
expectedRevision: description.revision,
|
|
action: 'edit',
|
|
writeScopes: ['src/nested'],
|
|
})
|
|
expect(scopes).toMatchObject({
|
|
subject: 'edited subject',
|
|
description: 'edited description',
|
|
writeScopes: ['src/nested'],
|
|
})
|
|
const unassigned = await ctx.agentTeams.updateTask(lead, {
|
|
taskId: task.id, expectedRevision: scopes.revision, action: 'reassign', owner: ' ',
|
|
})
|
|
expect(unassigned).toMatchObject({ status: 'pending' })
|
|
expect('ownerId' in unassigned).toBe(false)
|
|
|
|
const broad = await ctx.agentTeams.createTask(lead, {
|
|
subject: 'broad scope', description: 'broad scope', writeScopes: ['src'],
|
|
})
|
|
const narrow = await ctx.agentTeams.createTask(lead, {
|
|
subject: 'narrow scope', description: 'narrow scope', writeScopes: ['src/nested'],
|
|
})
|
|
const disjoint = await ctx.agentTeams.createTask(lead, {
|
|
subject: 'disjoint scope', description: 'disjoint scope', writeScopes: ['docs'],
|
|
})
|
|
await ctx.agentTeams.updateTask(lead, {
|
|
taskId: broad.id, expectedRevision: broad.revision, action: 'claim',
|
|
})
|
|
await ctx.agentTeams.updateTask(lead, {
|
|
taskId: narrow.id, expectedRevision: narrow.revision, action: 'claim',
|
|
})
|
|
await ctx.agentTeams.updateTask(lead, {
|
|
taskId: disjoint.id, expectedRevision: disjoint.revision, action: 'claim',
|
|
})
|
|
expect(ctx.agentTeams.getTask(lead, broad.id).writeScopeWarnings)
|
|
.toEqual([`write scopes overlap with ${narrow.id}`])
|
|
|
|
ctx.agentTeams.interrupt(lead, 'editor')
|
|
await waitNoAgent(ctx, editor.id)
|
|
})
|
|
})
|
|
|
|
describe('Team mailbox and waiting', () => {
|
|
it('injects a quiet message addressed to the Lead and checkpoints its receipt', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
const message: TeamMessageSnapshot = {
|
|
id: TeamMessageId('quiet-lead-message'),
|
|
senderId: SessionId('team-worker'),
|
|
senderName: 'worker',
|
|
targetId: lead.id,
|
|
delivery: 'quiet',
|
|
content: content('quiet report'),
|
|
}
|
|
lead.session.append('team/message/queued', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
message,
|
|
})
|
|
|
|
await expect(teamInternals(ctx).mailbox.tryDispatch(lead, message, SIGNAL)).resolves.toBe(true)
|
|
expect(lead.inbox.nextStep.some(input => input.source.kind === 'team-message'
|
|
&& input.source.messageId === message.id)).toBe(true)
|
|
expect(durable(lead).pendingMessages).toEqual([])
|
|
})
|
|
|
|
it('acknowledges waking messages persisted by a busy Lead before model claim', async () => {
|
|
const { ctx, lead, teamFiber } = await setup(['hang', 'hang'], { maxPendingMessagesPerMember: 1 })
|
|
const started = await spawn(ctx, lead, 'lead-reporter')
|
|
const reporter = await waitRunning(ctx, started.member.id)
|
|
lead.followup(createUserMessage({ content: content('keep the Lead busy'), source: { kind: 'user' } }))
|
|
await waitRunning(ctx, lead.id)
|
|
|
|
const first = await ctx.agentTeams.sendMessage(reporter, {
|
|
target: 'lead', content: content('first wakeup report'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
const second = await ctx.agentTeams.sendMessage(reporter, {
|
|
target: 'lead', content: content('second wakeup report'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
expect([first.status, second.status]).toEqual(['accepted', 'accepted'])
|
|
expect(lead.status).toBe('running')
|
|
expect(durable(lead).pendingMessages).toEqual([])
|
|
|
|
const messageIds = new Set([first.messageId, second.messageId])
|
|
const persisted = await ctx.sessionPersistence.inspect(lead.id)
|
|
const receiptOrder = persisted.events.flatMap((event) => {
|
|
if (event.type === 'agent/inbox/spliced' && event.data.inserted.some(message =>
|
|
message.source.kind === 'team-message' && messageIds.has(message.source.messageId))) {
|
|
return ['agent/inbox/spliced']
|
|
}
|
|
if (event.type === 'team/message/delivered' && messageIds.has(event.data.messageId)) {
|
|
return ['team/message/delivered']
|
|
}
|
|
return []
|
|
})
|
|
expect(receiptOrder).toEqual([
|
|
'agent/inbox/spliced',
|
|
'team/message/delivered',
|
|
'agent/inbox/spliced',
|
|
'team/message/delivered',
|
|
])
|
|
|
|
const receiptCount = lead.session.events.filter(event => event.type === 'agent/inbox/spliced'
|
|
&& event.data.inserted.some(message => message.source.kind === 'team-message'
|
|
&& messageIds.has(message.source.messageId))).length
|
|
await teamFiber.dispose()
|
|
await ctx.plugin(TeamService, { maxPendingMessagesPerMember: 1 })
|
|
await vi.waitFor(() => { expect(durable(lead).pendingMessages).toEqual([]) })
|
|
expect(lead.session.events.filter(event => event.type === 'agent/inbox/spliced'
|
|
&& event.data.inserted.some(message => message.source.kind === 'team-message'
|
|
&& messageIds.has(message.source.messageId)))).toHaveLength(receiptCount)
|
|
|
|
lead.cancel({ kind: 'parent' })
|
|
await lead.whenIdle()
|
|
})
|
|
|
|
it('flushes a live pending receipt before acknowledgement without inserting a duplicate', async () => {
|
|
const { ctx, lead } = await setup(['hang'])
|
|
const started = await spawn(ctx, lead, 'pending-target')
|
|
const target = await waitRunning(ctx, started.member.id)
|
|
const immediate = await ctx.agentTeams.sendMessage(lead, {
|
|
target: 'pending-target',
|
|
content: content('live quiet receipt'),
|
|
delivery: 'quiet',
|
|
signal: SIGNAL,
|
|
})
|
|
expect(immediate.status).toBe('accepted')
|
|
expect(durable(lead).pendingMessages).toEqual([])
|
|
expect(target.inbox.nextStep.some(item => item.source.kind === 'team-message'
|
|
&& item.source.messageId === immediate.messageId)).toBe(true)
|
|
|
|
const message: TeamMessageSnapshot = {
|
|
id: TeamMessageId('live-pending-message'),
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
targetId: target.id,
|
|
delivery: 'quiet',
|
|
content: content('durable pending receipt'),
|
|
}
|
|
lead.session.append('team/message/queued', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
message,
|
|
})
|
|
await ctx.sessions.flush(lead.session)
|
|
target.inject(createUserMessage({
|
|
content: content('durable pending receipt'),
|
|
source: {
|
|
kind: 'team-message',
|
|
teamId: TeamId(lead.id),
|
|
messageId: message.id,
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
},
|
|
}))
|
|
|
|
const flush = ctx.sessions.flush.bind(ctx.sessions)
|
|
const flushed: SessionId[] = []
|
|
const flushSpy = vi.spyOn(ctx.sessions, 'flush').mockImplementation(async (session) => {
|
|
flushed.push(session.id)
|
|
return flush(session)
|
|
})
|
|
const delivered = await teamInternals(ctx).mailbox.tryDispatch(lead, message, SIGNAL)
|
|
|
|
expect(delivered).toBe(true)
|
|
expect(flushed.slice(0, 2)).toEqual([target.id, lead.id])
|
|
expect(target.inbox.nextStep.filter(item => item.source.kind === 'team-message'
|
|
&& item.source.messageId === message.id)).toHaveLength(1)
|
|
expect(durable(lead).pendingMessages).toEqual([])
|
|
|
|
const disappearing: TeamMessageSnapshot = {
|
|
...message,
|
|
id: TeamMessageId('disappearing-pending-message'),
|
|
content: content('canceled before checkpoint'),
|
|
}
|
|
lead.session.append('team/message/queued', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
message: disappearing,
|
|
})
|
|
await flush(lead.session)
|
|
const disappearingInput = createUserMessage({
|
|
content: content('canceled before checkpoint'),
|
|
source: {
|
|
kind: 'team-message',
|
|
teamId: TeamId(lead.id),
|
|
messageId: disappearing.id,
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
},
|
|
})
|
|
target.inject(disappearingInput)
|
|
flushSpy.mockImplementationOnce(async (session) => {
|
|
target.inbox.remove(disappearingInput.id)
|
|
return flush(session)
|
|
})
|
|
await expect(teamInternals(ctx).mailbox.tryDispatch(lead, disappearing, SIGNAL)).resolves.toBe(false)
|
|
expect(durable(lead).pendingMessages.map(pending => pending.id)).toEqual([disappearing.id])
|
|
|
|
ctx.agentTeams.interrupt(lead, 'pending-target')
|
|
target.cancel({ kind: 'parent' })
|
|
await waitNoAgent(ctx, target.id)
|
|
})
|
|
|
|
it('acknowledges waking messages accepted by a busy target inbox', async () => {
|
|
const { ctx, lead } = await setup(['hang'], { maxPendingMessagesPerMember: 1 })
|
|
const started = await spawn(ctx, lead, 'busy-target')
|
|
const target = await waitRunning(ctx, started.member.id)
|
|
const flush = ctx.sessions.flush.bind(ctx.sessions)
|
|
const flushed: SessionId[] = []
|
|
vi.spyOn(ctx.sessions, 'flush').mockImplementation(async (session) => {
|
|
flushed.push(session.id)
|
|
return flush(session)
|
|
})
|
|
|
|
const first = await ctx.agentTeams.sendMessage(lead, {
|
|
target: 'busy-target', content: content('first waking message'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
|
|
expect(first.status).toBe('accepted')
|
|
expect(flushed).toEqual([lead.id, target.id, lead.id])
|
|
expect(durable(lead).pendingMessages).toEqual([])
|
|
expect(target.inbox.nextTurn.some(message => message.source.kind === 'team-message'
|
|
&& message.source.messageId === first.messageId)).toBe(true)
|
|
|
|
flushed.length = 0
|
|
const second = await ctx.agentTeams.sendMessage(lead, {
|
|
target: 'busy-target', content: content('second waking message'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
|
|
expect(second.status).toBe('accepted')
|
|
expect(flushed).toEqual([lead.id, target.id, lead.id])
|
|
expect(durable(lead).pendingMessages).toEqual([])
|
|
expect(target.inbox.nextTurn.filter(message => message.source.kind === 'team-message'
|
|
&& (message.source.messageId === first.messageId || message.source.messageId === second.messageId)))
|
|
.toHaveLength(2)
|
|
|
|
ctx.agentTeams.interrupt(lead, 'busy-target')
|
|
target.cancel({ kind: 'parent' })
|
|
await waitNoAgent(ctx, target.id)
|
|
})
|
|
|
|
it('serializes concurrent waking delivery admission for one target', async () => {
|
|
const { ctx, lead } = await setup([textResponse('target initial')])
|
|
const target = await spawn(ctx, lead, 'ordered-target')
|
|
await waitNoAgent(ctx, target.member.id)
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
const admitted: string[] = []
|
|
vi.spyOn(ctx.subagents, 'followup').mockImplementation(async (_parent, _childId, blocks) => {
|
|
const last = blocks.at(-1)
|
|
const text = last?.type === 'text' ? last.text : ''
|
|
admitted.push(text)
|
|
if (text === 'first waking') {
|
|
entered.resolve(undefined)
|
|
await release.promise
|
|
}
|
|
return createUserMessage({ content: blocks, source: { kind: 'user' } }).id
|
|
})
|
|
|
|
const first = ctx.agentTeams.sendMessage(lead, {
|
|
target: 'ordered-target', content: content('first waking'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
await entered.promise
|
|
let secondSettled = false
|
|
const second = ctx.agentTeams.sendMessage(lead, {
|
|
target: 'ordered-target', content: content('second waking'), delivery: 'wakeup', signal: SIGNAL,
|
|
}).finally(() => { secondSettled = true })
|
|
await new Promise<void>((resolve) => { setTimeout(resolve, 0) })
|
|
expect(admitted).toEqual(['first waking'])
|
|
expect(secondSettled).toBe(false)
|
|
|
|
release.resolve(undefined)
|
|
await expect(Promise.all([first, second])).resolves.toMatchObject([
|
|
{ status: 'accepted' },
|
|
{ status: 'accepted' },
|
|
])
|
|
expect(admitted).toEqual(['first waking', 'second waking'])
|
|
})
|
|
|
|
it('deduplicates live target history and contains inspection and delivery failures', async () => {
|
|
const { ctx, lead } = await setup(['hang', textResponse('inactive target initial')])
|
|
const liveStarted = await spawn(ctx, lead, 'live-target')
|
|
const live = await waitRunning(ctx, liveStarted.member.id)
|
|
const internal = teamInternals(ctx).mailbox
|
|
const message: TeamMessageSnapshot = {
|
|
id: TeamMessageId('live-recorded-message'),
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
targetId: live.id,
|
|
delivery: 'wakeup',
|
|
content: content('already in live history'),
|
|
}
|
|
lead.session.append('team/message/queued', {
|
|
version: 1, teamId: TeamId(lead.id), message,
|
|
})
|
|
await ctx.sessions.flush(lead.session)
|
|
live.session.append('user/message', createUserMessage({
|
|
content: content('different Team message first'),
|
|
source: {
|
|
kind: 'team-message',
|
|
teamId: TeamId(lead.id),
|
|
messageId: TeamMessageId('other-message'),
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
},
|
|
}), { surfaceOp: 'append' })
|
|
live.session.append('user/message', createUserMessage({
|
|
content: content('already in live history'),
|
|
source: {
|
|
kind: 'team-message',
|
|
teamId: TeamId(lead.id),
|
|
messageId: message.id,
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
},
|
|
}), { surfaceOp: 'append' })
|
|
await expect(internal.tryDispatch(lead, message, SIGNAL)).resolves.toBe(true)
|
|
await internal.markDelivered(lead, message.id, live.id)
|
|
|
|
const wrongTarget: TeamMessageSnapshot = {
|
|
...message,
|
|
id: TeamMessageId('wrong-target-message'),
|
|
}
|
|
lead.session.append('team/message/queued', {
|
|
version: 1, teamId: TeamId(lead.id), message: wrongTarget,
|
|
})
|
|
await ctx.sessions.flush(lead.session)
|
|
await internal.markDelivered(lead, wrongTarget.id, SessionId('wrong-target'))
|
|
await expect(internal.serializeDispatch(wrongTarget, async () => true)).resolves.toBe(true)
|
|
const serialEntered = Promise.withResolvers<undefined>()
|
|
const releaseSerial = Promise.withResolvers<undefined>()
|
|
const serialFirst = internal.serializeDispatch(wrongTarget, async () => {
|
|
serialEntered.resolve(undefined)
|
|
await releaseSerial.promise
|
|
return true
|
|
})
|
|
await serialEntered.promise
|
|
const serialSecond = internal.serializeDispatch({
|
|
...wrongTarget, id: TeamMessageId('second-serialized-message'),
|
|
}, async () => true)
|
|
releaseSerial.resolve(undefined)
|
|
await expect(Promise.all([serialFirst, serialSecond])).resolves.toEqual([true, true])
|
|
|
|
const warnings: string[] = []
|
|
ctx.logger.warn = ((value: unknown) => { warnings.push(String(value)) }) as typeof ctx.logger.warn
|
|
const failedAck = vi.spyOn(ctx.sessions, 'flush').mockRejectedValueOnce(new Error('acknowledgement flush failed'))
|
|
live.session.append('user/message', createUserMessage({
|
|
content: content('acknowledgement failure'),
|
|
source: {
|
|
kind: 'team-message',
|
|
teamId: TeamId(lead.id),
|
|
messageId: wrongTarget.id,
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
},
|
|
}), { surfaceOp: 'append' })
|
|
await vi.waitFor(() => {
|
|
expect(warnings.some(warning => warning.includes('acknowledgement flush failed'))).toBe(true)
|
|
})
|
|
failedAck.mockRestore()
|
|
|
|
const inactiveStarted = await spawn(ctx, lead, 'inactive-target')
|
|
await waitNoAgent(ctx, inactiveStarted.member.id)
|
|
const inspect = vi.spyOn(ctx.sessionPersistence, 'inspect').mockRejectedValueOnce(new Error('inspect unavailable'))
|
|
const uncertain = await ctx.agentTeams.sendMessage(lead, {
|
|
target: 'inactive-target', content: content('inspection failure'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
expect(uncertain.status).toBe('queued')
|
|
inspect.mockRestore()
|
|
|
|
vi.spyOn(ctx.subagents, 'followup').mockRejectedValueOnce(new Error('delivery unavailable'))
|
|
const failed = await ctx.agentTeams.sendMessage(lead, {
|
|
target: 'inactive-target', content: content('delivery failure'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
expect(failed.status).toBe('queued')
|
|
expect(warnings.some(warning => warning.includes('inspect unavailable'))).toBe(true)
|
|
expect(warnings.some(warning => warning.includes('delivery unavailable'))).toBe(true)
|
|
|
|
ctx.agentTeams.interrupt(lead, 'live-target')
|
|
await waitNoAgent(ctx, live.id)
|
|
})
|
|
|
|
it('keeps quiet mail dormant, wakes on follow-up, preserves FIFO, and de-duplicates delivery', async () => {
|
|
const { ctx, lead } = await setup(['hang', textResponse('beta first'), textResponse('beta resumed')])
|
|
const alphaStarted = await spawn(ctx, lead, 'alpha')
|
|
const alpha = await waitRunning(ctx, alphaStarted.member.id)
|
|
const betaStarted = await spawn(ctx, lead, 'beta')
|
|
await waitNoAgent(ctx, betaStarted.member.id)
|
|
|
|
const quiet = await ctx.agentTeams.sendMessage(alpha, {
|
|
target: 'beta', content: content('quiet info'), delivery: 'quiet', signal: SIGNAL,
|
|
})
|
|
expect(quiet.status).toBe('queued')
|
|
expect(ctx.agents.get(betaStarted.member.id)).toBeUndefined()
|
|
const waking = await ctx.agentTeams.sendMessage(alpha, {
|
|
target: 'beta', content: content('do another turn'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
expect(waking.status).toBe('accepted')
|
|
await waitNoAgent(ctx, betaStarted.member.id)
|
|
await vi.waitFor(() => { expect(durable(lead).pendingMessages).toEqual([]) })
|
|
|
|
const stored = await ctx.sessionPersistence.inspect(betaStarted.member.id)
|
|
const peerMessages = stored.events.filter(event => event.type === 'user/message'
|
|
&& event.data.source.kind === 'team-message')
|
|
expect(peerMessages.map((event) => {
|
|
if (event.type !== 'user/message') return undefined
|
|
const block = event.data.content.at(-1)
|
|
return block?.type === 'text' ? block.text : undefined
|
|
})).toEqual(['quiet info', 'do another turn'])
|
|
expect(peerMessages.map(event => event.type === 'user/message'
|
|
? event.data.content[0]?.type === 'text' && event.data.content[0].text
|
|
: undefined)).toEqual([
|
|
expect.stringMatching(/^Team message .* from alpha:$/u),
|
|
expect.stringMatching(/^Team message .* from alpha:$/u),
|
|
])
|
|
expect(peerMessages.map(event => event.type === 'user/message' && event.data.source.kind === 'team-message'
|
|
? [event.data.source.messageId, event.data.source.senderName]
|
|
: undefined)).toEqual([
|
|
[quiet.messageId, 'alpha'],
|
|
[waking.messageId, 'alpha'],
|
|
])
|
|
|
|
ctx.agentTeams.interrupt(lead, 'alpha')
|
|
await waitNoAgent(ctx, alpha.id)
|
|
})
|
|
|
|
it('enforces message byte and pending-count limits without encouraging retry after enqueue', async () => {
|
|
const { ctx, lead } = await setup([textResponse('idle')], {
|
|
maxMessageBytes: 256,
|
|
maxPendingMessagesPerMember: 1,
|
|
})
|
|
const target = await spawn(ctx, lead, 'target')
|
|
await waitNoAgent(ctx, target.member.id)
|
|
await expect(ctx.agentTeams.sendMessage(lead, {
|
|
target: 'target', content: content('x'.repeat(300)), delivery: 'quiet', signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_MESSAGE_TOO_LARGE' })
|
|
const queued = await ctx.agentTeams.sendMessage(lead, {
|
|
target: 'target', content: content('one'), delivery: 'quiet', signal: SIGNAL,
|
|
})
|
|
expect(queued.status).toBe('queued')
|
|
await expect(ctx.agentTeams.sendMessage(lead, {
|
|
target: 'target', content: content('two'), delivery: 'quiet', signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_MAILBOX_FULL' })
|
|
await expect(ctx.agentTeams.sendMessage(lead, {
|
|
target: 'lead', content: content('self'), delivery: 'quiet', signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_SELF_MESSAGE' })
|
|
await expect(ctx.agentTeams.sendMessage(lead, {
|
|
target: 'missing', content: content('unknown target'), delivery: 'quiet', signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_MEMBER_NOT_FOUND' })
|
|
const controller = new AbortController()
|
|
controller.abort(new TeamError('cancelled before queue', 'TEST_CANCELLED'))
|
|
await expect(ctx.agentTeams.sendMessage(lead, {
|
|
target: 'target', content: content('cancelled'), delivery: 'quiet', signal: controller.signal,
|
|
})).rejects.toMatchObject({ code: 'TEST_CANCELLED' })
|
|
})
|
|
|
|
it('interrupts only the current turn and retains an already accepted follow-up', async () => {
|
|
const { ctx, lead } = await setup(['hang', textResponse('after interrupt')])
|
|
const started = await spawn(ctx, lead, 'worker')
|
|
const worker = await waitRunning(ctx, started.member.id)
|
|
const followup = await ctx.agentTeams.sendMessage(lead, {
|
|
target: 'worker', content: content('retained follow-up'), delivery: 'wakeup', signal: SIGNAL,
|
|
})
|
|
expect(followup.status).toBe('accepted')
|
|
expect(ctx.agentTeams.interrupt(lead, 'worker')).toEqual({ previousStatus: 'running' })
|
|
await vi.waitFor(() => { expect(worker.status).toBe('idle') })
|
|
expect(worker.inbox.nextTurn.some(message => message.source.kind === 'team-message'
|
|
&& message.source.messageId === followup.messageId)).toBe(true)
|
|
worker.cancel({ kind: 'parent' })
|
|
await waitNoAgent(ctx, worker.id)
|
|
})
|
|
|
|
it('waits for one change, supports cancellation, times out, and releases waiters on HMR disposal', async () => {
|
|
const ctx = new Context()
|
|
await mountAgentLoopTestDependencies(ctx)
|
|
const storageRoot = mkdtempSync(join(tmpdir(), 'dsh-team-wait-'))
|
|
roots.push(storageRoot)
|
|
await ctx.plugin(JsonlSessionPersistence, { root: storageRoot })
|
|
await ctx.plugin(AgentLoop, { agents: [] })
|
|
await ctx.plugin(SubagentService)
|
|
const fiber = await ctx.plugin(TeamService)
|
|
const service = ctx.agentTeams
|
|
const lead = ctx.agentLoop.create(SessionId('wait-lead'), {})
|
|
|
|
await expect(service.waitForChange(lead, 9_999, SIGNAL))
|
|
.rejects.toMatchObject({ code: 'TEAM_INVALID_TIMEOUT' })
|
|
const alreadyAborted = new AbortController()
|
|
alreadyAborted.abort(new TeamError('cancelled before wait', 'TEST_CANCELLED'))
|
|
await expect(service.waitForChange(lead, 10_000, alreadyAborted.signal))
|
|
.rejects.toMatchObject({ code: 'TEST_CANCELLED' })
|
|
|
|
const changed = service.waitForChange(lead, 10_000, SIGNAL)
|
|
const flush = ctx.sessions.flush.bind(ctx.sessions)
|
|
const flushEntered = Promise.withResolvers<undefined>()
|
|
const releaseFlush = Promise.withResolvers<undefined>()
|
|
vi.spyOn(ctx.sessions, 'flush').mockImplementationOnce(async (session) => {
|
|
flushEntered.resolve(undefined)
|
|
await releaseFlush.promise
|
|
return await flush(session)
|
|
})
|
|
let waitSettled = false
|
|
void changed.finally(() => { waitSettled = true })
|
|
const creating = service.createTask(lead, { subject: 'wake', description: 'wake waiter' })
|
|
await flushEntered.promise
|
|
expect(waitSettled).toBe(false)
|
|
releaseFlush.resolve(undefined)
|
|
await creating
|
|
await expect(changed).resolves.toEqual({ timedOut: false })
|
|
|
|
const controller = new AbortController()
|
|
const cancelled = service.waitForChange(lead, 10_000, controller.signal)
|
|
controller.abort(new TeamError('cancelled', 'TEST_CANCELLED'))
|
|
await expect(cancelled).rejects.toMatchObject({ code: 'TEST_CANCELLED' })
|
|
|
|
const stringAbort = new AbortController()
|
|
const firstWaiter = service.waitForChange(lead, 10_000, stringAbort.signal)
|
|
const secondWaiter = service.waitForChange(lead, 10_000, SIGNAL)
|
|
stringAbort.abort('string cancellation')
|
|
await expect(firstWaiter).rejects.toMatchObject({
|
|
code: 'TEAM_WAIT_ABORTED',
|
|
message: 'wait_agent aborted: string cancellation',
|
|
})
|
|
await service.createTask(lead, { subject: 'second waiter', description: 'second waiter remains registered' })
|
|
await expect(secondWaiter).resolves.toEqual({ timedOut: false })
|
|
|
|
const objectAbort = new AbortController()
|
|
const objectCancelled = service.waitForChange(lead, 10_000, objectAbort.signal)
|
|
objectAbort.abort({ kind: 'user' })
|
|
await expect(objectCancelled).rejects.toMatchObject({
|
|
code: 'TEAM_WAIT_ABORTED',
|
|
message: "wait_agent aborted: { kind: 'user' }",
|
|
})
|
|
|
|
await service.createTask(lead, { subject: 'already changed', description: 'edge-triggered wait' })
|
|
vi.useFakeTimers()
|
|
const timeout = service.waitForChange(lead, 10_000, SIGNAL)
|
|
await vi.advanceTimersByTimeAsync(10_000)
|
|
await expect(timeout).resolves.toEqual({ timedOut: true })
|
|
vi.useRealTimers()
|
|
|
|
const disposed = service.waitForChange(lead, 10_000, SIGNAL)
|
|
await fiber.dispose()
|
|
await expect(disposed).resolves.toEqual({ timedOut: false })
|
|
expect(ctx.get('agentTeams')).toBeUndefined()
|
|
})
|
|
|
|
it('disposes live teammate Activations and their waits when the Team service unloads', async () => {
|
|
const { ctx, lead, teamFiber } = await setup(['hang'])
|
|
const started = await spawn(ctx, lead, 'dispose-worker')
|
|
await waitRunning(ctx, started.member.id)
|
|
const waiting = ctx.agentTeams.waitForChange(lead, 10_000, SIGNAL)
|
|
|
|
await teamFiber.dispose()
|
|
|
|
await expect(waiting).resolves.toEqual({ timedOut: false })
|
|
expect(ctx.agents.get(started.member.id)).toBeUndefined()
|
|
expect(ctx.get('agentTeams')).toBeUndefined()
|
|
})
|
|
|
|
it('closes creation admission and drains an in-flight spawn before unload completes', async () => {
|
|
const { ctx, lead, teamFiber } = await setup(['hang'])
|
|
const service = ctx.agentTeams
|
|
const start = ctx.subagents.startContinuable.bind(ctx.subagents)
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
let childId: SessionId | undefined
|
|
vi.spyOn(ctx.subagents, 'startContinuable').mockImplementation(async (spec) => {
|
|
childId = spec.childId
|
|
entered.resolve(undefined)
|
|
await release.promise
|
|
return start(spec)
|
|
})
|
|
const spawning = spawn(ctx, lead, 'disposing-worker')
|
|
const rejected = expect(spawning).rejects.toMatchObject({ code: 'TEAM_DISPOSED' })
|
|
await entered.promise
|
|
|
|
const disposal = teamFiber.dispose()
|
|
await Promise.resolve()
|
|
await expect(service.waitForChange(lead, 3_600_000, SIGNAL)).resolves.toEqual({ timedOut: false })
|
|
await expect(service.spawnTeammate(lead, {
|
|
name: 'late-worker',
|
|
description: 'must not enter after disposal',
|
|
prompt: content('late task'),
|
|
context: 'fresh',
|
|
provider: 'spawn',
|
|
signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_DISPOSED' })
|
|
release.resolve(undefined)
|
|
|
|
await rejected
|
|
await disposal
|
|
if (childId !== undefined) expect(ctx.agents.get(childId)).toBeUndefined()
|
|
expect(ctx.get('agentTeams')).toBeUndefined()
|
|
})
|
|
|
|
it('retains an in-flight creation cleanup failure during disposal', async () => {
|
|
const { ctx } = await setup([])
|
|
const internal = teamInternals(ctx)
|
|
const cleanupFailure = new Error('creation cleanup failed')
|
|
const rejected = Promise.reject(cleanupFailure)
|
|
void rejected.catch(() => undefined)
|
|
internal.roster.inFlightCreations.add(rejected)
|
|
|
|
await expect(internal.disposeRuntime()).rejects.toMatchObject({ errors: [cleanupFailure] })
|
|
})
|
|
|
|
it('recognizes wrapped and coded runtime cancellation during disposal settlement', async () => {
|
|
const open = new TeamRuntimeLifecycle(100)
|
|
const ordinaryFailure = new Error('ordinary failure before disposal')
|
|
const openFailures: unknown[] = []
|
|
await open.settle([Promise.reject(ordinaryFailure)], openFailures)
|
|
expect(openFailures).toEqual([ordinaryFailure])
|
|
|
|
const lifecycle = new TeamRuntimeLifecycle(100)
|
|
lifecycle.close()
|
|
const failures: unknown[] = []
|
|
await lifecycle.settle([
|
|
Promise.reject(new Error('wrapped cancellation', { cause: lifecycle.reason })),
|
|
Promise.reject(new TeamError('translated cancellation', 'TEAM_DISPOSED')),
|
|
], failures)
|
|
expect(failures).toEqual([])
|
|
|
|
const cyclic = new Error('unrelated cyclic failure')
|
|
cyclic.cause = cyclic
|
|
await lifecycle.settle([Promise.reject(cyclic)], failures)
|
|
expect(failures).toEqual([cyclic])
|
|
})
|
|
|
|
it('disposes a live child even after its durable member edge becomes failed', async () => {
|
|
const { ctx, lead } = await setup(['hang'])
|
|
const childId = SessionId('failed-live-child')
|
|
const member = {
|
|
id: childId,
|
|
name: 'failed-live-worker',
|
|
description: 'failed-live-worker responsibility',
|
|
provider: 'spawn',
|
|
context: 'fresh' as const,
|
|
phase: 'provisioning' as const,
|
|
}
|
|
lead.session.append('team/member', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
member,
|
|
})
|
|
await ctx.subagents.startContinuable({
|
|
childId,
|
|
provider: 'spawn',
|
|
label: member.description,
|
|
request: { prompt: content('failed child task'), parent: lead },
|
|
signal: SIGNAL,
|
|
})
|
|
await waitRunning(ctx, childId)
|
|
lead.session.append('team/member', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
member: {
|
|
...member,
|
|
phase: 'failed',
|
|
error: 'creation cleanup is pending',
|
|
},
|
|
})
|
|
await ctx.sessions.flush(lead.session)
|
|
expect(ctx.agentTeams.listMembers(lead)[1]?.status).toBe('failed')
|
|
|
|
const internal = ctx.agentTeams as unknown as { disposeRuntime(): Promise<void> }
|
|
await internal.disposeRuntime()
|
|
expect(ctx.agents.get(childId)).toBeUndefined()
|
|
})
|
|
|
|
it('aborts and awaits an admitted cold mailbox dispatch during disposal', async () => {
|
|
const { ctx, lead } = await setup([textResponse('worker done')])
|
|
const started = await spawn(ctx, lead, 'mailbox-worker')
|
|
await waitNoAgent(ctx, started.member.id)
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const aborted = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
vi.spyOn(ctx.subagents, 'followup').mockImplementation(async (_parent, _childId, _content, options) => {
|
|
entered.resolve(undefined)
|
|
return await new Promise<never>((_resolve, reject) => {
|
|
options.signal.addEventListener('abort', () => {
|
|
aborted.resolve(undefined)
|
|
void release.promise.then(() => {
|
|
const reason: unknown = options.signal.reason
|
|
reject(reason instanceof Error ? reason : new Error(String(reason)))
|
|
})
|
|
}, { once: true })
|
|
})
|
|
})
|
|
|
|
const sending = ctx.agentTeams.sendMessage(lead, {
|
|
target: 'mailbox-worker',
|
|
content: content('resume during disposal'),
|
|
delivery: 'wakeup',
|
|
signal: SIGNAL,
|
|
})
|
|
await entered.promise
|
|
const internal = ctx.agentTeams as unknown as { disposeRuntime(): Promise<void> }
|
|
let disposed = false
|
|
const disposal = internal.disposeRuntime().then(() => { disposed = true })
|
|
await aborted.promise
|
|
await Promise.resolve()
|
|
expect(disposed).toBe(false)
|
|
release.resolve(undefined)
|
|
|
|
await expect(sending).resolves.toMatchObject({ status: 'queued' })
|
|
await disposal
|
|
expect(disposed).toBe(true)
|
|
expect(ctx.agents.get(started.member.id)).toBeUndefined()
|
|
})
|
|
|
|
it('awaits an admitted asynchronous acknowledgement before disposal completes', async () => {
|
|
const { ctx, lead } = await setup([])
|
|
const message: TeamMessageSnapshot = {
|
|
id: TeamMessageId('dispose-ack-message'),
|
|
senderId: SessionId('sender'),
|
|
senderName: 'sender',
|
|
targetId: lead.id,
|
|
delivery: 'wakeup',
|
|
content: content('acknowledge before disposal'),
|
|
}
|
|
lead.session.append('team/message/queued', {
|
|
version: 1,
|
|
teamId: TeamId(lead.id),
|
|
message,
|
|
})
|
|
await ctx.sessions.flush(lead.session)
|
|
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
const flush = ctx.sessions.flush.bind(ctx.sessions)
|
|
let blockReceipt = true
|
|
const flushSpy = vi.spyOn(ctx.sessions, 'flush').mockImplementation(async (session) => {
|
|
if (blockReceipt && session === lead.session) {
|
|
blockReceipt = false
|
|
entered.resolve(undefined)
|
|
await release.promise
|
|
}
|
|
return flush(session)
|
|
})
|
|
lead.session.append('user/message', createUserMessage({
|
|
content: content('acknowledge before disposal'),
|
|
source: {
|
|
kind: 'team-message',
|
|
teamId: TeamId(lead.id),
|
|
messageId: message.id,
|
|
senderId: message.senderId,
|
|
senderName: message.senderName,
|
|
},
|
|
}), { surfaceOp: 'append' })
|
|
|
|
const internal = ctx.agentTeams as unknown as { disposeRuntime(): Promise<void> }
|
|
let disposed = false
|
|
const disposal = internal.disposeRuntime().then(() => { disposed = true })
|
|
await entered.promise
|
|
await Promise.resolve()
|
|
const disposedBeforeRelease = disposed
|
|
release.resolve(undefined)
|
|
await disposal
|
|
|
|
expect(disposedBeforeRelease).toBe(false)
|
|
expect(disposed).toBe(true)
|
|
expect(durable(lead).pendingMessages).toEqual([])
|
|
flushSpy.mockRestore()
|
|
})
|
|
|
|
it('bounds Team runtime disposal when a continuation drain never settles', async () => {
|
|
const { ctx, lead, teamFiber } = await setup(['hang'], { disposalTimeoutMs: 25 })
|
|
const started = await spawn(ctx, lead, 'stuck-worker')
|
|
await waitRunning(ctx, started.member.id)
|
|
const drain = vi.spyOn(ctx.subagents, 'drainContinuableChildren')
|
|
.mockImplementation(() => new Promise(() => {}))
|
|
|
|
const outcome = await Promise.race([
|
|
teamFiber.dispose().then(() => 'disposed'),
|
|
new Promise<'hung'>((resolve) => { setTimeout(() => { resolve('hung') }, 1_000) }),
|
|
])
|
|
expect(outcome).toBe('disposed')
|
|
expect(drain).toHaveBeenCalledWith(lead, [started.member.id])
|
|
expect(ctx.get('agentTeams')).toBeUndefined()
|
|
})
|
|
|
|
it('bounds disposal while an admitted creation ignores cancellation', async () => {
|
|
const { ctx, lead } = await setup([], { disposalTimeoutMs: 25 })
|
|
const internal = teamInternals(ctx)
|
|
internal.roster.inFlightCreations.add(new Promise(() => {}))
|
|
|
|
await expect(internal.disposeRuntime()).rejects.toBeInstanceOf(AggregateError)
|
|
await expect(ctx.agentTeams.spawnTeammate(lead, {
|
|
name: 'after-timeout',
|
|
description: 'admission remains closed',
|
|
prompt: content('must reject'),
|
|
context: 'fresh',
|
|
provider: 'spawn',
|
|
signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_DISPOSED' })
|
|
await expect(ctx.agentTeams.sendMessage(lead, {
|
|
target: 'nobody', content: content('must reject'), delivery: 'quiet', signal: SIGNAL,
|
|
})).rejects.toMatchObject({ code: 'TEAM_DISPOSED' })
|
|
await expect(internal.mailbox.tryDispatch(lead, {
|
|
id: TeamMessageId('post-disposal-message'),
|
|
senderId: lead.id,
|
|
senderName: 'lead',
|
|
targetId: lead.id,
|
|
delivery: 'quiet',
|
|
content: content('must not dispatch'),
|
|
}, SIGNAL)).resolves.toBe(false)
|
|
})
|
|
|
|
it('contains recovery callback failures and ignores work scheduled after disposal', async () => {
|
|
const { ctx, lead, teamFiber } = await setup([])
|
|
const warnings: string[] = []
|
|
ctx.logger.warn = ((value: unknown) => { warnings.push(String(value)) }) as typeof ctx.logger.warn
|
|
const internal = teamInternals(ctx)
|
|
internal.recoverFor = async () => { throw new Error('forced recovery failure') }
|
|
internal.scheduleRecovery(lead)
|
|
await Promise.resolve()
|
|
await Promise.resolve()
|
|
expect(warnings.some(warning => warning.includes('forced recovery failure'))).toBe(true)
|
|
|
|
lead.session.append('user/message', createUserMessage({
|
|
content: content('orphan Team source'),
|
|
source: {
|
|
kind: 'team-message',
|
|
teamId: TeamId('absent-team'),
|
|
messageId: TeamMessageId('absent-team-message'),
|
|
senderId: SessionId('absent-sender'),
|
|
senderName: 'absent',
|
|
},
|
|
}), { surfaceOp: 'append' })
|
|
await Promise.resolve()
|
|
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
internal.recoverFor = async () => {
|
|
entered.resolve(undefined)
|
|
await release.promise
|
|
throw new Error('failure after disposal')
|
|
}
|
|
internal.scheduleRecovery(lead)
|
|
await entered.promise
|
|
await teamFiber.dispose()
|
|
release.resolve(undefined)
|
|
await Promise.resolve()
|
|
await Promise.resolve()
|
|
internal.scheduleRecovery(lead)
|
|
await Promise.resolve()
|
|
})
|
|
|
|
it('reports contained teardown failures without retaining the Team service', async () => {
|
|
const { ctx, lead, teamFiber } = await setup(['hang'])
|
|
const started = await spawn(ctx, lead, 'failing-drain')
|
|
await waitRunning(ctx, started.member.id)
|
|
vi.spyOn(ctx.subagents, 'drainContinuableDescendants').mockRejectedValueOnce(new Error('drain failure'))
|
|
|
|
await teamFiber.dispose()
|
|
expect(ctx.get('agentTeams')).toBeUndefined()
|
|
})
|
|
|
|
it('reconciles mismatched persisted children and ignores a concurrently settled member', async () => {
|
|
const first = await setup([])
|
|
const liveId = SessionId('live-provisioning-child')
|
|
const live = await first.ctx.agents.create({
|
|
sessionId: liveId,
|
|
meta: { parentSession: first.lead.id },
|
|
agentOptions: { provider: 'mock', model: 'mock' },
|
|
})
|
|
const provisioning = {
|
|
id: liveId,
|
|
name: 'mismatched-child',
|
|
description: 'mismatched persisted child',
|
|
provider: 'spawn',
|
|
context: 'fresh' as const,
|
|
phase: 'provisioning' as const,
|
|
}
|
|
first.lead.session.append('team/member', {
|
|
version: 1, teamId: TeamId(first.lead.id), member: provisioning,
|
|
})
|
|
const reconcileFirst = teamInternals(first.ctx).roster
|
|
await reconcileFirst.reconcileProvisioning(first.lead, SIGNAL)
|
|
expect(durable(first.lead).members[0]?.phase).toBe('provisioning')
|
|
live.agent.session.append('user/message', createUserMessage({
|
|
content: content('persist mismatched child'), source: { kind: 'user' },
|
|
}), { surfaceOp: 'append' })
|
|
await first.ctx.sessions.flush(live.agent.session)
|
|
await live.dispose()
|
|
await reconcileFirst.reconcileProvisioning(first.lead, SIGNAL)
|
|
expect(durable(first.lead).members[0]).toMatchObject({
|
|
phase: 'failed',
|
|
error: 'persisted child Session does not match the provisioned continuation',
|
|
})
|
|
|
|
const second = await setup([])
|
|
const childId = SessionId('concurrently-settled-child')
|
|
const member = { ...provisioning, id: childId, name: 'concurrent-child' }
|
|
second.lead.session.append('team/member', {
|
|
version: 1, teamId: TeamId(second.lead.id), member,
|
|
})
|
|
const entered = Promise.withResolvers<undefined>()
|
|
const release = Promise.withResolvers<undefined>()
|
|
vi.spyOn(second.ctx.sessionPersistence, 'inspect').mockImplementationOnce(async () => {
|
|
entered.resolve(undefined)
|
|
await release.promise
|
|
throw new Error('late inspection failure')
|
|
})
|
|
const reconcileSecond = teamInternals(second.ctx).roster
|
|
const reconciling = reconcileSecond.reconcileProvisioning(second.lead, SIGNAL)
|
|
await entered.promise
|
|
second.lead.session.append('team/member', {
|
|
version: 1,
|
|
teamId: TeamId(second.lead.id),
|
|
member: { ...member, phase: 'failed', error: 'settled elsewhere' },
|
|
})
|
|
release.resolve(undefined)
|
|
await reconciling
|
|
expect(durable(second.lead).members[0]).toMatchObject({
|
|
phase: 'failed', error: 'settled elsewhere',
|
|
})
|
|
})
|
|
})
|