refactor(tasks): declare-then-execute — ctx.tasks.start() replaces register()

start({ kind, label, owner, run }) preflights everything that can fail
(the attachSurface fence, validation, the owner-cleanup attach) BEFORE
invoking the producer's run() starter, then commits atomically —
'work started but never got a collectable id' is now structurally
impossible instead of a producer try/catch rollback obligation (the
P1 review fix, rebuilt on #185's declare/execute split). Producers
lose their catch-wraps; the leak tests now pin the stronger property
that a failed preflight never spawns anything. TaskRegistration splits
into TaskStart (identity + run) and TaskHooks (cancel/done/readOutput);
docs, type-equiv manifest, catalogs, and both RFCs move with it.
This commit is contained in:
Yichen Jiang
2026-07-09 21:53:48 +08:00
parent f858c647a6
commit bd59fddacd
22 changed files with 281 additions and 240 deletions

View File

@@ -7,4 +7,4 @@ The shared background-task runtime: ONE home for task ids, owner isolation, poll
| [`tasks`](tasks/README.md) (`@deepseek-ai/dsh-tasks`) | `ctx.tasks` | The registry service: branded `<kind>-N` ids, owner-fenced read/kill/wait/list, settlement bookkeeping, the awaited owner-cleanup path, and the `attachSurface` misconfiguration fence |
| [`tool-tasks`](tool-tasks/README.md) (`@deepseek-ai/dsh-tool-tasks`) | — | The model-facing control surface: `task_output`, `task_list`, `task_kill`, the completion-notice injection, and the background-habit prompt section |
The split is the state/surface boundary: the registry holds task state (an HMR reload of any tool plugin never orphans or kills a running task), while the tool surface is stateless presentation. Producers (`dsh-tool-bash`, `dsh-tool-subagent`) register running work via `ctx.tasks.register` and keep their own execution concerns; whether a producer exposes `run_in_background` is that producer's own `enableRunInBackground` config, never rewritten by this family.
The split is the state/surface boundary: the registry holds task state (an HMR reload of any tool plugin never orphans or kills a running task), while the tool surface is stateless presentation. Producers (`dsh-tool-bash`, `dsh-tool-subagent`) hand their work to `ctx.tasks.start` (preflight, then the producer's starter, then an atomic commit) and keep their own execution concerns; whether a producer exposes `run_in_background` is that producer's own `enableRunInBackground` config, never rewritten by this family.

View File

@@ -4,7 +4,7 @@ The background task registry (`ctx.tasks`): a runtime-global, CONCRETE service (
## Service API
- `register(registration): TaskId` — a producer hands over running work: `kind` (also the id prefix), `label`, optional `owner: Agent`, `cancel(reason?)`, `done: Promise<TaskOutcome>` (settles at QUIESCENCE, never rejects), optional `readOutput()` (stream kinds; absence = final-output-only). Throws while no control surface is attached — the loud fence against a deployment exposing `run_in_background` with no way to collect or stop the work — and is ATOMIC: a failed registration mutates nothing (no stored task, no counter bump, no owner-cleanup bookkeeping), so producers can reliably cancel their just-started work and rethrow.
- `start(spec): TaskId` — declare-then-execute: the producer hands identity (`kind` also the id prefix `label`, optional `owner: Agent`) plus `run()`, the starter that returns the work's `TaskHooks` (`cancel(reason?)`, `done: Promise<TaskOutcome>` settling at QUIESCENCE and never rejecting, optional `readOutput()` for stream kinds; absence = final-output-only). Every check that can fail — the control-surface fence (the loud guard against a deployment exposing `run_in_background` with no way to collect or stop the work), validation, the owner-cleanup attach — runs BEFORE `run()` starts the actual work, and nothing can fail after it returns: work started without a collectable id is structurally impossible, not a producer rollback obligation.
- `get(id, caller?)` / `list(caller?)` — non-consuming snapshots; `list` returns only caller-owned plus unowned tasks (a global listing would leak foreign labels).
- `read(id, caller?): TaskRead` — stream kinds consume the per-task cursor (v1's single intended reader is the owning model — a non-consuming multi-reader surface would be a cursor/snapshot API extension, not a `read` change); final kinds read the terminal output idempotently.
- `kill(id, caller?, reason?)``'requested'` (live task: producer `cancel` runs first — a throw fails the kill loud and leaves the task untouched — then `stopping`) or `'already-terminal'`. Every successful kill marks the task `reported` (the killer saw the end → completion notice suppressed).

View File

@@ -4,8 +4,9 @@
* snapshots, incremental/final output reads, cancellation, wait-for-terminal,
* completion listeners, and the awaited owner-cleanup path. Producers
* (`dsh-tool-bash` background commands, `dsh-tool-subagent` background
* delegations, future long-running tools) register running work via
* {@link TaskService.register} and keep their own execution concerns; the
* delegations, future long-running tools) hand their work to
* {@link TaskService.start} — preflight, then the producer's starter, then an
* atomic commit — and keep their own execution concerns; the
* model-facing control surface (`@deepseek-ai/dsh-tool-tasks`) drives the
* generic read/list/kill/wait operations.
*
@@ -32,15 +33,16 @@ import { Context, Service } from 'cordis'
import type { Agent, AgentId } from '@deepseek-ai/dsh-agent'
import { deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
import { TaskId } from './types.ts'
import type { TaskDoneListener, TaskOutcome, TaskRead, TaskRegistration, TaskSnapshot, TaskStatus } from './types.ts'
import type { TaskDoneListener, TaskOutcome, TaskRead, TaskSnapshot, TaskStart, TaskStatus } from './types.ts'
export { TaskId } from './types.ts'
export type {
TaskDoneListener,
TaskHooks,
TaskOutcome,
TaskRead,
TaskRegistration,
TaskSnapshot,
TaskStart,
TaskStatus,
} from './types.ts'
@@ -114,44 +116,48 @@ export class TaskService extends Service {
}
/**
* Register running background work and receive its task id (`<kind>-N`,
* per-kind counter). The registry attaches ONE continuation to
* `registration.done` that records the terminal snapshot, notifies
* {@link onTaskDone} listeners, and releases waiters; an owned task also
* gets the owner's awaited disposal cleanup attached (once per owner agent)
* through `ctx.agents.onCleanup`. Throws when no control surface is
* attached ({@link attachSurface}) — a task the model could never read or
* stop must fail loud at the start, not dangle — and for an empty
* kind/label. ATOMIC: a throw mutates no registry state, so a producer can
* cancel its just-started work and rethrow without leaving a stored task
* behind.
* @param registration - the producer's task contract (see {@link TaskRegistration}).
* PREFLIGHT, start, then atomically register background work; returns its
* task id (`<kind>-N`, per-kind counter). Every check that can fail — the
* control-surface fence ({@link attachSurface}; a task the model could
* never read or stop must fail loud before it exists), kind/label
* validation, and the owner's awaited disposal-cleanup attach (once per
* owner agent, through `ctx.agents.onCleanup`) — runs BEFORE
* `spec.run()` starts the actual work, and nothing in the runtime can fail
* after it returns: "work started but never got a collectable id" is
* structurally impossible, not a producer rollback obligation. The runtime
* attaches ONE continuation to the returned `done` that records the
* terminal snapshot, notifies {@link onTaskDone} listeners, and releases
* waiters. A throwing `run()` propagates with nothing registered (the
* producer owns any partial cleanup of its own failed start).
* @param spec - the task's identity/owner plus the `run()` starter (see {@link TaskStart}).
* @returns the registry-issued task id.
*/
register(registration: TaskRegistration): TaskId {
start(spec: TaskStart): TaskId {
// -- Preflight: everything that can throw, before any work or mutation. --
if (this.surfaces.size === 0) {
throw new Error('background tasks unavailable: no control surface is attached (load @deepseek-ai/dsh-tool-tasks)')
}
if (registration.kind.length === 0) throw new Error('invalid task kind: expected a non-empty string')
if (registration.label.length === 0) throw new Error('invalid task label: expected a non-empty string')
// EVERYTHING that can throw runs before any mutation (counter, store):
// a failed registration must leave the registry exactly as it was — no
// stored-but-unreturned task the producer could never read or kill.
if (registration.owner !== undefined) this.ensureOwnerCleanup(registration.owner)
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')
if (spec.owner !== undefined) this.ensureOwnerCleanup(spec.owner)
const count = (this.counters.get(registration.kind) ?? 0) + 1
this.counters.set(registration.kind, count)
const id = TaskId(`${registration.kind}-${count}`)
// -- Start: the producer's work begins only now, preflight-clean. --
const hooks = spec.run()
// -- Commit: pure mutations; nothing below can throw. --
const count = (this.counters.get(spec.kind) ?? 0) + 1
this.counters.set(spec.kind, count)
const id = TaskId(`${spec.kind}-${count}`)
let markSettled!: () => void
const settled = new Promise<void>((resolve) => { markSettled = resolve })
const task: TrackedTask = {
id,
kind: registration.kind,
label: registration.label,
ownerSession: registration.owner?.session.header.id,
cancel: registration.cancel.bind(registration),
readOutput: registration.readOutput?.bind(registration),
kind: spec.kind,
label: spec.label,
ownerSession: spec.owner?.session.header.id,
cancel: hooks.cancel.bind(hooks),
readOutput: hooks.readOutput?.bind(hooks),
status: 'running',
detail: undefined,
output: undefined,
@@ -164,7 +170,7 @@ export class TaskService extends Service {
}
this.store.set(id, task)
void registration.done.then(
void hooks.done.then(
(outcome) => { this.settle(task, outcome) },
(error: unknown) => {
// Producer contract violation (`done` must never reject) — contained
@@ -325,7 +331,7 @@ export class TaskService extends Service {
/**
* Declare that a control surface capable of reading/stopping tasks is
* loaded. {@link register} refuses to start a background task while NO
* loaded. {@link start} refuses to start a background task while NO
* surface is attached — the loud fence against a deployment exposing
* `run_in_background` without any way to collect or stop the work. The
* model-facing `@deepseek-ai/dsh-tool-tasks` attaches on load; a deployment

View File

@@ -1,7 +1,8 @@
/**
* Task-registry vocabulary: the registration a producer hands to
* {@link TaskService.register} and the snapshots/reads consumers get back.
* Types only — the service lives in `./index.ts`.
* Task-runtime vocabulary: the {@link TaskStart} a producer hands to
* {@link TaskService.start} (identity + the `run()` starter), the
* {@link TaskHooks} its work is driven through, and the snapshots/reads
* consumers get back. Types only — the service lives in `./index.ts`.
*
* @module @deepseek-ai/dsh-tasks/types
*/
@@ -36,7 +37,7 @@ export function TaskId(id: string): TaskId {
export type TaskStatus = 'running' | 'stopping' | 'completed' | 'killed' | 'failed'
/**
* The terminal result a producer's {@link TaskRegistration.done} resolves
* The terminal result a producer's {@link TaskHooks.done} resolves
* with, mapped from the producer's own vocabulary (a process exit, a subagent
* stop reason) into the registry's closed status set.
*/
@@ -46,7 +47,7 @@ export interface TaskOutcome {
/** Kind-specific detail rendered into status lines ('exit code: 3', 'max-tokens'). */
detail?: string
/**
* Final output for FINAL-OUTPUT-ONLY kinds (no {@link TaskRegistration.readOutput}),
* Final output for FINAL-OUTPUT-ONLY kinds (no {@link TaskHooks.readOutput}),
* read idempotently after the task settles. Stream kinds leave it unset —
* their output is consumed incrementally through `readOutput`.
*/
@@ -54,13 +55,15 @@ export interface TaskOutcome {
}
/**
* What a producer registers with {@link TaskService.register}: the running
* work's identity, its owner, and the three hooks the registry drives it
* through. The producer stays the owner of its execution concerns (process
* streams, child agents); the registry owns ids, isolation, status, and
* completion fan-out.
* What a producer hands to {@link TaskService.start}: the task's identity and
* owner (preflighted BEFORE any work starts), plus {@link run} — the starter
* the runtime invokes only once preflight cannot fail anymore. The producer
* stays the owner of its execution concerns (process streams, child agents);
* the runtime owns ids, isolation, status, and completion fan-out. This
* declare-then-execute split is what makes "work started but never got a
* collectable id" structurally impossible.
*/
export interface TaskRegistration {
export interface TaskStart {
/** Producer kind — also the id prefix (`bash`, `subagent`, …). Non-empty. */
kind: string
/** One-line model-facing label (the command; the delegation description). */
@@ -69,10 +72,27 @@ export interface TaskRegistration {
* The spawning agent. Its `session.header.id` becomes the task's owner
* token (read/kill/wait/list are fenced to that session), and its disposal
* cancels and awaits the task through the `ctx.agents.onCleanup` seam.
* `undefined` registers an UNOWNED task: open to any caller, alive until the
* `undefined` starts an UNOWNED task: open to any caller, alive until the
* tasks service disposes.
*/
owner?: Agent | undefined
/**
* Start the actual work and return its {@link TaskHooks}. Called EXACTLY
* once, synchronously, after every preflight check (control-surface fence,
* validation, owner-cleanup attach) has passed — nothing in the runtime can
* fail after it returns, so the started work is always registered. A throw
* here propagates with nothing registered; the producer owns any partial
* cleanup of its own failed start.
*/
run(): TaskHooks
}
/**
* The live-work hooks a {@link TaskStart.run} returns: how the runtime
* cancels the work, observes its settlement, and (for stream kinds) reads
* its incremental output.
*/
export interface TaskHooks {
/**
* Request termination. Idempotent, synchronous, and must lead to
* {@link done} settling; a throw propagates to the killer (fail loud — a

View File

@@ -4,7 +4,7 @@ import { Session, SessionId } from '@deepseek-ai/dsh-session'
import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import TaskService, { TaskId } from '@deepseek-ai/dsh-tasks'
import type { TaskOutcome, TaskRegistration, TaskSnapshot } from '@deepseek-ai/dsh-tasks'
import type { TaskHooks, TaskOutcome, TaskSnapshot, TaskStart } from '@deepseek-ai/dsh-tasks'
function stubAgent(rawId: string): Agent {
const id = AgentId(rawId)
@@ -21,19 +21,19 @@ function stubAgent(rawId: string): Agent {
}
}
/** A controllable producer: settle its `done` on demand, record cancels. */
function producer(overrides: Partial<TaskRegistration> = {}) {
/** A controllable producer start-spec: settle its `done` on demand, record cancels. */
function producer(overrides: Partial<Omit<TaskStart, 'run'> & TaskHooks> = {}) {
let settle!: (outcome: TaskOutcome) => void
let reject!: (error: unknown) => void
const cancels: (string | undefined)[] = []
const registration: TaskRegistration = {
kind: 'bash',
label: 'sleep 60',
const { kind = 'bash', label = 'sleep 60', owner, ...hookOverrides } = overrides
const hooks: TaskHooks = {
cancel(reason) { cancels.push(reason) },
done: new Promise<TaskOutcome>((res, rej) => { settle = res; reject = rej }),
...overrides,
...hookOverrides,
}
return { registration, settle, reject, cancels }
const spec: TaskStart = { kind, label, ...owner !== undefined ? { owner } : {}, run: () => hooks }
return { spec, settle, reject, cancels }
}
async function harness() {
@@ -47,25 +47,25 @@ async function harness() {
/** Let the settlement continuation (a `done.then`) run. */
const tick = () => new Promise<void>(r => setTimeout(r, 0))
describe('TaskService.register', () => {
describe('TaskService.start', () => {
it('refuses to register while no control surface is attached', async () => {
const ctx = new Context()
await ctx.plugin(TaskService)
expect(() => ctx.tasks.register(producer().registration))
expect(() => ctx.tasks.start(producer().spec))
.toThrow('background tasks unavailable: no control surface is attached (load @deepseek-ai/dsh-tool-tasks)')
})
it('rejects an empty kind and an empty label', async () => {
const ctx = await harness()
expect(() => ctx.tasks.register(producer({ kind: '' }).registration)).toThrow('invalid task kind')
expect(() => ctx.tasks.register(producer({ label: '' }).registration)).toThrow('invalid task label')
expect(() => ctx.tasks.start(producer({ kind: '' }).spec)).toThrow('invalid task kind')
expect(() => ctx.tasks.start(producer({ label: '' }).spec)).toThrow('invalid task label')
})
it('issues kind-prefixed ids from per-kind counters', async () => {
const ctx = await harness()
expect(ctx.tasks.register(producer().registration)).toBe('bash-1')
expect(ctx.tasks.register(producer().registration)).toBe('bash-2')
expect(ctx.tasks.register(producer({ kind: 'subagent' }).registration)).toBe('subagent-1')
expect(ctx.tasks.start(producer().spec)).toBe('bash-1')
expect(ctx.tasks.start(producer().spec)).toBe('bash-2')
expect(ctx.tasks.start(producer({ kind: 'subagent' }).spec)).toBe('subagent-1')
})
})
@@ -74,7 +74,7 @@ describe('TaskService reads and settlement', () => {
const ctx = await harness()
const chunks = ['first', '', 'rest']
const p = producer({ readOutput: () => chunks.shift() ?? '' })
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
expect(ctx.tasks.read(id)).toMatchObject({ text: 'first', snapshot: { status: 'running', reported: false } })
expect(ctx.tasks.read(id).text).toBe('')
@@ -90,7 +90,7 @@ describe('TaskService reads and settlement', () => {
it('final-output kinds read empty while live, the outcome output idempotently once settled', async () => {
const ctx = await harness()
const p = producer({ kind: 'subagent', label: 'research task' })
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
expect(ctx.tasks.read(id)).toMatchObject({ text: '', snapshot: { status: 'running' } })
@@ -103,7 +103,7 @@ describe('TaskService reads and settlement', () => {
it('a settled task without output reads as empty text', async () => {
const ctx = await harness()
const p = producer({ kind: 'subagent' })
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
p.settle({ status: 'failed', detail: 'max-tokens' })
await tick()
expect(ctx.tasks.read(id)).toMatchObject({ text: '', snapshot: { status: 'failed', detail: 'max-tokens' } })
@@ -122,7 +122,7 @@ describe('TaskService reads and settlement', () => {
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
const p = producer()
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
p.settle({ status: 'completed', detail: 'exit code: 0' })
await tick()
@@ -135,7 +135,7 @@ describe('TaskService reads and settlement', () => {
const ctx = await harness()
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
const p = producer()
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
p.reject(new Error('transport exploded'))
await tick()
@@ -155,7 +155,7 @@ describe('TaskService reads and settlement', () => {
detach()
const p = producer()
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
await tick()
expect(seen).toEqual([])
@@ -168,7 +168,7 @@ describe('TaskService.kill', () => {
const seen: TaskSnapshot[] = []
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
const p = producer()
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
expect(ctx.tasks.kill(id, undefined, 'no longer needed')).toBe('requested')
expect(p.cancels).toEqual(['no longer needed'])
@@ -184,7 +184,7 @@ describe('TaskService.kill', () => {
it('reports an already-terminal task instead of failing', async () => {
const ctx = await harness()
const p = producer()
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
await tick()
expect(ctx.tasks.kill(id)).toBe('already-terminal')
@@ -196,11 +196,13 @@ describe('TaskService.kill', () => {
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
let broken = true
let settle!: (outcome: TaskOutcome) => void
const id = ctx.tasks.register({
const id = ctx.tasks.start({
kind: 'bash',
label: 'flaky cancel',
cancel() { if (broken) throw new Error('cancel boom') },
done: new Promise<TaskOutcome>((res) => { settle = res }),
run: () => ({
cancel() { if (broken) throw new Error('cancel boom') },
done: new Promise<TaskOutcome>((res) => { settle = res }),
}),
})
expect(() => ctx.tasks.kill(id)).toThrow('cancel boom')
// The failed kill mutated NOTHING: still running, notice not suppressed,
@@ -221,7 +223,7 @@ describe('TaskService.wait', () => {
const seen: TaskSnapshot[] = []
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
const p = producer()
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
const wait = ctx.tasks.wait(id, 5_000)
p.settle({ status: 'completed', detail: 'exit code: 0' })
@@ -232,14 +234,14 @@ describe('TaskService.wait', () => {
it('returns the live snapshot on timeout without marking reported', async () => {
const ctx = await harness()
const id = ctx.tasks.register(producer().registration)
const id = ctx.tasks.start(producer().spec)
expect(await ctx.tasks.wait(id, 5)).toMatchObject({ status: 'running', reported: false })
})
it('returns immediately for an already-terminal task', async () => {
const ctx = await harness()
const p = producer()
const id = ctx.tasks.register(p.registration)
const id = ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
await tick()
expect(await ctx.tasks.wait(id, 5_000)).toMatchObject({ status: 'completed', reported: true })
@@ -247,14 +249,14 @@ describe('TaskService.wait', () => {
it('rejects a non-positive or non-finite timeout', async () => {
const ctx = await harness()
const id = ctx.tasks.register(producer().registration)
const id = ctx.tasks.start(producer().spec)
await expect(ctx.tasks.wait(id, 0)).rejects.toThrow('invalid wait timeout')
await expect(ctx.tasks.wait(id, Number.NaN)).rejects.toThrow('invalid wait timeout')
})
it('an aborted signal rejects the wait only — the task stays alive', async () => {
const ctx = await harness()
const id = ctx.tasks.register(producer().registration)
const id = ctx.tasks.start(producer().spec)
const controller = new AbortController()
const wait = ctx.tasks.wait(id, 5_000, undefined, controller.signal)
@@ -275,8 +277,8 @@ describe('TaskService owner isolation', () => {
ctx.agents.register(owner)
const other = stubAgent('other')
const owned = ctx.tasks.register(producer({ owner }).registration)
const open = ctx.tasks.register(producer().registration)
const owned = ctx.tasks.start(producer({ owner }).spec)
const open = ctx.tasks.start(producer().spec)
// The owner and the unowned task are reachable.
expect(ctx.tasks.read(owned, owner).snapshot.id).toBe(owned)
@@ -296,9 +298,9 @@ describe('TaskService owner isolation', () => {
ctx.agents.register(alice)
ctx.agents.register(bob)
const aliceTask = ctx.tasks.register(producer({ owner: alice }).registration)
const bobTask = ctx.tasks.register(producer({ owner: bob }).registration)
const openTask = ctx.tasks.register(producer({ kind: 'subagent' }).registration)
const aliceTask = ctx.tasks.start(producer({ owner: alice }).spec)
const bobTask = ctx.tasks.start(producer({ owner: bob }).spec)
const openTask = ctx.tasks.start(producer({ kind: 'subagent' }).spec)
expect(ctx.tasks.list(alice).map(t => t.id)).toEqual([aliceTask, openTask])
expect(ctx.tasks.list(bob).map(t => t.id)).toEqual([bobTask, openTask])
@@ -309,11 +311,11 @@ describe('TaskService owner isolation', () => {
const ctx = new Context()
await ctx.plugin(TaskService)
ctx.tasks.attachSurface('test-surface')
expect(() => ctx.tasks.register(producer({ owner: stubAgent('a') }).registration))
expect(() => ctx.tasks.start(producer({ owner: stubAgent('a') }).spec))
.toThrow('background task ownership requires the agent registry')
// The failed registration mutated nothing: no stored task, counter untouched.
expect(ctx.tasks.list()).toEqual([])
expect(ctx.tasks.register(producer().registration)).toBe('bash-1')
expect(ctx.tasks.start(producer().spec)).toBe('bash-1')
})
it('a failed owner-cleanup attach leaves the registry unchanged and does not poison the owner', async () => {
@@ -321,7 +323,7 @@ describe('TaskService owner isolation', () => {
const ghost = stubAgent('ghost') // never registered in ctx.agents
// onCleanup rejects the unregistered agent BEFORE any registry mutation.
expect(() => ctx.tasks.register(producer({ owner: ghost }).registration))
expect(() => ctx.tasks.start(producer({ owner: ghost }).spec))
.toThrow('is not registered')
expect(ctx.tasks.list(ghost)).toEqual([])
@@ -330,12 +332,14 @@ describe('TaskService owner isolation', () => {
ctx.agents.register(ghost)
const cancels: (string | undefined)[] = []
let settle!: (outcome: TaskOutcome) => void
const id = ctx.tasks.register({
const id = ctx.tasks.start({
kind: 'bash',
label: 'after retry',
owner: ghost,
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
run: () => ({
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
}),
})
expect(id).toBe('bash-1') // the failed attempt burned no counter
await ctx.agents.drainCleanups(ghost.id)
@@ -353,15 +357,17 @@ describe('TaskService owner cleanup', () => {
// The producer settles only when cancelled — models a child that stops on request.
let settle!: (outcome: TaskOutcome) => void
const cancels: (string | undefined)[] = []
ctx.tasks.register({
ctx.tasks.start({
kind: 'subagent',
label: 'long research',
owner,
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
run: () => ({
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
}),
})
const terminal = producer({ owner })
ctx.tasks.register(terminal.registration)
ctx.tasks.start(terminal.spec)
terminal.settle({ status: 'completed' })
await tick()
@@ -378,8 +384,8 @@ describe('TaskService owner cleanup', () => {
const first = producer({ owner })
const second = producer({ owner })
ctx.tasks.register(first.registration)
ctx.tasks.register(second.registration)
ctx.tasks.start(first.spec)
ctx.tasks.start(second.spec)
first.settle({ status: 'completed' })
second.settle({ status: 'completed' })
await tick()
@@ -387,7 +393,7 @@ describe('TaskService owner cleanup', () => {
// A fresh task after the drain gets a fresh cleanup (the set was consumed).
const third = producer({ owner })
ctx.tasks.register(third.registration)
ctx.tasks.start(third.spec)
third.settle({ status: 'completed' })
await tick()
expect(ctx.tasks.list(owner)).toHaveLength(1)
@@ -402,12 +408,14 @@ describe('TaskService owner cleanup', () => {
ctx.agents.register(owner)
let settle!: (outcome: TaskOutcome) => void
ctx.tasks.register({
ctx.tasks.start({
kind: 'bash',
label: 'broken producer',
owner,
cancel() { throw new Error('cancel boom') },
done: new Promise<TaskOutcome>((res) => { settle = res }),
run: () => ({
cancel() { throw new Error('cancel boom') },
done: new Promise<TaskOutcome>((res) => { settle = res }),
}),
})
const drain = ctx.agents.drainCleanups(owner.id)
@@ -432,11 +440,13 @@ describe('TaskService disposal', () => {
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot.id))
let settle!: (outcome: TaskOutcome) => void
const cancels: (string | undefined)[] = []
ctx.tasks.register({
ctx.tasks.start({
kind: 'bash',
label: 'sleep 600',
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
run: () => ({
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
}),
})
await fiber.dispose()
@@ -456,10 +466,10 @@ describe('TaskService disposal', () => {
detachA1()
detachA1() // second call of the same disposer is a no-op
expect(() => ctx.tasks.register(producer().registration)).not.toThrow() // a ×1 + b remain
expect(() => ctx.tasks.start(producer().spec)).not.toThrow() // a ×1 + b remain
detachA2()
expect(() => ctx.tasks.register(producer().registration)).not.toThrow() // b remains
expect(() => ctx.tasks.start(producer().spec)).not.toThrow() // b remains
await fiber.dispose() // detaches b with its fiber (HMR safety)
expect(() => ctx.tasks.register(producer().registration)).toThrow('no control surface is attached')
expect(() => ctx.tasks.start(producer().spec)).toThrow('no control surface is attached')
})
})

View File

@@ -1,6 +1,6 @@
# @deepseek-ai/dsh-tool-tasks
The model-facing background task control surface over `ctx.tasks`: three kind-agnostic tools, the completion-notice injection, and the prompt section that teaches the background habit. Loading this plugin calls `ctx.tasks.attachSurface('tool-tasks')`, which is what arms producers' `register()`.
The model-facing background task control surface over `ctx.tasks`: three kind-agnostic tools, the completion-notice injection, and the prompt section that teaches the background habit. Loading this plugin calls `ctx.tasks.attachSurface('tool-tasks')`, which is what arms producers' `ctx.tasks.start()`.
## Tools

View File

@@ -6,8 +6,8 @@
* registry (`@deepseek-ai/dsh-tasks`).
*
* This plugin IS the control surface: it calls `ctx.tasks.attachSurface()` on
* load, which is what re-arms producers' `register()` (the registry refuses
* background work while no surface could collect or stop it).
* load, which is what arms producers' `ctx.tasks.start()` (the runtime's
* preflight refuses background work while no surface could collect or stop it).
*
* Completion notices: when a task settles, a short notice is injected into
* the owning agent's session (`agent.inject()` — durable context for the NEXT

View File

@@ -6,7 +6,7 @@ import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import TaskService from '@deepseek-ai/dsh-tasks'
import type { TaskOutcome, TaskRegistration, TaskSnapshot } from '@deepseek-ai/dsh-tasks'
import type { TaskHooks, TaskOutcome, TaskSnapshot, TaskStart } from '@deepseek-ai/dsh-tasks'
import * as ToolTasks from '@deepseek-ai/dsh-tool-tasks'
import { statusLine } from '@deepseek-ai/dsh-tool-tasks'
@@ -32,18 +32,18 @@ function fakeAgent(ctx: Context, sessionId: string, inject: (...args: unknown[])
return agent
}
/** A controllable producer registration (settle `done` on demand, record cancels). */
function producer(overrides: Partial<TaskRegistration> = {}) {
/** A controllable producer start-spec (settle `done` on demand, record cancels). */
function producer(overrides: Partial<Omit<TaskStart, 'run'> & TaskHooks> = {}) {
let settle!: (outcome: TaskOutcome) => void
const cancels: (string | undefined)[] = []
const registration: TaskRegistration = {
kind: 'bash',
label: 'sleep 60',
const { kind = 'bash', label = 'sleep 60', owner, ...hookOverrides } = overrides
const hooks: TaskHooks = {
cancel(reason) { cancels.push(reason) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
...overrides,
...hookOverrides,
}
return { registration, settle, cancels }
const spec: TaskStart = { kind, label, ...owner !== undefined ? { owner } : {}, run: () => hooks }
return { spec, settle, cancels }
}
let callCounter = 0
@@ -60,9 +60,9 @@ const tick = () => new Promise<void>(r => setTimeout(r, 0))
describe('tool-tasks setup', () => {
it('attaches the control surface on load and detaches it with the fiber', async () => {
const { ctx, toolsFiber } = await setup()
expect(() => ctx.tasks.register(producer().registration)).not.toThrow()
expect(() => ctx.tasks.start(producer().spec)).not.toThrow()
await toolsFiber.dispose()
expect(() => ctx.tasks.register(producer().registration)).toThrow('no control surface is attached')
expect(() => ctx.tasks.start(producer().spec)).toThrow('no control surface is attached')
})
it('rejects a config whose default wait exceeds the cap', async () => {
@@ -89,7 +89,7 @@ describe('tool-tasks setup', () => {
await ctx.plugin(TaskService)
ToolTasks.apply(ctx, {})
expect(ctx.tools.get('task_output')).toBeDefined()
expect(() => ctx.tasks.register(producer().registration)).not.toThrow()
expect(() => ctx.tasks.start(producer().spec)).not.toThrow()
})
})
@@ -97,7 +97,7 @@ describe('task_output', () => {
it('reads a consuming delta with a trailing status line', async () => {
const { ctx } = await setup()
const chunks = ['line one\n', '']
ctx.tasks.register(producer({ readOutput: () => chunks.shift() ?? '' }).registration)
ctx.tasks.start(producer({ readOutput: () => chunks.shift() ?? '' }).spec)
// A body already ending in a newline gets no doubled separator.
expect(text(await call(ctx, 'task_output', { task_id: 'bash-1' }))).toBe('line one\n[status: running]')
@@ -107,7 +107,7 @@ describe('task_output', () => {
it('returns the final output of a settled final-output task', async () => {
const { ctx } = await setup()
const p = producer({ kind: 'subagent', label: 'research' })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
expect(text(await call(ctx, 'task_output', { task_id: 'subagent-1' }))).toBe('(no new output)\n[status: running]')
p.settle({ status: 'completed', detail: 'completed', output: 'the answer' })
@@ -118,7 +118,7 @@ describe('task_output', () => {
it('wait: true blocks until settlement and reports the terminal state', async () => {
const { ctx } = await setup()
const p = producer({ kind: 'subagent', label: 'research' })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
const pending = call(ctx, 'task_output', { task_id: 'subagent-1', wait: true })
p.settle({ status: 'completed', output: 'done deal' })
@@ -127,7 +127,7 @@ describe('task_output', () => {
it('wait: true times out against the configured cap and leaves the task alive', async () => {
const { ctx } = await setup({ waitTimeoutMs: 10, maxWaitTimeoutMs: 20 })
ctx.tasks.register(producer().registration)
ctx.tasks.start(producer().spec)
// A model-supplied timeout far above the cap is clamped: this returns
// promptly (≤ the 20ms cap), not after ten minutes.
@@ -150,10 +150,10 @@ describe('task_list', () => {
expect(text(await call(ctx, 'task_list', {}))).toBe('(no background tasks)')
const alice = fakeAgent(ctx, 'sess-alice')
ctx.tasks.register(producer({ owner: alice, label: 'pnpm test' }).registration)
ctx.tasks.register(producer({ kind: 'subagent', label: 'open research' }).registration)
ctx.tasks.start(producer({ owner: alice, label: 'pnpm test' }).spec)
ctx.tasks.start(producer({ kind: 'subagent', label: 'open research' }).spec)
const p = producer({ owner: alice, label: 'build' })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
p.settle({ status: 'completed', detail: 'exit code: 0' })
await tick()
@@ -172,7 +172,7 @@ describe('task_kill', () => {
it('requests cancellation with the forwarded reason', async () => {
const { ctx } = await setup()
const p = producer()
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
const result = await call(ctx, 'task_kill', { task_id: 'bash-1', reason: 'superseded' })
expect(text(result)).toBe('requested cancellation of task bash-1')
@@ -183,7 +183,7 @@ describe('task_kill', () => {
const { ctx } = await setup()
let delta = 'unread tail'
const p = producer({ readOutput: () => { const d = delta; delta = ''; return d } })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
p.settle({ status: 'completed', detail: 'exit code: 0' })
await tick()
@@ -217,7 +217,7 @@ describe('completion notices', () => {
const inject = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', inject)
const p = producer({ owner, label: 'pnpm test' })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
p.settle({ status: 'completed', detail: 'exit code: 0' })
await tick()
@@ -233,7 +233,7 @@ describe('completion notices', () => {
const inject = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', inject)
const p = producer({ owner })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
await call(ctx, 'task_kill', { task_id: 'bash-1' }, owner)
p.settle({ status: 'killed' })
@@ -246,7 +246,7 @@ describe('completion notices', () => {
const inject = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', inject)
const p = producer({ owner, kind: 'subagent' })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
const pending = call(ctx, 'task_output', { task_id: 'subagent-1', wait: true }, owner)
p.settle({ status: 'completed', output: 'answer' })
@@ -258,7 +258,7 @@ describe('completion notices', () => {
const { ctx } = await setup()
// Unowned: settles with nobody to notify — nothing throws.
const unowned = producer()
ctx.tasks.register(unowned.registration)
ctx.tasks.start(unowned.spec)
unowned.settle({ status: 'completed' })
await tick()
@@ -266,7 +266,7 @@ describe('completion notices', () => {
const inject = vi.fn(() => { throw new Error('agent "agent-sess-1" is disposed') })
const owner = fakeAgent(ctx, 'sess-1', inject)
const p = producer({ owner })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
await tick()
expect(inject).toHaveBeenCalledTimes(1)
@@ -277,7 +277,7 @@ describe('completion notices', () => {
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
const owner = fakeAgent(ctx, 'sess-1', () => { throw new Error('unexpected inject bug') })
const p = producer({ owner })
ctx.tasks.register(p.registration)
ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
await tick()
// The throw escapes the notice listener and is contained (logged) by the
@@ -292,10 +292,10 @@ describe('completion notices', () => {
// Owner known at registration, unregistered before settlement → no match.
const p1 = producer({ owner })
ctx.tasks.register(p1.registration)
ctx.tasks.start(p1.spec)
// A second task whose settlement happens after the whole registry is gone.
const p2 = producer({ owner })
ctx.tasks.register(p2.registration)
ctx.tasks.start(p2.spec)
await agentsFiber.dispose()
p1.settle({ status: 'completed' })