fix(tasks): bound per-session background work

packages/tasks/tasks-local/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/tasks/tasks-local/README.md
-README.md: 80d7932466188955ade1c14e4968d51b739ba818
-README.zh.md: 5e4263e4685be64f91ec2a7c74edbf89e1148866
+README.md: dcfaf397ccfb61be72c2846cb3f678469645a2c7
+README.zh.md: 60ddb2901a51974cfeaccd8c820bc084b9ca0949

packages/tasks/tasks-local/README.md

@@ -2,7 +2,13 @@
 
 English | [中文](README.zh.md)
 
-Process-local implementation of the [`@deepseek-ai/dsh-tasks`](../tasks/README.md) registry contract: `LocalTaskService` keeps every record in memory, issues per-kind `<kind>-N` ids, and hands out fresh snapshots, never live state. It has no config; load it as a plugin and it registers as `ctx.tasks`.
+Process-local implementation of the [`@deepseek-ai/dsh-tasks`](../tasks/README.md) registry contract: `LocalTaskService` keeps every record in memory, issues per-kind `<kind>-N` ids, and hands out fresh snapshots, never live state. Load it as a plugin and it registers as `ctx.tasks`.
+
+## Admission
+
+`maxConcurrentTasksPerOwner` is a positive safe integer and defaults to `10`. Before invoking a producer, `start()` counts the exact owner's `running` and `stopping` records; all unowned tasks share one separate service bucket. Terminal history does not occupy capacity, and only producer `done` settlement releases a stopping task's place.
+
+At capacity, `start()` fails before producer execution and id allocation with an error that names the limit and tells the model to use `task_kill`, wait for the task to finish stopping, and retry. The registry does not queue, preempt, or maintain a second mutable counter.
 
 ## Lifecycle
 

packages/tasks/tasks-local/README.zh.md

@@ -2,7 +2,13 @@
 
 [English](README.md) | 中文
 
-[`@deepseek-ai/dsh-tasks`](../tasks/README.md) 注册表约定的进程本地实现：`LocalTaskService` 把每条记录保存在内存中，按 kind 签发 `<kind>-N` id，并且只交出全新快照，从不交出实时状态。它没有配置；作为插件加载后即注册为 `ctx.tasks`。
+[`@deepseek-ai/dsh-tasks`](../tasks/README.md) 注册表约定的进程本地实现：`LocalTaskService` 把每条记录保存在内存中，按 kind 签发 `<kind>-N` id，并且只交出全新快照，从不交出实时状态。作为插件加载后即注册为 `ctx.tasks`。
+
+## 准入
+
+`maxConcurrentTasksPerOwner` 必须是正的安全整数，默认值为 `10`。调用生产方之前，`start()` 会统计确切 owner 的 `running` 与 `stopping` 记录；所有无 owner 任务共享另一个独立的服务级桶。终止历史不占用容量，处于 `stopping` 的任务只有在生产方 `done` 结算后才释放名额。
+
+达到容量时，`start()` 会在生产方执行和 id 分配前失败；错误会给出上限，并告诉模型使用 `task_kill`、等待任务完全停稳后再重试。注册表不会排队或抢占任务，也不会维护第二份可变计数。
 
 ## 生命周期
 

packages/tasks/tasks-local/package.json

@@ -39,7 +39,12 @@
     "@deepseek-ai/dsh-timeout": "workspace:^",
     "@deepseek-ai/cordis": "workspace:^"
   },
+  "dependencies": {
+    "@deepseek-ai/schemastery": "workspace:^"
+  },
   "devDependencies": {
+    "@deepseek-ai/cordis-plugin-include": "workspace:^",
+    "@deepseek-ai/cordis-plugin-loader": "workspace:^",
     "@deepseek-ai/dsh-agent": "workspace:^",
     "@deepseek-ai/dsh-brand": "workspace:^",
     "@deepseek-ai/dsh-invariants": "workspace:^",

packages/tasks/tasks-local/src/index.ts

@@ -10,6 +10,7 @@
  */
 
 import { Context } from '@deepseek-ai/cordis'
+import z from '@deepseek-ai/schemastery'
 import type { Agent } from '@deepseek-ai/dsh-agent'
 import { AnonymousEntries, ScopedLayers, scopeOf } from '@deepseek-ai/dsh-scope'
 import type { ScopeLayer } from '@deepseek-ai/dsh-scope'
@@ -23,6 +24,18 @@ import type {
 /** Timeout code that distinguishes a bounded wait from caller cancellation. */
 export const TASK_WAIT_TIMEOUT = 'TASK_WAIT_TIMEOUT'
 
+/** Default maximum number of active tasks in one exact-owner bucket. */
+const DEFAULT_MAX_CONCURRENT_TASKS_PER_OWNER = 10
+
+/** Configuration for the process-local task registry. */
+export interface Config {
+  /** Maximum `running` plus `stopping` tasks per exact owner; omission defaults to 10. */
+  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
@@ -76,6 +89,16 @@ class TaskLayer implements ScopeLayer {
  * semantics this implementation honors.
  */
 export class LocalTaskService extends TaskService {
+  static Config: z<Config> = z.object({
+    maxConcurrentTasksPerOwner: z.number()
+      .step(1)
+      .min(1)
+      .max(Number.MAX_SAFE_INTEGER)
+      .default(DEFAULT_MAX_CONCURRENT_TASKS_PER_OWNER),
+  })
+
+  /** Validated registry configuration. */
+  readonly config: ResolvedConfig
   private store = new Map<TaskId, TrackedTask>()
   private counters = new Map<string, number>()
   /**
@@ -97,8 +120,14 @@ export class LocalTaskService extends TaskService {
   /** Service context used by detached settlement continuations and teardown. */
   private readonly selfCtx: Context
 
-  constructor(ctx: Context) {
+  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 }
     this.selfCtx = ctx
     ctx.effect(() => () => this.disposeAll(), 'tasks teardown')
   }
@@ -115,6 +144,13 @@ export class LocalTaskService extends TaskService {
     }
     if (spec.owner !== undefined) this.ensureOwnerCleanup(spec.owner)
 
+    const active = this.activeTaskCount(spec.owner)
+    if (active >= this.config.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`,
+      )
+    }
+
     const hooks = spec.run()
     const count = (this.counters.get(spec.kind) ?? 0) + 1
     this.counters.set(spec.kind, count)
@@ -285,6 +321,15 @@ export class LocalTaskService extends TaskService {
       .some(layer => !layer.surfaces.isEmpty())
   }
 
+  /** Count authoritative active records for one exact owner or the shared unowned bucket. */
+  private activeTaskCount(owner: Agent | undefined): number {
+    let count = 0
+    for (const task of this.store.values()) {
+      if (task.owner === owner && (task.status === 'running' || task.status === 'stopping')) count += 1
+    }
+    return count
+  }
+
   /**
    * The completion listeners that own `owner`'s notices: the global layer's
    * first, then each scoped layer along the owner's chain. A listener outside

packages/tasks/tasks-local/tests/loader-composition.spec.ts

@@ -0,0 +1,52 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import { mkdtemp, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { pathToFileURL } from 'node:url'
+import { Context } from '@deepseek-ai/cordis'
+import Include from '@deepseek-ai/cordis-plugin-include'
+import Loader from '@deepseek-ai/cordis-plugin-loader'
+import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
+
+let root: string | undefined
+let context: Context | undefined
+
+afterEach(async () => {
+  await context?.fiber.dispose()
+  context = undefined
+  if (root !== undefined) await rm(root, { recursive: true, force: true })
+  root = undefined
+})
+
+describe('tasks-local through a real Loader composition', () => {
+  it('applies the provider-owned admission config from a Cordis row', async () => {
+    root = await mkdtemp(join(tmpdir(), 'dsh-tasks-local-loader-'))
+    const configPath = join(root, 'cordis.yml')
+    await writeFile(configPath, [
+      "- name: '@deepseek-ai/dsh-tasks-local'",
+      '  config:',
+      '    maxConcurrentTasksPerOwner: 2',
+      '',
+    ].join('\n'))
+
+    context = new Context()
+    context.baseUrl = pathToFileURL(root).href + '/'
+    await context.plugin(Loader)
+    context.loader.builtins.include = Include
+    context.loader.internal = {
+      version: 'v2',
+      async import(specifier: string) {
+        if (specifier === '@deepseek-ai/dsh-tasks-local') return LocalTaskService
+        throw new Error(`unexpected Loader import: ${specifier}`)
+      },
+    } as unknown as NonNullable<typeof context.loader.internal>
+    await context.loader.create({
+      name: 'cordis:include',
+      config: { path: pathToFileURL(configPath).href },
+    })
+    await context.loader.await()
+
+    expect(context.tasks).toBeInstanceOf(LocalTaskService)
+    expect((context.tasks as LocalTaskService).config.maxConcurrentTasksPerOwner).toBe(2)
+  })
+})

packages/tasks/tasks-local/tests/tasks.spec.ts

@@ -7,7 +7,7 @@ import { bindScopeParent, createScope, scopeOf } from '@deepseek-ai/dsh-scope'
 import type { ScopeKey } from '@deepseek-ai/dsh-scope'
 import { TaskId } from '@deepseek-ai/dsh-tasks'
 import type { TaskHooks, TaskKind, TaskOutcome, TaskSnapshot, TaskStart } from '@deepseek-ai/dsh-tasks'
-import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
+import LocalTaskService, { type Config as TasksConfig } from '@deepseek-ai/dsh-tasks-local'
 
 declare module '@deepseek-ai/dsh-tasks' {
   interface TaskKindMap {
@@ -76,10 +76,10 @@ function producer(overrides: Partial<Omit<TaskStart, 'run'> & TaskHooks> = {}) {
   return { spec, settle, reject, cancels }
 }
 
-async function harness() {
+async function harness(config: TasksConfig = {}) {
   const ctx = new Context()
   await ctx.plugin(AgentRegistry)
-  await ctx.plugin(LocalTaskService)
+  await ctx.plugin(LocalTaskService, config)
   ctx.tasks.attachSurface('test-surface')
   return ctx
 }
@@ -166,6 +166,105 @@ describe('LocalTaskService.start', () => {
     expect(() => ctx.tasks.start(producer({ outputLimitBytes: 0 }).spec)).toThrow('outputLimitBytes')
   })
 
+  it.each([0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, Number.MAX_SAFE_INTEGER + 1])(
+    'rejects invalid maxConcurrentTasksPerOwner config: %s',
+    async (maxConcurrentTasksPerOwner) => {
+      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)
+  })
+
+  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)')
+    expect(run).not.toHaveBeenCalled()
+    for (const task of live) task.settle({ status: 'completed' })
+  })
+
+  it('rejects before producer start and id allocation, then admits immediately after settlement', async () => {
+    const ctx = await harness({ maxConcurrentTasksPerOwner: 1 })
+    const first = producer()
+    expect(ctx.tasks.start(first.spec)).toBe('bash-1')
+
+    const blocked = producer()
+    const run = vi.fn(() => blocked.spec.run())
+    expect(() => ctx.tasks.start({ ...blocked.spec, run }))
+      .toThrow('use task_kill to stop an unneeded task, wait for it to finish, then retry')
+    expect(run).not.toHaveBeenCalled()
+
+    first.settle({ status: 'completed' })
+    await tick()
+    expect(ctx.tasks.start(blocked.spec)).toBe('bash-2')
+  })
+
+  it('keeps a stopping task in the bucket until producer settlement', async () => {
+    const ctx = await harness({ maxConcurrentTasksPerOwner: 1 })
+    const first = producer()
+    const id = ctx.tasks.start(first.spec)
+    expect(ctx.tasks.kill(id)).toBe('requested')
+
+    const replacement = producer()
+    expect(() => ctx.tasks.start(replacement.spec)).toThrow('(1/1 active)')
+
+    first.settle({ status: 'killed' })
+    await tick()
+    expect(ctx.tasks.start(replacement.spec)).toBe('bash-2')
+  })
+
+  it.each(['completed', 'killed', 'failed'] as const)(
+    'releases the bucket after a %s terminal outcome',
+    async (status) => {
+      const ctx = await harness({ maxConcurrentTasksPerOwner: 1 })
+      const first = producer()
+      ctx.tasks.start(first.spec)
+      first.settle({ status })
+      await tick()
+      expect(() => ctx.tasks.start(producer().spec)).not.toThrow()
+    },
+  )
+
+  it('isolates exact owners, replacement objects with the same session id, and the unowned bucket', async () => {
+    const ctx = await harness({ maxConcurrentTasksPerOwner: 1 })
+    const oldOwner = stubAgent(ctx, 'shared-session')
+    const detachOld = ctx.agents.register(oldOwner)
+    const oldTask = producer({ owner: oldOwner })
+    ctx.tasks.start(oldTask.spec)
+
+    const otherOwner = stubAgent(ctx, 'other-session')
+    ctx.agents.register(otherOwner)
+    expect(() => ctx.tasks.start(producer({ owner: otherOwner }).spec)).not.toThrow()
+
+    detachOld()
+    const replacement = stubAgent(ctx, 'shared-session')
+    ctx.agents.register(replacement)
+    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({ owner: oldOwner }).spec))
+      .toThrow('is not the registered agent instance')
+
+    oldTask.settle({ status: 'completed' })
+    await tick()
+    await disposeAgentScope(oldOwner)
+  })
+
   it('issues kind-prefixed ids from per-kind counters', async () => {
     const ctx = await harness()
     expect(ctx.tasks.start(producer().spec)).toBe('bash-1')

packages/tasks/tasks/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/tasks/tasks/README.md
-README.md: 053e407d2e28cb175ebe9de15c7e16ef04e95cb8
-README.zh.md: dd39b1098cd01d7db9db6e49210c2b1149b37ecb
+README.md: 17579e417690dd34641e59f66cc8196465d08d68
+README.zh.md: 8f6ed21bb8ebcdcda4e0e3799aba5b88e5179393

packages/tasks/tasks/README.md

@@ -6,7 +6,7 @@ The background task registry contract (`ctx.tasks`). The abstract `TaskService`
 
 ## Service contract
 
-- `start(spec): TaskId` validates the control surface, spec, exact live owner, and optional positive `outputLimitBytes` before calling the producer's `run()` once. A starter throw leaves nothing registered; successful return commits without another failable step.
+- `start(spec): TaskId` validates the control surface, spec, exact live owner, optional positive `outputLimitBytes`, and any provider-owned admission policy before calling the producer's `run()` once. A preflight rejection or starter throw leaves no task id or registered work; successful return commits without another failable step.
 - `get(id, caller?)` and `list(caller?)` return non-consuming snapshots. Listing includes only caller-owned and unowned tasks.
 - `read(id, caller?)` consumes the single cursor for stream tasks and reads terminal output idempotently for final-output tasks.
 - `kill(id, caller?, reason?)` invokes producer cancellation before changing status. A cancellation throw leaves the task running; success changes it to `stopping` and marks terminal delivery reported.

packages/tasks/tasks/README.zh.md

@@ -6,7 +6,7 @@
 
 ## 服务约定
 
-- `start(spec): TaskId` 验证控制表层、spec、确切且仍存活的 owner，以及可选的 `outputLimitBytes`（如提供则须为正数），然后只调用生产方的 `run()` 一次。启动方抛出异常时不注册任何内容；成功返回会直接提交，不再执行其他可能失败的步骤。
+- `start(spec): TaskId` 验证控制表层、spec、确切且仍存活的 owner、可选的正数 `outputLimitBytes`，以及 Service provider 所拥有的准入策略，然后只调用生产方的 `run()` 一次。预检拒绝或启动方抛出异常时都不会生成 task id 或注册工作；成功返回会直接提交，不再执行其他可能失败的步骤。
 - `get(id, caller?)` 和 `list(caller?)` 返回非消费式快照。列表只包含调用方拥有及无 owner 的任务。
 - `read(id, caller?)` 消费流任务的唯一游标；对于最终输出任务，则以幂等方式读取终止输出。
 - `kill(id, caller?, reason?)` 在更改状态前调用生产方取消。取消抛出异常时任务保持运行；成功则把状态改为 `stopping`，并将终止交付标记为已报告。

packages/tasks/tasks/src/index.ts

@@ -67,10 +67,11 @@ export abstract class TaskService extends Service {
   }
 
   /**
-   * Preflight access, validation, and owner cleanup before starting and
-   * atomically registering work. A throwing starter leaves nothing registered;
-   * after it returns, registration cannot fail. Settlement records the outcome,
-   * notifies listeners, and releases waiters.
+   * Preflight access, validation, owner cleanup, and implementation-owned
+   * admission before starting and atomically registering work. Any preflight
+   * rejection leaves no task id or execution resource. A throwing starter
+   * leaves nothing registered; after it returns, registration cannot fail.
+   * Settlement records the outcome, notifies listeners, and releases waiters.
    * @param spec - task identity, owner, and synchronous starter.
    * @returns the registry-issued `<kind>-N` id.
    */

packages/tasks/tool-tasks/tests/tool-tasks.spec.ts

@@ -572,6 +572,7 @@ describe('completion notices', () => {
       const prior = producer({ kind: 'pty-send' })
       ctx.tasks.start(prior.spec)
       prior.settle({ status: 'completed' })
+      await tick()
     }
     const inject = vi.fn()
     const owner = fakeAgent(ctx, 'sess-1', inject)