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

feat(tasks): wake an idle owner when a background task completes

Browse Source 
Completion notices went through agent.inject(), which never reserves a
driver, so a task settling after its turn closed left the notice parked
until unrelated input woke the agent — while the same prompt told the
model not to poll for it.

An unreported completion now picks its lane from the owner's state: a
busy owner is injected as before, an idle owner is woken with
followup(). This adopts the delivery rule the subagent continuation
manager already ships. maxConsecutiveWakes bounds the self-exciting
chain and is reset by user-authored input; completionDelivery: quiet
restores the old lane for deterministic transcripts.

Teardown cancellation now claims the terminal report the way kill()
already does, so an owner being destroyed is never woken, and settle()
announces completion last so a reporter that opens a turn synchronously
sees a committed record.
This commit is contained in:
Yichen Jiang
2026-08-11 19:36:49 +08:00
parent 1d12ae62e7
commit d78b796a51
25 changed files with 531 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

157
packages/tasks/tool-tasks/tests/tool-tasks.spec.ts
View File

@@ -3,8 +3,9 @@ import { Context } from '@deepseek-ai/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 AgentRegistry, { emitAgentEvent } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import { SessionId } from '@deepseek-ai/dsh-session'
import { bindScopeParent, createScope, scopeOf } from '@deepseek-ai/dsh-scope'
import { TaskId } from '@deepseek-ai/dsh-tasks'
@@ -16,6 +17,7 @@ import { statusLine } from '@deepseek-ai/dsh-tool-tasks'
const testToolSignal = new AbortController().signal
const agentRegistryDisposers = new WeakMap<Agent, () => void>()
const agentScopeFibers = new WeakMap<Agent, { dispose: () => Promise<void> }>()
async function setup(config: ToolTasks.Config = {}) {
const ctx = new Context()
@@ -27,20 +29,31 @@ async function setup(config: ToolTasks.Config = {}) {
return { ctx, agentsFiber, toolsFiber }
}
/** The delivery surface a completion notice may reach on a fake owner. */
interface FakeDelivery {
inject?: (...args: unknown[]) => void
followup?: (...args: unknown[]) => void
/** Defaults to `running`, the lane that never wakes, so notice-content tests pin one lane. */
status?: 'idle' | 'running'
}
/**
* A fake agent with the shared agent/session identity, registered in
* `ctx.agents` with a dedicated lifecycle scope.
*/
function fakeAgent(ctx: Context, sessionId: string, inject: (...args: unknown[]) => void = () => {}): Agent {
function fakeAgent(ctx: Context, sessionId: string, delivery: FakeDelivery = {}): Agent {
const scopeFiber = ctx.plugin(() => {})
const id = SessionId(sessionId)
const agent = {
id,
ctx: scopeFiber.ctx,
inject,
inject: delivery.inject ?? (() => {}),
followup: delivery.followup ?? (() => {}),
status: delivery.status ?? 'running',
session: { id, header: { version: 0, id, createdAt: 0 } },
} as unknown as Agent
agentRegistryDisposers.set(agent, ctx.agents.register(agent))
agentScopeFibers.set(agent, scopeFiber)
return agent
}
@@ -50,6 +63,13 @@ function detachAgent(agent: Agent): void {
dispose()
}
/** Dispose the agent's own lifecycle scope, which is what drains its owned tasks. */
async function disposeAgentScope(agent: Agent): Promise<void> {
const fiber = agentScopeFibers.get(agent)
if (fiber === undefined) throw new Error(`missing scope fiber for agent "${agent.id}"`)
await fiber.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
@@ -81,6 +101,16 @@ function text(result: { content: { type: string; text?: string }[] }): string {
const tick = () => new Promise<void>(r => setTimeout(r, 0))
/** Start and settle `count` owned tasks one at a time, letting each notice land. */
async function settleTasks(ctx: Context, owner: Agent, count: number): Promise<void> {
for (let i = 0; i < count; i += 1) {
const p = producer({ owner })
ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
await tick()
}
}
describe('tool-tasks setup', () => {
it('attaches the task controller on load and detaches it with the fiber', async () => {
const { ctx, toolsFiber } = await setup()
@@ -494,11 +524,112 @@ describe('completion notices across scoped mounts', () => {
})
})
describe('completion notice delivery', () => {
it('opens a turn on an idle owner when a task settles', async () => {
const { ctx } = await setup()
const inject = vi.fn()
const followup = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' })
const p = producer({ owner, label: 'pnpm test' })
ctx.tasks.start(p.spec)
p.settle({ status: 'completed', detail: 'exit code: 0' })
await tick()
expect(followup).toHaveBeenCalledTimes(1)
expect(inject).not.toHaveBeenCalled()
})
it('never wakes an idle owner under quiet delivery', async () => {
const { ctx } = await setup({ completionDelivery: 'quiet' })
const inject = vi.fn()
const followup = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' })
const p = producer({ owner })
ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
await tick()
expect(inject).toHaveBeenCalledTimes(1)
expect(followup).not.toHaveBeenCalled()
})
it('degrades to injection once the consecutive wake budget is spent', async () => {
const { ctx } = await setup({ maxConsecutiveWakes: 2 })
const inject = vi.fn()
const followup = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' })
await settleTasks(ctx, owner, 3)
// A woken turn that starts another task is the self-exciting case: the
// budget stops the chain, and the notice still reaches the inbox.
expect(followup).toHaveBeenCalledTimes(2)
expect(inject).toHaveBeenCalledTimes(1)
})
it('restores the wake budget when the owner claims a user message', async () => {
const { ctx } = await setup({ maxConsecutiveWakes: 1 })
const inject = vi.fn()
const followup = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' })
await settleTasks(ctx, owner, 2)
expect(followup).toHaveBeenCalledTimes(1)
emitAgentEvent(ctx, owner, 'agent/inbox/claimed', {
message: createUserMessage({ content: [{ type: 'text', text: 'carry on' }], source: { kind: 'user' } }),
turn: 1,
})
await settleTasks(ctx, owner, 1)
expect(followup).toHaveBeenCalledTimes(2)
})
it('neither wakes nor injects into an owner its own teardown is draining', async () => {
const { ctx } = await setup()
const inject = vi.fn()
const followup = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' })
let settle!: (outcome: TaskOutcome) => void
ctx.tasks.start({
kind: 'bash',
label: 'sleep 60',
owner,
run: () => ({
cancel() { settle({ status: 'killed' }) },
done: new Promise<TaskOutcome>((res) => { settle = res }),
}),
})
// Disposal cancels and settles the owned task. Waking here would spend a
// model request on an agent the host is destroying, once per tree layer.
await disposeAgentScope(owner)
await tick()
expect(followup).not.toHaveBeenCalled()
expect(inject).not.toHaveBeenCalled()
})
it('keeps the budget spent when the owner only claims plugin notices', async () => {
const { ctx } = await setup({ maxConsecutiveWakes: 1 })
const followup = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', { followup, status: 'idle' })
await settleTasks(ctx, owner, 1)
emitAgentEvent(ctx, owner, 'agent/inbox/claimed', {
message: createUserMessage({
content: [{ type: 'text', text: 'background task bash-1 finished' }],
source: { kind: 'plugin', plugin: 'tool-tasks', form: 'notice', summary: 'bash' },
}),
turn: 1,
})
await settleTasks(ctx, owner, 1)
expect(followup).toHaveBeenCalledTimes(1)
})
})
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 owner = fakeAgent(ctx, 'sess-1', { inject })
const p = producer({ owner, label: 'pnpm test' })
ctx.tasks.start(p.spec)
@@ -521,7 +652,7 @@ describe('completion notices', () => {
it('preserves task ids and collection guidance in bounded completion notices', async () => {
const { ctx } = await setup()
const inject = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', inject)
const owner = fakeAgent(ctx, 'sess-1', { inject })
const first = producer({
owner,
kind: 'subagent',
@@ -574,7 +705,7 @@ describe('completion notices', () => {
prior.settle({ status: 'completed' })
}
const inject = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', inject)
const owner = fakeAgent(ctx, 'sess-1', { inject })
const target = producer({
owner,
kind: 'pty-send',
@@ -595,7 +726,7 @@ describe('completion notices', () => {
it('reserves the collection-action tail when a producer supplies a smaller budget', async () => {
const { ctx } = await setup()
const inject = vi.fn()
const owner = fakeAgent(ctx, 'sess-1', inject)
const owner = fakeAgent(ctx, 'sess-1', { inject })
const tiny = producer({ owner, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 8 })
const short = producer({ owner, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 32 })
ctx.tasks.start(tiny.spec)
@@ -616,7 +747,7 @@ describe('completion notices', () => {
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 owner = fakeAgent(ctx, 'sess-1', { inject })
const p = producer({ owner })
ctx.tasks.start(p.spec)
@@ -629,7 +760,7 @@ describe('completion notices', () => {
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 owner = fakeAgent(ctx, 'sess-1', { inject })
const p = producer({ owner, kind: 'subagent' })
ctx.tasks.start(p.spec)
@@ -654,13 +785,13 @@ describe('completion notices', () => {
// terminal state, so the notice lands in the old owner's (detached)
// session instead of throwing or re-routing.
const oldInject = vi.fn()
const oldOwner = fakeAgent(ctx, 'shared', oldInject)
const oldOwner = fakeAgent(ctx, 'shared', { inject: oldInject })
const p = producer({ owner: oldOwner })
ctx.tasks.start(p.spec)
detachAgent(oldOwner)
const replacementInject = vi.fn()
fakeAgent(ctx, 'shared', replacementInject)
fakeAgent(ctx, 'shared', { inject: replacementInject })
p.settle({ status: 'completed' })
await tick()
@@ -671,7 +802,7 @@ describe('completion notices', () => {
it('surfaces an inject failure through listener containment (a real bug must be visible)', 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 owner = fakeAgent(ctx, 'sess-1', { inject: () => { throw new Error('unexpected inject bug') } })
const p = producer({ owner })
ctx.tasks.start(p.spec)
p.settle({ status: 'completed' })
@@ -684,7 +815,7 @@ describe('completion notices', () => {
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)
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.