fix(tasks): correct the change-feed contract and its documentation

Review found the `onTasksChanged` teardown reasoning inverted. The comment
claimed every registration is an effect on the registry's own fiber, so
listeners would be gone before service disposal empties the store — but the
traceable proxy rebinds `this.ctx` to the CALLER, which this package's own
HMR-safety test already proves. The only shipped consumer registers from the
api-proxy mux stream, so it was still listening and simply kept the rows it
last received. Service disposal now announces the emptied set, and teardown
announces its stopping transition immediately instead of leaving an observer
on `running` for however long a slow producer takes to release.

Two documentation claims were false in the opposite direction: the Agent Note
and the ui-task README both said an unowned task is invisible in the header,
while `list(caller)` returns unowned tasks to every caller, the carrier fans
their changes out to every subscribed session, and this PR's own test asserts
exactly that. The note even contradicted itself two sections earlier. Both
sides now state the real asymmetries — another session's tasks, and the
process-local registry emptying on restart.

The "no Web path calls the consuming `ctx.tasks.read()`" invariant claimed a
test that did not exist; the carrier suite's producer had no `readOutput` at
all, so a stray read would have failed nothing. Its producer now counts cursor
consumption and the lifecycle and baseline paths both assert zero.

Also: a session created after the mux opened now receives the task baseline it
missed, the popover samples its clock when it opens rather than at mount, and
a failed task's unbounded producer detail elides instead of widening the row.
This commit is contained in:
Yichen Jiang
2026-08-10 13:37:18 +08:00
parent 0a0a75730f
commit 15d452d7ea
21 changed files with 192 additions and 29 deletions

View File

@@ -3111,6 +3111,14 @@ export function createApiProxy(ctx: Context, defaults: ApiProxyDefaults): ApiPro
}),
ctx.on('session/created', (session: Session) => {
subscribeSession(queue, session)
// The subscribe frame clears the client's task mirror, and a
// session born after the stream opened missed the baseline loop.
// Unowned tasks are visible to it from birth, so without this it
// would show none until the next registry change.
const views = tasks === undefined ? [] : taskViews(tasks.list(ctx.agents.get(session.id)))
if (views.length > 0) {
queue.push(frame({ type: 'session/tasks', sessionId: session.id, tasks: views }))
}
}),
ctx.on('session/disposed', (session: Session) => {
openCalls.delete(session.id)

View File

@@ -29,15 +29,19 @@ type TaskFrame = Extract<MuxFrame, { type: 'session/tasks' }>
*/
function producer(label = 'sleep 60') {
let settle!: (outcome: TaskOutcome) => void
// A stream producer, so the carrier CAN consume the cursor if it ever calls
// `read()`; `reads` is what proves it never does.
const reads = { count: 0 }
const spec = {
kind: 'bash' as const,
label,
run: () => ({
cancel: () => {},
done: new Promise<TaskOutcome>((resolve) => { settle = resolve }),
readOutput: () => { reads.count += 1; return 'stolen output' },
}),
}
return { spec, settle: (outcome: TaskOutcome) => { settle(outcome) } }
return { spec, reads, settle: (outcome: TaskOutcome) => { settle(outcome) } }
}
async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: Session; agent: Agent }> {
@@ -204,3 +208,56 @@ describe('session/tasks without the registry', () => {
expect(frames.some(frame => frame.type === 'session/tasks')).toBe(false)
})
})
describe('session/tasks never consumes model output', () => {
it('drives the whole lifecycle without calling the single consuming cursor', async () => {
// `ctx.tasks.read()` consumes the one output cursor, so a carrier read
// silently takes bytes the model's `task_output` will never see. The
// failure is invisible at the call site, which is why this asserts the
// count rather than trusting review.
const { ctx, agent } = await harness(true)
const proxy = api(ctx)
const abort = new AbortController()
const stream = proxy.events.mux({ rpcId: RpcId('t-tasks-no-read'), payload: {} }, abort.signal)
const collected = collect(stream, 3, abort)
const p = producer()
const id = ctx.tasks.start({ ...p.spec, owner: agent })
ctx.tasks.kill(id, agent, 'test')
p.settle({ status: 'killed', detail: 'signal: SIGTERM' })
await collected
expect(p.reads.count).toBe(0)
})
it('reads nothing while minting the subscription baseline either', async () => {
const { ctx, agent } = await harness(true)
const p = producer()
ctx.tasks.start({ ...p.spec, owner: agent })
const abort = new AbortController()
const stream = api(ctx).events.mux({ rpcId: RpcId('t-tasks-no-read-baseline'), payload: {} }, abort.signal)
const [baseline] = await collect(stream, 1, abort)
expect(baseline?.tasks).toHaveLength(1)
expect(p.reads.count).toBe(0)
})
})
describe('session/tasks baseline for a session born after the stream opened', () => {
it('carries the already-visible unowned set to the new session', async () => {
const { ctx } = await harness(true)
const proxy = api(ctx)
const abort = new AbortController()
const stream = proxy.events.mux({ rpcId: RpcId('t-tasks-late-session'), payload: {} }, abort.signal)
// One unowned task exists before the new session is created; the subscribe
// frame clears the client mirror, so the baseline has to follow it.
ctx.tasks.start(producer('visible to every caller').spec)
const created = ctx.sessions.create()
const frames = await collect(stream, 2, abort)
const forNew = frames.filter(frame => frame.sessionId === created.id)
expect(forNew.at(-1)?.tasks[0]?.label).toBe('visible to every caller')
})
})