diff --git a/packages/jobs/jobs-local/src/events.ts b/packages/jobs/jobs-local/src/events.ts index 2e39f7089f..333d5db5fc 100644 --- a/packages/jobs/jobs-local/src/events.ts +++ b/packages/jobs/jobs-local/src/events.ts @@ -62,10 +62,12 @@ export class JobEventHub { if ('owners' in filter && filter.owners === 'scope') { return this.layers.effect(ctx, layer => layer.scoped.append(subscription), { label: 'jobs.events.subscribe()' }) } - return ctx.effect(() => { + const dispose = ctx.effect(() => { this.unscoped.add(subscription) return () => { this.unscoped.delete(subscription) } }, 'jobs.events.subscribe()') + // oxlint-disable-next-line typescript/no-misused-promises -- exact synchronous disposer preserves Cordis effect identity + return dispose } /** diff --git a/packages/jobs/jobs-local/src/index.ts b/packages/jobs/jobs-local/src/index.ts index 165cf83aff..990c03209c 100644 --- a/packages/jobs/jobs-local/src/index.ts +++ b/packages/jobs/jobs-local/src/index.ts @@ -177,7 +177,7 @@ export class LocalJobRegistry extends JobRegistry { this.settledRetainBytes = resolved.settledRetainBytes this.pumpPollMs = resolved.pumpPollMs this.selfCtx = ctx - this.hub = new JobEventHub(this.layers, message => ctx.logger.warn(message)) + this.hub = new JobEventHub(this.layers, (message) => { ctx.logger.warn(message) }) ctx.effect(() => () => this.disposeAll(), 'jobs teardown') } diff --git a/packages/jobs/jobs-local/tests/jobs.spec.ts b/packages/jobs/jobs-local/tests/jobs.spec.ts index 6f65916157..55e5615a58 100644 --- a/packages/jobs/jobs-local/tests/jobs.spec.ts +++ b/packages/jobs/jobs-local/tests/jobs.spec.ts @@ -395,7 +395,7 @@ describe('LocalJobRegistry reads and settlement', () => { p.settle({ status: 'completed', detail: 'exit code: 0' }) await tick() - expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed', detail: 'exit code: 0' }) }]) + expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed', detail: 'exit code: 0' }) as unknown }]) expect(warn).toHaveBeenCalledWith(expect.stringContaining('listener boom')) }) @@ -440,14 +440,14 @@ describe('LocalJobRegistry.kill', () => { expect(jobsOf(ctx).kill(id, { reason: 'no longer needed' })).toBe('requested') expect(p.cancels).toEqual(['no longer needed']) expect(jobsOf(ctx).list()[0]).toMatchObject({ status: 'stopping' }) - expect(seen).toEqual([{ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping' }) }]) + expect(seen).toEqual([{ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping' }) as unknown }]) p.settle({ status: 'killed', detail: 'signal: SIGTERM' }) await tick() expect(seen[1]).toEqual({ type: 'settled', cause: 'kill', - job: expect.objectContaining({ id, status: 'killed', detail: 'signal: SIGTERM; no longer needed' }), + job: expect.objectContaining({ id, status: 'killed', detail: 'signal: SIGTERM; no longer needed' }) as unknown, }) }) @@ -495,7 +495,7 @@ describe('LocalJobRegistry.kill', () => { expect(seen).toEqual([]) settle({ status: 'completed' }) await tick() - expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id }) }]) + expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id }) as unknown }]) broken = false expect(jobsOf(ctx).kill(id)).toBe('already-finished') @@ -587,7 +587,7 @@ describe('LocalJobRegistry.wait', () => { p.settle({ status: 'completed', detail: 'exit code: 0' }) controller.abort() await expect(wait).rejects.toThrow('wait aborted') - expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed' }) }]) + expect(seen).toEqual([{ type: 'settled', cause: 'producer', job: expect.objectContaining({ id, status: 'completed' }) as unknown }]) }) it('an abort landing after settlement still delivers the terminal projection it owes', async () => { @@ -751,7 +751,7 @@ describe('LocalJobRegistry owner cleanup', () => { }) await disposeAgentScope(owner) - expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'killed' }) }]) + expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'killed' }) as unknown }]) }) it('attaches one cleanup per owner and drains all owned jobs with the scope', async () => { @@ -933,7 +933,7 @@ describe('LocalJobRegistry disposal', () => { expect(disposedWithoutProducerDone).toBe(true) expect(warn).toHaveBeenCalledWith(expect.stringContaining('work may be orphaned')) - expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'failed' }) }]) + expect(seen).toEqual([{ type: 'settled', cause: 'teardown', job: expect.objectContaining({ status: 'failed' }) as unknown }]) }) it('detaches owner effects from still-live agent scopes when the service unloads', async () => { @@ -1012,14 +1012,14 @@ describe('LocalJobRegistry events', () => { const p = producer({ owner }) const id = ctx.jobs.start(p.spec) // Registration is announced only once the record is readable. - expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id, owner: 'alice', status: 'running' }) }]) + expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id, owner: 'alice', status: 'running' }) as unknown }]) expect(jobsOf(ctx, owner).list()).toHaveLength(1) p.job().updateProgress('3/10') - expect(seen[1]).toEqual({ type: 'progress', job: expect.objectContaining({ id, progress: '3/10' }) }) + expect(seen[1]).toEqual({ type: 'progress', job: expect.objectContaining({ id, progress: '3/10' }) as unknown }) expect(jobsOf(ctx, owner).kill(id)).toBe('requested') - expect(seen[2]).toEqual({ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping', progress: '3/10' }) }) + expect(seen[2]).toEqual({ type: 'stopping', job: expect.objectContaining({ id, status: 'stopping', progress: '3/10' }) as unknown }) p.settle({ status: 'killed' }) await tick() @@ -1033,7 +1033,7 @@ describe('LocalJobRegistry events', () => { const ctx = await harness() const seen = collect(ctx, { owner: SessionId('anyone') }, ['registered']) ctx.jobs.start(producer().spec) - expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id: 'bash-1' }) }]) + expect(seen).toEqual([{ type: 'registered', job: expect.objectContaining({ id: 'bash-1' }) as unknown }]) expect((seen[0] as { job: JobView }).job.owner).toBeUndefined() }) diff --git a/packages/jobs/jobs/src/view.ts b/packages/jobs/jobs/src/view.ts index 033fcfbe76..d3c73ae018 100644 --- a/packages/jobs/jobs/src/view.ts +++ b/packages/jobs/jobs/src/view.ts @@ -62,6 +62,8 @@ export interface JobView { readonly label: string /** Owning session; absent for an unowned job, which every caller can see. */ readonly owner?: SessionId + /** Producer-owned cap for complete model-facing notices and reads, in UTF-8 bytes. */ + readonly outputLimitBytes?: number /** Current lifecycle state. */ readonly status: JobStatus /** The producer's live progress line (`3/10`, the current phase); cleared at settlement. */ diff --git a/packages/jobs/jobs/tests/service.spec.ts b/packages/jobs/jobs/tests/service.spec.ts index f3b456acbc..5bcde91862 100644 --- a/packages/jobs/jobs/tests/service.spec.ts +++ b/packages/jobs/jobs/tests/service.spec.ts @@ -77,7 +77,7 @@ describe('JobRegistry seam', () => { expect(jobs.list()).toHaveLength(1) expect(jobs.get(id).status).toBe('running') expect(jobs.get(id).owner).toBe(caller) - expect(jobs.read(id)).toEqual({ chunks: [], lossy: false, job: expect.objectContaining({ id }) }) + expect(jobs.read(id)).toEqual({ chunks: [], lossy: false, job: expect.objectContaining({ id }) as unknown }) expect(jobs.readAt(id, 7)).toEqual({ chunks: [], next: 7, lossy: false }) expect(jobs.kill(id, { reason: 'seam test' })).toBe('requested') await expect(jobs.wait(id, 5)).resolves.toMatchObject({ id }) diff --git a/packages/jobs/tool-jobs/src/index.ts b/packages/jobs/tool-jobs/src/index.ts index 325766ee2b..d189182b72 100644 --- a/packages/jobs/tool-jobs/src/index.ts +++ b/packages/jobs/tool-jobs/src/index.ts @@ -1,9 +1,10 @@ /** * Model-facing `job_output`, `job_list`, and `job_kill` tools over * `ctx.jobs`. Loading the plugin attaches the controller required by - * producers. It also delivers unreported completions to the owning agent: - * injected into a busy owner's next step, or opening a turn on an idle one - * under the default `wakeup` delivery, bounded per owner. + * producers. It also delivers completions the model has not already + * collected to the owning agent: injected into a busy owner's next step, or + * opening a turn on an idle one under the default `wakeup` delivery, bounded + * per owner. * @module @deepseek-ai/dsh-tool-jobs */ @@ -14,14 +15,19 @@ import { TextRetainer } from '@deepseek-ai/dsh-output-retention' import { defineTool } from '@deepseek-ai/dsh-tools' import type { GenericCallView, ToolDefinition, ToolExecution } from '@deepseek-ai/dsh-tools' import { JobId } from '@deepseek-ai/dsh-jobs' -import type { JobSnapshot } from '@deepseek-ai/dsh-jobs' +import type { JobView, VisibleJobs } from '@deepseek-ai/dsh-jobs' import type { Agent } from '@deepseek-ai/dsh-agent' +import { jobDetail, publicJob, renderModelDelta, statusLine } from './render.ts' +import type { PublicJobSnapshot } from './render.ts' + +export { publicJob, renderModelDelta, statusLine } from './render.ts' +export type { PublicJobSnapshot } from './render.ts' export const name = 'tool-jobs' -export const inject = ['tools', 'jobs', 'systemPrompt'] +export const inject = ['tools', 'jobs', 'agents', 'systemPrompt'] /** - * How an unreported completion reaches an owner that is already idle: `wakeup` + * How an uncollected completion reaches an owner that is already idle: `wakeup` * opens a turn for it, `quiet` leaves it pending until something else wakes the * owner. A busy owner is injected either way. */ @@ -51,17 +57,6 @@ export const Config: z = z.object({ maxConsecutiveWakes: z.number().min(1).default(3), }) -/** Task state safe for model-authored programs; ownership/bookkeeping fields are omitted. */ -export interface PublicJobSnapshot { - id: string - kind: string - label: string - status: JobSnapshot['status'] - detail?: string - startedAt: number - finishedAt?: number -} - /** Shared schema for job-control outputs. */ const PUBLIC_TASK_SCHEMA = { type: 'object', @@ -81,30 +76,6 @@ const PUBLIC_TASK_SCHEMA = { }, } as const -/** Remove job ownership and notification bookkeeping from a registry snapshot. */ -function publicJob(snapshot: JobSnapshot): PublicJobSnapshot { - return { - id: snapshot.id, - kind: snapshot.kind, - label: snapshot.label, - status: snapshot.status, - ...snapshot.detail !== undefined ? { detail: snapshot.detail } : {}, - startedAt: snapshot.startedAt, - ...snapshot.finishedAt !== undefined ? { finishedAt: snapshot.finishedAt } : {}, - } -} - -/** - * Render generic status with optional producer detail. - * @param snapshot - job state to render. - * @returns a bracketed status line. - */ -export function statusLine(snapshot: Pick): string { - return snapshot.detail !== undefined - ? `[status: ${snapshot.status}, ${snapshot.detail}]` - : `[status: ${snapshot.status}]` -} - const encoder = new TextEncoder() function retainTail(text: string, maxBytes: number): string { @@ -135,19 +106,19 @@ function fitWithSuffix( /** * One-line account of a settled job for the `notice` form's collapsed row. - * @param snapshot - the settled job. + * @param job - the settled job. * @returns its kind, label, and status, bounded like every notice summary. */ -function completionSummary(snapshot: JobSnapshot): string { - return boundContextSummary(`${snapshot.kind} ${snapshot.label} ${statusLine(snapshot)}`) +function completionSummary(job: JobView): string { + return boundContextSummary(`${job.kind} ${job.label} ${statusLine({ status: job.status, detail: jobDetail(job) })}`) } -function fitCompletionNotice(snapshot: JobSnapshot): string { - const prefix = `background job ${snapshot.id}` - const detail = ` (${snapshot.kind}: ${snapshot.label}) finished ${statusLine(snapshot)}` +function fitCompletionNotice(job: JobView): string { + const prefix = `background job ${job.id}` + const detail = ` (${job.kind}: ${job.label}) finished ${statusLine({ status: job.status, detail: jobDetail(job) })}` const action = '\nDone; job_output.' const complete = `${prefix}${detail}. Read its output with job_output.` - const maxBytes = snapshot.outputLimitBytes + const maxBytes = job.outputLimitBytes if (maxBytes === undefined || encoder.encode(complete).byteLength <= maxBytes) return complete const omitted = '\n[notice truncated]' const fixed = `${prefix}${omitted}${action}` @@ -181,11 +152,12 @@ function boundSingleText(content: readonly ContentBlock[], maxBytes: number): Co }] } +/** The producer's cap for the job a `job_output` or `job_kill` call names, when it is visible to the caller. */ function visibleOutputLimit(ctx: Context, exec: ToolExecution): number | undefined { if (exec.name !== 'job_output' && exec.name !== 'job_kill') return undefined const jobId = (exec.arguments as { job_id?: unknown } | null | undefined)?.job_id if (typeof jobId !== 'string' || jobId.length === 0) return undefined - return ctx.jobs.list(exec.agent).find(snapshot => snapshot.id === jobId)?.outputLimitBytes + return ctx.jobs.visibleTo(exec.agent?.id).list().find(job => job.id === jobId)?.outputLimitBytes } /** Validate the non-empty constraint that ParameterSchemaSpec cannot express. */ @@ -201,6 +173,21 @@ function presentTaskCall(title: string, kind: 'read' | 'execute', rawInput?: str return { card: 'generic', title, kind, ...rawInput !== undefined ? { rawInput } : {} } } +/** True for the three terminal statuses. */ +function isTerminal(job: JobView): boolean { + return job.status === 'completed' || job.status === 'killed' || job.status === 'failed' +} + +/** The consuming read as the model sees it: the delta, then the result once, then the status line. */ +function readBody(jobs: VisibleJobs, id: JobId): { text: string; job: PublicJobSnapshot } { + const read = jobs.read(id) + const delta = renderModelDelta(read.chunks, read.lossy) + const text = read.result === undefined + ? delta + : `${delta}${delta.length > 0 && !delta.endsWith('\n') ? '\n' : ''}${read.result}` + return { text, job: publicJob(read.job) } +} + export function apply(ctx: Context, config: Config): void { const waitDefault = config.waitTimeoutMs ?? 30_000 const waitCap = config.maxWaitTimeoutMs ?? 600_000 @@ -262,31 +249,52 @@ export function apply(ctx: Context, config: Config): void { ctx.systemPrompt.section({ name: 'tool:jobs', order: ctx.systemPrompt.getSectionOrder('TOOL_JOBS'), - text: 'Track every background job id you start. You are notified in-session when a job finishes — do not busy-poll or sleep on one; keep working on independent steps and do not duplicate a running job\'s work. Before giving a final answer, collect every still-relevant job with job_output (set wait: true only when you are genuinely blocked on it), and job_kill jobs that stopped mattering.', + text: 'Track every background job id you start. You are notified in-session when a job finishes — do not busy-poll or sleep on one; keep working on independent steps and do not stop your turn just to wait. Use job_output when you need its output now. Use job_kill to stop a job you no longer need.', }) - // Use the exact lifecycle owner; reusable ids could resolve to a replacement. + // Jobs whose terminal state this plugin already handed to the model through + // a tool result: a terminal `job_output`, a wait that returned it, or a + // `job_kill` it requested. A wait claims when it starts, not when it + // returns — settlement releases waiters before it announces, but their + // continuations run after the event — and withdraws the claim if the job + // outlives the wait. The settlement event drops the entry, so the ledger + // holds only live jobs the model already touched. + const delivered = new Set() + const claim = (id: JobId): (() => void) => { + delivered.add(id) + return () => { delivered.delete(id) } + } + // A busy owner is injected: the notice waits in its next-step inbox, which // the turn cannot close over, so jobs settling together cost one step. An // idle owner is woken instead, because an unclaimed notice is a completion // the model never learns about. Either way, disposal before the claim - // discards it with the owner, and teardown settlements arrive `reported`. + // discards it with the owner, and a teardown settlement has no reader left. // - // The registry routes each settlement to the listeners its owner's scope - // chain reaches, so a mount under one preset never sees another preset's - // agents; this listener owns delivery, not the choice of whom to deliver to. - ctx.jobs.onJobDone((snapshot, owner) => { - if (snapshot.reported || owner === undefined) return + // The registry routes each settlement to the scope this plugin was mounted + // under, so a mount under one preset never sees another preset's agents; + // this listener owns delivery, not the choice of whom to deliver to. + ctx.jobs.events.subscribe({ owners: 'scope' }, (event) => { + if (event.type === 'removed') { + delivered.delete(event.job.id) + return + } + if (event.type !== 'settled') return + const claimed = delivered.delete(event.job.id) + if (claimed || event.cause === 'teardown' || event.job.owner === undefined) return + // Use the exact lifecycle owner; a session whose agent left has no inbox. + const owner = ctx.agents.get(event.job.owner) + if (owner === undefined) return const message = createUserMessage({ content: [{ type: 'text', - text: fitCompletionNotice(snapshot), + text: fitCompletionNotice(event.job), }], source: { kind: 'plugin', plugin: 'tool-jobs', form: 'notice', - summary: completionSummary(snapshot), + summary: completionSummary(event.job), }, }) const spent = spentWakes.get(owner) ?? 0 @@ -328,12 +336,19 @@ export function apply(ctx: Context, config: Config): void { }, async execute(args, exec) { const id = validateJobId(args.job_id) + const jobs = ctx.jobs.visibleTo(exec.agent?.id) if (args.wait === true) { const timeout = Math.min(args.timeout_ms ?? waitDefault, waitCap) - await ctx.jobs.wait(id, timeout, exec.agent, exec.signal) + const withdraw = claim(id) + try { + const view = await jobs.wait(id, timeout, exec.signal) + if (!isTerminal(view)) withdraw() + } catch (error: unknown) { + withdraw() + throw error + } } - const read = ctx.jobs.read(id, exec.agent) - return { text: read.text, job: publicJob(read.snapshot) } + return readBody(jobs, id) }, presentCall: args => presentTaskCall(`Read output from background job ${args.job_id}`, 'read', args.job_id), })) @@ -352,7 +367,7 @@ export function apply(ctx: Context, config: Config): void { }], }, execute(_args, exec) { - const jobs = ctx.jobs.list(exec.agent) + const jobs = ctx.jobs.visibleTo(exec.agent?.id).list() return Promise.resolve(jobs.map(publicJob)) }, presentCall: () => presentTaskCall('List background jobs', 'read'), @@ -388,12 +403,16 @@ export function apply(ctx: Context, config: Config): void { }, execute(args, exec) { const id = validateJobId(args.job_id) - const result = ctx.jobs.kill(id, exec.agent, args.reason) - // A snapshot describes current state without consuming pending output. - const snapshot = publicJob(ctx.jobs.get(id, exec.agent)) + const jobs = ctx.jobs.visibleTo(exec.agent?.id) + const result = jobs.kill(id, args.reason === undefined ? {} : { reason: args.reason }) + // The model's own kill is its delivery: the settlement notice would only + // repeat what this tool result already said. + if (result === 'requested') claim(id) + // A projection describes current state without consuming pending output. + const job = publicJob(jobs.get(id)) return Promise.resolve({ outcome: result === 'already-finished' ? 'already-finished' as const : 'cancellation-requested' as const, - job: snapshot, + job, }) }, presentCall: args => presentTaskCall(`Kill background job ${args.job_id}`, 'execute', args.job_id), diff --git a/packages/jobs/tool-jobs/src/render.ts b/packages/jobs/tool-jobs/src/render.ts new file mode 100644 index 0000000000..74b0255cbd --- /dev/null +++ b/packages/jobs/tool-jobs/src/render.ts @@ -0,0 +1,80 @@ +/** + * Model-facing rendering of registry reads: the consuming delta as the + * shell tools have always shown it (stdout, then one marked stderr section), + * the `[status: …]` line, and the public job projection the tool schemas + * expose. + * @module @deepseek-ai/dsh-tool-jobs/render + */ + +import type { JobChunk, JobView } from '@deepseek-ai/dsh-jobs' + +/** Job state safe for model-authored programs; ownership and offsets are omitted. */ +export interface PublicJobSnapshot { + id: string + kind: string + label: string + status: JobView['status'] + /** The live progress line while running, the terminal reason once settled. */ + detail?: string + startedAt: number + finishedAt?: number +} + +/** + * The one-line qualifier the model reads beside a status: live progress while + * the job runs, the terminal reason once it settled. + * @param job - the job projection. + * @returns the qualifier, or undefined when the job carries neither. + */ +export function jobDetail(job: Pick): string | undefined { + return job.progress ?? job.detail +} + +/** + * Remove ownership, offsets, and limits from a registry projection. + * @param job - the registry projection. + * @returns the model-safe projection. + */ +export function publicJob(job: JobView): PublicJobSnapshot { + const detail = jobDetail(job) + return { + id: job.id, + kind: job.kind, + label: job.label, + status: job.status, + ...detail !== undefined ? { detail } : {}, + startedAt: job.startedAt, + ...job.finishedAt !== undefined ? { finishedAt: job.finishedAt } : {}, + } +} + +/** + * Render generic status with optional detail. + * @param snapshot - job state to render. + * @returns a bracketed status line. + */ +export function statusLine(snapshot: Pick): string { + return snapshot.detail !== undefined + ? `[status: ${snapshot.status}, ${snapshot.detail}]` + : `[status: ${snapshot.status}]` +} + +/** + * Render one consuming read for the model: stdout and unlabeled chunks in + * order, then every stderr chunk in one `[stderr]` section, exactly as the + * shell tools render a foreground result. `log` chunks are producer + * narration for observers and never reach the model. A lossy read — the + * cursor fell behind the ring's retention — ends with a dropped-output + * notice. + * @param chunks - the chunks since the model cursor, in offset order. + * @param lossy - whether bytes before `chunks` were evicted unread. + * @returns the delta text, possibly empty. + */ +export function renderModelDelta(chunks: readonly JobChunk[], lossy: boolean): string { + const out = chunks.filter(chunk => chunk.channel === undefined || chunk.channel === 'stdout').map(chunk => chunk.text).join('') + const err = chunks.filter(chunk => chunk.channel === 'stderr').map(chunk => chunk.text).join('') + const separator = out.length > 0 && !out.endsWith('\n') ? '\n' : '' + const body = out + (err.length > 0 ? `${separator}[stderr]\n${err}` : '') + if (!lossy) return body + return `${body}${body.length > 0 && !body.endsWith('\n') ? '\n' : ''}[some output was dropped from memory before this read]` +} diff --git a/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts b/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts index 4c6a81e48a..6c792226d4 100644 --- a/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts +++ b/packages/jobs/tool-jobs/tests/tool-jobs.spec.ts @@ -10,21 +10,26 @@ import { SessionId } from '@deepseek-ai/dsh-session' import { bindScopeParent, createScope, scopeOf } from '@deepseek-ai/dsh-scope' import { JobId } from '@deepseek-ai/dsh-jobs' import LocalJobRegistry from '@deepseek-ai/dsh-jobs-local' -import type { JobHooks, JobOutcome, JobSnapshot, JobStart } from '@deepseek-ai/dsh-jobs' +import type { JobAppendOptions, JobHandle, JobHooks, JobOutcome, JobSpec, JobView } from '@deepseek-ai/dsh-jobs' import * as ToolTasks from '@deepseek-ai/dsh-tool-jobs' -import { statusLine } from '@deepseek-ai/dsh-tool-jobs' +import { publicJob, renderModelDelta, statusLine } from '@deepseek-ai/dsh-tool-jobs' const testToolSignal = new AbortController().signal const agentRegistryDisposers = new WeakMap void>() const agentScopeFibers = new WeakMap Promise }>() -async function setup(config: ToolTasks.Config = {}) { +/** Registry knobs a test may narrow; every producer here pushes, so the pump stays idle. */ +interface RegistryConfig { + retainBytes?: number +} + +async function setup(config: ToolTasks.Config = {}, registry: RegistryConfig = {}) { const ctx = new Context() await ctx.plugin(SystemPrompt) await ctx.plugin(ToolRuntime) const agentsFiber = await ctx.plugin(AgentRegistry) - await ctx.plugin(LocalJobRegistry) + await ctx.plugin(LocalJobRegistry, registry) const toolsFiber = await ctx.plugin(ToolTasks, config) return { ctx, agentsFiber, toolsFiber } } @@ -70,9 +75,13 @@ async function disposeAgentScope(agent: Agent): Promise { await fiber.dispose() } -/** A controllable producer start-spec (settle `done` on demand, record cancels). */ -function producer(overrides: Partial & JobHooks> = {}) { +/** + * A controllable producer spec: pushes output and progress through the handle + * the registry hands its starter, settles `done` on demand, records cancels. + */ +function producer(overrides: Partial & JobHooks> = {}) { let settle!: (outcome: JobOutcome) => void + let handle: JobHandle | undefined const cancels: (string | undefined)[] = [] const { kind = 'bash', label = 'sleep 60', owner, outputLimitBytes, ...hookOverrides } = overrides const hooks: JobHooks = { @@ -80,14 +89,24 @@ function producer(overrides: Partial & JobHooks> = {}) { done: new Promise((res) => { settle = res }), ...hookOverrides, } - const spec: JobStart = { + const spec: JobSpec = { kind, label, ...owner !== undefined ? { owner } : {}, ...outputLimitBytes !== undefined ? { outputLimitBytes } : {}, - run: () => hooks, + run: (job) => { handle = job; return hooks }, + } + const started = (): JobHandle => { + if (handle === undefined) throw new Error('producer not started') + return handle + } + return { + spec, + settle, + cancels, + append: (text: string, options?: JobAppendOptions) => { started().append(text, options) }, + progress: (line: string) => { started().updateProgress(line) }, } - return { spec, settle, cancels } } let callCounter = 0 @@ -104,7 +123,7 @@ const tick = () => new Promise(r => setTimeout(r, 0)) /** Start and settle `count` owned jobs one at a time, letting each notice land. */ async function settleTasks(ctx: Context, owner: Agent, count: number): Promise { for (let i = 0; i < count; i += 1) { - const p = producer({ owner }) + const p = producer({ owner: owner.id }) ctx.jobs.start(p.spec) p.settle({ status: 'completed' }) await tick() @@ -119,10 +138,18 @@ describe('tool-jobs setup', () => { expect(() => ctx.jobs.start(producer().spec)).toThrow('no job controller serves this agent') }) + it('leaves with the agent registry it resolves owners through', async () => { + const { ctx, agentsFiber } = await setup() + await agentsFiber.dispose() + expect(ctx.tools.get('job_output')).toBeUndefined() + expect(() => ctx.jobs.start(producer().spec)).toThrow('no job controller serves this agent') + }) + it('rejects a config whose default wait exceeds the cap', async () => { const ctx = new Context() await ctx.plugin(SystemPrompt) await ctx.plugin(ToolRuntime) + await ctx.plugin(AgentRegistry) await ctx.plugin(LocalJobRegistry) await expect(ctx.plugin(ToolTasks, { waitTimeoutMs: 100, maxWaitTimeoutMs: 50 })) .rejects.toThrow('waitTimeoutMs (100) exceeds maxWaitTimeoutMs (50)') @@ -141,6 +168,7 @@ describe('tool-jobs setup', () => { const ctx = new Context() await ctx.plugin(SystemPrompt) await ctx.plugin(ToolRuntime) + await ctx.plugin(AgentRegistry) await ctx.plugin(LocalJobRegistry) try { await ctx.plugin(ToolTasks, { maxConsecutiveWakes }) @@ -158,9 +186,35 @@ describe('tool-jobs setup', () => { }) it('renders status lines with and without producer detail', () => { - const base = { id: 'bash-1', kind: 'bash', label: 'x', startedAt: 0, reported: false } as unknown as JobSnapshot - expect(statusLine({ ...base, status: 'running' })).toBe('[status: running]') - expect(statusLine({ ...base, status: 'completed', detail: 'exit code: 0' })).toBe('[status: completed, exit code: 0]') + expect(statusLine({ status: 'running' })).toBe('[status: running]') + expect(statusLine({ status: 'completed', detail: 'exit code: 0' })).toBe('[status: completed, exit code: 0]') + }) + + it('projects the public job: live progress or terminal detail, never ownership or offsets', () => { + const live: JobView = { + id: JobId('bash-1'), kind: 'bash', label: 'x', owner: SessionId('sess-1'), outputLimitBytes: 8, + status: 'running', progress: '3/10 files', startedAt: 5, output: { total: 12, earliest: 0 }, + } + expect(publicJob(live)).toEqual({ id: 'bash-1', kind: 'bash', label: 'x', status: 'running', detail: '3/10 files', startedAt: 5 }) + const settled: JobView = { + id: JobId('bash-1'), kind: 'bash', label: 'x', status: 'completed', detail: 'exit code: 0', + startedAt: 5, finishedAt: 9, output: { total: 12, earliest: 0 }, + } + expect(publicJob(settled)).toEqual({ id: 'bash-1', kind: 'bash', label: 'x', status: 'completed', detail: 'exit code: 0', startedAt: 5, finishedAt: 9 }) + }) + + it('renders a model delta as stdout, one stderr section, and a dropped-output notice', () => { + expect(renderModelDelta([], false)).toBe('') + expect(renderModelDelta([], true)).toBe('[some output was dropped from memory before this read]') + expect(renderModelDelta([ + { at: 0, text: 'a' }, + { at: 1, text: 'e1\n', channel: 'stderr' }, + { at: 4, text: 'narration', channel: 'log' }, + { at: 13, text: 'b', channel: 'stdout' }, + { at: 14, text: 'e2', channel: 'stderr' }, + ], false)).toBe('ab\n[stderr]\ne1\ne2') + expect(renderModelDelta([{ at: 0, text: 'tail' }], true)).toBe('tail\n[some output was dropped from memory before this read]') + expect(renderModelDelta([{ at: 0, text: 'line\n' }], true)).toBe('line\n[some output was dropped from memory before this read]') }) it('applies the built-in wait bounds when apply() receives a bare config', async () => { @@ -169,6 +223,7 @@ describe('tool-jobs setup', () => { const ctx = new Context() await ctx.plugin(SystemPrompt) await ctx.plugin(ToolRuntime) + await ctx.plugin(AgentRegistry) await ctx.plugin(LocalJobRegistry) ToolTasks.apply(ctx, {}) expect(ctx.tools.get('job_output')).toBeDefined() @@ -179,8 +234,9 @@ describe('tool-jobs setup', () => { describe('job_output', () => { it('reads a consuming delta with a trailing status line', async () => { const { ctx } = await setup() - const chunks = ['line one\n', ''] - ctx.jobs.start(producer({ readOutput: () => chunks.shift() ?? '' }).spec) + const p = producer() + ctx.jobs.start(p.spec) + p.append('line one\n') // A body already ending in a newline gets no doubled separator. const first = await call(ctx, 'job_output', { job_id: 'bash-1' }) @@ -190,29 +246,75 @@ describe('job_output', () => { text: 'line one\n', job: { id: 'bash-1', kind: 'bash', label: 'sleep 60', status: 'running' }, }) - expect(firstValue.job).not.toHaveProperty('ownerSession') - expect(firstValue.job).not.toHaveProperty('reported') + expect(firstValue.job).not.toHaveProperty('owner') + expect(firstValue.job).not.toHaveProperty('output') expect(text(first)).toBe('line one\n[status: running]') expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running]') }) - it('returns the final output of a settled final-output job', async () => { + it('renders stderr in a trailing section and never shows log narration', async () => { + const { ctx } = await setup() + const p = producer() + ctx.jobs.start(p.spec) + p.append('out\n') + p.append('warn\n', { channel: 'stderr' }) + p.append('observer-only narration', { channel: 'log' }) + expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('out\n[stderr]\nwarn\n[status: running]') + expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running]') + }) + + it('shows live progress beside the status until settlement replaces it with the terminal detail', async () => { + const { ctx } = await setup() + const p = producer() + ctx.jobs.start(p.spec) + p.progress('3/10 files') + expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running, 3/10 files]') + p.settle({ status: 'completed', detail: 'exit code: 0' }) + await tick() + expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: completed, exit code: 0]') + }) + + it('flags a read whose cursor fell behind the ring', async () => { + const { ctx } = await setup({}, { retainBytes: 16 }) + const p = producer() + ctx.jobs.start(p.spec) + p.append('x'.repeat(40)) + const first = text(await call(ctx, 'job_output', { job_id: 'bash-1' })) + expect(first).toContain('[some output was dropped from memory before this read]') + expect(first).toContain('[status: running]') + expect(first).not.toContain('x'.repeat(17)) + // The cursor now sits at the ring's end: the next read is clean. + expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('(no new output)\n[status: running]') + }) + + it('returns the result of a settled job exactly once, after the pending delta', async () => { const { ctx } = await setup() const p = producer({ kind: 'subagent', label: 'research' }) ctx.jobs.start(p.spec) expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('(no new output)\n[status: running]') - p.settle({ status: 'completed', detail: 'completed', output: 'the answer' }) + p.append('partial') + p.settle({ status: 'completed', detail: 'completed', result: 'the answer' }) await tick() - expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('the answer\n[status: completed, completed]') + expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('partial\nthe answer\n[status: completed, completed]') + expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('(no new output)\n[status: completed, completed]') + }) + + it('keeps a newline-terminated delta and the result on separate lines', async () => { + const { ctx } = await setup() + const p = producer({ kind: 'subagent', label: 'research' }) + ctx.jobs.start(p.spec) + p.append('progress log\n') + p.settle({ status: 'completed', result: 'the answer' }) + await tick() + expect(text(await call(ctx, 'job_output', { job_id: 'subagent-1' }))).toBe('progress log\nthe answer\n[status: completed]') }) it('applies a producer limit to the complete body and status result', async () => { const { ctx } = await setup() - ctx.jobs.start(producer({ - outputLimitBytes: 48, - readOutput: () => '界'.repeat(100), - }).spec) + const p = producer({ outputLimitBytes: 48 }) + ctx.jobs.start(p.spec) + p.append('界'.repeat(100)) const output = text(await call(ctx, 'job_output', { job_id: 'bash-1' })) expect(Buffer.byteLength(output)).toBeLessThanOrEqual(48) @@ -221,24 +323,21 @@ describe('job_output', () => { it('preserves empty and newline-terminated output under a producer limit', async () => { const { ctx } = await setup() - const chunks = ['', 'line\n'] - ctx.jobs.start(producer({ - outputLimitBytes: 64, - readOutput: () => chunks.shift() ?? '', - }).spec) + const p = producer({ outputLimitBytes: 64 }) + ctx.jobs.start(p.spec) expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))) .toBe('(no new output)\n[status: running]') + p.append('line\n') expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))) .toBe('line\n[status: running]') }) it('bounds post-policy output without restoring the canonical status rendering', async () => { const { ctx } = await setup() - ctx.jobs.start(producer({ - outputLimitBytes: 64, - readOutput: () => 'canonical output', - }).spec) + const p = producer({ outputLimitBytes: 64 }) + ctx.jobs.start(p.spec) + p.append('canonical output') ctx.on('tools/post-execute', (exec, _result, next) => { if (exec.name !== 'job_output') return next() return Promise.resolve({ kind: 'accept', content: [{ type: 'text', text: 'p'.repeat(1_000) }] }) @@ -250,17 +349,30 @@ describe('job_output', () => { expect(text(result)).not.toContain('[status: running]') }) - it('applies a producer limit to a normalized read failure', async () => { + it('does not double a truncation marker a policy already appended', async () => { const { ctx } = await setup() - ctx.jobs.start(producer({ - outputLimitBytes: 64, - readOutput: () => { throw new Error('read failed: '.repeat(100)) }, - }).spec) + ctx.jobs.start(producer({ outputLimitBytes: 64 }).spec) + ctx.on('tools/post-execute', (exec, _result, next) => { + if (exec.name !== 'job_output') return next() + return Promise.resolve({ kind: 'accept', content: [{ type: 'text', text: `${'z'.repeat(1_000)}\n[result truncated]` }] }) + }) const result = await call(ctx, 'job_output', { job_id: 'bash-1' }) - expect(result.isError).toBe(true) expect(Buffer.byteLength(text(result))).toBeLessThanOrEqual(64) - expect(text(result)).toContain('[result truncated]') + expect(text(result).match(/\[result truncated\]/g)).toHaveLength(1) + }) + + it('refuses a job owned by another session as an errored result', async () => { + const { ctx } = await setup() + const alice = fakeAgent(ctx, 'sess-alice') + ctx.jobs.start(producer({ owner: alice.id }).spec) + const bob = fakeAgent(ctx, 'sess-bob') + + const result = await call(ctx, 'job_output', { job_id: 'bash-1' }, bob) + expect(result.isError).toBe(true) + expect(text(result)).toContain('belongs to another session') + expect((await call(ctx, 'job_output', { job_id: 'bash-1' })).isError).toBe(true) + expect((await call(ctx, 'job_output', { job_id: 'bash-1' }, alice)).isError).toBe(false) }) it('bounds pre-, around-, and post-execute policy outcomes and failures', async () => { @@ -325,7 +437,7 @@ describe('job_output', () => { ctx.jobs.start(p.spec) const pending = call(ctx, 'job_output', { job_id: 'subagent-1', wait: true }) - p.settle({ status: 'completed', output: 'done deal' }) + p.settle({ status: 'completed', result: 'done deal' }) expect(text(await pending)).toBe('done deal\n[status: completed]') }) @@ -354,9 +466,9 @@ describe('job_list', () => { expect(text(await call(ctx, 'job_list', {}))).toBe('(no background jobs)') const alice = fakeAgent(ctx, 'sess-alice') - ctx.jobs.start(producer({ owner: alice, label: 'pnpm test' }).spec) + ctx.jobs.start(producer({ owner: alice.id, label: 'pnpm test' }).spec) ctx.jobs.start(producer({ kind: 'subagent', label: 'open research' }).spec) - const p = producer({ owner: alice, label: 'build' }) + const p = producer({ owner: alice.id, label: 'build' }) ctx.jobs.start(p.spec) p.settle({ status: 'completed', detail: 'exit code: 0' }) await tick() @@ -368,8 +480,8 @@ describe('job_list', () => { expect(listedValue[0]).toMatchObject({ id: 'bash-1', kind: 'bash', label: 'pnpm test', status: 'running' }) expect(listedValue[2]).toMatchObject({ id: 'bash-2', kind: 'bash', label: 'build', status: 'completed', detail: 'exit code: 0' }) for (const job of listedValue) { - expect(job).not.toHaveProperty('ownerSession') - expect(job).not.toHaveProperty('reported') + expect(job).not.toHaveProperty('owner') + expect(job).not.toHaveProperty('output') } expect(text(listed)).toBe([ 'bash-1 [bash] running — pnpm test', @@ -395,8 +507,8 @@ describe('job_kill', () => { outcome: 'cancellation-requested', job: { id: 'bash-1', kind: 'bash', label: 'sleep 60', status: 'stopping' }, }) - expect(killValue.job).not.toHaveProperty('ownerSession') - expect(killValue.job).not.toHaveProperty('reported') + expect(killValue.job).not.toHaveProperty('owner') + expect(killValue.job).not.toHaveProperty('output') expect(text(result)).toBe('requested cancellation of job bash-1') expect(p.cancels).toEqual(['superseded']) }) @@ -422,7 +534,7 @@ describe('job_kill', () => { expect(result.isError).toBe(true) expect(Buffer.byteLength(text(result))).toBeLessThanOrEqual(64) expect(text(result)).toContain('[result truncated]') - expect(ctx.jobs.get(JobId('bash-1'))).toMatchObject({ status: 'running', reported: false }) + expect(ctx.jobs.visibleTo().get(JobId('bash-1')).status).toBe('running') }) it('bounds single-text post policy while preserving structured policy results', async () => { @@ -470,9 +582,9 @@ describe('job_kill', () => { it('reports an already-finished job without consuming its pending delta', async () => { const { ctx } = await setup() - let delta = 'unread tail' - const p = producer({ readOutput: () => { const d = delta; delta = ''; return d } }) + const p = producer() ctx.jobs.start(p.spec) + p.append('unread tail') p.settle({ status: 'completed', detail: 'exit code: 0' }) await tick() @@ -483,7 +595,7 @@ describe('job_kill', () => { job: { id: 'bash-1', kind: 'bash', label: 'sleep 60', status: 'completed', detail: 'exit code: 0' }, }) expect(text(killed)).toBe('job bash-1 had already finished [status: completed, exit code: 0]') - // The kill described the job via a non-consuming snapshot: the delta is intact. + // The kill described the job via a non-consuming projection: the delta is intact. expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }))).toBe('unread tail\n[status: completed, exit code: 0]') }) @@ -508,9 +620,9 @@ describe('tool-owned UI presentation (presentCall)', () => { describe('completion notices across scoped mounts', () => { /** * Two agent presets mounting `tool-jobs` over ONE host registry: each mount - * registers its own `onJobDone` listener on the shared service, and - * `settle()` broadcasts one snapshot to every listener with no scope filter. - * Only the mount whose scope the owner belongs to may deliver the notice. + * subscribes with `{ owners: 'scope' }`, which the registry files into the + * mount's own scope layer and reaches along the owner's scope chain. Only + * the mount whose scope the owner belongs to sees the settlement. */ it('delivers one notice from the owning scope when two mounts share the registry', async () => { const ctx = new Context() @@ -539,9 +651,7 @@ describe('completion notices across scoped mounts', () => { const dispose = ctx.agents.register(owner) try { - // No waiter: `settle()` leaves `reported` false, which is the only path - // that reaches the notice listeners at all. - const p = producer({ owner, label: 'pnpm test' }) + const p = producer({ owner: owner.id, label: 'pnpm test' }) ctx.jobs.start(p.spec) p.settle({ status: 'completed', detail: 'exit code: 0' }) await tick() @@ -559,7 +669,7 @@ describe('completion notice delivery', () => { 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' }) + const p = producer({ owner: owner.id, label: 'pnpm test' }) ctx.jobs.start(p.spec) p.settle({ status: 'completed', detail: 'exit code: 0' }) @@ -573,7 +683,7 @@ describe('completion notice delivery', () => { const inject = vi.fn() const followup = vi.fn() const owner = fakeAgent(ctx, 'sess-1', { inject, followup, status: 'idle' }) - const p = producer({ owner }) + const p = producer({ owner: owner.id }) ctx.jobs.start(p.spec) p.settle({ status: 'completed' }) @@ -621,15 +731,15 @@ describe('completion notice delivery', () => { ctx.jobs.start({ kind: 'bash', label: 'sleep 60', - owner, + owner: owner.id, run: () => ({ cancel() { settle({ status: 'killed' }) }, done: new Promise((res) => { settle = res }), }), }) - // Disposal cancels and settles the owned job. Waking here would spend a - // model request on an agent the host is destroying, once per tree layer. + // Disposal cancels and settles the owned job with a teardown cause. Waking + // here would spend a model request on an agent the host is destroying. await disposeAgentScope(owner) await tick() expect(followup).not.toHaveBeenCalled() @@ -645,17 +755,16 @@ describe('completion notice delivery', () => { ctx.jobs.start({ kind: 'bash', label: 'broken producer', - owner, + owner: owner.id, run: () => ({ cancel() { throw new Error('cancel boom') }, done: new Promise(() => {}), }), }) - // The registry force-fails the record instead of deadlocking. That path - // settles the job too, so it must claim the report as the ordinary - // teardown cancel does — otherwise a throwing producer is all it takes to - // spend a model request on an owner being destroyed. + // The registry force-fails the record instead of deadlocking. That + // settlement also announces a teardown cause, so a throwing producer is + // not enough to spend a model request on an owner being destroyed. await disposeAgentScope(owner) await tick() expect(warn).toHaveBeenCalledWith(expect.stringContaining('work may be orphaned')) @@ -682,11 +791,11 @@ describe('completion notice delivery', () => { }) describe('completion notices', () => { - it('injects a notice into the owning agent when an unreported job settles', async () => { + it('injects a notice into the owning agent when an uncollected job settles', async () => { const { ctx } = await setup() const inject = vi.fn() const owner = fakeAgent(ctx, 'sess-1', { inject }) - const p = producer({ owner, label: 'pnpm test' }) + const p = producer({ owner: owner.id, label: 'pnpm test' }) ctx.jobs.start(p.spec) p.settle({ status: 'completed', detail: 'exit code: 0' }) @@ -710,7 +819,7 @@ describe('completion notices', () => { const inject = vi.fn() const owner = fakeAgent(ctx, 'sess-1', { inject }) const first = producer({ - owner, + owner: owner.id, kind: 'subagent', label: 'x'.repeat(1_000), outputLimitBytes: 61, @@ -737,7 +846,7 @@ describe('completion notices', () => { ) const second = producer({ - owner, + owner: owner.id, kind: 'subagent', label: 'x'.repeat(1_000), outputLimitBytes: 80, @@ -764,7 +873,7 @@ describe('completion notices', () => { const inject = vi.fn() const owner = fakeAgent(ctx, 'sess-1', { inject }) const target = producer({ - owner, + owner: owner.id, kind: 'pty-send', label: 'x'.repeat(1_000), outputLimitBytes: 64, @@ -784,8 +893,8 @@ describe('completion notices', () => { const { ctx } = await setup() const inject = vi.fn() 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 }) + const tiny = producer({ owner: owner.id, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 8 }) + const short = producer({ owner: owner.id, kind: 'pty-send', label: 'x'.repeat(100), outputLimitBytes: 32 }) ctx.jobs.start(tiny.spec) ctx.jobs.start(short.spec) @@ -805,7 +914,7 @@ describe('completion notices', () => { const { ctx } = await setup() const inject = vi.fn() const owner = fakeAgent(ctx, 'sess-1', { inject }) - const p = producer({ owner }) + const p = producer({ owner: owner.id }) ctx.jobs.start(p.spec) await call(ctx, 'job_kill', { job_id: 'bash-1' }, owner) @@ -818,15 +927,55 @@ describe('completion notices', () => { const { ctx } = await setup() const inject = vi.fn() const owner = fakeAgent(ctx, 'sess-1', { inject }) - const p = producer({ owner, kind: 'subagent' }) + const p = producer({ owner: owner.id, kind: 'subagent' }) ctx.jobs.start(p.spec) const pending = call(ctx, 'job_output', { job_id: 'subagent-1', wait: true }, owner) - p.settle({ status: 'completed', output: 'answer' }) + p.settle({ status: 'completed', result: 'answer' }) expect(text(await pending)).toContain('answer') + await tick() expect(inject).not.toHaveBeenCalled() }) + it('withdraws the wait claim when the wait times out, so the later settlement still notifies', async () => { + const { ctx } = await setup({ waitTimeoutMs: 10, maxWaitTimeoutMs: 20 }) + const inject = vi.fn() + const owner = fakeAgent(ctx, 'sess-1', { inject }) + const p = producer({ owner: owner.id }) + ctx.jobs.start(p.spec) + + expect(text(await call(ctx, 'job_output', { job_id: 'bash-1', wait: true }, owner))).toBe('(no new output)\n[status: running]') + p.settle({ status: 'completed' }) + await tick() + expect(inject).toHaveBeenCalledTimes(1) + }) + + it('withdraws the wait claim when the caller aborts the wait', async () => { + const { ctx } = await setup() + const inject = vi.fn() + const owner = fakeAgent(ctx, 'sess-1', { inject }) + const p = producer({ owner: owner.id }) + ctx.jobs.start(p.spec) + + const aborter = new AbortController() + const pending = ctx.tools.execute({ + signal: aborter.signal, + callId: ToolCallId('call-aborted-wait'), + name: 'job_output', + arguments: { job_id: 'bash-1', wait: true }, + agent: owner, + }) + await tick() + aborter.abort() + const result = await pending + expect(result.isError).toBe(true) + expect(text(result)).toContain('wait aborted') + + p.settle({ status: 'completed' }) + await tick() + expect(inject).toHaveBeenCalledTimes(1) + }) + it('drops the notice for unowned jobs without throwing', async () => { const { ctx } = await setup() // Unowned: settles with nobody to notify — nothing throws. @@ -836,31 +985,45 @@ describe('completion notices', () => { await tick() }) - it('does not route an old owner completion notice to a same-session replacement', async () => { + it('delivers to the agent currently registered for the owner session', async () => { const { ctx } = await setup() - // Delivery into a tearing-down owner is a plain inject: the loop has no - // terminal state, so the notice lands in the old owner's (detached) - // session instead of throwing or re-routing. + // The settlement names the owner session; the notice goes to whichever + // agent holds that session when the job settles, which can read the job + // through the same session id. const oldInject = vi.fn() const oldOwner = fakeAgent(ctx, 'shared', { inject: oldInject }) - const p = producer({ owner: oldOwner }) + const p = producer({ owner: oldOwner.id }) ctx.jobs.start(p.spec) detachAgent(oldOwner) const replacementInject = vi.fn() - fakeAgent(ctx, 'shared', { inject: replacementInject }) + const replacement = fakeAgent(ctx, 'shared', { inject: replacementInject }) p.settle({ status: 'completed' }) await tick() - expect(oldInject).toHaveBeenCalledTimes(1) - expect(replacementInject).not.toHaveBeenCalled() + expect(oldInject).not.toHaveBeenCalled() + expect(replacementInject).toHaveBeenCalledTimes(1) + expect(text(await call(ctx, 'job_output', { job_id: 'bash-1' }, replacement))).toBe('(no new output)\n[status: completed]') + }) + + it('drops the notice when the owner session has no live agent at settlement', async () => { + const { ctx } = await setup() + const inject = vi.fn() + const owner = fakeAgent(ctx, 'sess-1', { inject }) + const p = producer({ owner: owner.id }) + ctx.jobs.start(p.spec) + + detachAgent(owner) + p.settle({ status: 'completed' }) + await tick() + expect(inject).not.toHaveBeenCalled() }) 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', { inject: () => { throw new Error('unexpected inject bug') } }) - const p = producer({ owner }) + const p = producer({ owner: owner.id }) ctx.jobs.start(p.spec) p.settle({ status: 'completed' }) await tick() @@ -868,23 +1031,4 @@ describe('completion notices', () => { // registry's per-listener containment — visible, not swallowed. expect(warn).toHaveBeenCalledWith(expect.stringContaining('unexpected inject bug')) }) - - 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 }) - - // Settlement must not depend on a later registry lookup: the exact owner - // supplied at start remains the destination while its own scope is live. - const p1 = producer({ owner }) - ctx.jobs.start(p1.spec) - const p2 = producer({ owner }) - ctx.jobs.start(p2.spec) - - await agentsFiber.dispose() - p1.settle({ status: 'completed' }) - p2.settle({ status: 'failed' }) - await tick() - expect(inject).toHaveBeenCalledTimes(2) - }) })