refactor(tasks): simplify admission configuration
This commit is contained in:
@@ -33,9 +33,6 @@ export interface Config {
|
||||
maxConcurrentTasksPerOwner?: number
|
||||
}
|
||||
|
||||
/** Configuration after defaults and load-time validation. */
|
||||
type ResolvedConfig = Required<Config>
|
||||
|
||||
/** The registry's mutable per-task record (never handed out — see {@link LocalTaskService.snapshot}). */
|
||||
interface TrackedTask {
|
||||
id: TaskId
|
||||
@@ -97,8 +94,8 @@ export class LocalTaskService extends TaskService {
|
||||
.default(DEFAULT_MAX_CONCURRENT_TASKS_PER_OWNER),
|
||||
})
|
||||
|
||||
/** Validated registry configuration. */
|
||||
readonly config: ResolvedConfig
|
||||
/** Schemastery-defaulted active-task limit. */
|
||||
private readonly maxConcurrentTasksPerOwner: number
|
||||
private store = new Map<TaskId, TrackedTask>()
|
||||
private counters = new Map<string, number>()
|
||||
/**
|
||||
@@ -120,14 +117,10 @@ export class LocalTaskService extends TaskService {
|
||||
/** Service context used by detached settlement continuations and teardown. */
|
||||
private readonly selfCtx: Context
|
||||
|
||||
constructor(ctx: Context, config: Config = {}) {
|
||||
constructor(ctx: Context, config: Config) {
|
||||
super(ctx)
|
||||
const maxConcurrentTasksPerOwner = config.maxConcurrentTasksPerOwner
|
||||
?? DEFAULT_MAX_CONCURRENT_TASKS_PER_OWNER
|
||||
if (!Number.isSafeInteger(maxConcurrentTasksPerOwner) || maxConcurrentTasksPerOwner <= 0) {
|
||||
throw new TypeError('tasks-local: maxConcurrentTasksPerOwner must be a positive safe integer')
|
||||
}
|
||||
this.config = { maxConcurrentTasksPerOwner }
|
||||
// Schemastery validates and fills the default before constructing the service.
|
||||
this.maxConcurrentTasksPerOwner = (config as Required<Config>).maxConcurrentTasksPerOwner
|
||||
this.selfCtx = ctx
|
||||
ctx.effect(() => () => this.disposeAll(), 'tasks teardown')
|
||||
}
|
||||
@@ -145,9 +138,9 @@ export class LocalTaskService extends TaskService {
|
||||
if (spec.owner !== undefined) this.ensureOwnerCleanup(spec.owner)
|
||||
|
||||
const active = this.activeTaskCount(spec.owner)
|
||||
if (active >= this.config.maxConcurrentTasksPerOwner) {
|
||||
if (active >= this.maxConcurrentTasksPerOwner) {
|
||||
throw new Error(
|
||||
`background task limit reached for this owner (${active}/${this.config.maxConcurrentTasksPerOwner} active); use task_kill to stop an unneeded task, wait for it to finish, then retry`,
|
||||
`background task limit reached for this owner (limit: ${this.maxConcurrentTasksPerOwner}); use task_kill to stop an unneeded task, wait for it to finish, then retry`,
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
@@ -25,7 +25,7 @@ describe('tasks-local through a real Loader composition', () => {
|
||||
await writeFile(configPath, [
|
||||
"- name: '@deepseek-ai/dsh-tasks-local'",
|
||||
' config:',
|
||||
' maxConcurrentTasksPerOwner: 2',
|
||||
' maxConcurrentTasksPerOwner: 1',
|
||||
'',
|
||||
].join('\n'))
|
||||
|
||||
@@ -47,6 +47,20 @@ describe('tasks-local through a real Loader composition', () => {
|
||||
await context.loader.await()
|
||||
|
||||
expect(context.tasks).toBeInstanceOf(LocalTaskService)
|
||||
expect((context.tasks as LocalTaskService).config.maxConcurrentTasksPerOwner).toBe(2)
|
||||
context.tasks.attachController('loader-test')
|
||||
let settle!: (outcome: { status: 'killed' }) => void
|
||||
context.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'hold loader slot',
|
||||
run: () => ({
|
||||
cancel: () => { settle({ status: 'killed' }) },
|
||||
done: new Promise((resolve) => { settle = resolve }),
|
||||
}),
|
||||
})
|
||||
expect(() => context!.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'blocked loader task',
|
||||
run: () => ({ cancel: () => {}, done: Promise.resolve({ status: 'completed' }) }),
|
||||
})).toThrow('(limit: 1)')
|
||||
})
|
||||
})
|
||||
|
||||
@@ -172,35 +172,27 @@ describe('LocalTaskService.start', () => {
|
||||
const ctx = new Context()
|
||||
await expect(ctx.plugin(LocalTaskService, { maxConcurrentTasksPerOwner }))
|
||||
.rejects.toThrow()
|
||||
expect(() => new LocalTaskService(new Context(), { maxConcurrentTasksPerOwner }))
|
||||
.toThrow('maxConcurrentTasksPerOwner must be a positive safe integer')
|
||||
},
|
||||
)
|
||||
|
||||
it('accepts the largest safe integer limit', async () => {
|
||||
const ctx = await harness({ maxConcurrentTasksPerOwner: Number.MAX_SAFE_INTEGER })
|
||||
expect((ctx.tasks as LocalTaskService).config.maxConcurrentTasksPerOwner)
|
||||
.toBe(Number.MAX_SAFE_INTEGER)
|
||||
expect(ctx.tasks).toBeInstanceOf(LocalTaskService)
|
||||
})
|
||||
|
||||
it('defaults each owner bucket to ten active tasks', async () => {
|
||||
const ctx = await harness()
|
||||
expect((ctx.tasks as LocalTaskService).config.maxConcurrentTasksPerOwner).toBe(10)
|
||||
const live = Array.from({ length: 10 }, () => producer())
|
||||
for (const task of live) ctx.tasks.start(task.spec)
|
||||
|
||||
const blocked = producer()
|
||||
const run = vi.fn(() => blocked.spec.run())
|
||||
expect(() => ctx.tasks.start({ ...blocked.spec, run }))
|
||||
.toThrow('background task limit reached for this owner (10/10 active)')
|
||||
.toThrow('background task limit reached for this owner (limit: 10)')
|
||||
expect(run).not.toHaveBeenCalled()
|
||||
for (const task of live) task.settle({ status: 'completed' })
|
||||
})
|
||||
|
||||
it('defaults direct construction when the config schema is bypassed', () => {
|
||||
expect(new LocalTaskService(new Context()).config.maxConcurrentTasksPerOwner).toBe(10)
|
||||
})
|
||||
|
||||
it('rejects before producer start and id allocation, then admits immediately after settlement', async () => {
|
||||
const ctx = await harness({ maxConcurrentTasksPerOwner: 1 })
|
||||
const first = producer()
|
||||
@@ -224,7 +216,7 @@ describe('LocalTaskService.start', () => {
|
||||
expect(ctx.tasks.kill(id)).toBe('requested')
|
||||
|
||||
const replacement = producer()
|
||||
expect(() => ctx.tasks.start(replacement.spec)).toThrow('(1/1 active)')
|
||||
expect(() => ctx.tasks.start(replacement.spec)).toThrow('(limit: 1)')
|
||||
|
||||
first.settle({ status: 'killed' })
|
||||
await tick()
|
||||
@@ -260,7 +252,7 @@ describe('LocalTaskService.start', () => {
|
||||
expect(() => ctx.tasks.start(producer({ owner: replacement }).spec)).not.toThrow()
|
||||
|
||||
ctx.tasks.start(producer().spec)
|
||||
expect(() => ctx.tasks.start(producer().spec)).toThrow('(1/1 active)')
|
||||
expect(() => ctx.tasks.start(producer().spec)).toThrow('(limit: 1)')
|
||||
expect(() => ctx.tasks.start(producer({ owner: oldOwner }).spec))
|
||||
.toThrow('is not the registered agent instance')
|
||||
|
||||
|
||||
Reference in New Issue
Block a user