Merge remote-tracking branch 'origin/master' into worktree/web-background-tasks-display-258f7e
# Conflicts: # docs/subsystems/tasks.i18n.yaml # docs/subsystems/tasks.md # docs/subsystems/tasks.zh.md # packages/host/apiproxy/README.i18n.yaml # packages/host/apiproxy/README.md # packages/host/apiproxy/README.zh.md # packages/host/apiproxy/src/api-proxy.ts # packages/tasks/tasks-local/src/index.ts # packages/tasks/tasks/README.i18n.yaml # packages/tasks/tasks/README.md # packages/tasks/tasks/README.zh.md # packages/tasks/tasks/src/index.ts
This commit is contained in:
Yichen Jiang
@@ -9,8 +9,10 @@
|
||||
* @module @deepseek-ai/dsh-tasks-local
|
||||
*/
|
||||
|
||||
import { Context } from 'cordis'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import type { Agent } from '@deepseek-ai/dsh-agent'
|
||||
import { AnonymousEntries, ScopedLayers, scopeOf } from '@deepseek-ai/dsh-scope'
|
||||
import type { ScopeLayer } from '@deepseek-ai/dsh-scope'
|
||||
import { deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
|
||||
import { TaskService, TaskId } from '@deepseek-ai/dsh-tasks'
|
||||
import type {
|
||||
@@ -52,6 +54,22 @@ function isTerminal(status: TaskStatus): boolean {
|
||||
return status === 'completed' || status === 'killed' || status === 'failed'
|
||||
}
|
||||
|
||||
/**
|
||||
* One scope's contributions: the control surfaces attached from it and the
|
||||
* completion listeners registered there. Both tables are anonymous because a
|
||||
* contribution is identified by its own disposer, never by a name a second
|
||||
* registrant could shadow.
|
||||
*/
|
||||
class TaskLayer implements ScopeLayer {
|
||||
readonly surfaces = new AnonymousEntries<symbol>()
|
||||
readonly listeners = new AnonymousEntries<TaskDoneListener>()
|
||||
readonly changed = new AnonymousEntries<TasksChangedListener>()
|
||||
|
||||
isEmpty(): boolean {
|
||||
return this.surfaces.isEmpty() && this.listeners.isEmpty() && this.changed.isEmpty()
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The in-memory `tasks` registry. See the Service Definition contract in
|
||||
* `@deepseek-ai/dsh-tasks` for the ownership, isolation, and lifecycle
|
||||
@@ -60,9 +78,19 @@ function isTerminal(status: TaskStatus): boolean {
|
||||
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 changeListeners = new Set<TasksChangedListener>()
|
||||
/**
|
||||
* Surfaces and listeners layered by the scope that registered them, in the
|
||||
* tools-registry shape: a contribution files into its registering context's
|
||||
* scope, and a read unions the global layer with the reader's scope chain.
|
||||
*
|
||||
* The registry is one process-wide instance serving every composition, so a
|
||||
* flat table would answer a per-owner question process-wide: one preset's
|
||||
* task controls would hold `start()` open for an agent whose own composition
|
||||
* loads none, and one settlement would reach every preset's notice listener.
|
||||
* Layers make both reads owner-relative. Nothing derives a cache from a
|
||||
* layer, so change notification is a no-op.
|
||||
*/
|
||||
private readonly layers = new ScopedLayers<TaskLayer>(() => new TaskLayer(), () => {})
|
||||
private listenersClosed = false
|
||||
/** Owner agents with attached scope cleanup, mapped to the exact disposer. */
|
||||
private ownerCleanups = new Map<Agent, () => Promise<void> | void>()
|
||||
@@ -76,8 +104,8 @@ export class LocalTaskService extends TaskService {
|
||||
}
|
||||
|
||||
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 (!this.servesOwner(spec.owner)) {
|
||||
throw new Error('background tasks unavailable: no control surface serves this agent (load @deepseek-ai/dsh-tool-tasks in its composition)')
|
||||
}
|
||||
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')
|
||||
@@ -118,8 +146,8 @@ export class LocalTaskService extends TaskService {
|
||||
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)}`)
|
||||
// Contain a producer contract violation (`done` rejected) so cleanup and waiters cannot hang.
|
||||
this.selfCtx.logger.warn(`tasks: task ${task.id} producer done promise rejected (producer contract violation): ${String(error)}`)
|
||||
this.settle(task, { status: 'failed', detail: String(error) })
|
||||
},
|
||||
)
|
||||
@@ -218,29 +246,57 @@ export class LocalTaskService extends TaskService {
|
||||
}
|
||||
|
||||
onTaskDone(listener: TaskDoneListener): () => void {
|
||||
const dispose = this.ctx.effect(() => {
|
||||
this.listeners.add(listener)
|
||||
return () => this.listeners.delete(listener)
|
||||
}, 'tasks.onTaskDone()')
|
||||
return () => void dispose()
|
||||
return this.layers.effect(
|
||||
this.ctx,
|
||||
layer => layer.listeners.append(listener),
|
||||
{ label: 'tasks.onTaskDone()' },
|
||||
)
|
||||
}
|
||||
|
||||
onTasksChanged(listener: TasksChangedListener): () => void {
|
||||
const dispose = this.ctx.effect(() => {
|
||||
this.changeListeners.add(listener)
|
||||
return () => this.changeListeners.delete(listener)
|
||||
}, 'tasks.onTasksChanged()')
|
||||
return () => void dispose()
|
||||
return this.layers.effect(
|
||||
this.ctx,
|
||||
layer => layer.changed.append(listener),
|
||||
{ label: 'tasks.onTasksChanged()' },
|
||||
)
|
||||
}
|
||||
|
||||
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()
|
||||
return this.layers.effect(
|
||||
this.ctx,
|
||||
layer => layer.surfaces.append(token),
|
||||
{ label: 'tasks.attachSurface()' },
|
||||
)
|
||||
}
|
||||
|
||||
/**
|
||||
* Whether an attached control surface can collect and stop work owned by
|
||||
* `owner`. The global layer holds every surface attached from an unscoped
|
||||
* context — a host composition's own controls — and therefore serves every
|
||||
* owner; a scoped surface serves exactly the agents composed under it.
|
||||
* @param owner - the task's owner, or undefined for unowned work.
|
||||
* @returns whether some reachable surface serves the owner.
|
||||
*/
|
||||
private servesOwner(owner?: Agent): boolean {
|
||||
if (!this.layers.global.surfaces.isEmpty()) return true
|
||||
return this.layers.chainLayers(owner === undefined ? undefined : scopeOf(owner.ctx))
|
||||
.some(layer => !layer.surfaces.isEmpty())
|
||||
}
|
||||
|
||||
/**
|
||||
* The completion listeners that own `owner`'s notices: the global layer's
|
||||
* first, then each scoped layer along the owner's chain. A listener outside
|
||||
* that chain belongs to another composition and must not deliver, or the
|
||||
* owner reads one notice per mounted preset.
|
||||
* @param owner - the settled task's owner, or undefined for unowned work.
|
||||
* @returns the listeners to notify, in registration order per layer.
|
||||
*/
|
||||
private *listenersFor(owner?: Agent): IterableIterator<TaskDoneListener> {
|
||||
yield* this.layers.global.listeners.values()
|
||||
const scope = owner === undefined ? undefined : scopeOf(owner.ctx)
|
||||
for (const layer of this.layers.chainLayers(scope)) yield* layer.listeners.values()
|
||||
}
|
||||
|
||||
/** Look up a task or fail loud. */
|
||||
@@ -278,12 +334,27 @@ export class LocalTaskService extends TaskService {
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The change observers that own `owner`'s updates, resolved exactly like
|
||||
* {@link listenersFor}: the global layer — a host composition's own carrier,
|
||||
* which serves every owner — then each scoped layer along the owner's chain.
|
||||
* An observer outside that chain belongs to another composition and would
|
||||
* otherwise be told about agents it does not compose.
|
||||
* @param owner - the owner whose visible set moved, or undefined for unowned work.
|
||||
* @returns the observers to notify, in registration order per layer.
|
||||
*/
|
||||
private *changedFor(owner?: Agent): IterableIterator<TasksChangedListener> {
|
||||
yield* this.layers.global.changed.values()
|
||||
const scope = owner === undefined ? undefined : scopeOf(owner.ctx)
|
||||
for (const layer of this.layers.chainLayers(scope)) yield* layer.changed.values()
|
||||
}
|
||||
|
||||
/**
|
||||
* Announce that one owner's visible set changed. Each listener is contained
|
||||
* so an observer cannot break a lifecycle commit that already happened.
|
||||
*/
|
||||
private notifyChanged(owner: Agent | undefined): void {
|
||||
for (const listener of this.changeListeners) {
|
||||
for (const listener of this.changedFor(owner)) {
|
||||
try {
|
||||
listener(owner)
|
||||
} catch (error: unknown) {
|
||||
@@ -306,7 +377,7 @@ export class LocalTaskService extends TaskService {
|
||||
if (task.waiters > 0) task.reported = true
|
||||
if (!this.listenersClosed) {
|
||||
const snapshot = this.snapshot(task)
|
||||
for (const listener of this.listeners) {
|
||||
for (const listener of this.listenersFor(task.owner)) {
|
||||
try {
|
||||
const returned = listener(snapshot, task.owner)
|
||||
void Promise.resolve(returned).catch((error: unknown) => {
|
||||
@@ -364,20 +435,20 @@ export class LocalTaskService extends TaskService {
|
||||
* effects. Throwing cancels are force-failed to avoid teardown deadlock.
|
||||
*/
|
||||
private async disposeAll(): Promise<void> {
|
||||
// The flag is the whole guard: each layer entry's undo belongs to the fiber
|
||||
// that registered it, so this service may not drop them on its own way out.
|
||||
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))
|
||||
// Distinct owners whose records just disappeared. `onTasksChanged` binds to
|
||||
// the CALLING fiber (the traceable proxy rebinds `this.ctx`), so a consumer
|
||||
// mounted outside this service — the api-proxy carrier reads `ctx.get` from
|
||||
// the mux stream — is still listening here. Without this it keeps the rows
|
||||
// it last received after a registry reload.
|
||||
// Distinct owners whose records just disappeared. A change observer files
|
||||
// into the layer of the context that registered it, so a consumer mounted
|
||||
// outside this service — the api-proxy carrier registers from the mux
|
||||
// stream — is still reachable here. Without this it keeps the rows it last
|
||||
// received after a registry reload.
|
||||
const emptied = new Set(all.map(task => task.owner))
|
||||
this.store.clear()
|
||||
for (const owner of emptied) this.notifyChanged(owner)
|
||||
this.changeListeners.clear()
|
||||
// Detach cross-fiber owner effects after the shared store is quiescent.
|
||||
const ownerCleanups = [...this.ownerCleanups.values()]
|
||||
this.ownerCleanups.clear()
|
||||
|
||||
@@ -4,7 +4,7 @@
|
||||
*/
|
||||
|
||||
/* jscpd:ignore-start */
|
||||
import type { Context } from 'cordis'
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
|
||||
|
||||
const PACKAGE_NAME = '@deepseek-ai/dsh-tasks-local'
|
||||
|
||||
Reference in New Issue
Block a user