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

Merge origin/master into worktree/archive-agent-notes-20260726

Browse Source
This commit is contained in:
Tianyi Cui
2026-07-26 23:11:33 +08:00
parent 0dca9684d5 4dfbd2cbd7
commit 32ae16fa59
58 changed files with 918 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
packages/bash/tool-bash/package.json
View File

@@ -60,6 +60,7 @@
"@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tasks": "workspace:^",
"@deepseek-ai/dsh-tasks-local": "workspace:^",
"@deepseek-ai/dsh-tool-tasks": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"@deepseek-ai/dsh-user-approval": "workspace:^",

2
packages/bash/tool-bash/src/index.ts
View File

@@ -535,7 +535,7 @@ export function apply(ctx: Context, config: Config = {}): void {
if (tasks === undefined) {
throw new Error('background tasks unavailable: load @deepseek-ai/dsh-tasks and @deepseek-ai/dsh-tool-tasks')
}
// The caller owns cancellation until TaskService commits detached ownership.
// The caller owns cancellation until ctx.tasks commits detached ownership.
if (exec.signal.aborted) {
const error = new HarnessError('tool call aborted', TOOL_ABORTED)
error.name = 'AbortError'

6
packages/bash/tool-bash/tests/integration.spec.ts
View File

@@ -8,7 +8,7 @@ import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
import type { Agent } from '@deepseek-ai/dsh-agent'
import AgentLoop from '@deepseek-ai/dsh-agent-loop'
import { mountAgentLoopTestDependencies } from '@deepseek-ai/dsh-agent-loop-testkit'
import TaskService from '@deepseek-ai/dsh-tasks'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
import * as ToolTasks from '@deepseek-ai/dsh-tool-tasks'
import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local'
import * as ToolBash from '@deepseek-ai/dsh-tool-bash'
@@ -27,7 +27,7 @@ async function harness(adapter: MockAdapter, sessionRoot?: string, dshHome?: str
await ctx.plugin(SessionPersistenceJsonl, { root: sessionRoot, compression: 'none' })
}
await ctx.plugin(AgentLoop, { agents: [] })
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks)
await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000 })
await ctx.plugin(ToolBash, dshHome === undefined ? {} : { dshHome })
@@ -169,7 +169,7 @@ describe('bash tool through the agent loop', () => {
})
it('background: start ack → completion notice as user/message → task_output collects it', async () => {
// The task id is deterministic (a fresh TaskService counts per kind from 1),
// The task id is deterministic (a fresh LocalTaskService counts per kind from 1),
// so the script can name `bash-1` without threading a generated id.
const adapter = new MockAdapter([
toolCallResponse('call-1', 'bash', { command: 'echo bg-ok', description: 'test command', run_in_background: true }),

14
packages/bash/tool-bash/tests/tools.spec.ts
View File

@@ -12,7 +12,7 @@ import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
import TaskService from '@deepseek-ai/dsh-tasks'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
import * as ToolTasks from '@deepseek-ai/dsh-tool-tasks'
import ApprovalService from '@deepseek-ai/dsh-user-approval'
import type { ApprovalOutcome } from '@deepseek-ai/dsh-user-approval'
@@ -44,7 +44,7 @@ async function setupWithTasks() {
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks)
await ctx.plugin(LocalBashExecutor, { timeoutMs: 10_000, graceMs: 200 })
;(ctx.bash as LocalBashExecutor).internals = { spillDir }
@@ -180,7 +180,7 @@ async function setupSandboxed(withApproval = false) {
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks)
await ctx.plugin(SandboxPolicyService, {})
await ctx.plugin(RecordingSandboxExecutor)
@@ -472,7 +472,7 @@ describe('background execution through the task runtime', () => {
})
it('fails loud when the task runtime is not loaded', async () => {
const ctx = await setup() // no TaskService / ToolTasks
const ctx = await setup() // no LocalTaskService / ToolTasks
const result = await call(ctx, 'bash', { command: 'sleep 60', description: 'test command', run_in_background: true })
expect(result.isError).toBe(true)
expect(text(result)).toContain('background tasks unavailable: load @deepseek-ai/dsh-tasks and @deepseek-ai/dsh-tool-tasks')
@@ -483,7 +483,7 @@ describe('background execution through the task runtime', () => {
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks)
await ctx.plugin(CountingStartExecutor)
await ctx.plugin(ToolBash)
@@ -511,7 +511,7 @@ describe('background execution through the task runtime', () => {
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(CountingStartExecutor)
await ctx.plugin(ToolBash)
@@ -1073,7 +1073,7 @@ describe('the model-facing bash tool builds its request from named args only (no
await ctx.plugin(SessionStore)
await ctx.plugin(SessionPersistenceJsonl, { root: join(spillDir, 'jsonl') })
}
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks)
await ctx.plugin(RecordingBashExecutor)
await ctx.plugin(ToolBash, { dshHome: recordingDshHome })

20
packages/cordis/tool-cordis/src/api-catalog.ts
View File

@@ -768,38 +768,38 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
},
{
key: 'tasks',
summary: 'The `tasks` service: the runtime-global background task registry.',
summary: 'Abstract background task registry.',
methods: [
{
signature: 'start(spec: TaskStart): TaskId',
signature: 'abstract start(spec: TaskStart): TaskId',
jsDoc: '/**\n * Preflight access, validation, and owner cleanup before starting and\n * atomically registering work. A throwing starter leaves nothing registered;\n * after it returns, registration cannot fail. Settlement records the outcome,\n * notifies listeners, and releases waiters.\n * @param spec - task identity, owner, and synchronous starter.\n * @returns the registry-issued `<kind>-N` id.\n */',
},
{
signature: 'list(caller?: Agent): TaskSnapshot[]',
signature: 'abstract list(caller?: Agent): TaskSnapshot[]',
jsDoc: '/**\n * List caller-owned and unowned tasks in registration order without exposing\n * another session\'s labels.\n * @param caller - reading agent; a non-agent caller sees only unowned tasks.\n * @returns fresh snapshots.\n */',
},
{
signature: 'get(id: TaskId, caller?: Agent): TaskSnapshot',
signature: 'abstract get(id: TaskId, caller?: Agent): TaskSnapshot',
jsDoc: '/**\n * Return a non-consuming snapshot without changing its read cursor or notice\n * state. Throws for an unknown or foreign task.\n * @param id - task to look up.\n * @param caller - reading agent checked against the owner.\n * @returns a fresh snapshot.\n */',
},
{
signature: 'read(id: TaskId, caller?: Agent): TaskRead',
signature: 'abstract read(id: TaskId, caller?: Agent): TaskRead',
jsDoc: '/**\n * Read the next stream delta, or the idempotent final output after settlement.\n * A terminal read marks the task reported. Throws for an unknown or foreign\n * task.\n * @param id - task to read.\n * @param caller - reading agent checked against the owner.\n * @returns output text and the post-read snapshot.\n */',
},
{
signature: 'kill(id: TaskId, caller?: Agent, reason?: string): \'requested\' | \'already-finished\'',
signature: 'abstract kill(id: TaskId, caller?: Agent, reason?: string): \'requested\' | \'already-finished\'',
jsDoc: '/**\n * Request cancellation, then mark the task stopping and reported. A producer\n * throw propagates without changing task state. Throws for an unknown or\n * foreign task.\n * @param id - task to cancel.\n * @param caller - killing agent checked against the owner.\n * @param reason - logged reason forwarded to the producer.\n * @returns `requested` for live work, otherwise `already-finished`.\n */',
},
{
signature: 'async wait(id: TaskId, timeoutMs: number, caller?: Agent, signal?: AbortSignal): Promise<TaskSnapshot>',
jsDoc: '/**\n * Wait for settlement or timeout without cancelling the task. Caller abort\n * rejects only while the task is live; after settlement it returns the\n * terminal snapshot so a notice suppressed for this waiter is still delivered.\n * Timed-out and aborted waits detach their resolvers. Throws for invalid,\n * unknown, or foreign input.\n * @param id - task to wait for.\n * @param timeoutMs - positive finite wait bound in milliseconds.\n * @param caller - waiting agent checked against the owner.\n * @param signal - optional cancellation of the wait itself.\n * @returns snapshot at settlement or timeout.\n */',
signature: 'abstract wait(id: TaskId, timeoutMs: number, caller?: Agent, signal?: AbortSignal): Promise<TaskSnapshot>',
jsDoc: '/**\n * Wait for settlement or timeout without cancelling the task. Caller abort\n * rejects only while the task is live; after settlement the terminal\n * snapshot wins so a notice suppressed for this waiter is still delivered.\n * Throws for invalid, unknown, or foreign input.\n * @param id - task to wait for.\n * @param timeoutMs - positive finite wait bound in milliseconds.\n * @param caller - waiting agent checked against the owner.\n * @param signal - optional cancellation of the wait itself.\n * @returns snapshot at settlement or timeout.\n */',
},
{
signature: 'onTaskDone(listener: TaskDoneListener): () => void',
signature: 'abstract onTaskDone(listener: TaskDoneListener): () => void',
jsDoc: '/**\n * Register an effect-scoped completion listener. Each listener is contained;\n * returned promises are observed but not awaited. No listener runs after\n * service disposal.\n * @param listener - receives each terminal snapshot and its exact owner.\n * @returns disposer that unregisters the listener.\n */',
},
{
signature: 'attachSurface(name: string): () => void',
signature: 'abstract attachSurface(name: string): () => void',
jsDoc: '/**\n * Attach an effect-scoped surface that can read and stop tasks. {@link start}\n * refuses work while none is attached.\n * @param name - diagnostic label; duplicate names remain independent.\n * @returns disposer that detaches this surface.\n */',
},
],

4
packages/examples/agent-spine-demo/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write
README.md: 736de2ea01e1524854c57f91d128b82a9fe0c9e8
README.zh.md: 4ffe47ba82539d12c9b74b1690392d58d21a24b1
README.md: 32874bf2839c194572ddde8c4ed007297f763ccc
README.zh.md: 57a06a00203b8e67f2f33c87d7450d1a0789d7e6

2
packages/examples/agent-spine-demo/README.md
View File

@@ -24,7 +24,7 @@ Read this package for the whole plugin tree and its composition order.
@deepseek-ai/dsh-tool-goal optional model-facing goal controls
@deepseek-ai/dsh-goal-session optional same-session goal-round driver
@deepseek-ai/dsh-llm-retry bounded transient request retry policy
@deepseek-ai/dsh-tasks generic background-task registry
@deepseek-ai/dsh-tasks-local generic background-task registry
@deepseek-ai/dsh-invariants configurable invariant registry service
@deepseek-ai/dsh-session/invariant
@deepseek-ai/dsh-agent/invariant

2
packages/examples/agent-spine-demo/README.zh.md
View File

@@ -24,7 +24,7 @@
@deepseek-ai/dsh-tool-goal optional model-facing goal controls
@deepseek-ai/dsh-goal-session optional same-session goal-round driver
@deepseek-ai/dsh-llm-retry bounded transient request retry policy
@deepseek-ai/dsh-tasks generic background-task registry
@deepseek-ai/dsh-tasks-local generic background-task registry
@deepseek-ai/dsh-invariants configurable invariant registry service
@deepseek-ai/dsh-session/invariant
@deepseek-ai/dsh-agent/invariant

3
packages/examples/agent-spine-demo/package.json
View File

@@ -42,7 +42,7 @@
"@deepseek-ai/dsh-skill": "^0.0.1",
"@deepseek-ai/dsh-skill-local": "^0.0.1",
"@deepseek-ai/dsh-system-prompt": "^0.0.1",
"@deepseek-ai/dsh-tasks": "^0.0.1",
"@deepseek-ai/dsh-tasks-local": "^0.0.1",
"@deepseek-ai/dsh-tool-bash": "^0.0.1",
"@deepseek-ai/dsh-tool-goal": "^0.0.1",
"@deepseek-ai/dsh-tool-skill": "^0.0.1",
@@ -74,6 +74,7 @@
"@deepseek-ai/dsh-skill-local": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tasks": "workspace:^",
"@deepseek-ai/dsh-tasks-local": "workspace:^",
"@deepseek-ai/dsh-tool-bash": "workspace:^",
"@deepseek-ai/dsh-tool-fs": "workspace:^",
"@deepseek-ai/dsh-tool-goal": "workspace:^",

4
packages/examples/agent-spine-demo/src/index.ts
View File

@@ -22,7 +22,7 @@ import AgentRegistry from '@deepseek-ai/dsh-agent'
import GoalService, { type Config as GoalDomainConfig } from '@deepseek-ai/dsh-goal'
import * as goalSession from '@deepseek-ai/dsh-goal-session'
import * as toolGoal from '@deepseek-ai/dsh-tool-goal'
import TaskService from '@deepseek-ai/dsh-tasks'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
import InvariantService, { type Config as InvariantConfig } from '@deepseek-ai/dsh-invariants'
import * as sessionInvariant from '@deepseek-ai/dsh-session/invariant'
import * as agentInvariant from '@deepseek-ai/dsh-agent/invariant'
@@ -223,7 +223,7 @@ export function apply(ctx: Context, config: Config): void {
ctx.plugin(toolGoal, config.goals.tool ?? {})
ctx.plugin(goalSession)
}
ctx.plugin(TaskService)
ctx.plugin(LocalTaskService)
ctx.plugin(InvariantService, config.invariants ?? {})
ctx.plugin(sessionInvariant)
ctx.plugin(agentInvariant)

3
packages/examples/agent-spine-demo/tsconfig.json
View File

@@ -74,6 +74,9 @@
{
"path": "../../tasks/tasks"
},
{
"path": "../../tasks/tasks-local"
},
{
"path": "../../tasks/tool-tasks"
}

1
packages/pty/tool-pty/package.json
View File

@@ -54,6 +54,7 @@
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tasks": "workspace:^",
"@deepseek-ai/dsh-tasks-local": "workspace:^",
"@deepseek-ai/dsh-tool-tasks": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"cordis": "^4.0.0-rc.7"

4
packages/pty/tool-pty/tests/tools.spec.ts
View File

@@ -9,7 +9,7 @@ import ToolRegistry, { renderToolsSdk } from '@deepseek-ai/dsh-tools'
import type { ToolSdkSchema } from '@deepseek-ai/dsh-tools/src/ts-types.ts'
import PtyService, { PtySessionId } from '@deepseek-ai/dsh-pty'
import type { PtyBackend, PtyBackendSession, PtySendOperation, PtySendRequest, PtySessionStatus, PtySignal } from '@deepseek-ai/dsh-pty'
import TaskService from '@deepseek-ai/dsh-tasks'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
import * as ToolTasks from '@deepseek-ai/dsh-tool-tasks'
import * as ToolPty from '@deepseek-ai/dsh-tool-pty'
@@ -106,7 +106,7 @@ async function setupBase(tasks: boolean) {
const stub = stubBackend()
ctx.pty.registerBackend(stub.backend)
if (tasks) {
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks)
}
return { ctx, stub, agent: fakeAgent(ctx, tasks ? 'with-tasks' : 'foreground') }

1
packages/subagent/tool-subagent/package.json
View File

@@ -46,6 +46,7 @@
"@deepseek-ai/dsh-subagent": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tasks": "workspace:^",
"@deepseek-ai/dsh-tasks-local": "workspace:^",
"@deepseek-ai/dsh-tool-tasks": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"cordis": "^4.0.0-rc.7"

6
packages/subagent/tool-subagent/tests/tool-subagent.spec.ts
View File

@@ -8,7 +8,7 @@ import { type Agent } from '@deepseek-ai/dsh-agent'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import SubagentService from '@deepseek-ai/dsh-subagent'
import type { SubagentStartRequest } from '@deepseek-ai/dsh-subagent'
import TaskService from '@deepseek-ai/dsh-tasks'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
import * as ToolTasks from '@deepseek-ai/dsh-tool-tasks'
import * as mock from './scripted-provider.ts'
import * as tool from '../src/index.ts'
@@ -641,7 +641,7 @@ describe('dsh-tool-subagent background mode', () => {
async function backgroundSetup(toolConfig: tool.Config, mockConfig: Partial<mock.Config> = {}) {
const ctx = await setup(toolConfig, mockConfig)
await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await ctx.plugin(ToolTasks, {})
return ctx
}
@@ -868,7 +868,7 @@ describe('background preflight failure (no orphaned child, by construction)', ()
// With no control surface, task preflight fails before the provider can spawn.
const ctx = await setup({ provider: 'mock' })
await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
const scopeFiber = ctx.plugin(() => {})
const id = SessionId('sess-p')
const parent = {

4
packages/tasks/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write
README.md: f1c224345c94a833c44cbafb635be7617e8c42bf
README.zh.md: 610a84a1506b4bb780297322f7827e6f04533bc1
README.md: 9bafe5633bb7e57a5404ffb41fad04b621832b6d
README.zh.md: 73c87a2c95ccebf70558a2051149eca4ba41f60e

5
packages/tasks/README.md
View File

@@ -2,11 +2,12 @@
English | [中文](README.zh.md)
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 Agent Note](../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md).
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 Agent Note](../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md) and the [task-registry seam Agent Note](../../.agents/notes/implemented/architecture/2026-07-26-task-registry-seam.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 |
| [`tasks`](tasks/README.md) (`@deepseek-ai/dsh-tasks`) | `ctx.tasks` | The registry seam: branded `<kind>-N` ids, the owner-fenced read/kill/wait/list contract, snapshot vocabulary, the `attachSurface` misconfiguration fence, and the snapshot invariant companion |
| [`tasks-local`](tasks-local/README.md) (`@deepseek-ai/dsh-tasks-local`) | — | The process-local registry implementation: in-memory records, first-wins settlement bookkeeping, and the awaited owner-cleanup and teardown paths |
| [`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`.

5
packages/tasks/README.zh.md
View File

@@ -2,11 +2,12 @@
[English](README.md) | 中文
后台 task id、拥有者隔离、读取、取消、等待和完成通知的共用归属位置。Bash、subagent 及未来的长时间运行工具共用一套面向模型的协议。参见[后台任务运行时 Agent Note（agent 决策记录）](../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md)。
后台 task id、拥有者隔离、读取、取消、等待和完成通知的共用归属位置。Bash、subagent 及未来的长时间运行工具共用一套面向模型的协议。参见[后台任务运行时 Agent Note（agent 决策记录）](../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md)和[任务注册表 seam Agent Note](../../.agents/notes/implemented/architecture/2026-07-26-task-registry-seam.md)。
| 包（package） | ctx 键 | 角色 |
|---|---|---|
| [`tasks`](tasks/README.md)（`@deepseek-ai/dsh-tasks`） | `ctx.tasks` | 注册表服务：品牌化 `<kind>-N` id、按拥有者设防的 read／kill／wait／list、结算记账、等待完成的拥有者清理路径，以及防止 `attachSurface` 配置错误的防线 |
| [`tasks`](tasks/README.md)（`@deepseek-ai/dsh-tasks`） | `ctx.tasks` | 注册表 seam：品牌化 `<kind>-N` id、按拥有者设防的 read／kill／wait／list 契约、快照词汇、防止 `attachSurface` 配置错误的防线，以及快照不变式配套插件 |
| [`tasks-local`](tasks-local/README.md)（`@deepseek-ai/dsh-tasks-local`） | 无 | 进程局部的注册表实现：内存记录、首次结果优先的结算簿记，以及等待完成的拥有者清理与拆卸路径 |
| [`tool-tasks`](tool-tasks/README.md)（`@deepseek-ai/dsh-tool-tasks`） | 无 | 面向模型的控制接口：`task_output`、`task_list`、`task_kill`、完成通知注入和后台工作习惯提示词段落 |
注册表拥有跨生产方或接口重载的状态；工具包拥有呈现。生产方通过 `ctx.tasks.start` 注册执行钩子，并自行决定其配置是否公开 `run_in_background`。

6
packages/tasks/tasks-local/README.i18n.yaml Normal file
View File

@@ -0,0 +1,6 @@
# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write
README.md: 23ca6fca61ccb59c855e5d6da6b0a2e23e7cb632
README.zh.md: c5553a76690278f5b6d5ec40a55d213ef7e1e2d9

26
packages/tasks/tasks-local/README.md Normal file
View File

@@ -0,0 +1,26 @@
# @deepseek-ai/dsh-tasks-local
English | [中文](README.zh.md)
Process-local implementation of the [`@deepseek-ai/dsh-tasks`](../tasks/README.md) registry seam: `LocalTaskService` keeps every record in memory, issues per-kind `<kind>-N` ids, and hands out fresh snapshots, never live state. It has no config; load it as a plugin and it registers as `ctx.tasks`.
## 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.
Settlement is first-wins: the earliest terminal outcome — producer settlement, a rejected `done` contained as `failed`, or a teardown force-failure — records once, notifies listeners once with per-listener containment, and releases waiters. Pending waits mark the task reported before listeners run so completion surfaces do not duplicate notices.
## Model Experience
Indirectly, through producer plugins and [`dsh-tool-tasks`](../tool-tasks/README.md), which render task ids, output, status, cancellation, and completion notices.
#### KV Cache effect
No direct invalidation; the named consumer owns any request-prefix changes.
## Known Limitations and Deferred Work
- **Tasks are process-local** — records die with the harness process; durable or cross-restart execution needs a separate backend implementing the seam.
- **A silently ineffective cancel can stall teardown** — only an explicit throw can be force-failed safely.

26
packages/tasks/tasks-local/README.zh.md Normal file
View File

@@ -0,0 +1,26 @@
# @deepseek-ai/dsh-tasks-local
[English](README.md) | 中文
[`@deepseek-ai/dsh-tasks`](../tasks/README.md) 注册表 seam 的进程局部实现：`LocalTaskService` 把每条记录保存在内存中，按 kind 签发 `<kind>-N` id，并且只交出全新快照，从不交出实时状态。它没有配置；作为插件加载后即注册为 `ctx.tasks`。
## 生命周期
任务属于其 owner 和后端，而不是生产方工具 fiber，因此重载生产方或表层不会停止任务。某个 owner 的第一个任务会把一个受等待的 effect 附加到精确的 `Agent` scope。owner 释放会取消该对象的任务，等待生产方完全停稳，并移除其快照；复用 agent 或 Session id 无法重定向旧清理。
服务释放会关闭监听器、取消所有存活任务、等待其记录，并从仍存活的 owner scope 分离 effect。如果拆卸取消抛出异常，服务会强制把记录标为失败，并警告工作可能遗留，而不会死锁。取消已返回但始终不终止 `done` 时，系统无法将其与缓慢停止区分开，拆卸可能因此停滞。
结算遵循首次结果优先：最早出现的终止结果（生产方结算、被隔离为 `failed` 的 `done` 拒绝，或拆卸强制失败）只记录一次，只通知监听器一次并对每个监听器单独隔离故障，然后释放等待方。挂起的等待会在监听器运行前把任务标记为已报告，因此呈现完成情况的表层不会重复发出通知。
## 模型体验
通过生产方插件和 [`dsh-tool-tasks`](../tool-tasks/README.md) 间接影响；它们会渲染 task id、输出、状态、取消和完成通知。
#### KV Cache 影响
不会直接失效；请求前缀变更由命名消费方负责。
## 已知限制与暂缓事项
- **任务只存在于进程本地**：记录随 harness 进程一起消亡；持久或跨重启执行需要一个单独实现该 seam 的后端。
- **静默无效的取消可能使拆卸停滞**：只有显式抛出异常才能安全地强制标为失败。

45
packages/tasks/tasks-local/package.json Normal file
View File

@@ -0,0 +1,45 @@
{
"name": "@deepseek-ai/dsh-tasks-local",
"description": "Process-local implementation of the DeepSeek Harness background task registry seam",
"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"
},
"./invariant": {
"types": "./lib/types/invariant.d.ts",
"default": "./lib/invariant.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/invariant.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-invariants": "^0.0.1",
"@deepseek-ai/dsh-tasks": "^0.0.1",
"@deepseek-ai/dsh-timeout": "^0.0.1",
"cordis": "^4.0.0-rc.7"
},
"devDependencies": {
"@deepseek-ai/dsh-agent": "workspace:^",
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-tasks": "workspace:^",
"@deepseek-ai/dsh-timeout": "workspace:^",
"cordis": "^4.0.0-rc.7"
}
}

365
packages/tasks/tasks-local/src/index.ts Normal file
View File

@@ -0,0 +1,365 @@
/**
* Process-local implementation of the background task registry seam
* (`ctx.tasks`). It keeps every record in memory and hands out fresh
* snapshots, never live state.
*
* 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-local
*/
import { Context } from 'cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
import { TaskService, TaskId } from '@deepseek-ai/dsh-tasks'
import type { TaskDoneListener, TaskKind, TaskOutcome, TaskRead, TaskSnapshot, TaskStart, TaskStatus } from '@deepseek-ai/dsh-tasks'
/** 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 LocalTaskService.snapshot}). */
interface TrackedTask {
id: TaskId
kind: TaskKind
label: string
outputLimitBytes: number | undefined
/** 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 in-memory `tasks` registry. See the seam contract in
* `@deepseek-ai/dsh-tasks` for the ownership, isolation, and lifecycle
* semantics this implementation honors.
*/
export class LocalTaskService extends TaskService {
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)
this.selfCtx = ctx
ctx.effect(() => () => this.disposeAll(), 'tasks teardown')
}
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.outputLimitBytes !== undefined
&& (!Number.isSafeInteger(spec.outputLimitBytes) || spec.outputLimitBytes <= 0)) {
throw new Error(`invalid outputLimitBytes: expected a positive safe integer, got ${JSON.stringify(spec.outputLimitBytes)}`)
}
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,
outputLimitBytes: spec.outputLimitBytes,
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?: Agent): TaskSnapshot[] {
const session = caller?.id
return [...this.store.values()]
.filter(task => task.owner === undefined || task.owner.id === session)
.map(task => this.snapshot(task))
}
get(id: TaskId, caller?: Agent): TaskSnapshot {
const task = this.expect(id)
this.assertAccess(task, caller)
return this.snapshot(task)
}
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) }
}
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'
}
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)
}
onTaskDone(listener: TaskDoneListener): () => void {
const dispose = this.ctx.effect(() => {
this.listeners.add(listener)
return () => this.listeners.delete(listener)
}, 'tasks.onTaskDone()')
return () => void dispose()
}
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,
...task.outputLimitBytes !== undefined ? { outputLimitBytes: task.outputLimitBytes } : {},
...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 LocalTaskService

30
packages/tasks/tasks-local/src/invariant.ts Normal file
View File

@@ -0,0 +1,30 @@
/**
* Package-owned invariant companion for `@deepseek-ai/dsh-tasks-local`.
* @module @deepseek-ai/dsh-tasks-local/invariant
*/
/* jscpd:ignore-start */
import type { Context } from 'cordis'
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
const PACKAGE_NAME = '@deepseek-ai/dsh-tasks-local'
/** Cordis companion plugin name. */
export const name = 'tasks-local-invariant'
/** Service required before the companion can reserve package ownership. */
export const inject = ['invariants']
/**
* No runtime invariant: the seam companion in `@deepseek-ai/dsh-tasks` already
* validates every registry snapshot this implementation publishes.
*/
const install: InvariantInstaller = () => {}
/**
* Register this package's invariant companion.
* @param ctx - Cordis context carrying the invariant service.
* @returns the installed registration's disposer after setup succeeds.
*/
export const apply = (ctx: Context): Promise<() => void> =>
Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
/* jscpd:ignore-end */

33
packages/tasks/tasks/tests/tasks.spec.ts → packages/tasks/tasks-local/tests/tasks.spec.ts
View File

@@ -3,8 +3,9 @@ import { Context } from 'cordis'
import { Session, SessionId } from '@deepseek-ai/dsh-session'
import AgentRegistry, { AgentMessageId } from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import TaskService, { TaskId } from '@deepseek-ai/dsh-tasks'
import { TaskId } from '@deepseek-ai/dsh-tasks'
import type { TaskHooks, TaskKind, TaskOutcome, TaskSnapshot, TaskStart } from '@deepseek-ai/dsh-tasks'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
declare module '@deepseek-ai/dsh-tasks' {
interface TaskKindMap {
@@ -65,7 +66,7 @@ function producer(overrides: Partial<Omit<TaskStart, 'run'> & TaskHooks> = {}) {
async function harness() {
const ctx = new Context()
await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
ctx.tasks.attachSurface('test-surface')
return ctx
}
@@ -81,14 +82,14 @@ function waitResolverCount(ctx: Context, id: TaskId): number {
return task.waitResolvers.size
}
describe('TaskService.start', () => {
describe('LocalTaskService.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)
await ctx.plugin(LocalTaskService)
expect(() => ctx.tasks.start(producer().spec))
.toThrow('background tasks unavailable: no control surface is attached (load @deepseek-ai/dsh-tool-tasks)')
})
@@ -109,7 +110,7 @@ describe('TaskService.start', () => {
})
})
describe('TaskService reads and settlement', () => {
describe('LocalTaskService reads and settlement', () => {
it('stream kinds read a consuming delta; terminal reads mark reported', async () => {
const ctx = await harness()
const chunks = ['first', '', 'rest']
@@ -229,7 +230,7 @@ describe('TaskService reads and settlement', () => {
})
})
describe('TaskService.kill', () => {
describe('LocalTaskService.kill', () => {
it('cancels a live task with the forwarded reason and suppresses the notice', async () => {
const ctx = await harness()
const seen: TaskSnapshot[] = []
@@ -284,7 +285,7 @@ describe('TaskService.kill', () => {
})
})
describe('TaskService.wait', () => {
describe('LocalTaskService.wait', () => {
it('resolves with the terminal snapshot when the task settles, marked reported', async () => {
const ctx = await harness()
const seen: TaskSnapshot[] = []
@@ -394,7 +395,7 @@ describe('TaskService.wait', () => {
})
})
describe('TaskService owner isolation', () => {
describe('LocalTaskService 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')
@@ -433,7 +434,7 @@ describe('TaskService owner isolation', () => {
it('rejects an owned registration when no agent registry is mounted', async () => {
const ctx = new Context()
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
ctx.tasks.attachSurface('test-surface')
expect(() => ctx.tasks.start(producer({ owner: stubAgent(ctx, 'a') }).spec))
.toThrow('background task ownership requires the agent registry')
@@ -498,7 +499,7 @@ describe('TaskService owner isolation', () => {
})
})
describe('TaskService owner cleanup', () => {
describe('LocalTaskService owner cleanup', () => {
it('drains the owner: cancels live tasks, awaits settlement, drops snapshots', async () => {
const ctx = await harness()
const owner = stubAgent(ctx, 'owner')
@@ -580,7 +581,7 @@ describe('TaskService owner cleanup', () => {
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)
const tasksFiber = await ctx.plugin(LocalTaskService)
ctx.tasks.attachSurface('test-surface')
const owner = stubAgent(ctx, 'owner')
ctx.agents.register(owner)
@@ -646,11 +647,11 @@ describe('TaskService owner cleanup', () => {
})
})
describe('TaskService disposal', () => {
describe('LocalTaskService 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 fiber = await ctx.plugin(LocalTaskService)
const surface = await ctx.plugin(Object.assign((inner: Context) => {
inner.tasks.attachSurface('test-surface')
}, { inject: ['tasks'] }))
@@ -678,7 +679,7 @@ describe('TaskService disposal', () => {
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)
const fiber = await ctx.plugin(LocalTaskService)
ctx.tasks.attachSurface('test-surface')
const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => {})
const seen: TaskSnapshot[] = []
@@ -716,7 +717,7 @@ describe('TaskService disposal', () => {
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)
const tasksFiber = await ctx.plugin(LocalTaskService)
ctx.tasks.attachSurface('test-surface')
const owner = stubAgent(ctx, 'owner')
ctx.agents.register(owner)
@@ -741,7 +742,7 @@ describe('TaskService disposal', () => {
it('detaching the last surface re-arms the register fence', async () => {
const ctx = new Context()
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
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) => {

30
packages/tasks/tasks-local/tsconfig.json Normal file
View 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": "../../core/agent"
},
{
"path": "../../util/timeout"
},
{
"path": "../tasks"
},
{
"path": "../../support/invariants"
}
]
}

4
packages/tasks/tasks/README.i18n.yaml
View File

@@ -2,5 +2,5 @@
# side as of the last confirmed-consistent state. Both languages carry equal authority;
# after editing either side, bring the other along and re-record with:
# pnpm run verify-translation-pairing --write
README.md: 1a073add0fde8f2e519cc83b087af6a531a6cbb8
README.zh.md: 795602701f072068f05bbf16ee98bdeea57548af
README.md: 2f822bad139020f0ebae0165aa4e8893853f635d
README.zh.md: 4adb249f31241d5c61c3f8cbee638e8243e4a92e

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

@@ -2,9 +2,9 @@
English | [中文](README.zh.md)
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.
The background task registry seam (`ctx.tasks`). The abstract `TaskService` and its vocabulary types give long-running producers shared ids, owner isolation, reads, cancellation, waiting, notices, and cleanup under one contract; the process-local registry lives in [`dsh-tasks-local`](../tasks-local/README.md). Producer plugins extend `TaskKindMap` with their opaque id namespace.
## Service API
## Service contract
- `start(spec): TaskId` validates the control surface, spec, exact live owner, and optional positive `outputLimitBytes` 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.
@@ -18,13 +18,9 @@ Owned access compares the task's `SessionId` with the caller's. Ids such as `bas
`outputLimitBytes` is producer-owned model-presentation policy carried unchanged into snapshots. A control surface applies it after adding status or notice metadata; the registry does not rewrite producer output or invent a default for producers that omit it.
## Lifecycle
Implementations also owe the lifecycle semantics of the contract: registrations outlive producer and control-surface fibers, owner and service disposal cancel live work and await compliant producers, and settlement is first-wins — one terminal record, one round of contained listener notification, released waiters.
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 Agent Note](../../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md).
See the [task type catalog](../../../docs/core-data-structures/tasks.md), the [runtime Agent Note](../../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md), and the [seam Agent Note](../../../.agents/notes/implemented/architecture/2026-07-26-task-registry-seam.md).
## Model Experience
@@ -36,8 +32,6 @@ No direct invalidation; the named consumer owns any request-prefix changes.
## 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.
- **The contract is in-process** — `TaskStart.run()` passes callbacks and exact `Agent` objects; a durable or cross-process backend must reshape identity, restart, ownership, and observation semantics before it can implement this seam.

16
packages/tasks/tasks/README.zh.md
View File

@@ -2,9 +2,9 @@
[English](README.md) | 中文
进程局部的后台任务注册表（`ctx.tasks`）。它为长时间运行的生产方提供共享 id、owner 隔离、读取、取消、等待、通知和清理。生产方插件使用其不透明 id namespace 扩展 `TaskKindMap`。
后台任务注册表 seam（`ctx.tasks`）。抽象的 `TaskService` 及其词汇类型在同一份契约下为长时间运行的生产方提供共享 id、owner 隔离、读取、取消、等待、通知和清理；进程局部注册表位于 [`dsh-tasks-local`](../tasks-local/README.md)。生产方插件使用其不透明 id namespace 扩展 `TaskKindMap`。
## 服务 API
## 服务契约
- `start(spec): TaskId` 验证控制表层、spec、精确的存活 owner，以及可选的正 `outputLimitBytes`，然后只调用生产方的 `run()` 一次。启动方抛出异常时不注册任何内容；成功返回会直接提交，不再执行其他可能失败的步骤。
- `get(id, caller?)` 和 `list(caller?)` 返回非消费式快照。列表只包含调用方拥有及无 owner 的任务。
@@ -18,13 +18,9 @@
`outputLimitBytes` 是生产方拥有的模型呈现策略，会原样携带到快照中。控制表层在添加状态或通知元数据后应用它；注册表不会重写生产方输出，也不会为省略此字段的生产方虚构默认值。
## 生命周期
实现还必须兑现契约的生命周期语义：注册的存续期长于生产方与控制表层的 fiber，owner 释放和服务释放会取消存活工作并等待守约的生产方，结算遵循首次结果优先（一条终止记录、一轮故障隔离的监听器通知，然后释放等待方）。
任务属于其 owner 和后端，而不是生产方工具 fiber，因此重载生产方或表层不会停止任务。某个 owner 的第一个任务会把一个受等待的 effect 附加到精确的 `Agent` scope。owner 释放会取消该对象的任务，等待生产方完全停稳，并移除其快照；复用 agent 或 Session id 无法重定向旧清理。
服务释放会关闭监听器、取消所有存活任务、等待其记录，并从仍存活的 owner scope 分离 effect。如果拆卸取消抛出异常，服务会强制把记录标为失败，并警告工作可能遗留，而不会死锁。取消已返回但始终不终止 `done` 时，系统无法将其与缓慢停止区分开，拆卸可能因此停滞。
参见[任务类型目录](../../../docs/core-data-structures/tasks.md)和[运行时 Agent Note](../../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md)。
参见[任务类型目录](../../../docs/core-data-structures/tasks.md)、[运行时 Agent Note](../../../.agents/notes/implemented/architecture/2026-06-20-generic-long-running-tool-runtime.md)和 [seam Agent Note](../../../.agents/notes/implemented/architecture/2026-07-26-task-registry-seam.md)。
## 模型体验
@@ -36,8 +32,6 @@
## 已知限制与暂缓事项
- **任务只存在于进程本地**：持久或跨重启执行需要独立生命周期。
- **服务与实现没有拆分**：第二个后端必须先定义塑造该边界的生命周期。
- **流输出只有一个消费游标**：独立观察者需要游标或快照 API。
- **前台工作无法提升**：生产方在启动前选择前台或后台。
- **静默无效的取消可能使拆卸停滞**：只有显式抛出异常才能安全地强制标为失败。
- **契约是进程内的**：`TaskStart.run()` 传入回调和确切的 `Agent` 对象；持久或跨进程后端必须先重塑身份、重启、所有权与观察语义，才能实现此 seam。

2
packages/tasks/tasks/package.json
View File

@@ -31,7 +31,6 @@
"@deepseek-ai/dsh-brand": "^0.0.1",
"@deepseek-ai/dsh-invariants": "^0.0.1",
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-timeout": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"devDependencies": {
@@ -39,7 +38,6 @@
"@deepseek-ai/dsh-brand": "workspace:^",
"@deepseek-ai/dsh-invariants": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-timeout": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

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

@@ -1,19 +1,14 @@
/**
* 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.
* The background task registry seam (`ctx.tasks`). It owns the contract for
* task ids, session-scoped access, lifecycle state, completion listeners, and
* owner cleanup while producers retain their execution resources. The
* process-local registry lives in `@deepseek-ai/dsh-tasks-local`.
* @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'
import type { TaskDoneListener, TaskId, TaskRead, TaskSnapshot, TaskStart } from './types.ts'
export { TaskId } from './types.ts'
export type {
@@ -34,61 +29,34 @@ declare module 'cordis' {
}
}
/** 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
outputLimitBytes: number | undefined
/** 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.
* Abstract background task registry. Subclass, implement the abstract methods,
* and load the subclass as a plugin — it registers as `ctx.tasks` (one
* implementation per context; loading a second throws, which is cordis'
* standard duplicate-service behavior).
*
* Implementations must honor these semantics:
* - Registrations outlive producer and control-surface fibers. Owner and
* service disposal cancel live work and await compliant producers; a
* throwing teardown cancel force-fails only the record.
* - Owned-task access is fenced by the owner's session id. Ids are
* predictable, so authorization — not secrecy — is the boundary.
* - Settlement is first-wins: one terminal record, one round of contained
* listener notification, and released waiters, even against a late
* producer outcome.
* - {@link start} refuses work while no control surface is attached, so a
* producer cannot start work that callers cannot collect or stop.
*/
// 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
export abstract class TaskService extends Service {
constructor(ctx: Context) {
// `abstract` erases at runtime, and this package name used to be the
// mountable concrete registry — a stale composition row would otherwise
// register a ctx.tasks with no method implementations and fail far from
// the misconfiguration. Fail loud at load instead.
if (new.target === TaskService) {
throw new Error('@deepseek-ai/dsh-tasks is the abstract task registry seam; load an implementation such as @deepseek-ai/dsh-tasks-local instead')
}
super(ctx, 'tasks')
this.selfCtx = ctx
ctx.effect(() => () => this.disposeAll(), 'tasks teardown')
}
/**
@@ -99,56 +67,7 @@ export class TaskService extends Service {
* @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.outputLimitBytes !== undefined
&& (!Number.isSafeInteger(spec.outputLimitBytes) || spec.outputLimitBytes <= 0)) {
throw new Error(`invalid outputLimitBytes: expected a positive safe integer, got ${JSON.stringify(spec.outputLimitBytes)}`)
}
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,
outputLimitBytes: spec.outputLimitBytes,
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
}
abstract start(spec: TaskStart): TaskId
/**
* List caller-owned and unowned tasks in registration order without exposing
@@ -156,12 +75,7 @@ export class TaskService extends Service {
* @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))
}
abstract list(caller?: Agent): TaskSnapshot[]
/**
* Return a non-consuming snapshot without changing its read cursor or notice
@@ -170,11 +84,7 @@ export class TaskService extends Service {
* @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)
}
abstract get(id: TaskId, caller?: Agent): TaskSnapshot
/**
* Read the next stream delta, or the idempotent final output after settlement.
@@ -184,15 +94,7 @@ export class TaskService extends Service {
* @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) }
}
abstract read(id: TaskId, caller?: Agent): TaskRead
/**
* Request cancellation, then mark the task stopping and reported. A producer
@@ -203,81 +105,20 @@ export class TaskService extends Service {
* @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'
}
abstract kill(id: TaskId, caller?: Agent, reason?: string): 'requested' | 'already-finished'
/**
* 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.
* rejects only while the task is live; after settlement the terminal
* snapshot wins so a notice suppressed for this waiter is still delivered.
* 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)
}
abstract wait(id: TaskId, timeoutMs: number, caller?: Agent, signal?: AbortSignal): Promise<TaskSnapshot>
/**
* Register an effect-scoped completion listener. Each listener is contained;
@@ -286,13 +127,7 @@ export class TaskService extends Service {
* @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()
}
abstract onTaskDone(listener: TaskDoneListener): () => void
/**
* Attach an effect-scoped surface that can read and stop tasks. {@link start}
@@ -300,149 +135,7 @@ export class TaskService extends Service {
* @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,
...task.outputLimitBytes !== undefined ? { outputLimitBytes: task.outputLimitBytes } : {},
...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 })
}
}
}
abstract attachSurface(name: string): () => void
}
export default TaskService

88
packages/tasks/tasks/tests/service.spec.ts Normal file
View File

@@ -0,0 +1,88 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { TaskId, TaskService } from '@deepseek-ai/dsh-tasks'
import type { TaskDoneListener, TaskRead, TaskSnapshot, TaskStart } from '@deepseek-ai/dsh-tasks'
/**
* Minimal concrete registry: one canned record. The seam owns the contract
* only (ids, snapshots, authorization-shaped signatures); the registry
* behavior suite lives with `@deepseek-ai/dsh-tasks-local`.
*/
class StubTaskService extends TaskService {
snapshotOf(id: TaskId): TaskSnapshot {
return {
id,
kind: 'bash',
label: 'sleep 60',
status: 'running',
startedAt: 0,
reported: false,
}
}
start(spec: TaskStart): TaskId {
spec.run()
return TaskId(`${spec.kind}-1`)
}
list(): TaskSnapshot[] {
return [this.snapshotOf(TaskId('bash-1'))]
}
get(id: TaskId): TaskSnapshot {
return this.snapshotOf(id)
}
read(id: TaskId): TaskRead {
return { text: '', snapshot: this.snapshotOf(id) }
}
kill(): 'requested' | 'already-finished' {
return 'requested'
}
wait(id: TaskId, _timeoutMs: number, _caller?: Agent, _signal?: AbortSignal): Promise<TaskSnapshot> {
return Promise.resolve(this.snapshotOf(id))
}
onTaskDone(_listener: TaskDoneListener): () => void {
return () => {}
}
attachSurface(_name: string): () => void {
return () => {}
}
}
describe('TaskService seam', () => {
it('a concrete subclass registers as ctx.tasks and serves the abstract API', async () => {
const ctx = new Context()
await ctx.plugin(StubTaskService)
const detachSurface = ctx.tasks.attachSurface('seam-test')
const id = ctx.tasks.start({ kind: 'bash', label: 'sleep 60', run: () => ({ cancel() {}, done: new Promise(() => {}) }) })
expect(id).toBe('bash-1')
expect(ctx.tasks.list()).toHaveLength(1)
expect(ctx.tasks.get(id).status).toBe('running')
expect(ctx.tasks.read(id).text).toBe('')
expect(ctx.tasks.kill(id)).toBe('requested')
await expect(ctx.tasks.wait(id, 5)).resolves.toMatchObject({ id })
const detachListener = ctx.tasks.onTaskDone(() => {})
detachListener()
detachSurface()
})
it('loading a second implementation throws (one tasks service per context — cordis standard)', async () => {
const ctx = new Context()
await ctx.plugin(StubTaskService)
class SecondTaskService extends StubTaskService {}
await expect(ctx.plugin(SecondTaskService)).rejects.toThrow(/service "tasks" has been registered/)
})
it('mounting the abstract seam directly fails loudly at load (stale-composition fence)', async () => {
const ctx = new Context()
await expect(ctx.plugin(TaskService as unknown as typeof StubTaskService))
.rejects.toThrow(/abstract task registry seam; load an implementation such as @deepseek-ai\/dsh-tasks-local/)
})
})

3
packages/tasks/tasks/tsconfig.json
View File

@@ -23,9 +23,6 @@
{
"path": "../../core/session"
},
{
"path": "../../util/timeout"
},
{
"path": "../../support/invariants"
}

1
packages/tasks/tool-tasks/package.json
View File

@@ -46,6 +46,7 @@
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-system-prompt": "workspace:^",
"@deepseek-ai/dsh-tasks": "workspace:^",
"@deepseek-ai/dsh-tasks-local": "workspace:^",
"@deepseek-ai/dsh-tools": "workspace:^",
"cordis": "^4.0.0-rc.6"
}

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

@@ -6,7 +6,8 @@ import ToolRegistry from '@deepseek-ai/dsh-tools'
import AgentRegistry from '@deepseek-ai/dsh-agent'
import type { Agent } from '@deepseek-ai/dsh-agent'
import { SessionId } from '@deepseek-ai/dsh-session'
import TaskService, { TaskId } from '@deepseek-ai/dsh-tasks'
import { TaskId } from '@deepseek-ai/dsh-tasks'
import LocalTaskService from '@deepseek-ai/dsh-tasks-local'
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'
@@ -20,7 +21,7 @@ async function setup(config: ToolTasks.Config = {}) {
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
const agentsFiber = await ctx.plugin(AgentRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
const toolsFiber = await ctx.plugin(ToolTasks, config)
return { ctx, agentsFiber, toolsFiber }
}
@@ -91,7 +92,7 @@ describe('tool-tasks setup', () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
await expect(ctx.plugin(ToolTasks, { waitTimeoutMs: 100, maxWaitTimeoutMs: 50 }))
.rejects.toThrow('waitTimeoutMs (100) exceeds maxWaitTimeoutMs (50)')
})
@@ -108,7 +109,7 @@ describe('tool-tasks setup', () => {
const ctx = new Context()
await ctx.plugin(SystemPrompt)
await ctx.plugin(ToolRegistry)
await ctx.plugin(TaskService)
await ctx.plugin(LocalTaskService)
ToolTasks.apply(ctx, {})
expect(ctx.tools.get('task_output')).toBeDefined()
expect(() => ctx.tasks.start(producer().spec)).not.toThrow()