mirror of
https://github.com/deepseek-ai/deepseek-harness
synced 2026-08-15 21:04:50 +00:00
fix(tasks): bind cleanup and notices to exact owners
Task records previously retained only ownerSession. If an old agent scope unwound after another agent reused the same agent and session ids, cleanup selected both records and could cancel replacement work. The completion surface also re-resolved the session at settlement, which could inject an old task notice into the replacement agent. Retain the exact Agent instance for lifecycle work, select owner cleanup by object identity, and pass that exact owner to completion listeners. Keep read, list, kill, and wait authorization session-based as the runtime RFC intends. Add regressions for cleanup and notice routing under id reuse, then update the public docs and generated API catalogs.
This commit is contained in:
Tianyi Cui
@@ -259,7 +259,7 @@ attachSurface(name: string): () => void
|
||||
|
||||
Types: [Agent](../core-data-structures/core.md)
|
||||
|
||||
Source: [`packages/tasks/tasks/src/index.ts:99`](../../packages/tasks/tasks/src/index.ts)
|
||||
Source: [`packages/tasks/tasks/src/index.ts:98`](../../packages/tasks/tasks/src/index.ts)
|
||||
|
||||
## `ctx.tools` — `ToolRegistry`
|
||||
|
||||
|
||||
@@ -99,11 +99,12 @@ interface TaskSnapshot {
|
||||
/** The producer-supplied one-line label. */
|
||||
label: string
|
||||
/**
|
||||
* The owner's session id (`session.header.id`), for surfaces that must
|
||||
* reach the owning agent (the completion-notice injector); absent for
|
||||
* unowned tasks. Session ids are runtime-shared identifiers, not secrets —
|
||||
* the read/kill/wait/list FENCE is what isolation rests on. The shared
|
||||
* {@link SessionId} brand is preserved across this package boundary.
|
||||
* The owner's session id (`session.header.id`), for authorization and
|
||||
* correlation; absent for unowned tasks. A listener that must reach the
|
||||
* lifecycle owner receives the exact Agent separately through
|
||||
* {@link TaskDoneListener}. Session ids are runtime-shared identifiers, not
|
||||
* secrets — the read/kill/wait/list FENCE is what isolation rests on. The
|
||||
* shared {@link SessionId} brand is preserved across this package boundary.
|
||||
*/
|
||||
ownerSession?: SessionId
|
||||
/** Current lifecycle state. */
|
||||
@@ -140,4 +141,4 @@ interface TaskRead {
|
||||
|
||||
## The service
|
||||
|
||||
`TaskService` (`ctx.tasks` — [`packages/tasks/tasks/src/index.ts`](../../packages/tasks/tasks/src/index.ts)): `start` (preflight → producer `run()` → atomic commit, fenced by `attachSurface`), non-consuming `get`/`list` (caller-scoped — owned-by-caller plus unowned only), `read` (consuming for stream kinds), `kill` (producer `cancel` first; a throw leaves the task untouched), `wait` (bounded, abort cancels the wait only), and `onTaskDone` (a `TaskDoneListener` per terminal record, effect-scoped, contained). Start validates that an owned task names the exact live Agent instance currently registered under its id, so an old reference cannot bind work to a replacement agent's cleanup after id reuse. Every read/kill/wait/get separately compares the task's owner session with the caller's and rejects a foreign one. Owned tasks register one async cleanup through the exact owner's `agent.ctx`; scope disposal cancels them and normally awaits producer quiescence. A teardown cancel that throws force-fails only the registry record and reports that the underlying work may be orphaned, preventing disposal deadlock without claiming quiescence. The model-facing surface over all of this is [dsh-tool-tasks](../../packages/tasks/tool-tasks/README.md).
|
||||
`TaskService` (`ctx.tasks` — [`packages/tasks/tasks/src/index.ts`](../../packages/tasks/tasks/src/index.ts)): `start` (preflight → producer `run()` → atomic commit, fenced by `attachSurface`), non-consuming `get`/`list` (caller-scoped — owned-by-caller plus unowned only), `read` (consuming for stream kinds), `kill` (producer `cancel` first; a throw leaves the task untouched), `wait` (bounded, abort cancels the wait only), and `onTaskDone` (a `TaskDoneListener` per terminal record, given the exact lifecycle owner, effect-scoped, contained). Start retains the exact live Agent instance validated under its id; owner-scope cleanup selects by that identity, so a reused agent/session id cannot make an old scope cancel replacement work. Read/kill/wait/get authorization remains session-based and rejects a foreign session. A teardown cancel that throws force-fails only the registry record and reports that the underlying work may be orphaned, preventing disposal deadlock without claiming quiescence. The model-facing surface over all of this is [dsh-tool-tasks](../../packages/tasks/tool-tasks/README.md).
|
||||
|
||||
@@ -65,7 +65,7 @@ Registrations are NOT effect-scoped to the registering fiber: a task belongs to
|
||||
|
||||
## Authorization and the service surface
|
||||
|
||||
Cross-session isolation lives IN the runtime so every consumer gets the same rule for free: read/kill/wait/get take the caller (`Agent | undefined`), and a task whose owner session differs from the caller's session is rejected (`!== undefined` comparison — an unowned task is open, a no-agent caller cannot match an owned one). `list(caller)` returns only the caller-visible tasks (owned-by-caller or unowned) — a global listing would leak other sessions' labels. The snapshot carries that owner as the canonical branded `SessionId`, not a package-local token or bare string. Lifecycle ownership is checked independently at start: the supplied owner must be the exact live `Agent` instance currently registered under its id, so an old object cannot attach its session's work to a replacement agent's cleanup after id reuse.
|
||||
Cross-session isolation lives IN the runtime so every consumer gets the same rule for free: read/kill/wait/get take the caller (`Agent | undefined`), and a task whose owner session differs from the caller's session is rejected (`!== undefined` comparison — an unowned task is open, a no-agent caller cannot match an owned one). `list(caller)` returns only the caller-visible tasks (owned-by-caller or unowned) — a global listing would leak other sessions' labels. The snapshot carries that authorization identity as the canonical branded `SessionId`, not a package-local token or bare string. Lifecycle ownership is independent: start retains the exact live `Agent` instance, owner cleanup selects by object identity, and completion listeners receive that exact owner, so id reuse cannot redirect cleanup or notices to a replacement.
|
||||
|
||||
```ts ignore-check
|
||||
class TaskService extends Service { // ctx.tasks
|
||||
@@ -75,7 +75,7 @@ class TaskService extends Service { // ctx.tasks
|
||||
read(id: TaskId, caller?: Agent): TaskRead // delta (stream kinds, consuming) or final output (final kinds, idempotent) + snapshot
|
||||
kill(id: TaskId, caller?: Agent, reason?: string): 'requested' | 'already-terminal'
|
||||
wait(id: TaskId, timeoutMs: number, caller?: Agent, signal?: AbortSignal): Promise<TaskSnapshot>
|
||||
onTaskDone(listener: (snapshot: TaskSnapshot) => void): () => void // effect-scoped, contained, never fires after dispose
|
||||
onTaskDone(listener: (snapshot: TaskSnapshot, owner: Agent | undefined) => void): () => void // exact lifecycle owner; effect-scoped, contained
|
||||
attachSurface(name: string): () => void // the misconfiguration fence, below
|
||||
}
|
||||
```
|
||||
@@ -96,7 +96,7 @@ class TaskService extends Service { // ctx.tasks
|
||||
|
||||
One system-prompt section (order 106, next to `tool:bash`) teaches the cross-call habit the per-tool descriptions cannot: track every returned task id; you are notified in-session when a task finishes, so do not busy-poll or sleep on one — keep working on independent steps and do not duplicate a running task's work; do not produce a final answer while a relevant task still runs — call `task_output` (with `wait` when blocked) to collect it first; `task_kill` tasks that stopped mattering. The do-not-poll and do-not-duplicate sentences are near-verbatim convergent across Claude Code, Kimi Code, and OpenCode — they are the two failure modes every peer engineered against.
|
||||
|
||||
Completion notices stay durable context, not a wake-up (`agent.inject()` appends a logged `context/message` the next model request sees; it does not run the model): on `onTaskDone`, `dsh-tool-tasks` injects `background task <id> (<kind>: <label>) finished [status: …]. Read its output with task_output.` into the owning agent's session, with the same disposed-race containment `dsh-tool-bash` used to carry. Notices are deduplicated the way Claude Code and Kimi Code both learned to: a task the model explicitly killed, or whose terminal state a read/wait already returned (including a wait pending at the moment of settlement), is marked `reported` and its notice suppressed — never a redundant "finished" for work the model just collected or ended. The model-visible ⟺ logged invariant holds with no new session event type.
|
||||
Completion notices stay durable context, not a wake-up (`agent.inject()` appends a logged `context/message` the next model request sees; it does not run the model): on `onTaskDone`, `dsh-tool-tasks` injects `background task <id> (<kind>: <label>) finished [status: …]. Read its output with task_output.` through the exact owner captured at start, never a replacement found through a reused agent/session id, with the disposed-race contained. Notices are deduplicated the way Claude Code and Kimi Code both learned to: a task the model explicitly killed, or whose terminal state a read/wait already returned (including a wait pending at the moment of settlement), is marked `reported` and its notice suppressed — never a redundant "finished" for work the model just collected or ended. The model-visible ⟺ logged invariant holds with no new session event type.
|
||||
|
||||
## Producer opt-in and schema exposure
|
||||
|
||||
@@ -106,7 +106,7 @@ Whether a producer tool offers `run_in_background` is that producer's own defaul
|
||||
|
||||
A contract-compliant background task must not outlive its owner: the subagent case leaks live child agents/sessions otherwise, and `agent/disposed` is an observe-only notification rather than a quiescence seam. Every live agent already owns an awaited structural registration scope ([agent-scope contract](2026-07-08-agent-scope-contexts.md)), so the task runtime uses that single lifecycle mechanism:
|
||||
|
||||
- After validating the exact registered owner instance, the first task for that owner registers an async effect through `owner.ctx`. The effect belongs to the agent scope, survives producer reloads, cancels that owner's live tasks, awaits each terminal record, and drops the snapshots.
|
||||
- After validating and retaining the exact registered owner instance, the first task for that owner registers an async effect through `owner.ctx`. The effect belongs to the agent scope, survives producer reloads, selects tasks by exact owner identity, cancels them, awaits each terminal record, and drops the snapshots; a replacement reusing the same ids is outside that set.
|
||||
- `AgentHandle.dispose()` stops and drains the driver, detaches the agent and session, then awaits scope disposal. The task cleanup therefore participates in the same memoized quiescence boundary as every other agent-owned registration; no task-specific link exists in `AgentRegistry` or the loop.
|
||||
- The tasks service retains each exact owner-effect disposer so service reload can detach callbacks from still-live scopes after global task teardown, rather than leaving a dead service retained until every agent exits.
|
||||
|
||||
|
||||
@@ -939,7 +939,7 @@ export const TYPE_API: readonly TypeApiEntry[] = [
|
||||
},
|
||||
{
|
||||
name: 'TaskDoneListener',
|
||||
declaration: 'export type TaskDoneListener = (snapshot: TaskSnapshot) => void;',
|
||||
declaration: 'export type TaskDoneListener = (snapshot: TaskSnapshot, owner: Agent | undefined) => void;',
|
||||
},
|
||||
{
|
||||
name: 'TaskHooks',
|
||||
|
||||
@@ -9,7 +9,7 @@ The background task registry (`ctx.tasks`): a runtime-global, CONCRETE service (
|
||||
- `read(id, caller?): TaskRead` — stream kinds consume the per-task cursor (v1's single intended reader is the owning model — a non-consuming multi-reader surface would be a cursor/snapshot API extension, not a `read` change); final kinds read the terminal output idempotently.
|
||||
- `kill(id, caller?, reason?)` — `'requested'` (live task: producer `cancel` runs first — a throw fails the kill loud and leaves the task untouched — then `stopping`) or `'already-terminal'`. Every successful kill marks the task `reported` (the killer saw the end → completion notice suppressed).
|
||||
- `wait(id, timeoutMs, caller?, signal?)` — resolves with the terminal snapshot (marked `reported`), or the live snapshot at timeout; an aborted signal rejects the WAIT only — unless the task already settled, in which case the wait still resolves and delivers the terminal snapshot (settlement suppressed the completion notice on this waiter's behalf, so rejecting would leave the finish both unreported and un-noticed). Timing is a [`dsh-timeout`](../../util/timeout/README.md) `deadline()` scoped to the `TASK_WAIT_TIMEOUT` code, so a nested foreign deadline never misreads as a wait timeout; timeout and abort detach their settlement resolver immediately, keeping retention bounded while the task remains live.
|
||||
- `onTaskDone(listener)` — exactly once per terminal task record; effect-scoped, per-listener containment, silent after service disposal.
|
||||
- `onTaskDone(listener)` — exactly once per terminal task record, with the snapshot plus its exact lifecycle owner (or `undefined`); effect-scoped, per-listener containment, silent after service disposal.
|
||||
- `attachSurface(name)` — declares a control surface exists (the model tools, or a deployment's custom surface); effect-scoped.
|
||||
|
||||
Every read/kill/wait/get compares the task's owner session (`owner.session.header.id`) with the caller's and rejects a foreign one — ids are predictable (`bash-1`), so the fence, not id secrecy, is the isolation boundary.
|
||||
@@ -17,7 +17,7 @@ Every read/kill/wait/get compares the task's owner session (`owner.session.heade
|
||||
## Lifecycle
|
||||
|
||||
- Registrations are NOT effect-scoped to the registering fiber: tasks belong to their owning agent + producing backend, so producer/surface HMR reloads never touch them.
|
||||
- An owned task must name the exact live `Agent` instance currently registered under its id (stale objects are rejected after id reuse), then attaches one awaited cleanup through `owner.ctx`: agent-scope disposal cancels live tasks, awaits contract-compliant producers to quiescence, and drops their snapshots. If a teardown cancel throws, it force-fails the record and logs that the underlying work may be orphaned rather than deadlocking `AgentHandle.dispose()`.
|
||||
- An owned task retains the exact live `Agent` instance validated at start and attaches one awaited cleanup through `owner.ctx`: agent-scope disposal selects only that instance's tasks, cancels them, awaits contract-compliant producers to quiescence, and drops their snapshots. Reused agent/session ids cannot make an old cleanup sweep replacement work. If a teardown cancel throws, it force-fails the record and logs that the underlying work may be orphaned rather than deadlocking `AgentHandle.dispose()`.
|
||||
- Service disposal closes the listener registry first, applies the same cancellation rule to every live task, awaits terminal records, then detaches its effects from still-live agent scopes so a reloaded tasks service is not retained until those agents exit.
|
||||
- A producer whose `cancel` returns but never causes `done` to settle remains indistinguishable from a slow stop and can stall teardown; solving that residual requires an explicit bounded-lifetime or forced-disposal design.
|
||||
|
||||
|
||||
@@ -32,7 +32,6 @@
|
||||
|
||||
import { Context, Service } from 'cordis'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import type { SessionId } from '@deepseek-ai/dsh-session'
|
||||
import { deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
|
||||
import { TaskId } from './types.ts'
|
||||
import type { TaskDoneListener, TaskOutcome, TaskRead, TaskSnapshot, TaskStart, TaskStatus } from './types.ts'
|
||||
@@ -67,8 +66,8 @@ interface TrackedTask {
|
||||
id: TaskId
|
||||
kind: string
|
||||
label: string
|
||||
/** The owner's session id (`session.header.id`), or undefined for an unowned task. */
|
||||
ownerSession: SessionId | undefined
|
||||
/** Exact lifecycle owner; session-id authorization is derived from it. */
|
||||
owner: Agent | undefined
|
||||
cancel: (reason?: string) => void
|
||||
readOutput: (() => string) | undefined
|
||||
status: TaskStatus
|
||||
@@ -159,7 +158,7 @@ export class TaskService extends Service {
|
||||
id,
|
||||
kind: spec.kind,
|
||||
label: spec.label,
|
||||
ownerSession: spec.owner?.session.header.id,
|
||||
owner: spec.owner,
|
||||
cancel: hooks.cancel.bind(hooks),
|
||||
readOutput: hooks.readOutput?.bind(hooks),
|
||||
status: 'running',
|
||||
@@ -197,7 +196,7 @@ export class TaskService extends Service {
|
||||
list(caller?: Agent): TaskSnapshot[] {
|
||||
const session = caller?.session.header.id
|
||||
return [...this.store.values()]
|
||||
.filter(task => task.ownerSession === undefined || task.ownerSession === session)
|
||||
.filter(task => task.owner === undefined || task.owner.session.header.id === session)
|
||||
.map(task => this.snapshot(task))
|
||||
}
|
||||
|
||||
@@ -349,10 +348,11 @@ export class TaskService extends Service {
|
||||
|
||||
/**
|
||||
* Register a completion listener, called exactly once per terminal task
|
||||
* record with its snapshot. Effect-scoped (disposed with the calling fiber);
|
||||
* per-listener containment (one throwing listener is logged, never starves
|
||||
* the rest); never fires after this service is disposed.
|
||||
* @param listener - called with each settling task's terminal snapshot.
|
||||
* record with its snapshot and exact lifecycle owner (or `undefined` for an
|
||||
* unowned task). Effect-scoped (disposed with the calling fiber); per-listener
|
||||
* containment (one throwing listener is logged, never starves the rest);
|
||||
* never fires after this service is disposed.
|
||||
* @param listener - called with each terminal snapshot and its exact owner.
|
||||
* @returns the disposer that unregisters the listener.
|
||||
*/
|
||||
onTaskDone(listener: TaskDoneListener): () => void {
|
||||
@@ -398,18 +398,19 @@ export class TaskService extends Service {
|
||||
* open, and a no-agent caller can never match an owned one).
|
||||
*/
|
||||
private assertAccess(task: TrackedTask, caller?: Agent): void {
|
||||
if (task.ownerSession !== undefined && task.ownerSession !== caller?.session.header.id) {
|
||||
if (task.owner !== undefined && task.owner.session.header.id !== caller?.session.header.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?.session.header.id
|
||||
return {
|
||||
id: task.id,
|
||||
kind: task.kind,
|
||||
label: task.label,
|
||||
...task.ownerSession !== undefined ? { ownerSession: task.ownerSession } : {},
|
||||
...ownerSession !== undefined ? { ownerSession } : {},
|
||||
status: task.status,
|
||||
...task.detail !== undefined ? { detail: task.detail } : {},
|
||||
startedAt: task.startedAt,
|
||||
@@ -439,7 +440,7 @@ export class TaskService extends Service {
|
||||
const snapshot = this.snapshot(task)
|
||||
for (const listener of this.listeners) {
|
||||
try {
|
||||
listener(snapshot)
|
||||
listener(snapshot, task.owner)
|
||||
} catch (error: unknown) {
|
||||
this.selfCtx.logger.warn(`tasks: onTaskDone listener threw for ${task.id}: ${String(error)}`)
|
||||
}
|
||||
@@ -474,19 +475,18 @@ export class TaskService extends Service {
|
||||
throw new Error(`agent "${ownerId}" is not the registered agent instance (background task owner must be live)`)
|
||||
}
|
||||
if (this.ownerCleanups.has(owner)) return
|
||||
const ownerSession = owner.session.header.id
|
||||
// Attach FIRST, record after: an already-disposing scope rejects effects,
|
||||
// and marking the owner as covered before that would poison later starts.
|
||||
const detach = owner.ctx.effect(() => async () => {
|
||||
this.ownerCleanups.delete(owner)
|
||||
await this.disposeOwned(ownerSession)
|
||||
await this.disposeOwned(owner)
|
||||
}, 'tasks.ownerCleanup()')
|
||||
this.ownerCleanups.set(owner, detach)
|
||||
}
|
||||
|
||||
/** Cancel, await terminal records, and drop every task owned by one session. */
|
||||
private async disposeOwned(ownerSession: SessionId): Promise<void> {
|
||||
const owned = [...this.store.values()].filter(task => task.ownerSession === ownerSession)
|
||||
/** 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)
|
||||
|
||||
@@ -136,11 +136,12 @@ export interface TaskSnapshot {
|
||||
/** The producer-supplied one-line label. */
|
||||
label: string
|
||||
/**
|
||||
* The owner's session id (`session.header.id`), for surfaces that must
|
||||
* reach the owning agent (the completion-notice injector); absent for
|
||||
* unowned tasks. Session ids are runtime-shared identifiers, not secrets —
|
||||
* the read/kill/wait/list FENCE is what isolation rests on. The shared
|
||||
* {@link SessionId} brand is preserved across this package boundary.
|
||||
* The owner's session id (`session.header.id`), for authorization and
|
||||
* correlation; absent for unowned tasks. A listener that must reach the
|
||||
* lifecycle owner receives the exact Agent separately through
|
||||
* {@link TaskDoneListener}. Session ids are runtime-shared identifiers, not
|
||||
* secrets — the read/kill/wait/list FENCE is what isolation rests on. The
|
||||
* shared {@link SessionId} brand is preserved across this package boundary.
|
||||
*/
|
||||
ownerSession?: SessionId
|
||||
/** Current lifecycle state. */
|
||||
@@ -177,5 +178,9 @@ export interface TaskRead {
|
||||
snapshot: TaskSnapshot
|
||||
}
|
||||
|
||||
/** Completion callback registered via {@link TaskService.onTaskDone}. */
|
||||
export type TaskDoneListener = (snapshot: TaskSnapshot) => void
|
||||
/**
|
||||
* Completion callback registered via {@link TaskService.onTaskDone}.
|
||||
* `owner` is the exact lifecycle instance supplied at start, not a registry
|
||||
* lookup by reusable agent or session id; it is absent for unowned tasks.
|
||||
*/
|
||||
export type TaskDoneListener = (snapshot: TaskSnapshot, owner: Agent | undefined) => void
|
||||
|
||||
@@ -506,6 +506,39 @@ describe('TaskService owner cleanup', () => {
|
||||
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', 'shared-session')
|
||||
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', 'shared-session')
|
||||
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)
|
||||
|
||||
@@ -12,7 +12,7 @@ ACP render intent: all three are `generic` cards (`read`/`read`/`execute`) — a
|
||||
|
||||
## Completion notices
|
||||
|
||||
On `onTaskDone`, injects `background task <id> (<kind>: <label>) finished [status: …]. Read its output with task_output.` into the owning agent's session (`agent.inject()` — durable context for the next request, not a wake-up). Suppressed when the snapshot is `reported` (the model already killed it, or a read/wait returned the end) — never a redundant "finished". The disposed-owner race is contained; a missing agent registry drops the notice.
|
||||
On `onTaskDone`, injects `background task <id> (<kind>: <label>) finished [status: …]. Read its output with task_output.` through the exact owner `Agent` captured at task start (`agent.inject()` — durable context for the next request, not a wake-up). It never re-resolves a reusable agent/session id to a replacement. Suppressed when the snapshot is `reported` (the model already killed it, or a read/wait returned the end) — never a redundant "finished"; the disposed-owner race is contained.
|
||||
|
||||
## Config
|
||||
|
||||
|
||||
@@ -93,19 +93,15 @@ export function apply(ctx: Context, config: Config): void {
|
||||
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.',
|
||||
})
|
||||
|
||||
// Background completion → inject a notice into the owning agent's session.
|
||||
// `ctx.get('agents')` (not static inject): this listener runs from a
|
||||
// detached settlement continuation on the tasks fiber — a foreign fiber —
|
||||
// where the `ctx.agents` property proxy would throw; `ctx.get` is the
|
||||
// topology-independent lookup. No registry mounted → drop the notice.
|
||||
ctx.tasks.onTaskDone((snapshot) => {
|
||||
// Background completion → inject a notice through the exact lifecycle owner.
|
||||
// Re-resolving by a reusable agent/session id could target a replacement
|
||||
// while the old owner's scope is still unwinding.
|
||||
ctx.tasks.onTaskDone((snapshot, owner) => {
|
||||
// A reported terminal state was already surfaced by an explicit
|
||||
// read/wait/kill response — a notice would be a redundant "finished".
|
||||
if (snapshot.reported || snapshot.ownerSession === undefined) return
|
||||
const agent = ctx.get('agents')?.list().find(a => a.session.header.id === snapshot.ownerSession)
|
||||
if (!agent) return
|
||||
if (snapshot.reported || owner === undefined) return
|
||||
try {
|
||||
agent.inject(
|
||||
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' } },
|
||||
)
|
||||
|
||||
@@ -10,6 +10,8 @@ import type { TaskHooks, TaskOutcome, TaskSnapshot, TaskStart } from '@deepseek-
|
||||
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)
|
||||
@@ -21,10 +23,9 @@ async function setup(config: ToolTasks.Config = {}) {
|
||||
}
|
||||
|
||||
/**
|
||||
* A fake agent whose session token is `sessionId`, registered in `ctx.agents`
|
||||
* (the notice path finds the owner by scanning the registry for a matching
|
||||
* `session.header.id` — the agent id is deliberately DIFFERENT so a
|
||||
* wrong-field match fails the test).
|
||||
* 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(() => {})
|
||||
@@ -34,10 +35,16 @@ function fakeAgent(ctx: Context, sessionId: string, inject: (...args: unknown[])
|
||||
inject,
|
||||
session: { header: { version: 0, id: sessionId, createdAt: 0 } },
|
||||
} as unknown as Agent
|
||||
ctx.agents.register(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
|
||||
@@ -278,6 +285,23 @@ describe('completion notices', () => {
|
||||
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(() => {})
|
||||
@@ -291,15 +315,15 @@ describe('completion notices', () => {
|
||||
expect(warn).toHaveBeenCalledWith(expect.stringContaining('unexpected inject bug'))
|
||||
})
|
||||
|
||||
it('drops the notice when no live agent matches and when the agent registry is gone', async () => {
|
||||
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)
|
||||
|
||||
// Owner known at registration, unregistered before settlement → no match.
|
||||
// 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)
|
||||
// A second task whose settlement happens after the whole registry is gone.
|
||||
const p2 = producer({ owner })
|
||||
ctx.tasks.start(p2.spec)
|
||||
|
||||
@@ -307,6 +331,6 @@ describe('completion notices', () => {
|
||||
p1.settle({ status: 'completed' })
|
||||
p2.settle({ status: 'failed' })
|
||||
await tick()
|
||||
expect(inject).not.toHaveBeenCalled()
|
||||
expect(inject).toHaveBeenCalledTimes(2)
|
||||
})
|
||||
})
|
||||
|
||||
Reference in New Issue
Block a user