fix(tasks-local): layer control surfaces and listeners by registering scope

One host registry serves every composition in the process, so its two
service-wide collections answered per-owner questions process-wide. `start()`
asked only whether SOME surface was attached, so an agent whose own composition
loads no `tool-tasks` could start work it has no tool to collect or stop as soon
as any other preset attached one — and the answer changed depending on which
sessions happened to be open. `settle()` walked every registered listener, so a
task settling without a waiter injected one completion notice per mounted
preset into the same owner.

Both collections now sit in `ScopedLayers`, the layered-registry primitive
`tools` and `skills` already use: a registration files into its registering
context's scope, and a read unions the global layer with the owner's scope
chain. A surface or listener registered from an unscoped context lands in the
global layer and serves every owner, which is exactly the host-plane
composition's own controls, so the TUI path is unchanged without a special
case.

This supersedes the consumer-side filter in the previous commit. That filter
produced the right notices but sat in the wrong layer: it left the `start()`
gate process-wide, it could not be enforced against a producer that resolves
the registry directly, and it made a Consumer carry scope knowledge that the
other layered registries keep in the registry. `tool-tasks` is scope-agnostic
again and the `dsh-scope` edge moves to `tasks-local`.

`start()`'s refusal is now owner-relative, so its model-visible text names the
agent rather than the process. The shipped `minimal` preset keeps
`enableRunInBackground: false`, no longer as the safety boundary — the registry
owns that now — but so an agent that could never collect a task is not offered
the parameter at all.

Refs #2141
This commit is contained in:
Yichen Jiang
2026-08-10 16:12:41 +08:00
parent 37ebe87087
commit 59e759ce13
36 changed files with 232 additions and 97 deletions

View File

@@ -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: 663ce0d333c6df0f84900a2570d5487d8d7abe55
README.zh.md: a5d1acaa50f63dc95b7607657b157272c15a4f11
README.md: 80d7932466188955ade1c14e4968d51b739ba818
README.zh.md: 5e4263e4685be64f91ec2a7c74edbf89e1148866

View File

@@ -12,6 +12,8 @@ Service disposal closes listeners, cancels all live tasks, awaits their records,
Settlement is first-wins: the earliest terminal outcome — producer settlement, a rejected `done` contained as `failed`, or a teardown force-failure — records once, notifies listeners once with per-listener containment, and releases waiters. Pending waits mark the task reported before listeners run so completion surfaces do not duplicate notices.
Surfaces and listeners are layered by the scope that registered them, in the tools-registry shape: a registration files into its registering context's scope, and a read unions the global layer with the owner's scope chain. One process-wide registry therefore answers per-owner questions per owner — `start()` refuses `background tasks unavailable: no control surface serves this agent (load @deepseek-ai/dsh-tool-tasks in its composition)` for an owner whose own composition attaches none, however many other compositions attach theirs, and a settlement reaches only the listeners its owner's composition registered.
## Model Experience
Indirectly, through producer plugins and [`dsh-tool-tasks`](../tool-tasks/README.md), which render task ids, output, status, cancellation, and completion notices.

View File

@@ -12,6 +12,8 @@
结算遵循首次结算优先原则:最早出现的终止结果(生产方结算、作为 `failed` 隔离处理的 `done` 拒绝,或销毁时的强制失败)只记录一次,也只通知监听器一次;各监听器的故障会单独隔离,随后释放等待方。挂起的等待会在监听器运行前把任务标记为已报告,因此呈现完成情况的表层不会重复发出通知。
表层与监听器按注册方所在的 scope 分层,形状与 tools 注册表一致:一次注册归档到其注册上下文的 scope,一次读取则把全局层与所有者的 scope 链求并集。因此一个进程级注册表能逐所有者地回答逐所有者的问题——对自身组合未附加任何表层的所有者,无论其他组合附加了多少,`start()` 都会拒绝并抛出 `background tasks unavailable: no control surface serves this agent (load @deepseek-ai/dsh-tool-tasks in its composition)`;一次结算也只会抵达其所有者所属组合注册的监听器。
## 模型体验
通过生产方插件和 [`dsh-tool-tasks`](../tool-tasks/README.md) 间接影响;它们会呈现任务 id、输出、状态、取消和完成通知。

View File

@@ -27,6 +27,7 @@
"peerDependencies": {
"@deepseek-ai/dsh-agent": "^0.0.1",
"@deepseek-ai/dsh-invariants": "^0.0.1",
"@deepseek-ai/dsh-scope": "^0.0.1",
"@deepseek-ai/dsh-tasks": "^0.0.1",
"@deepseek-ai/dsh-timeout": "^0.0.1",
"cordis": "^4.0.0-rc.7"
@@ -35,6 +36,7 @@
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-scope": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-tasks": "workspace:^",
"@deepseek-ai/dsh-timeout": "workspace:^",

View File

@@ -11,6 +11,8 @@
import { Context } from 'cordis'
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'
import { deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
import { TaskService, TaskId } from '@deepseek-ai/dsh-tasks'
import type { TaskDoneListener, TaskKind, TaskOutcome, TaskRead, TaskSnapshot, TaskStart, TaskStatus } from '@deepseek-ai/dsh-tasks'
@@ -49,6 +51,21 @@ function isTerminal(status: TaskStatus): boolean {
return status === 'completed' || status === 'killed' || status === 'failed'
}
/**
* One scope's contributions: the control surfaces attached from it and the
* completion listeners registered there. Both tables are anonymous because a
* contribution is identified by its own disposer, never by a name a second
* registrant could shadow.
*/
class TaskLayer implements ScopeLayer {
readonly surfaces = new AnonymousEntries<symbol>()
readonly listeners = new AnonymousEntries<TaskDoneListener>()
isEmpty(): boolean {
return this.surfaces.isEmpty() && this.listeners.isEmpty()
}
}
/**
* The in-memory `tasks` registry. See the Service Definition contract in
* `@deepseek-ai/dsh-tasks` for the ownership, isolation, and lifecycle
@@ -57,8 +74,19 @@ function isTerminal(status: TaskStatus): boolean {
export class LocalTaskService extends TaskService {
private store = new Map<TaskId, TrackedTask>()
private counters = new Map<string, number>()
private surfaces = new Set<symbol>()
private listeners = new Set<TaskDoneListener>()
/**
* Surfaces and listeners layered by the scope that registered them, in the
* tools-registry shape: a contribution files into its registering context's
* scope, and a read unions the global layer with the reader's scope chain.
*
* The registry is one process-wide instance serving every composition, so a
* flat table would answer a per-owner question process-wide: one preset's
* task controls would hold `start()` open for an agent whose own composition
* loads none, and one settlement would reach every preset's notice listener.
* Layers make both reads owner-relative. Nothing derives a cache from a
* layer, so change notification is a no-op.
*/
private readonly layers = new ScopedLayers<TaskLayer>(() => new TaskLayer(), () => {})
private listenersClosed = false
/** Owner agents with attached scope cleanup, mapped to the exact disposer. */
private ownerCleanups = new Map<Agent, () => Promise<void> | void>()
@@ -72,8 +100,8 @@ export class LocalTaskService extends TaskService {
}
start(spec: TaskStart): TaskId {
if (this.surfaces.size === 0) {
throw new Error('background tasks unavailable: no control surface is attached (load @deepseek-ai/dsh-tool-tasks)')
if (!this.servesOwner(spec.owner)) {
throw new Error('background tasks unavailable: no control surface serves this agent (load @deepseek-ai/dsh-tool-tasks in its composition)')
}
if (spec.kind.length === 0) throw new Error('invalid task kind: expected a non-empty string')
if (spec.label.length === 0) throw new Error('invalid task label: expected a non-empty string')
@@ -210,23 +238,53 @@ export class LocalTaskService extends TaskService {
}
onTaskDone(listener: TaskDoneListener): () => void {
const dispose = this.ctx.effect(() => {
this.listeners.add(listener)
return () => this.listeners.delete(listener)
}, 'tasks.onTaskDone()')
const dispose = this.layers.effect(
this.ctx,
layer => layer.listeners.append(listener),
{ label: 'tasks.onTaskDone()' },
)
return () => void dispose()
}
attachSurface(name: string): () => void {
// One token per call keeps duplicate labels independently disposable.
const token = Symbol(name)
const dispose = this.ctx.effect(() => {
this.surfaces.add(token)
return () => this.surfaces.delete(token)
}, 'tasks.attachSurface()')
const dispose = this.layers.effect(
this.ctx,
layer => layer.surfaces.append(token),
{ label: 'tasks.attachSurface()' },
)
return () => void dispose()
}
/**
* Whether an attached control surface can collect and stop work owned by
* `owner`. The global layer holds every surface attached from an unscoped
* context — a host composition's own controls — and therefore serves every
* owner; a scoped surface serves exactly the agents composed under it.
* @param owner - the task's owner, or undefined for unowned work.
* @returns whether some reachable surface serves the owner.
*/
private servesOwner(owner?: Agent): boolean {
if (!this.layers.global.surfaces.isEmpty()) return true
return this.layers.chainLayers(owner === undefined ? undefined : scopeOf(owner.ctx))
.some(layer => !layer.surfaces.isEmpty())
}
/**
* 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
* that chain belongs to another composition and must not deliver, or the
* owner reads one notice per mounted preset.
* @param owner - the settled task's owner, or undefined for unowned work.
* @returns the listeners to notify, in registration order per layer.
*/
private *listenersFor(owner?: Agent): IterableIterator<TaskDoneListener> {
yield* this.layers.global.listeners.values()
const scope = owner === undefined ? undefined : scopeOf(owner.ctx)
for (const layer of this.layers.chainLayers(scope)) yield* layer.listeners.values()
}
/** Look up a task or fail loud. */
private expect(id: TaskId): TrackedTask {
const task = this.store.get(id)
@@ -276,7 +334,7 @@ export class LocalTaskService extends TaskService {
if (task.waiters > 0) task.reported = true
if (!this.listenersClosed) {
const snapshot = this.snapshot(task)
for (const listener of this.listeners) {
for (const listener of this.listenersFor(task.owner)) {
try {
const returned = listener(snapshot, task.owner)
void Promise.resolve(returned).catch((error: unknown) => {
@@ -330,8 +388,9 @@ export class LocalTaskService extends TaskService {
* effects. Throwing cancels are force-failed to avoid teardown deadlock.
*/
private async disposeAll(): Promise<void> {
// The flag is the whole guard: each layer entry's undo belongs to the fiber
// that registered it, so this service may not drop them on its own way out.
this.listenersClosed = true
this.listeners.clear()
const all = [...this.store.values()]
this.cancelForTeardown(all, 'tasks service disposed')
await Promise.all(all.map(task => task.settled))

View File

@@ -3,6 +3,8 @@ import { Context } from 'cordis'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
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'
@@ -15,9 +17,18 @@ declare module '@deepseek-ai/dsh-tasks' {
const agentScopeDisposers = new WeakMap<Agent, () => Promise<void>>()
function stubAgent(ctx: Context, rawId: string): Agent {
function stubAgent(ctx: Context, rawId: string, presetScope?: ScopeKey): Agent {
const id = SessionId(rawId)
const scopeFiber = ctx.plugin(() => {})
// `presetScope` reproduces what `agentPresets.compose` does: the agent gets
// its own key parented to the standing mount's, so the registry's chain walk
// reaches that preset's layer.
let agentCtx = scopeFiber.ctx
if (presetScope !== undefined) {
const key = {}
bindScopeParent(key, presetScope)
agentCtx = createScope(scopeFiber.ctx, key).ctx
}
const session = Session.create(id)
const agent = {
id,
@@ -25,7 +36,7 @@ function stubAgent(ctx: Context, rawId: string): Agent {
session,
inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }),
status: 'idle' as const,
ctx: scopeFiber.ctx,
ctx: agentCtx,
send: () => {},
followup: () => {},
steer: () => ({ outcome: Promise.resolve({ status: 'rejected' as const }) }),
@@ -73,6 +84,21 @@ async function harness() {
return ctx
}
/**
* Attach a control surface the way `tool-tasks` does: from a plugin whose own
* `inject` resolves `ctx.tasks`, so the service method binds to the REGISTERING
* context and the surface files into that context's scope layer. Reading the
* service off a bare scoped context instead throws `cannot get property "tasks"
* without inject`, which is the same rule the shipped plugin obeys.
* @param ctx - the context whose scope should own the surface.
*/
async function attachSurfaceIn(ctx: Context): Promise<void> {
await ctx.plugin({
inject: ['tasks'],
apply(pluginCtx: Context) { pluginCtx.tasks.attachSurface('tool-tasks') },
})
}
/** Let the settlement continuation (a `done.then`) run. */
const tick = () => new Promise<void>(r => setTimeout(r, 0))
@@ -89,11 +115,48 @@ describe('LocalTaskService.start', () => {
expectTypeOf<TaskSnapshot['ownerSession']>().toEqualTypeOf<SessionId | undefined>()
})
it('refuses to register while no control surface is attached', async () => {
it('refuses to register while no control surface serves the owner', async () => {
const ctx = new Context()
await ctx.plugin(LocalTaskService)
expect(() => ctx.tasks.start(producer().spec))
.toThrow('background tasks unavailable: no control surface is attached (load @deepseek-ai/dsh-tool-tasks)')
.toThrow('background tasks unavailable: no control surface serves this agent (load @deepseek-ai/dsh-tool-tasks in its composition)')
})
it('refuses an owner whose own composition attaches no surface', async () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
await ctx.plugin(LocalTaskService)
// Two standing preset mounts over one registry; only the first loads the
// task controls. The second must not inherit the first's open gate.
const withControls = createScope(ctx, {})
const withoutControls = createScope(ctx, {})
await attachSurfaceIn(withControls.ctx)
const served = stubAgent(ctx, 'served', scopeOf(withControls.ctx))
const unserved = stubAgent(ctx, 'unserved', scopeOf(withoutControls.ctx))
ctx.agents.register(served)
ctx.agents.register(unserved)
expect(() => ctx.tasks.start(producer({ owner: served }).spec)).not.toThrow()
expect(() => ctx.tasks.start(producer({ owner: unserved }).spec))
.toThrow('no control surface serves this agent')
// An unowned producer has no chain to walk, so only a global surface serves it.
expect(() => ctx.tasks.start(producer().spec))
.toThrow('no control surface serves this agent')
})
it('lets a surface attached without a scope serve every owner', async () => {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
await ctx.plugin(LocalTaskService)
// The host-plane composition's own controls: no scope, so the global layer
// holds them and every owner's read includes it.
await attachSurfaceIn(ctx)
const scoped = stubAgent(ctx, 'scoped', scopeOf(createScope(ctx, {}).ctx))
ctx.agents.register(scoped)
expect(() => ctx.tasks.start(producer({ owner: scoped }).spec)).not.toThrow()
expect(() => ctx.tasks.start(producer().spec)).not.toThrow()
})
it('rejects an empty kind, empty label, and invalid output limit', async () => {
@@ -757,6 +820,6 @@ describe('LocalTaskService disposal', () => {
detachA2()
expect(() => ctx.tasks.start(producer().spec)).not.toThrow() // b remains
await fiber.dispose() // detaches b with its fiber (HMR safety)
expect(() => ctx.tasks.start(producer().spec)).toThrow('no control surface is attached')
expect(() => ctx.tasks.start(producer().spec)).toThrow('no control surface serves this agent')
})
})

View File

@@ -17,6 +17,9 @@
{
"path": "../../core/agent"
},
{
"path": "../../core/scope"
},
{
"path": "../../util/timeout"
},