Coder/deepseek-harness
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(tasks): an aborted wait must not leave a settled task notice-suppressed (review finding)

Browse Source 
Two windows where settle() could read a live waiter count, mark the
task reported (suppressing the completion notice), and then watch that
waiter reject with 'wait aborted' delivering nothing — leaving the
owning session with no terminal notification at all:

- abort and settlement in the same tick, settle continuation ordered
  first: the waiter now un-counts itself SYNCHRONOUSLY inside onAbort
  (the finally decrement alone lands a microtask too late), so the
  settle path sees no live waiter and the notice fires;
- abort landing after settlement but before the wait's resolve
  microtask: the wait now resolves and DELIVERS the terminal snapshot
  it owes instead of rejecting (settlement suppressed the notice on
  this waiter's behalf).

Both windows pinned by deterministic tests that fail on the previous
implementation; wait()'s abort contract updated in JSDoc/README/RFC.
This commit is contained in:
Yichen Jiang
2026-07-09 23:08:26 +08:00
parent 032510a85a
commit 43d60286e3
4 changed files with 74 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/rfc/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md
View File

@@ -80,7 +80,7 @@ class TaskService extends Service { // ctx.tasks
}
```
`TaskSnapshot` is the read-only projection: id, kind, label, owner session, status, detail, started/finished timestamps, and the `reported` notice-suppression flag (below). `wait` resolves with the terminal snapshot, or with the still-`running` snapshot on timeout; aborting the wait cancels only the wait.
`TaskSnapshot` is the read-only projection: id, kind, label, owner session, status, detail, started/finished timestamps, and the `reported` notice-suppression flag (below). `wait` resolves with the terminal snapshot, or with the still-`running` snapshot on timeout; aborting the wait cancels only the wait — unless the task already settled, in which case the wait still delivers the terminal snapshot (settlement suppressed the completion notice on this live waiter's behalf, and an aborted waiter un-counts itself synchronously so a same-tick settlement never suppresses a notice nobody will deliver).
**Misconfiguration fails loud**: a deployment that loads a background-capable producer without any control surface would let the model start tasks it can never read or stop — the half-loaded failure mode the subagent RFC's first draft reshaped a whole plugin to avoid. The fence is `attachSurface()`: `dsh-tool-tasks` attaches (effect-scoped) on load, and `start()` throws `background tasks unavailable: no control surface is attached (load @deepseek-ai/dsh-tool-tasks)` when none is attached — the earliest self-contained moment, since concurrent plugin start makes a load-time check racy. The registry stays ignorant of tool names; a deployment with a custom (non-model) surface attaches its own.

2
packages/tasks/tasks/README.md
View File

@@ -8,7 +8,7 @@ The background task registry (`ctx.tasks`): a runtime-global, CONCRETE service (
- `get(id, caller?)` / `list(caller?)` — non-consuming snapshots; `list` returns only caller-owned plus unowned tasks (a global listing would leak foreign labels).
- `read(id, caller?): TaskRead` — stream kinds consume the per-task cursor (v1's single intended reader is the owning model — a non-consuming multi-reader surface would be a cursor/snapshot API extension, not a `read` change); final kinds read the terminal output idempotently.
- `kill(id, caller?, reason?)` — `'requested'` (live task: producer `cancel` runs first — a throw fails the kill loud and leaves the task untouched — then `stopping`) or `'already-terminal'`. Every successful kill marks the task `reported` (the killer saw the end → completion notice suppressed).
- `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. 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.
- `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.
- `onTaskDone(listener)` — exactly once per task with the terminal snapshot; 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.

34
packages/tasks/tasks/src/index.ts
View File

@@ -270,8 +270,12 @@ export class TaskService extends Service {
* terminal snapshot (marked {@link TaskSnapshot.reported} — the wait
* response reports the end, so the completion notice is suppressed), or
* with the still-live snapshot when the timeout expires first. An abort of
* `signal` rejects the WAIT only — the task keeps running. Throws for an
* unknown id, a task owned by another session, or a non-positive timeout.
* `signal` rejects the WAIT only (the task keeps running) — UNLESS the task
* has already settled: settlement saw this live waiter and suppressed the
* completion notice on its behalf, so the wait still resolves and delivers
* the terminal snapshot it owes (an abort must never leave a finished task
* both unreported and notice-suppressed). Throws for an unknown id, a task
* owned by another session, or a non-positive timeout.
* @param id - the task to wait for.
* @param timeoutMs - max wait in milliseconds (positive, finite; the surface caps it).
* @param caller - the waiting agent, checked against the task's owner.
@@ -286,7 +290,19 @@ export class TaskService extends Service {
}
if (!isTerminal(task.status)) {
if (signal?.aborted) throw new Error('wait aborted')
// `waiters` is the settle-path heuristic "someone WILL deliver the
// terminal snapshot, suppress the notice". An abort breaks that promise,
// so the un-count must happen SYNCHRONOUSLY inside onAbort — the
// `finally` decrement alone runs a microtask later, after a same-tick
// settlement could already have read the stale count and suppressed the
// notice for a waiter that then rejects and delivers nothing.
task.waiters += 1
let counted = true
const uncount = (): void => {
if (!counted) return
counted = false
task.waiters -= 1
}
try {
// The dsh-timeout deadline fits wait() exactly because both only
// NOTIFY: a wait timeout returns the live snapshot (the task keeps
@@ -296,8 +312,16 @@ export class TaskService extends Service {
using d = deadline(signal, timeoutMs, TASK_WAIT_TIMEOUT)
await new Promise<void>((resolve, reject) => {
const onAbort = (): void => {
if (timeoutOf(d.signal, TASK_WAIT_TIMEOUT) !== undefined) resolve()
else reject(new Error('wait aborted'))
if (timeoutOf(d.signal, TASK_WAIT_TIMEOUT) !== undefined) {
resolve()
} else if (isTerminal(task.status)) {
// Settlement already ran and suppressed the notice for this
// waiter — deliver the terminal snapshot instead of rejecting.
resolve()
} else {
uncount()
reject(new Error('wait aborted'))
}
}
d.signal.addEventListener('abort', onAbort, { once: true })
void task.settled.then(() => {
@@ -306,7 +330,7 @@ export class TaskService extends Service {
})
})
} finally {
task.waiters -= 1
uncount()
}
}
if (isTerminal(task.status)) task.reported = true

43
packages/tasks/tasks/tests/tasks.spec.ts
View File

@@ -268,6 +268,49 @@ describe('TaskService.wait', () => {
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 (review finding)', 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)
// Same synchronous tick, settlement QUEUED first: the settle continuation
// will run before the rejected wait's `finally`, so only the SYNCHRONOUS
// un-count inside onAbort keeps it from reading a stale waiter count,
// marking the task reported, and suppressing the completion notice for a
// wait that then delivers 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 runs synchronously inside settle — aborting HERE lands the
// abort after settlement marked this waiter reported (notice suppressed)
// but before the wait's own resolve microtask. Rejecting now would leave
// the finished task both unreported and notice-suppressed, so the wait
// must resolve and deliver instead.
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', () => {