Merge branch 'codex/simp-agent-entry-state' into codex/simp-unify-agent-session-id
# Conflicts: # docs/cordis-catalog/events.md # docs/cordis-catalog/services.md # docs/core-data-structures/bash.md # docs/core-data-structures/core.md # docs/event-producer-consumer.md # docs/module-graph.md # packages/bash/bash/src/types.ts # packages/bash/tool-bash/README.md # packages/bash/tool-bash/src/index.ts # packages/bash/tool-bash/tests/tools.spec.ts # packages/examples/agent-spine-demo/README.md # packages/examples/agent-spine-demo/tests/agent-core.spec.ts # packages/subagent/tool-subagent/tests/tool-subagent.spec.ts # packages/util/brand/README.md # packages/util/brand/src/index.ts
This commit is contained in:
Tianyi Cui
10
packages/tasks/README.md
Normal file
10
packages/tasks/README.md
Normal file
@@ -0,0 +1,10 @@
|
||||
# tasks/ — background task capability family
|
||||
|
||||
The shared home for background-task ids, owner isolation, reads, cancellation, waiting, and completion notices. Bash, subagents, and future long-running tools use one model-facing protocol. See the [background-task runtime RFC](../../docs/rfc/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md).
|
||||
|
||||
| Package | ctx key | Role |
|
||||
|---|---|---|
|
||||
| [`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 registry owns state across producer or surface reloads; the tool package owns presentation. Producers register execution hooks through `ctx.tasks.start` and own whether their config exposes `run_in_background`.
|
||||
35
packages/tasks/tasks/README.md
Normal file
35
packages/tasks/tasks/README.md
Normal file
@@ -0,0 +1,35 @@
|
||||
# @deepseek-ai/dsh-tasks
|
||||
|
||||
The process-local background task registry (`ctx.tasks`). It gives long-running producers shared ids, owner isolation, reads, cancellation, waiting, notices, and cleanup. Producer plugins extend `TaskKindMap` with their opaque id namespace.
|
||||
|
||||
## Service API
|
||||
|
||||
- `start(spec): TaskId` validates the control surface, spec, and exact live owner before calling the producer's `run()` once. A starter throw leaves nothing registered; 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.
|
||||
- `wait(id, timeoutMs, caller?, signal?)` returns a terminal snapshot or the live snapshot at timeout. Aborting stops only the wait; settlement wins once it has committed terminal delivery to that waiter.
|
||||
- `onTaskDone(listener)` observes each terminal record with the exact owner. Listener throws and rejections are contained; listener work is not awaited.
|
||||
- `attachSurface(name)` declares a control surface for its effect lifetime. `start()` fails before producer execution when none is attached.
|
||||
|
||||
Owned access compares the task's `SessionId` with the caller's. Ids such as `bash-1` are predictable, so this fence is the boundary. Unowned tasks are open to callers and last until service disposal.
|
||||
|
||||
## Lifecycle
|
||||
|
||||
Tasks belong to their owner and backend, not the producer tool fiber, so producer and surface reloads do not stop them. The first task for an owner attaches one awaited effect to the exact `Agent` scope. Owner disposal cancels that object's tasks, awaits producer quiescence, and removes their snapshots; reused agent or session ids cannot redirect an old cleanup.
|
||||
|
||||
Service disposal closes listeners, cancels all live tasks, awaits their records, and detaches effects from surviving owner scopes. If teardown cancellation throws, the service force-fails the record and warns that work may be orphaned instead of deadlocking. A cancellation that returns but never settles `done` remains indistinguishable from a slow stop and can stall teardown.
|
||||
|
||||
See the [task type catalog](../../../docs/core-data-structures/tasks.md) and [runtime RFC](../../../docs/rfc/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md).
|
||||
|
||||
## Model Experience
|
||||
|
||||
Indirectly, through producer plugins and [`dsh-tool-tasks`](../tool-tasks/README.md), which render task ids, output, status, cancellation, and completion notices.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Tasks are process-local** — durable or cross-restart execution needs a separate lifecycle.
|
||||
- **The service and implementation are not split** — a second backend must define the lifecycle that shapes that boundary.
|
||||
- **Stream output has one consuming cursor** — independent observers need a cursor or snapshot API.
|
||||
- **Foreground work cannot be promoted** — producers choose foreground or background before starting.
|
||||
- **A silently ineffective cancel can stall teardown** — only an explicit throw can be force-failed safely.
|
||||
38
packages/tasks/tasks/package.json
Normal file
38
packages/tasks/tasks/package.json
Normal file
@@ -0,0 +1,38 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-tasks",
|
||||
"description": "Background task registry (ctx.tasks) for the DeepSeek Harness \u2014 shared ids, owner isolation, polling, cancellation, and completion listeners for long-running tool work",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "lib/index.js",
|
||||
"types": "lib/types/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./lib/types/index.d.ts",
|
||||
"default": "./lib/index.js"
|
||||
},
|
||||
"./src/*": "./src/*",
|
||||
"./package.json": "./package.json"
|
||||
},
|
||||
"files": [
|
||||
"lib/index.js",
|
||||
"lib/types/**/*.d.ts",
|
||||
"lib/types/**/*.d.ts.map",
|
||||
"src"
|
||||
],
|
||||
"license": "BSD-3-Clause",
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-brand": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"@deepseek-ai/dsh-timeout": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-brand": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-timeout": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
441
packages/tasks/tasks/src/index.ts
Normal file
441
packages/tasks/tasks/src/index.ts
Normal file
@@ -0,0 +1,441 @@
|
||||
/**
|
||||
* The in-process background task registry (`ctx.tasks`). It owns task ids,
|
||||
* session-scoped access, lifecycle state, completion listeners, and owner
|
||||
* cleanup while producers retain their execution resources.
|
||||
*
|
||||
* Registrations outlive producer and control-surface fibers. Agent or service
|
||||
* disposal cancels live work and awaits compliant producers; a throwing
|
||||
* teardown cancel force-fails only the record and reports a possible orphan.
|
||||
* @module @deepseek-ai/dsh-tasks
|
||||
*/
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
|
||||
import { TaskId } from './types.ts'
|
||||
import type { TaskDoneListener, TaskKind, TaskOutcome, TaskRead, TaskSnapshot, TaskStart, TaskStatus } from './types.ts'
|
||||
|
||||
export { TaskId } from './types.ts'
|
||||
export type {
|
||||
TaskDoneListener,
|
||||
TaskHooks,
|
||||
TaskKind,
|
||||
TaskKindMap,
|
||||
TaskOutcome,
|
||||
TaskRead,
|
||||
TaskSnapshot,
|
||||
TaskStart,
|
||||
TaskStatus,
|
||||
} from './types.ts'
|
||||
|
||||
declare module 'cordis' {
|
||||
interface Context {
|
||||
tasks: TaskService
|
||||
}
|
||||
}
|
||||
|
||||
/** Timeout code that distinguishes a bounded wait from caller cancellation. */
|
||||
export const TASK_WAIT_TIMEOUT = 'TASK_WAIT_TIMEOUT'
|
||||
|
||||
/** The registry's mutable per-task record (never handed out — see {@link TaskService.snapshot}). */
|
||||
interface TrackedTask {
|
||||
id: TaskId
|
||||
kind: TaskKind
|
||||
label: string
|
||||
/** Exact lifecycle owner; session-id authorization is derived from it. */
|
||||
owner: Agent | undefined
|
||||
cancel: (reason?: string) => void
|
||||
readOutput: (() => string) | undefined
|
||||
status: TaskStatus
|
||||
detail: string | undefined
|
||||
output: string | undefined
|
||||
startedAt: number
|
||||
finishedAt: number | undefined
|
||||
reported: boolean
|
||||
/** Resolves once the terminal snapshot is recorded and listeners notified. */
|
||||
settled: Promise<void>
|
||||
/** Resolver for {@link settled}, called by the first effective settlement. */
|
||||
markSettled: () => void
|
||||
/** Live waits; settlement with a waiter marks the task reported. */
|
||||
waiters: number
|
||||
/** Removable resolvers for live waits; timeout/abort unregister before the task settles. */
|
||||
waitResolvers: Set<() => void>
|
||||
}
|
||||
|
||||
/** True for the three terminal {@link TaskStatus} values. */
|
||||
function isTerminal(status: TaskStatus): boolean {
|
||||
return status === 'completed' || status === 'killed' || status === 'failed'
|
||||
}
|
||||
|
||||
/**
|
||||
* The `tasks` service: the runtime-global background task registry. See the
|
||||
* module doc for the ownership, isolation, and lifecycle contracts.
|
||||
*/
|
||||
// TODO(task-service-backend): Separate the service contract from this
|
||||
// process-local implementation when a second backend defines its lifecycle.
|
||||
export class TaskService extends Service {
|
||||
private store = new Map<TaskId, TrackedTask>()
|
||||
private counters = new Map<string, number>()
|
||||
private surfaces = new Set<symbol>()
|
||||
private listeners = new Set<TaskDoneListener>()
|
||||
private listenersClosed = false
|
||||
/** Owner agents with attached scope cleanup, mapped to the exact disposer. */
|
||||
private ownerCleanups = new Map<Agent, () => Promise<void> | void>()
|
||||
/** Service context used by detached settlement continuations and teardown. */
|
||||
private readonly selfCtx: Context
|
||||
|
||||
constructor(ctx: Context) {
|
||||
super(ctx, 'tasks')
|
||||
this.selfCtx = ctx
|
||||
ctx.effect(() => () => this.disposeAll(), 'tasks teardown')
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* @param spec - task identity, owner, and synchronous starter.
|
||||
* @returns the registry-issued `<kind>-N` id.
|
||||
*/
|
||||
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 (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 hooks = spec.run()
|
||||
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: spec.kind,
|
||||
label: spec.label,
|
||||
owner: spec.owner,
|
||||
cancel: hooks.cancel.bind(hooks),
|
||||
readOutput: hooks.readOutput?.bind(hooks),
|
||||
status: 'running',
|
||||
detail: undefined,
|
||||
output: undefined,
|
||||
startedAt: Date.now(),
|
||||
finishedAt: undefined,
|
||||
reported: false,
|
||||
settled,
|
||||
markSettled,
|
||||
waiters: 0,
|
||||
waitResolvers: new Set(),
|
||||
}
|
||||
this.store.set(id, task)
|
||||
|
||||
void hooks.done.then(
|
||||
(outcome) => { this.settle(task, outcome) },
|
||||
(error: unknown) => {
|
||||
// Contain a producer contract violation so cleanup and waiters cannot hang.
|
||||
this.selfCtx.logger.warn(`tasks: task ${task.id} 'done' rejected (producer contract violation): ${String(error)}`)
|
||||
this.settle(task, { status: 'failed', detail: String(error) })
|
||||
},
|
||||
)
|
||||
return id
|
||||
}
|
||||
|
||||
/**
|
||||
* List caller-owned and unowned tasks in registration order without exposing
|
||||
* another session's labels.
|
||||
* @param caller - reading agent; a non-agent caller sees only unowned tasks.
|
||||
* @returns fresh snapshots.
|
||||
*/
|
||||
list(caller?: Agent): TaskSnapshot[] {
|
||||
const session = caller?.id
|
||||
return [...this.store.values()]
|
||||
.filter(task => task.owner === undefined || task.owner.id === session)
|
||||
.map(task => this.snapshot(task))
|
||||
}
|
||||
|
||||
/**
|
||||
* Return a non-consuming snapshot without changing its read cursor or notice
|
||||
* state. Throws for an unknown or foreign task.
|
||||
* @param id - task to look up.
|
||||
* @param caller - reading agent checked against the owner.
|
||||
* @returns a fresh snapshot.
|
||||
*/
|
||||
get(id: TaskId, caller?: Agent): TaskSnapshot {
|
||||
const task = this.expect(id)
|
||||
this.assertAccess(task, caller)
|
||||
return this.snapshot(task)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read the next stream delta, or the idempotent final output after settlement.
|
||||
* A terminal read marks the task reported. Throws for an unknown or foreign
|
||||
* task.
|
||||
* @param id - task to read.
|
||||
* @param caller - reading agent checked against the owner.
|
||||
* @returns output text and the post-read snapshot.
|
||||
*/
|
||||
read(id: TaskId, caller?: Agent): TaskRead {
|
||||
const task = this.expect(id)
|
||||
this.assertAccess(task, caller)
|
||||
const text = task.readOutput !== undefined
|
||||
? task.readOutput()
|
||||
: isTerminal(task.status) ? task.output ?? '' : ''
|
||||
if (isTerminal(task.status)) task.reported = true
|
||||
return { text, snapshot: this.snapshot(task) }
|
||||
}
|
||||
|
||||
/**
|
||||
* Request cancellation, then mark the task stopping and reported. A producer
|
||||
* throw propagates without changing task state. Throws for an unknown or
|
||||
* foreign task.
|
||||
* @param id - task to cancel.
|
||||
* @param caller - killing agent checked against the owner.
|
||||
* @param reason - logged reason forwarded to the producer.
|
||||
* @returns `requested` for live work, otherwise `already-finished`.
|
||||
*/
|
||||
kill(id: TaskId, caller?: Agent, reason?: string): 'requested' | 'already-finished' {
|
||||
const task = this.expect(id)
|
||||
this.assertAccess(task, caller)
|
||||
if (isTerminal(task.status)) {
|
||||
task.reported = true
|
||||
return 'already-finished'
|
||||
}
|
||||
// Cancel first so a throw leaves both lifecycle and notice state unchanged.
|
||||
task.cancel(reason)
|
||||
task.status = 'stopping'
|
||||
task.reported = true
|
||||
return 'requested'
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait for settlement or timeout without cancelling the task. Caller abort
|
||||
* rejects only while the task is live; after settlement it returns the
|
||||
* terminal snapshot so a notice suppressed for this waiter is still delivered.
|
||||
* Timed-out and aborted waits detach their resolvers. Throws for invalid,
|
||||
* unknown, or foreign input.
|
||||
* @param id - task to wait for.
|
||||
* @param timeoutMs - positive finite wait bound in milliseconds.
|
||||
* @param caller - waiting agent checked against the owner.
|
||||
* @param signal - optional cancellation of the wait itself.
|
||||
* @returns snapshot at settlement or timeout.
|
||||
*/
|
||||
async wait(id: TaskId, timeoutMs: number, caller?: Agent, signal?: AbortSignal): Promise<TaskSnapshot> {
|
||||
const task = this.expect(id)
|
||||
this.assertAccess(task, caller)
|
||||
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
|
||||
throw new Error(`invalid wait timeout: expected a positive number of milliseconds, got ${JSON.stringify(timeoutMs)}`)
|
||||
}
|
||||
if (!isTerminal(task.status)) {
|
||||
if (signal?.aborted) throw new Error('wait aborted')
|
||||
// Abort removes the waiter synchronously so same-tick settlement cannot
|
||||
// suppress a notice for a wait that will reject.
|
||||
task.waiters += 1
|
||||
let counted = true
|
||||
const uncount = (): void => {
|
||||
if (!counted) return
|
||||
counted = false
|
||||
task.waiters -= 1
|
||||
}
|
||||
try {
|
||||
// The scoped deadline distinguishes a successful wait timeout from
|
||||
// caller cancellation and clears its timer on every exit.
|
||||
using d = deadline(signal, timeoutMs, TASK_WAIT_TIMEOUT)
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const onSettled = (): void => {
|
||||
task.waitResolvers.delete(onSettled)
|
||||
d.signal.removeEventListener('abort', onAbort)
|
||||
resolve()
|
||||
}
|
||||
const onAbort = (): void => {
|
||||
task.waitResolvers.delete(onSettled)
|
||||
if (timeoutOf(d.signal, TASK_WAIT_TIMEOUT) !== undefined) {
|
||||
resolve()
|
||||
} else if (isTerminal(task.status)) {
|
||||
// Settlement suppressed the notice for this waiter; deliver it.
|
||||
resolve()
|
||||
} else {
|
||||
uncount()
|
||||
reject(new Error('wait aborted'))
|
||||
}
|
||||
}
|
||||
task.waitResolvers.add(onSettled)
|
||||
d.signal.addEventListener('abort', onAbort, { once: true })
|
||||
})
|
||||
} finally {
|
||||
uncount()
|
||||
}
|
||||
}
|
||||
if (isTerminal(task.status)) task.reported = true
|
||||
return this.snapshot(task)
|
||||
}
|
||||
|
||||
/**
|
||||
* Register an effect-scoped completion listener. Each listener is contained;
|
||||
* returned promises are observed but not awaited. No listener runs after
|
||||
* service disposal.
|
||||
* @param listener - receives each terminal snapshot and its exact owner.
|
||||
* @returns disposer that unregisters the listener.
|
||||
*/
|
||||
onTaskDone(listener: TaskDoneListener): () => void {
|
||||
const dispose = this.ctx.effect(() => {
|
||||
this.listeners.add(listener)
|
||||
return () => this.listeners.delete(listener)
|
||||
}, 'tasks.onTaskDone()')
|
||||
return () => void dispose()
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach an effect-scoped surface that can read and stop tasks. {@link start}
|
||||
* refuses work while none is attached.
|
||||
* @param name - diagnostic label; duplicate names remain independent.
|
||||
* @returns disposer that detaches this surface.
|
||||
*/
|
||||
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()')
|
||||
return () => void dispose()
|
||||
}
|
||||
|
||||
/** Look up a task or fail loud. */
|
||||
private expect(id: TaskId): TrackedTask {
|
||||
const task = this.store.get(id)
|
||||
if (task === undefined) throw new Error(`unknown task ${id}`)
|
||||
return task
|
||||
}
|
||||
|
||||
/**
|
||||
* The isolation fence: a task with an owner is reachable only by callers
|
||||
* whose session id matches (`!== undefined` semantics — an unowned task is
|
||||
* open, and a no-agent caller can never match an owned one).
|
||||
*/
|
||||
private assertAccess(task: TrackedTask, caller?: Agent): void {
|
||||
if (task.owner !== undefined && task.owner.id !== caller?.id) {
|
||||
throw new Error(`task ${task.id} belongs to another session`)
|
||||
}
|
||||
}
|
||||
|
||||
/** Project a fresh read-only snapshot from the mutable record. */
|
||||
private snapshot(task: TrackedTask): TaskSnapshot {
|
||||
const ownerSession = task.owner?.id
|
||||
return {
|
||||
id: task.id,
|
||||
kind: task.kind,
|
||||
label: task.label,
|
||||
...ownerSession !== undefined ? { ownerSession } : {},
|
||||
status: task.status,
|
||||
...task.detail !== undefined ? { detail: task.detail } : {},
|
||||
startedAt: task.startedAt,
|
||||
...task.finishedAt !== undefined ? { finishedAt: task.finishedAt } : {},
|
||||
reported: task.reported,
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Record the first terminal outcome, notify contained listeners, and release
|
||||
* waiters. First-wins preserves a teardown force-failure against late producer
|
||||
* settlement. Pending waits mark the task reported before listeners run.
|
||||
*/
|
||||
private settle(task: TrackedTask, outcome: TaskOutcome): void {
|
||||
if (isTerminal(task.status)) return
|
||||
task.status = outcome.status
|
||||
task.detail = outcome.detail
|
||||
task.output = outcome.output
|
||||
task.finishedAt = Date.now()
|
||||
if (task.waiters > 0) task.reported = true
|
||||
if (!this.listenersClosed) {
|
||||
const snapshot = this.snapshot(task)
|
||||
for (const listener of this.listeners) {
|
||||
try {
|
||||
const returned = listener(snapshot, task.owner)
|
||||
void Promise.resolve(returned).catch((error: unknown) => {
|
||||
this.selfCtx.logger.warn(`tasks: onTaskDone listener rejected for ${task.id}: ${String(error)}`)
|
||||
})
|
||||
} catch (error: unknown) {
|
||||
this.selfCtx.logger.warn(`tasks: onTaskDone listener threw for ${task.id}: ${String(error)}`)
|
||||
}
|
||||
}
|
||||
}
|
||||
const waitResolvers = [...task.waitResolvers]
|
||||
task.waitResolvers.clear()
|
||||
for (const resolveWait of waitResolvers) resolveWait()
|
||||
task.markSettled()
|
||||
}
|
||||
|
||||
/**
|
||||
* Attach one awaited cleanup through the exact owner's scope. This survives
|
||||
* producer reloads and joins agent quiescence; the retained disposer lets
|
||||
* service teardown detach the cross-fiber effect. Fails when the registry is
|
||||
* absent or the owner is not its currently registered instance.
|
||||
*/
|
||||
private ensureOwnerCleanup(owner: Agent): void {
|
||||
const ownerId = owner.id
|
||||
const agents = this.selfCtx.get('agents')
|
||||
if (agents === undefined) {
|
||||
throw new Error('background task ownership requires the agent registry (load @deepseek-ai/dsh-agent)')
|
||||
}
|
||||
if (agents.get(ownerId) !== owner) {
|
||||
throw new Error(`agent "${ownerId}" is not the registered agent instance (background task owner must be live)`)
|
||||
}
|
||||
if (this.ownerCleanups.has(owner)) return
|
||||
// Record only after attach succeeds; a disposing scope rejects new effects.
|
||||
const detach = owner.ctx.effect(() => async () => {
|
||||
this.ownerCleanups.delete(owner)
|
||||
await this.disposeOwned(owner)
|
||||
}, 'tasks.ownerCleanup()')
|
||||
this.ownerCleanups.set(owner, detach)
|
||||
}
|
||||
|
||||
/** Cancel, await terminal records, and drop every task owned by one exact agent lifecycle. */
|
||||
private async disposeOwned(owner: Agent): Promise<void> {
|
||||
const owned = [...this.store.values()].filter(task => task.owner === owner)
|
||||
this.cancelForTeardown(owned, 'owner disposed')
|
||||
await Promise.all(owned.map(task => task.settled))
|
||||
for (const task of owned) this.store.delete(task.id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Close listeners, cancel live tasks, await settlement, and detach owner
|
||||
* effects. Throwing cancels are force-failed to avoid teardown deadlock.
|
||||
*/
|
||||
private async disposeAll(): Promise<void> {
|
||||
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))
|
||||
this.store.clear()
|
||||
// Detach cross-fiber owner effects after the shared store is quiescent.
|
||||
const ownerCleanups = [...this.ownerCleanups.values()]
|
||||
this.ownerCleanups.clear()
|
||||
await Promise.all(ownerCleanups.map(cleanup => Promise.resolve(cleanup())))
|
||||
}
|
||||
|
||||
/**
|
||||
* Cancel tasks during teardown with per-task containment. A throwing cancel
|
||||
* force-fails the record and reports a possible orphan; a cancel that returns
|
||||
* without settling remains indistinguishable from a slow stop and may stall.
|
||||
*/
|
||||
private cancelForTeardown(tasks: TrackedTask[], reason: string): void {
|
||||
for (const task of tasks) {
|
||||
if (isTerminal(task.status)) continue
|
||||
try {
|
||||
task.cancel(reason)
|
||||
task.status = 'stopping'
|
||||
} catch (error: unknown) {
|
||||
const detail = `cancel threw during teardown; work may be orphaned: ${String(error)}`
|
||||
this.selfCtx.logger.warn(`tasks: cancel of ${task.id} threw during teardown; task record forced failed and work may be orphaned: ${String(error)}`)
|
||||
this.settle(task, { status: 'failed', detail })
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export default TaskService
|
||||
152
packages/tasks/tasks/src/types.ts
Normal file
152
packages/tasks/tasks/src/types.ts
Normal file
@@ -0,0 +1,152 @@
|
||||
/**
|
||||
* Types shared by task producers, the registry, and control surfaces. The
|
||||
* service implementation lives in `./index.ts`.
|
||||
* @module @deepseek-ai/dsh-tasks/types
|
||||
*/
|
||||
|
||||
import type { Branded } from '@deepseek-ai/dsh-brand'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import type { SessionId } from '@deepseek-ai/dsh-session'
|
||||
|
||||
/**
|
||||
* Identifies a background task. The registry generates `<kind>-N`; predictable
|
||||
* ids rely on owner authorization rather than secrecy.
|
||||
*/
|
||||
export type TaskId = Branded<'TaskId'>
|
||||
|
||||
/**
|
||||
* Brand a string as a {@link TaskId}.
|
||||
* @param id - the raw task-id string (the registry generates `<kind>-N`).
|
||||
* @returns the same string, branded; no validation is performed.
|
||||
*/
|
||||
export function TaskId(id: string): TaskId {
|
||||
return id as TaskId
|
||||
}
|
||||
|
||||
/**
|
||||
* Task lifecycle: `running`, optionally `stopping`, then exactly one terminal
|
||||
* status. Producer-specific facts belong in {@link TaskSnapshot.detail}.
|
||||
*/
|
||||
export type TaskStatus = 'running' | 'stopping' | 'completed' | 'killed' | 'failed'
|
||||
|
||||
/**
|
||||
* Producer-defined task kinds. Plugins extend this map by declaration merging;
|
||||
* the registry treats every value as an opaque id namespace.
|
||||
*/
|
||||
export interface TaskKindMap {
|
||||
bash: 'bash'
|
||||
subagent: 'subagent'
|
||||
}
|
||||
|
||||
/** The merge-extensible union of registered producer kind names. */
|
||||
export type TaskKind = TaskKindMap[keyof TaskKindMap]
|
||||
|
||||
/** Terminal result supplied by a producer through {@link TaskHooks.done}. */
|
||||
export interface TaskOutcome {
|
||||
/** How the task ended: finished (`completed`), cancelled (`killed`), or broke (`failed`). */
|
||||
status: 'completed' | 'killed' | 'failed'
|
||||
/** Kind-specific detail rendered into status lines ('exit code: 3', 'max-tokens'). */
|
||||
detail?: string
|
||||
/** Final output for tasks without `readOutput`; stream tasks leave it unset. */
|
||||
output?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Producer declaration passed to {@link TaskService.start}. The runtime
|
||||
* preflights access and cleanup before invoking {@link run}; the producer owns
|
||||
* execution resources while the runtime owns identity and lifecycle state.
|
||||
*/
|
||||
export interface TaskStart {
|
||||
/** Producer kind — also the id prefix (`bash`, `subagent`, …). */
|
||||
kind: TaskKind
|
||||
/** One-line model-facing label (the command; the delegation description). */
|
||||
label: string
|
||||
/**
|
||||
* Owning live agent. Access is fenced by its session id, and agent disposal
|
||||
* cancels and awaits the task. The instance must be the one currently
|
||||
* registered under its agent id. Omitting the owner creates an unowned task,
|
||||
* open to any caller until service disposal.
|
||||
*/
|
||||
owner?: Agent
|
||||
/**
|
||||
* Start the work after preflight and synchronously return its hooks. Called
|
||||
* once; a throw leaves nothing registered, and the producer must clean up any
|
||||
* partially started resources.
|
||||
*/
|
||||
run(): TaskHooks
|
||||
}
|
||||
|
||||
/** Hooks through which the runtime controls and observes producer work. */
|
||||
export interface TaskHooks {
|
||||
/**
|
||||
* Request termination. Must be synchronous, idempotent, and eventually settle
|
||||
* {@link done}; throws propagate. The optional reason is forwarded verbatim.
|
||||
*/
|
||||
cancel(reason?: string): void
|
||||
/**
|
||||
* Resolves after the producer releases its resources, not merely when work
|
||||
* finishes. Must not reject; the runtime converts a rejection to `failed`.
|
||||
* If teardown cancellation throws, the runtime may force-fail only the
|
||||
* registry record without claiming that the work stopped.
|
||||
*/
|
||||
done: Promise<TaskOutcome>
|
||||
/**
|
||||
* Consume output produced since the previous call. The producer formats
|
||||
* truncation and spill notices. Absence marks a final-output-only task; each
|
||||
* task has one consuming cursor.
|
||||
*/
|
||||
readOutput?(): string
|
||||
}
|
||||
|
||||
/**
|
||||
* A read-only projection of one task, safe to hand to listeners and tools —
|
||||
* a fresh object per call, never live registry state.
|
||||
*/
|
||||
export interface TaskSnapshot {
|
||||
/** The registry-issued id (`<kind>-N`). */
|
||||
id: TaskId
|
||||
/** The producer kind the task was registered with. */
|
||||
kind: TaskKind
|
||||
/** The producer-supplied one-line label. */
|
||||
label: string
|
||||
/**
|
||||
* Owner session id used for authorization and correlation; absent for
|
||||
* unowned tasks. Completion listeners receive the exact {@link Agent}
|
||||
* separately through {@link TaskDoneListener}.
|
||||
*/
|
||||
ownerSession?: SessionId
|
||||
/** Current lifecycle state. */
|
||||
status: TaskStatus
|
||||
/** Kind-specific status detail, present once the producer supplied one (usually terminal). */
|
||||
detail?: string
|
||||
/** Epoch ms when the task was registered. */
|
||||
startedAt: number
|
||||
/** Epoch ms when the task settled; absent while `running`/`stopping`. */
|
||||
finishedAt?: number
|
||||
/**
|
||||
* True when a kill, read, or wait has reported or committed to report the
|
||||
* terminal state. Completion surfaces suppress redundant notices when set.
|
||||
*/
|
||||
reported: boolean
|
||||
}
|
||||
|
||||
/** Output and post-read state returned by {@link TaskService.read}. */
|
||||
export interface TaskRead {
|
||||
/**
|
||||
* Stream kinds: the consuming delta since the previous read. Final-output
|
||||
* kinds: empty while live, the terminal {@link TaskOutcome.output} (or
|
||||
* empty) once settled — idempotent, never consumed.
|
||||
*/
|
||||
text: string
|
||||
/** The task's state at read time. */
|
||||
snapshot: TaskSnapshot
|
||||
}
|
||||
|
||||
/**
|
||||
* Completion callback with the exact owner supplied at start, or `undefined`
|
||||
* for an unowned task. Returned promises are observed but not awaited.
|
||||
*/
|
||||
export type TaskDoneListener = (
|
||||
snapshot: TaskSnapshot,
|
||||
owner: Agent | undefined,
|
||||
) => void | PromiseLike<void>
|
||||
740
packages/tasks/tasks/tests/tasks.spec.ts
Normal file
740
packages/tasks/tasks/tests/tasks.spec.ts
Normal file
@@ -0,0 +1,740 @@
|
||||
import { describe, expect, expectTypeOf, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { Session, SessionId } from '@deepseek-ai/dsh-session'
|
||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import TaskService, { TaskId } from '@deepseek-ai/dsh-tasks'
|
||||
import type { TaskHooks, TaskKind, TaskOutcome, TaskSnapshot, TaskStart } from '@deepseek-ai/dsh-tasks'
|
||||
|
||||
declare module '@deepseek-ai/dsh-tasks' {
|
||||
interface TaskKindMap {
|
||||
workflow: 'workflow'
|
||||
}
|
||||
}
|
||||
|
||||
const agentScopeDisposers = new WeakMap<Agent, () => Promise<void>>()
|
||||
|
||||
function stubAgent(ctx: Context, rawId: string): Agent {
|
||||
const id = SessionId(rawId)
|
||||
const scopeFiber = ctx.plugin(() => {})
|
||||
const agent = {
|
||||
id,
|
||||
options: {},
|
||||
session: new Session(id),
|
||||
status: 'idle' as const,
|
||||
ctx: scopeFiber.ctx,
|
||||
send() {},
|
||||
steer() {},
|
||||
inject() {},
|
||||
cancel() {},
|
||||
whenIdle() { return Promise.resolve() },
|
||||
}
|
||||
agentScopeDisposers.set(agent, async () => { await scopeFiber.dispose() })
|
||||
return agent
|
||||
}
|
||||
|
||||
async function disposeAgentScope(agent: Agent): Promise<void> {
|
||||
const dispose = agentScopeDisposers.get(agent)
|
||||
if (dispose === undefined) throw new Error(`missing test scope for agent "${agent.id}"`)
|
||||
await dispose()
|
||||
}
|
||||
|
||||
/** 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 { 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 }),
|
||||
...hookOverrides,
|
||||
}
|
||||
const spec: TaskStart = { kind, label, ...owner !== undefined ? { owner } : {}, run: () => hooks }
|
||||
return { spec, settle, reject, cancels }
|
||||
}
|
||||
|
||||
async function harness() {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(TaskService)
|
||||
ctx.tasks.attachSurface('test-surface')
|
||||
return ctx
|
||||
}
|
||||
|
||||
/** Let the settlement continuation (a `done.then`) run. */
|
||||
const tick = () => new Promise<void>(r => setTimeout(r, 0))
|
||||
|
||||
/** Inspect the internal resolver registry to pin bounded retention while a task stays live. */
|
||||
function waitResolverCount(ctx: Context, id: TaskId): number {
|
||||
const service = ctx.tasks as unknown as { store: Map<TaskId, { waitResolvers: Set<() => void> }> }
|
||||
const task = service.store.get(id)
|
||||
if (task === undefined) throw new Error(`missing test task ${id}`)
|
||||
return task.waitResolvers.size
|
||||
}
|
||||
|
||||
describe('TaskService.start', () => {
|
||||
it('preserves the SessionId brand on public owner snapshots', () => {
|
||||
expectTypeOf<TaskSnapshot['ownerSession']>().toEqualTypeOf<SessionId | undefined>()
|
||||
})
|
||||
|
||||
it('refuses to register while no control surface is attached', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(TaskService)
|
||||
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.start(producer({ kind: '' as TaskKind }).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.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')
|
||||
expect(ctx.tasks.start(producer({ kind: 'workflow' }).spec)).toBe('workflow-1')
|
||||
})
|
||||
})
|
||||
|
||||
describe('TaskService reads and settlement', () => {
|
||||
it('stream kinds read a consuming delta; terminal reads mark reported', async () => {
|
||||
const ctx = await harness()
|
||||
const chunks = ['first', '', 'rest']
|
||||
const p = producer({ readOutput: () => chunks.shift() ?? '' })
|
||||
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('')
|
||||
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
await tick()
|
||||
const read = ctx.tasks.read(id)
|
||||
expect(read.text).toBe('rest')
|
||||
expect(read.snapshot).toMatchObject({ status: 'completed', detail: 'exit code: 0', reported: true })
|
||||
expect(read.snapshot.finishedAt).toBeTypeOf('number')
|
||||
})
|
||||
|
||||
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.start(p.spec)
|
||||
|
||||
expect(ctx.tasks.read(id)).toMatchObject({ text: '', snapshot: { status: 'running' } })
|
||||
|
||||
p.settle({ status: 'completed', output: 'final answer' })
|
||||
await tick()
|
||||
expect(ctx.tasks.read(id).text).toBe('final answer')
|
||||
expect(ctx.tasks.read(id).text).toBe('final answer') // idempotent, not consumed
|
||||
})
|
||||
|
||||
it('a settled task without output reads as empty text', async () => {
|
||||
const ctx = await harness()
|
||||
const p = producer({ kind: 'subagent' })
|
||||
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' } })
|
||||
})
|
||||
|
||||
it('throws for unknown task ids', async () => {
|
||||
const ctx = await harness()
|
||||
expect(() => ctx.tasks.read(TaskId('bash-99'))).toThrow('unknown task bash-99')
|
||||
})
|
||||
|
||||
it('notifies onTaskDone once per task with containment across listeners', async () => {
|
||||
const ctx = await harness()
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const seen: TaskSnapshot[] = []
|
||||
ctx.tasks.onTaskDone(() => { throw new Error('listener boom') })
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
|
||||
|
||||
const p = producer()
|
||||
const id = ctx.tasks.start(p.spec)
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
await tick()
|
||||
|
||||
expect(seen).toHaveLength(1)
|
||||
expect(seen[0]).toMatchObject({ id, status: 'completed', reported: false })
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('listener boom'))
|
||||
})
|
||||
|
||||
it('contains a rejecting onTaskDone listener without starving later listeners', async () => {
|
||||
const ctx = await harness()
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const seen: TaskId[] = []
|
||||
ctx.tasks.onTaskDone(async () => { throw new Error('async listener boom') })
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot.id))
|
||||
|
||||
const p = producer()
|
||||
const id = ctx.tasks.start(p.spec)
|
||||
p.settle({ status: 'completed' })
|
||||
await tick()
|
||||
|
||||
expect(seen).toEqual([id])
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('onTaskDone listener rejected'))
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('async listener boom'))
|
||||
})
|
||||
|
||||
it('contains a rejecting done as a failed outcome (producer contract violation)', async () => {
|
||||
const ctx = await harness()
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const p = producer()
|
||||
const id = ctx.tasks.start(p.spec)
|
||||
p.reject(new Error('transport exploded'))
|
||||
await tick()
|
||||
|
||||
expect(ctx.tasks.read(id).snapshot).toMatchObject({ status: 'failed', detail: 'Error: transport exploded' })
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('producer contract violation'))
|
||||
})
|
||||
|
||||
it('unregisters onTaskDone listeners with the contributing fiber (HMR safety)', async () => {
|
||||
const ctx = await harness()
|
||||
const seen: string[] = []
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
inner.tasks.onTaskDone(snapshot => void seen.push(snapshot.id))
|
||||
}, { inject: ['tasks'] }))
|
||||
await fiber.dispose()
|
||||
// The returned disposer detaches too (the non-fiber path).
|
||||
const detach = ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot.id))
|
||||
detach()
|
||||
|
||||
const p = producer()
|
||||
ctx.tasks.start(p.spec)
|
||||
p.settle({ status: 'completed' })
|
||||
await tick()
|
||||
expect(seen).toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
describe('TaskService.kill', () => {
|
||||
it('cancels a live task with the forwarded reason and suppresses the notice', async () => {
|
||||
const ctx = await harness()
|
||||
const seen: TaskSnapshot[] = []
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
|
||||
const p = producer()
|
||||
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'])
|
||||
expect(ctx.tasks.list()[0]).toMatchObject({ status: 'stopping', reported: true })
|
||||
|
||||
p.settle({ status: 'killed' })
|
||||
await tick()
|
||||
// The listener still fires (telemetry may care), but carries reported: true
|
||||
// so the notice surface suppresses its redundant "finished".
|
||||
expect(seen[0]).toMatchObject({ id, status: 'killed', reported: true })
|
||||
})
|
||||
|
||||
it('reports an already-finished task instead of failing', async () => {
|
||||
const ctx = await harness()
|
||||
const p = producer()
|
||||
const id = ctx.tasks.start(p.spec)
|
||||
p.settle({ status: 'completed' })
|
||||
await tick()
|
||||
expect(ctx.tasks.kill(id)).toBe('already-finished')
|
||||
})
|
||||
|
||||
it('propagates a throwing producer cancel and leaves the task untouched', async () => {
|
||||
const ctx = await harness()
|
||||
const seen: TaskSnapshot[] = []
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
|
||||
let broken = true
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
const id = ctx.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'flaky cancel',
|
||||
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,
|
||||
// and a later (successful) kill still works.
|
||||
expect(ctx.tasks.get(id)).toMatchObject({ status: 'running', reported: false })
|
||||
settle({ status: 'completed' })
|
||||
await tick()
|
||||
expect(seen[0]).toMatchObject({ id, reported: false }) // notice would still fire
|
||||
|
||||
broken = false
|
||||
expect(ctx.tasks.kill(id)).toBe('already-finished')
|
||||
})
|
||||
})
|
||||
|
||||
describe('TaskService.wait', () => {
|
||||
it('resolves with the terminal snapshot when the task settles, marked reported', async () => {
|
||||
const ctx = await harness()
|
||||
const seen: TaskSnapshot[] = []
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
|
||||
const p = producer()
|
||||
const id = ctx.tasks.start(p.spec)
|
||||
|
||||
const wait = ctx.tasks.wait(id, 5_000)
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
expect(await wait).toMatchObject({ status: 'completed', reported: true })
|
||||
// A waiting reader claims delivery before completion listeners inspect the snapshot.
|
||||
expect(seen[0]).toMatchObject({ id, reported: true })
|
||||
})
|
||||
|
||||
it('returns the live snapshot on timeout without marking reported', async () => {
|
||||
const ctx = await harness()
|
||||
const id = ctx.tasks.start(producer().spec)
|
||||
expect(await ctx.tasks.wait(id, 5)).toMatchObject({ status: 'running', reported: false })
|
||||
})
|
||||
|
||||
it('unregisters timed-out and aborted wait resolvers while the task remains live', async () => {
|
||||
const ctx = await harness()
|
||||
const id = ctx.tasks.start(producer().spec)
|
||||
|
||||
for (let index = 0; index < 3; index += 1) {
|
||||
const wait = ctx.tasks.wait(id, 5)
|
||||
expect(waitResolverCount(ctx, id)).toBe(1)
|
||||
await expect(wait).resolves.toMatchObject({ status: 'running' })
|
||||
expect(waitResolverCount(ctx, id)).toBe(0)
|
||||
}
|
||||
|
||||
const controller = new AbortController()
|
||||
const wait = ctx.tasks.wait(id, 5_000, undefined, controller.signal)
|
||||
expect(waitResolverCount(ctx, id)).toBe(1)
|
||||
controller.abort()
|
||||
await expect(wait).rejects.toThrow('wait aborted')
|
||||
expect(waitResolverCount(ctx, id)).toBe(0)
|
||||
expect(ctx.tasks.get(id).status).toBe('running')
|
||||
})
|
||||
|
||||
it('returns immediately for an already-finished task', async () => {
|
||||
const ctx = await harness()
|
||||
const p = producer()
|
||||
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 })
|
||||
})
|
||||
|
||||
it('rejects a non-positive or non-finite timeout', async () => {
|
||||
const ctx = await harness()
|
||||
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.start(producer().spec)
|
||||
|
||||
const controller = new AbortController()
|
||||
const wait = ctx.tasks.wait(id, 5_000, undefined, controller.signal)
|
||||
controller.abort()
|
||||
await expect(wait).rejects.toThrow('wait aborted')
|
||||
expect(ctx.tasks.list()[0]).toMatchObject({ status: 'running' })
|
||||
|
||||
const preAborted = new AbortController()
|
||||
preAborted.abort()
|
||||
await expect(ctx.tasks.wait(id, 5_000, undefined, preAborted.signal)).rejects.toThrow('wait aborted')
|
||||
})
|
||||
|
||||
it('an abort racing settlement in the same tick does not swallow the notice', async () => {
|
||||
const ctx = await harness()
|
||||
const seen: TaskSnapshot[] = []
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
|
||||
const p = producer()
|
||||
const id = ctx.tasks.start(p.spec)
|
||||
|
||||
const controller = new AbortController()
|
||||
const wait = ctx.tasks.wait(id, 5_000, undefined, controller.signal)
|
||||
// Settlement is queued first, so abort must remove the waiter synchronously;
|
||||
// otherwise settlement suppresses the notice for a reader that receives nothing.
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
controller.abort()
|
||||
await expect(wait).rejects.toThrow('wait aborted')
|
||||
expect(seen).toHaveLength(1)
|
||||
expect(seen[0]).toMatchObject({ id, status: 'completed', reported: false })
|
||||
})
|
||||
|
||||
it('an abort landing after settlement still delivers the terminal snapshot it owes', async () => {
|
||||
const ctx = await harness()
|
||||
const controller = new AbortController()
|
||||
const seen: TaskSnapshot[] = []
|
||||
// The listener aborts after settlement has assigned delivery to this waiter
|
||||
// but before its resolve microtask; the waiter must still receive the result.
|
||||
ctx.tasks.onTaskDone((snapshot) => {
|
||||
seen.push(snapshot)
|
||||
controller.abort()
|
||||
})
|
||||
const p = producer()
|
||||
const id = ctx.tasks.start(p.spec)
|
||||
|
||||
const wait = ctx.tasks.wait(id, 5_000, undefined, controller.signal)
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
await expect(wait).resolves.toMatchObject({ status: 'completed', reported: true })
|
||||
expect(seen[0]).toMatchObject({ id, reported: true }) // suppression stays honest: the wait delivered
|
||||
})
|
||||
})
|
||||
|
||||
describe('TaskService owner isolation', () => {
|
||||
it('fences read/kill/wait to the owning session and keeps unowned tasks open', async () => {
|
||||
const ctx = await harness()
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
const other = stubAgent(ctx, 'other')
|
||||
|
||||
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)
|
||||
expect(ctx.tasks.read(open, other).snapshot.id).toBe(open)
|
||||
|
||||
// A different session and a no-agent caller are rejected.
|
||||
expect(() => ctx.tasks.read(owned, other)).toThrow(`task ${owned} belongs to another session`)
|
||||
expect(() => ctx.tasks.kill(owned, other)).toThrow('belongs to another session')
|
||||
await expect(ctx.tasks.wait(owned, 10, other)).rejects.toThrow('belongs to another session')
|
||||
expect(() => ctx.tasks.read(owned)).toThrow('belongs to another session')
|
||||
})
|
||||
|
||||
it('list() shows only caller-owned plus unowned tasks', async () => {
|
||||
const ctx = await harness()
|
||||
const alice = stubAgent(ctx, 'alice')
|
||||
const bob = stubAgent(ctx, 'bob')
|
||||
ctx.agents.register(alice)
|
||||
ctx.agents.register(bob)
|
||||
|
||||
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])
|
||||
expect(ctx.tasks.list().map(t => t.id)).toEqual([openTask])
|
||||
})
|
||||
|
||||
it('rejects an owned registration when no agent registry is mounted', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(TaskService)
|
||||
ctx.tasks.attachSurface('test-surface')
|
||||
expect(() => ctx.tasks.start(producer({ owner: stubAgent(ctx, '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.start(producer().spec)).toBe('bash-1')
|
||||
})
|
||||
|
||||
it('a failed owner-cleanup attach leaves the registry unchanged and does not poison the owner', async () => {
|
||||
const ctx = await harness()
|
||||
const ghost = stubAgent(ctx, 'ghost') // never registered in ctx.agents
|
||||
|
||||
// Exact-instance validation precedes registry mutation and cleanup attachment.
|
||||
expect(() => ctx.tasks.start(producer({ owner: ghost }).spec))
|
||||
.toThrow('is not the registered agent instance')
|
||||
expect(ctx.tasks.list(ghost)).toEqual([])
|
||||
|
||||
// A later valid registration must still attach cleanup for the same object.
|
||||
ctx.agents.register(ghost)
|
||||
const cancels: (string | undefined)[] = []
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
const id = ctx.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'after retry',
|
||||
owner: ghost,
|
||||
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 disposeAgentScope(ghost)
|
||||
expect(cancels).toEqual(['owner disposed'])
|
||||
expect(ctx.tasks.list(ghost)).toEqual([])
|
||||
})
|
||||
|
||||
it('rejects a stale owner instance after another agent reuses its id', async () => {
|
||||
const ctx = await harness()
|
||||
const staleOwner = stubAgent(ctx, 'owner')
|
||||
const unregisterStale = ctx.agents.register(staleOwner)
|
||||
unregisterStale()
|
||||
|
||||
const currentOwner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(currentOwner)
|
||||
const current = producer({ owner: currentOwner })
|
||||
ctx.tasks.start(current.spec) // Attach the current owner's cleanup first.
|
||||
|
||||
const stale = producer({ owner: staleOwner })
|
||||
const staleRun = vi.fn(() => stale.spec.run())
|
||||
expect(() => ctx.tasks.start({ ...stale.spec, run: staleRun }))
|
||||
.toThrow('is not the registered agent instance')
|
||||
expect(staleRun).not.toHaveBeenCalled()
|
||||
// Access is keyed by the unified session id, so a reconnect carrying the
|
||||
// same identity can observe the current task even though stale ownership
|
||||
// registration is rejected by exact-instance validation.
|
||||
expect(ctx.tasks.list(staleOwner)).toHaveLength(1)
|
||||
expect(ctx.tasks.list(currentOwner)).toHaveLength(1)
|
||||
|
||||
current.settle({ status: 'completed' })
|
||||
await tick()
|
||||
await disposeAgentScope(currentOwner)
|
||||
})
|
||||
})
|
||||
|
||||
describe('TaskService owner cleanup', () => {
|
||||
it('drains the owner: cancels live tasks, awaits settlement, drops snapshots', async () => {
|
||||
const ctx = await harness()
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
|
||||
// The producer settles only when cancelled — models a child that stops on request.
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
const cancels: (string | undefined)[] = []
|
||||
ctx.tasks.start({
|
||||
kind: 'subagent',
|
||||
label: 'long research',
|
||||
owner,
|
||||
run: () => ({
|
||||
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
|
||||
done: new Promise<TaskOutcome>((res) => { settle = res }),
|
||||
}),
|
||||
})
|
||||
const terminal = producer({ owner })
|
||||
ctx.tasks.start(terminal.spec)
|
||||
terminal.settle({ status: 'completed' })
|
||||
await tick()
|
||||
|
||||
await disposeAgentScope(owner)
|
||||
expect(cancels).toEqual(['owner disposed'])
|
||||
// Snapshots dropped: nothing of the owner's remains, listing is empty.
|
||||
expect(ctx.tasks.list(owner)).toEqual([])
|
||||
})
|
||||
|
||||
it('attaches one cleanup per owner and drains all owned tasks with the scope', async () => {
|
||||
const ctx = await harness()
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
|
||||
const first = producer({ owner })
|
||||
const second = producer({ owner })
|
||||
ctx.tasks.start(first.spec)
|
||||
ctx.tasks.start(second.spec)
|
||||
first.settle({ status: 'completed' })
|
||||
second.settle({ status: 'completed' })
|
||||
await tick()
|
||||
expect(owner.ctx.fiber.getEffects().filter(effect => effect.label === 'tasks.ownerCleanup()')).toHaveLength(1)
|
||||
await disposeAgentScope(owner)
|
||||
expect(ctx.tasks.list(owner)).toEqual([])
|
||||
})
|
||||
|
||||
it('does not let an old scope cleanup cancel a same-id/session replacement task', async () => {
|
||||
const ctx = await harness()
|
||||
const oldOwner = stubAgent(ctx, 'owner')
|
||||
const detachOld = ctx.agents.register(oldOwner)
|
||||
const cancels: string[] = []
|
||||
|
||||
function start(owner: Agent, label: string): TaskId {
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
return ctx.tasks.start({
|
||||
kind: 'bash',
|
||||
label,
|
||||
owner,
|
||||
run: () => ({
|
||||
cancel() { cancels.push(label); settle({ status: 'killed' }) },
|
||||
done: new Promise<TaskOutcome>((resolve) => { settle = resolve }),
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
start(oldOwner, 'old task')
|
||||
detachOld()
|
||||
const replacement = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(replacement)
|
||||
const replacementId = start(replacement, 'replacement task')
|
||||
|
||||
await disposeAgentScope(oldOwner)
|
||||
expect(cancels).toEqual(['old task'])
|
||||
expect(ctx.tasks.list(replacement).map(task => task.id)).toEqual([replacementId])
|
||||
|
||||
await disposeAgentScope(replacement)
|
||||
expect(cancels).toEqual(['old task', 'replacement task'])
|
||||
})
|
||||
|
||||
it('registers owner cleanup on the agent scope rather than the tasks fiber', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const tasksFiber = await ctx.plugin(TaskService)
|
||||
ctx.tasks.attachSurface('test-surface')
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
const ownerCleanupEffects = () => owner.ctx.fiber.getEffects()
|
||||
.filter(effect => effect.label === 'tasks.ownerCleanup()')
|
||||
|
||||
const first = producer({ owner })
|
||||
ctx.tasks.start(first.spec)
|
||||
expect(ownerCleanupEffects()).toHaveLength(1)
|
||||
first.settle({ status: 'completed' })
|
||||
await tick()
|
||||
expect(tasksFiber.getEffects().some(effect => effect.label === 'tasks.ownerCleanup()')).toBe(false)
|
||||
await disposeAgentScope(owner)
|
||||
|
||||
// Only the owner registration is released; the long-lived tasks service
|
||||
// and its own teardown effect remain active.
|
||||
expect(ownerCleanupEffects()).toHaveLength(0)
|
||||
expect(ctx.get('tasks')).toBeDefined()
|
||||
expect(tasksFiber.getEffects().some(effect => effect.label === 'tasks teardown')).toBe(true)
|
||||
|
||||
})
|
||||
|
||||
it('force-fails a throwing teardown cancel without awaiting producer done, first outcome wins', async () => {
|
||||
const ctx = await harness()
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
const seen: TaskSnapshot[] = []
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
|
||||
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
ctx.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'broken producer',
|
||||
owner,
|
||||
run: () => ({
|
||||
cancel() { throw new Error('cancel boom') },
|
||||
done: new Promise<TaskOutcome>((res) => { settle = res }),
|
||||
}),
|
||||
})
|
||||
|
||||
const drain = disposeAgentScope(owner)
|
||||
let drained = false
|
||||
void drain.then(() => { drained = true })
|
||||
await tick()
|
||||
const drainedWithoutProducerDone = drained
|
||||
if (!drainedWithoutProducerDone) {
|
||||
// Release the producer if the assertion fails so the test can finish.
|
||||
settle({ status: 'completed' })
|
||||
await drain
|
||||
} else {
|
||||
// A late producer completion must not replace the failure or notify twice.
|
||||
settle({ status: 'completed' })
|
||||
await tick()
|
||||
}
|
||||
|
||||
expect(drainedWithoutProducerDone).toBe(true)
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('work may be orphaned'))
|
||||
expect(seen).toHaveLength(1)
|
||||
expect(seen[0]?.status).toBe('failed')
|
||||
expect(seen[0]?.detail).toContain('cancel threw during teardown')
|
||||
expect(ctx.tasks.list(owner)).toEqual([])
|
||||
})
|
||||
})
|
||||
|
||||
describe('TaskService disposal', () => {
|
||||
it('cancels live tasks, awaits settlement, and silences listeners', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const fiber = await ctx.plugin(TaskService)
|
||||
const surface = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
inner.tasks.attachSurface('test-surface')
|
||||
}, { inject: ['tasks'] }))
|
||||
void surface
|
||||
|
||||
const seen: string[] = []
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot.id))
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
const cancels: (string | undefined)[] = []
|
||||
ctx.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'sleep 600',
|
||||
run: () => ({
|
||||
cancel(reason) { cancels.push(reason); settle({ status: 'killed' }) },
|
||||
done: new Promise<TaskOutcome>((res) => { settle = res }),
|
||||
}),
|
||||
})
|
||||
|
||||
await fiber.dispose()
|
||||
expect(cancels).toEqual(['tasks service disposed'])
|
||||
// The teardown kill settles AFTER the listener registry closed: silent.
|
||||
expect(seen).toEqual([])
|
||||
})
|
||||
|
||||
it('force-fails a throwing cancel so service disposal does not await producer done', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const fiber = await ctx.plugin(TaskService)
|
||||
ctx.tasks.attachSurface('test-surface')
|
||||
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
|
||||
const seen: TaskSnapshot[] = []
|
||||
ctx.tasks.onTaskDone(snapshot => void seen.push(snapshot))
|
||||
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
ctx.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'broken service task',
|
||||
run: () => ({
|
||||
cancel() { throw new Error('service cancel boom') },
|
||||
done: new Promise<TaskOutcome>((resolve) => { settle = resolve }),
|
||||
}),
|
||||
})
|
||||
|
||||
const disposal = fiber.dispose()
|
||||
let disposed = false
|
||||
void disposal.then(() => { disposed = true })
|
||||
await tick()
|
||||
const disposedWithoutProducerDone = disposed
|
||||
if (!disposedWithoutProducerDone) {
|
||||
// Release the producer if the assertion fails so the test can finish.
|
||||
settle({ status: 'completed' })
|
||||
await disposal
|
||||
} else {
|
||||
settle({ status: 'completed' })
|
||||
await tick()
|
||||
}
|
||||
|
||||
expect(disposedWithoutProducerDone).toBe(true)
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('work may be orphaned'))
|
||||
expect(seen).toEqual([])
|
||||
})
|
||||
|
||||
it('detaches owner effects from still-live agent scopes when the service unloads', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const tasksFiber = await ctx.plugin(TaskService)
|
||||
ctx.tasks.attachSurface('test-surface')
|
||||
const owner = stubAgent(ctx, 'owner')
|
||||
ctx.agents.register(owner)
|
||||
let settle!: (outcome: TaskOutcome) => void
|
||||
ctx.tasks.start({
|
||||
kind: 'bash',
|
||||
label: 'owned work',
|
||||
owner,
|
||||
run: () => ({
|
||||
cancel() { settle({ status: 'killed' }) },
|
||||
done: new Promise<TaskOutcome>((resolve) => { settle = resolve }),
|
||||
}),
|
||||
})
|
||||
const ownerEffects = () => owner.ctx.fiber.getEffects()
|
||||
.filter(effect => effect.label === 'tasks.ownerCleanup()')
|
||||
expect(ownerEffects()).toHaveLength(1)
|
||||
|
||||
await tasksFiber.dispose()
|
||||
|
||||
expect(ownerEffects()).toHaveLength(0)
|
||||
})
|
||||
|
||||
it('detaching the last surface re-arms the register fence', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(TaskService)
|
||||
const detachA1 = ctx.tasks.attachSurface('a')
|
||||
const detachA2 = ctx.tasks.attachSurface('a') // duplicate name counts independently
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
inner.tasks.attachSurface('b')
|
||||
}, { inject: ['tasks'] }))
|
||||
|
||||
detachA1()
|
||||
detachA1() // second call of the same disposer is a no-op
|
||||
expect(() => ctx.tasks.start(producer().spec)).not.toThrow() // a ×1 + b remain
|
||||
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')
|
||||
})
|
||||
})
|
||||
30
packages/tasks/tasks/tsconfig.json
Normal file
30
packages/tasks/tasks/tsconfig.json
Normal file
@@ -0,0 +1,30 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib/types"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../util/brand"
|
||||
},
|
||||
{
|
||||
"path": "../../core/agent"
|
||||
},
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../../util/timeout"
|
||||
}
|
||||
]
|
||||
}
|
||||
56
packages/tasks/tool-tasks/README.md
Normal file
56
packages/tasks/tool-tasks/README.md
Normal file
@@ -0,0 +1,56 @@
|
||||
# @deepseek-ai/dsh-tool-tasks
|
||||
|
||||
The model-facing control surface for `ctx.tasks`: three kind-independent tools, completion notices, and one background-work prompt section. Loading the plugin attaches the surface required by `ctx.tasks.start()`.
|
||||
|
||||
## Tools
|
||||
|
||||
- `task_output(task_id, wait?, timeout_ms?)` reads without blocking by default. Stream tasks return only the next delta; final-output tasks return their result after settlement. Every response ends with `[status: ...]`. `wait: true` waits up to the configured cap and leaves a still-running task alive on timeout.
|
||||
- `task_list()` returns caller-visible tasks as `<id> [<kind>] <status> — <label>`.
|
||||
- `task_kill(task_id, reason?)` requests cancellation immediately and forwards the logged reason. Terminal tasks return a non-consuming snapshot.
|
||||
|
||||
All three use generic ACP cards: `read` for output and list, `execute` for kill.
|
||||
|
||||
## Completion notices
|
||||
|
||||
An unreported completion injects `background task <id> (<kind>: <label>) finished [status: ...]. Read its output with task_output.` into the exact owner's session. Injection is durable context for the next request, not a wake-up. A kill or terminal read/wait marks delivery reported and suppresses the redundant notice; owner-disposal races are contained.
|
||||
|
||||
## Config
|
||||
|
||||
| key | default | meaning |
|
||||
|---|---|---|
|
||||
| `waitTimeoutMs` | `30000` | wait used when `wait: true` omits `timeout_ms` |
|
||||
| `maxWaitTimeoutMs` | `600000` | cap for model-supplied waits |
|
||||
|
||||
A default above the cap fails at load.
|
||||
|
||||
## Model Experience
|
||||
|
||||
### System prompt
|
||||
|
||||
**What the model sees**: Every request in this plugin's registration scope contains this guidance. Agent-scoped tool filtering may hide the tools without removing the independently registered prompt section.
|
||||
|
||||
**Token effect**: Small fixed input cost per request while active.
|
||||
|
||||
#### Background-task guidance
|
||||
|
||||
```markdown
|
||||
Track every background task id you start. You are notified in-session when a task finishes — do not busy-poll or sleep on one; keep working on independent steps and do not duplicate a running task's work. Before giving a final answer, collect every still-relevant task with task_output (set wait: true only when you are genuinely blocked on it), and task_kill tasks that stopped mattering.
|
||||
```
|
||||
|
||||
### Tool schemas
|
||||
|
||||
**What the model sees**: The generated [`task_output`, `task_list`, and `task_kill` schemas](../../../docs/tool-catalog.md#deepseek-aidsh-tool-tasks) while this surface is visible.
|
||||
|
||||
**Token effect**: Fixed schema cost on each request where the tools are visible.
|
||||
|
||||
### Results and notices
|
||||
|
||||
**What the model sees**: Reads return output or `(no new output)` followed by `[status: <status>]` and optional detail. An empty list returns `(no background tasks)`. Kill returns `requested cancellation of task <id>` or the existing terminal status. Unreported owned completion uses the notice above.
|
||||
|
||||
**Token effect**: Results and notices remain in parent history until compaction. Stream reads do not repeat consumed output.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **Completion notices do not wake idle agents** — callers needing an immediate result must use `task_output`.
|
||||
- **Stream reads are single-consumer** — independent observers need another runtime API.
|
||||
- **Unowned tasks have no session fence** — external surfaces must supply caller policy or avoid them.
|
||||
43
packages/tasks/tool-tasks/package.json
Normal file
43
packages/tasks/tool-tasks/package.json
Normal file
@@ -0,0 +1,43 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-tool-tasks",
|
||||
"description": "Model-facing background task control tools (task_output, task_list, task_kill) over the ctx.tasks registry",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "lib/index.js",
|
||||
"types": "lib/types/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./lib/types/index.d.ts",
|
||||
"default": "./lib/index.js"
|
||||
},
|
||||
"./src/*": "./src/*",
|
||||
"./package.json": "./package.json"
|
||||
},
|
||||
"files": [
|
||||
"lib/index.js",
|
||||
"lib/types/**/*.d.ts",
|
||||
"lib/types/**/*.d.ts.map",
|
||||
"src"
|
||||
],
|
||||
"license": "BSD-3-Clause",
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "^0.0.1",
|
||||
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
|
||||
"@deepseek-ai/dsh-tasks": "^0.0.1",
|
||||
"@deepseek-ai/dsh-tools": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"dependencies": {
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-agent": "workspace:^",
|
||||
"@deepseek-ai/dsh-llm": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"@deepseek-ai/dsh-system-prompt": "workspace:^",
|
||||
"@deepseek-ai/dsh-tasks": "workspace:^",
|
||||
"@deepseek-ai/dsh-tools": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
148
packages/tasks/tool-tasks/src/index.ts
Normal file
148
packages/tasks/tool-tasks/src/index.ts
Normal file
@@ -0,0 +1,148 @@
|
||||
/**
|
||||
* Model-facing `task_output`, `task_list`, and `task_kill` tools over
|
||||
* `ctx.tasks`. Loading the plugin attaches the control surface required by
|
||||
* producers. It also injects unreported completions as durable context for the
|
||||
* owner's next request; notices do not wake idle agents.
|
||||
* @module @deepseek-ai/dsh-tool-tasks
|
||||
*/
|
||||
|
||||
import type { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { defineTool } from '@deepseek-ai/dsh-tools'
|
||||
import type { GenericCallView } from '@deepseek-ai/dsh-tools'
|
||||
import { TaskId } from '@deepseek-ai/dsh-tasks'
|
||||
import type { TaskSnapshot } from '@deepseek-ai/dsh-tasks'
|
||||
import type {} from '@deepseek-ai/dsh-system-prompt'
|
||||
|
||||
export const name = 'tool-tasks'
|
||||
export const inject = ['tools', 'tasks', 'systemPrompt']
|
||||
|
||||
/** Configures bounded `task_output` waits. */
|
||||
export interface Config {
|
||||
/** Wait duration applied when `task_output` sets `wait` without `timeout_ms` (default 30s). */
|
||||
waitTimeoutMs?: number
|
||||
/** Hard cap on any single wait; a larger model-supplied `timeout_ms` is clamped down to it (default 10min). */
|
||||
maxWaitTimeoutMs?: number
|
||||
}
|
||||
|
||||
export const Config: z<Config> = z.object({
|
||||
waitTimeoutMs: z.number().min(1).default(30_000),
|
||||
maxWaitTimeoutMs: z.number().min(1).default(600_000),
|
||||
})
|
||||
|
||||
/**
|
||||
* Render generic status with optional producer detail.
|
||||
* @param snapshot - task state to render.
|
||||
* @returns a bracketed status line.
|
||||
*/
|
||||
export function statusLine(snapshot: TaskSnapshot): string {
|
||||
return snapshot.detail !== undefined
|
||||
? `[status: ${snapshot.status}, ${snapshot.detail}]`
|
||||
: `[status: ${snapshot.status}]`
|
||||
}
|
||||
|
||||
/** Validate the non-empty constraint that SchemaSpec cannot express. */
|
||||
function validateTaskId(value: string): TaskId {
|
||||
if (value.length === 0) {
|
||||
throw new Error(`invalid task_id: expected a non-empty string, got ${JSON.stringify(value)}`)
|
||||
}
|
||||
return TaskId(value)
|
||||
}
|
||||
|
||||
/** Pending presentation shared by the three generic task controls. */
|
||||
function presentTaskCall(title: string, kind: 'read' | 'execute', rawInput?: string): GenericCallView {
|
||||
return { card: 'generic', title, kind, ...rawInput !== undefined ? { rawInput } : {} }
|
||||
}
|
||||
|
||||
export function apply(ctx: Context, config: Config): void {
|
||||
const waitDefault = config.waitTimeoutMs ?? 30_000
|
||||
const waitCap = config.maxWaitTimeoutMs ?? 600_000
|
||||
if (waitDefault > waitCap) {
|
||||
throw new Error(`tool-tasks: waitTimeoutMs (${waitDefault}) exceeds maxWaitTimeoutMs (${waitCap})`)
|
||||
}
|
||||
|
||||
// Producers may start work only while a control surface is attached.
|
||||
ctx.tasks.attachSurface('tool-tasks')
|
||||
|
||||
// Cross-call guidance follows the bash section and precedes product sections.
|
||||
ctx.systemPrompt.section({
|
||||
name: 'tool:tasks',
|
||||
order: 106,
|
||||
text: 'Track every background task id you start. You are notified in-session when a task finishes — do not busy-poll or sleep on one; keep working on independent steps and do not duplicate a running task\'s work. Before giving a final answer, collect every still-relevant task with task_output (set wait: true only when you are genuinely blocked on it), and task_kill tasks that stopped mattering.',
|
||||
})
|
||||
|
||||
// Use the exact lifecycle owner; reusable ids could resolve to a replacement.
|
||||
ctx.tasks.onTaskDone((snapshot, owner) => {
|
||||
if (snapshot.reported || owner === undefined) return
|
||||
try {
|
||||
owner.inject(
|
||||
[{ type: 'text', text: `background task ${snapshot.id} (${snapshot.kind}: ${snapshot.label}) finished ${statusLine(snapshot)}. Read its output with task_output.` }],
|
||||
{ source: { kind: 'plugin', plugin: 'tool-tasks' } },
|
||||
)
|
||||
} catch (error: unknown) {
|
||||
// Disposal may win the race after settlement; other injection failures surface.
|
||||
if (error instanceof Error && error.message.includes('is disposed')) return
|
||||
throw error
|
||||
}
|
||||
})
|
||||
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'task_output',
|
||||
description: 'Read a background task. Stream tasks return only output since the previous read; '
|
||||
+ 'final-output tasks return their result after settlement. Every response ends with '
|
||||
+ '`[status: ...]`. Reads are non-blocking unless `wait: true`, which waits up to the configured cap.',
|
||||
// A timed-out wait returns task state rather than a TOOL_TIMEOUT error, so
|
||||
// this tool owns its deadline instead of using ToolDefinition.timeoutMs.
|
||||
parameters: {
|
||||
task_id: { type: 'string', required: true, description: 'Task id returned by the tool that started the background work.' },
|
||||
wait: { type: 'boolean', description: 'Block until the task reaches a terminal status or the timeout expires. A timed-out wait returns [status: running] and leaves the task alive.' },
|
||||
timeout_ms: { type: 'number', description: 'Max wait in milliseconds (only meaningful with wait: true). Defaults to the configured wait timeout; capped by the configured maximum.' },
|
||||
},
|
||||
async execute(args, exec) {
|
||||
const id = validateTaskId(args.task_id)
|
||||
if (args.wait === true) {
|
||||
const timeout = Math.min(args.timeout_ms ?? waitDefault, waitCap)
|
||||
await ctx.tasks.wait(id, timeout, exec.agent, exec.signal)
|
||||
}
|
||||
const read = ctx.tasks.read(id, exec.agent)
|
||||
const body = read.text.length > 0 ? read.text : '(no new output)'
|
||||
const separator = body.endsWith('\n') ? '' : '\n'
|
||||
return [{ type: 'text', text: `${body}${separator}${statusLine(read.snapshot)}` }]
|
||||
},
|
||||
presentCall: args => presentTaskCall(`Read output from background task ${args.task_id}`, 'read', args.task_id),
|
||||
}))
|
||||
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'task_list',
|
||||
description: 'List your background tasks (running and finished) with their ids, kinds, and statuses.',
|
||||
parameters: {},
|
||||
execute(_args, exec) {
|
||||
const tasks = ctx.tasks.list(exec.agent)
|
||||
const text = tasks.length === 0
|
||||
? '(no background tasks)'
|
||||
: tasks.map(t => `${t.id} [${t.kind}] ${t.status} — ${t.label}`).join('\n')
|
||||
return Promise.resolve([{ type: 'text', text }])
|
||||
},
|
||||
presentCall: () => presentTaskCall('List background tasks', 'read'),
|
||||
}))
|
||||
|
||||
ctx.tools.register(defineTool({
|
||||
name: 'task_kill',
|
||||
description: 'Request cancellation of a running background task by task id. Returns immediately; the task settles as killed once its work actually stops.',
|
||||
parameters: {
|
||||
task_id: { type: 'string', required: true, description: 'Task id returned by the tool that started the background work.' },
|
||||
reason: { type: 'string', description: 'Optional short reason, recorded in the log and forwarded to the task.' },
|
||||
},
|
||||
execute(args, exec) {
|
||||
const id = validateTaskId(args.task_id)
|
||||
const result = ctx.tasks.kill(id, exec.agent, args.reason)
|
||||
if (result === 'already-finished') {
|
||||
// A snapshot describes terminal state without consuming pending output.
|
||||
const snapshot = ctx.tasks.get(id, exec.agent)
|
||||
return Promise.resolve([{ type: 'text', text: `task ${id} had already finished ${statusLine(snapshot)}` }])
|
||||
}
|
||||
return Promise.resolve([{ type: 'text', text: `requested cancellation of task ${id}` }])
|
||||
},
|
||||
presentCall: args => presentTaskCall(`Kill background task ${args.task_id}`, 'execute', args.task_id),
|
||||
}))
|
||||
}
|
||||
336
packages/tasks/tool-tasks/tests/tool-tasks.spec.ts
Normal file
336
packages/tasks/tool-tasks/tests/tool-tasks.spec.ts
Normal file
@@ -0,0 +1,336 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { CallId } from '@deepseek-ai/dsh-llm'
|
||||
import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
|
||||
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 { 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'
|
||||
|
||||
const agentRegistryDisposers = new WeakMap<Agent, () => void>()
|
||||
|
||||
async function setup(config: ToolTasks.Config = {}) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
const agentsFiber = await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(TaskService)
|
||||
const toolsFiber = await ctx.plugin(ToolTasks, config)
|
||||
return { ctx, agentsFiber, toolsFiber }
|
||||
}
|
||||
|
||||
/**
|
||||
* A fake agent whose session token is `sessionId`, registered in `ctx.agents`.
|
||||
* The agent id is deliberately different so session authorization and exact
|
||||
* lifecycle ownership cannot be confused in tests.
|
||||
*/
|
||||
function fakeAgent(ctx: Context, sessionId: string, inject: (...args: unknown[]) => void = () => {}): Agent {
|
||||
const scopeFiber = ctx.plugin(() => {})
|
||||
const agent = {
|
||||
id: `agent-${sessionId}`,
|
||||
ctx: scopeFiber.ctx,
|
||||
inject,
|
||||
session: { header: { version: 0, id: sessionId, createdAt: 0 } },
|
||||
} as unknown as Agent
|
||||
agentRegistryDisposers.set(agent, ctx.agents.register(agent))
|
||||
return agent
|
||||
}
|
||||
|
||||
function detachAgent(agent: Agent): void {
|
||||
const dispose = agentRegistryDisposers.get(agent)
|
||||
if (dispose === undefined) throw new Error(`missing registry disposer for agent "${agent.id}"`)
|
||||
dispose()
|
||||
}
|
||||
|
||||
/** 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 { kind = 'bash', label = 'sleep 60', owner, ...hookOverrides } = overrides
|
||||
const hooks: TaskHooks = {
|
||||
cancel(reason) { cancels.push(reason) },
|
||||
done: new Promise<TaskOutcome>((res) => { settle = res }),
|
||||
...hookOverrides,
|
||||
}
|
||||
const spec: TaskStart = { kind, label, ...owner !== undefined ? { owner } : {}, run: () => hooks }
|
||||
return { spec, settle, cancels }
|
||||
}
|
||||
|
||||
let callCounter = 0
|
||||
function call(ctx: Context, name: string, args: unknown, agent?: Agent) {
|
||||
return ctx.tools.execute({ callId: CallId(`call-${++callCounter}`), name, arguments: args, ...agent ? { agent } : {} })
|
||||
}
|
||||
|
||||
function text(result: { content: { type: string; text?: string }[] }): string {
|
||||
return result.content.filter(block => block.type === 'text').map(block => block.text).join('')
|
||||
}
|
||||
|
||||
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.start(producer().spec)).not.toThrow()
|
||||
await toolsFiber.dispose()
|
||||
expect(() => ctx.tasks.start(producer().spec)).toThrow('no control surface is attached')
|
||||
})
|
||||
|
||||
it('rejects a config whose default wait exceeds the cap', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(TaskService)
|
||||
await expect(ctx.plugin(ToolTasks, { waitTimeoutMs: 100, maxWaitTimeoutMs: 50 }))
|
||||
.rejects.toThrow('waitTimeoutMs (100) exceeds maxWaitTimeoutMs (50)')
|
||||
})
|
||||
|
||||
it('renders status lines with and without producer detail', () => {
|
||||
const base = { id: 'bash-1', kind: 'bash', label: 'x', startedAt: 0, reported: false } as unknown as TaskSnapshot
|
||||
expect(statusLine({ ...base, status: 'running' })).toBe('[status: running]')
|
||||
expect(statusLine({ ...base, status: 'completed', detail: 'exit code: 0' })).toBe('[status: completed, exit code: 0]')
|
||||
})
|
||||
|
||||
it('applies the built-in wait bounds when apply() receives a bare config', async () => {
|
||||
// Bypasses the schemastery defaults on purpose: apply() must stand on its
|
||||
// own `??` fallbacks when embedded programmatically without the schema.
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SystemPrompt)
|
||||
await ctx.plugin(ToolRegistry)
|
||||
await ctx.plugin(TaskService)
|
||||
ToolTasks.apply(ctx, {})
|
||||
expect(ctx.tools.get('task_output')).toBeDefined()
|
||||
expect(() => ctx.tasks.start(producer().spec)).not.toThrow()
|
||||
})
|
||||
})
|
||||
|
||||
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.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]')
|
||||
expect(text(await call(ctx, 'task_output', { task_id: 'bash-1' }))).toBe('(no new output)\n[status: running]')
|
||||
})
|
||||
|
||||
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.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' })
|
||||
await tick()
|
||||
expect(text(await call(ctx, 'task_output', { task_id: 'subagent-1' }))).toBe('the answer\n[status: completed, completed]')
|
||||
})
|
||||
|
||||
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.start(p.spec)
|
||||
|
||||
const pending = call(ctx, 'task_output', { task_id: 'subagent-1', wait: true })
|
||||
p.settle({ status: 'completed', output: 'done deal' })
|
||||
expect(text(await pending)).toBe('done deal\n[status: completed]')
|
||||
})
|
||||
|
||||
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.start(producer().spec)
|
||||
|
||||
// A model-supplied timeout far above the cap is clamped: this returns
|
||||
// promptly (≤ the 20ms cap), not after ten minutes.
|
||||
const result = await call(ctx, 'task_output', { task_id: 'bash-1', wait: true, timeout_ms: 600_000 })
|
||||
expect(text(result)).toBe('(no new output)\n[status: running]')
|
||||
})
|
||||
|
||||
it('rejects an empty or unknown task id as an errored result', async () => {
|
||||
const { ctx } = await setup()
|
||||
expect((await call(ctx, 'task_output', { task_id: '' })).isError).toBe(true)
|
||||
const unknown = await call(ctx, 'task_output', { task_id: 'bash-99' })
|
||||
expect(unknown.isError).toBe(true)
|
||||
expect(text(unknown)).toContain('unknown task bash-99')
|
||||
})
|
||||
})
|
||||
|
||||
describe('task_list', () => {
|
||||
it('lists caller-visible tasks and renders the empty case', async () => {
|
||||
const { ctx } = await setup()
|
||||
expect(text(await call(ctx, 'task_list', {}))).toBe('(no background tasks)')
|
||||
|
||||
const alice = fakeAgent(ctx, 'sess-alice')
|
||||
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.start(p.spec)
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
await tick()
|
||||
|
||||
expect(text(await call(ctx, 'task_list', {}, alice))).toBe([
|
||||
'bash-1 [bash] running — pnpm test',
|
||||
'subagent-1 [subagent] running — open research',
|
||||
'bash-2 [bash] completed — build',
|
||||
].join('\n'))
|
||||
// A different caller sees only the unowned task.
|
||||
const bob = fakeAgent(ctx, 'sess-bob')
|
||||
expect(text(await call(ctx, 'task_list', {}, bob))).toBe('subagent-1 [subagent] running — open research')
|
||||
})
|
||||
})
|
||||
|
||||
describe('task_kill', () => {
|
||||
it('requests cancellation with the forwarded reason', async () => {
|
||||
const { ctx } = await setup()
|
||||
const p = producer()
|
||||
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')
|
||||
expect(p.cancels).toEqual(['superseded'])
|
||||
})
|
||||
|
||||
it('reports an already-finished task without consuming its pending delta', async () => {
|
||||
const { ctx } = await setup()
|
||||
let delta = 'unread tail'
|
||||
const p = producer({ readOutput: () => { const d = delta; delta = ''; return d } })
|
||||
ctx.tasks.start(p.spec)
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
await tick()
|
||||
|
||||
expect(text(await call(ctx, 'task_kill', { task_id: 'bash-1' })))
|
||||
.toBe('task bash-1 had already finished [status: completed, exit code: 0]')
|
||||
// The kill described the task via a non-consuming snapshot: the delta is intact.
|
||||
expect(text(await call(ctx, 'task_output', { task_id: 'bash-1' }))).toBe('unread tail\n[status: completed, exit code: 0]')
|
||||
})
|
||||
|
||||
it('rejects an empty task id as an errored result', async () => {
|
||||
const { ctx } = await setup()
|
||||
expect((await call(ctx, 'task_kill', { task_id: '' })).isError).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('tool-owned UI presentation (presentCall)', () => {
|
||||
it('renders generic cards for all three control tools', async () => {
|
||||
const { ctx } = await setup()
|
||||
expect(ctx.tools.get('task_output')?.presentCall?.({ task_id: 'bash-1' }))
|
||||
.toEqual({ card: 'generic', title: 'Read output from background task bash-1', kind: 'read', rawInput: 'bash-1' })
|
||||
expect(ctx.tools.get('task_list')?.presentCall?.({}))
|
||||
.toEqual({ card: 'generic', title: 'List background tasks', kind: 'read' })
|
||||
expect(ctx.tools.get('task_kill')?.presentCall?.({ task_id: 'subagent-2' }))
|
||||
.toEqual({ card: 'generic', title: 'Kill background task subagent-2', kind: 'execute', rawInput: 'subagent-2' })
|
||||
})
|
||||
})
|
||||
|
||||
describe('completion notices', () => {
|
||||
it('injects a notice into the owning agent when an unreported task settles', async () => {
|
||||
const { ctx } = await setup()
|
||||
const inject = vi.fn()
|
||||
const owner = fakeAgent(ctx, 'sess-1', inject)
|
||||
const p = producer({ owner, label: 'pnpm test' })
|
||||
ctx.tasks.start(p.spec)
|
||||
|
||||
p.settle({ status: 'completed', detail: 'exit code: 0' })
|
||||
await tick()
|
||||
expect(inject).toHaveBeenCalledTimes(1)
|
||||
expect(inject).toHaveBeenCalledWith(
|
||||
[{ type: 'text', text: 'background task bash-1 (bash: pnpm test) finished [status: completed, exit code: 0]. Read its output with task_output.' }],
|
||||
{ source: { kind: 'plugin', plugin: 'tool-tasks' } },
|
||||
)
|
||||
})
|
||||
|
||||
it('suppresses the notice for a task the model already killed', async () => {
|
||||
const { ctx } = await setup()
|
||||
const inject = vi.fn()
|
||||
const owner = fakeAgent(ctx, 'sess-1', inject)
|
||||
const p = producer({ owner })
|
||||
ctx.tasks.start(p.spec)
|
||||
|
||||
await call(ctx, 'task_kill', { task_id: 'bash-1' }, owner)
|
||||
p.settle({ status: 'killed' })
|
||||
await tick()
|
||||
expect(inject).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('suppresses the notice when a wait returned the terminal state', async () => {
|
||||
const { ctx } = await setup()
|
||||
const inject = vi.fn()
|
||||
const owner = fakeAgent(ctx, 'sess-1', inject)
|
||||
const p = producer({ owner, kind: 'subagent' })
|
||||
ctx.tasks.start(p.spec)
|
||||
|
||||
const pending = call(ctx, 'task_output', { task_id: 'subagent-1', wait: true }, owner)
|
||||
p.settle({ status: 'completed', output: 'answer' })
|
||||
expect(text(await pending)).toContain('answer')
|
||||
expect(inject).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('drops the notice for unowned tasks and for a disposed owner (benign race)', async () => {
|
||||
const { ctx } = await setup()
|
||||
// Unowned: settles with nobody to notify — nothing throws.
|
||||
const unowned = producer()
|
||||
ctx.tasks.start(unowned.spec)
|
||||
unowned.settle({ status: 'completed' })
|
||||
await tick()
|
||||
|
||||
// Disposed owner: inject throws the disposed message — contained.
|
||||
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.start(p.spec)
|
||||
p.settle({ status: 'completed' })
|
||||
await tick()
|
||||
expect(inject).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('does not route an old owner completion notice to a same-session replacement', async () => {
|
||||
const { ctx } = await setup()
|
||||
const oldInject = vi.fn(() => { throw new Error('agent "agent-shared" is disposed') })
|
||||
const oldOwner = fakeAgent(ctx, 'shared', oldInject)
|
||||
const p = producer({ owner: oldOwner })
|
||||
ctx.tasks.start(p.spec)
|
||||
|
||||
detachAgent(oldOwner)
|
||||
const replacementInject = vi.fn()
|
||||
fakeAgent(ctx, 'shared', replacementInject)
|
||||
p.settle({ status: 'completed' })
|
||||
await tick()
|
||||
|
||||
expect(oldInject).toHaveBeenCalledTimes(1)
|
||||
expect(replacementInject).not.toHaveBeenCalled()
|
||||
})
|
||||
|
||||
it('propagates a non-disposed inject failure (a real bug must surface)', async () => {
|
||||
const { ctx } = await setup()
|
||||
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.start(p.spec)
|
||||
p.settle({ status: 'completed' })
|
||||
await tick()
|
||||
// The throw escapes the notice listener and is contained (logged) by the
|
||||
// registry's per-listener containment — visible, not swallowed.
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('unexpected inject bug'))
|
||||
})
|
||||
|
||||
it('keeps using the exact owner after the agent registry is gone', async () => {
|
||||
const { ctx, agentsFiber } = await setup()
|
||||
const inject = vi.fn()
|
||||
const owner = fakeAgent(ctx, 'sess-1', inject)
|
||||
|
||||
// Settlement must not depend on a later registry lookup: the exact owner
|
||||
// supplied at start remains the destination while its own scope is live.
|
||||
const p1 = producer({ owner })
|
||||
ctx.tasks.start(p1.spec)
|
||||
const p2 = producer({ owner })
|
||||
ctx.tasks.start(p2.spec)
|
||||
|
||||
await agentsFiber.dispose()
|
||||
p1.settle({ status: 'completed' })
|
||||
p2.settle({ status: 'failed' })
|
||||
await tick()
|
||||
expect(inject).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
})
|
||||
33
packages/tasks/tool-tasks/tsconfig.json
Normal file
33
packages/tasks/tool-tasks/tsconfig.json
Normal file
@@ -0,0 +1,33 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib/types"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/schemastery"
|
||||
},
|
||||
{
|
||||
"path": "../../core/agent"
|
||||
},
|
||||
{
|
||||
"path": "../../core/system-prompt"
|
||||
},
|
||||
{
|
||||
"path": "../../core/tools"
|
||||
},
|
||||
{
|
||||
"path": "../tasks"
|
||||
}
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user