From d7ea02de688d016e764237e5dc9ef0fb560dbbf6 Mon Sep 17 00:00:00 2001 From: Jinwoo Hong <73622457+Jinwoo-H@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:25:15 -0500 Subject: [PATCH] Stop listing a worker as retained after its follow-up took its terminal (#36978) * Stop listing a worker as retained after its follow-up took its terminal * Read handed-on terminal history only for the listed workers * Detect a handed-on terminal from follow-up lineage instead of owner history * Count a terminal as handed on only when the follow-up runs in the same process --- .../worker-terminal-inventory-counts.ts | 9 +- .../worker-terminal-listing.ts | 9 +- .../worker/worker-follow-up.test.ts | 60 ++++++++ .../worker/worker-list-pagination.test.ts | 132 ++++++++++++++++++ 4 files changed, 206 insertions(+), 4 deletions(-) diff --git a/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-inventory-counts.ts b/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-inventory-counts.ts index 9fc45231426..1441005f0a0 100644 --- a/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-inventory-counts.ts +++ b/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-inventory-counts.ts @@ -43,6 +43,12 @@ function buildInventoryScope(params: WorkerTerminalInventoryParams): { return { where, values } } +/** With no terminal of its own left, a Dispatch whose follow-up runs in its exact process handed that terminal on. */ +export const WORKER_TERMINAL_HANDED_ON_SQL = `r.id IS NULL AND EXISTS ( + SELECT 1 FROM dispatch_contexts f + WHERE f.follow_up_of_dispatch_id = d.id + AND f.process_incarnation = d.process_incarnation AND f.host_scope IS d.host_scope)` + /** Derive all filtering/counting through the single state machine; newest-first order supports paging. */ export function scanWorkerTerminalStates( this: OrchestrationDb, @@ -55,7 +61,8 @@ export function scanWorkerTerminalStates( `SELECT d.id AS dispatch_id, d.rowid AS database_id, COALESCE(w.state, 'unsupervised') AS worker_state, - COALESCE(w.agent_terminal_handle, d.assignee_handle) AS agent_terminal_handle, + CASE WHEN ${WORKER_TERMINAL_HANDED_ON_SQL} THEN NULL + ELSE COALESCE(w.agent_terminal_handle, d.assignee_handle) END AS agent_terminal_handle, r.id AS resource_id, r.ownership_state, r.release_state FROM dispatch_contexts d LEFT JOIN worker_dispatches w ON w.dispatch_id = d.id diff --git a/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-listing.ts b/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-listing.ts index faf41369670..ea484a9df58 100644 --- a/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-listing.ts +++ b/src/main/runtime/orchestration/db/worker-terminal/worker-terminal-listing.ts @@ -14,7 +14,8 @@ import { } from './worker-terminal-attention-query' import { countWorkerTerminalInventory, - scanWorkerTerminalStates + scanWorkerTerminalStates, + WORKER_TERMINAL_HANDED_ON_SQL } from './worker-terminal-inventory-counts' import { allowWorkerTerminalAutoClose, @@ -168,8 +169,10 @@ export function listWorkerTerminalResources( d.rowid AS database_id, ${orderExpression} AS created_at, COALESCE(w.state, 'unsupervised') AS worker_state, - COALESCE(w.agent_terminal_handle, d.assignee_handle) AS agent_terminal_handle, - COALESCE(r.pane_key, d.assignee_pane_key) AS pane_key, + CASE WHEN ${WORKER_TERMINAL_HANDED_ON_SQL} THEN NULL + ELSE COALESCE(w.agent_terminal_handle, d.assignee_handle) END AS agent_terminal_handle, + CASE WHEN ${WORKER_TERMINAL_HANDED_ON_SQL} THEN NULL + ELSE COALESCE(r.pane_key, d.assignee_pane_key) END AS pane_key, COALESCE(w.worktree_id, r.worktree_id) AS worktree_id, w.stage AS worker_stage, t.parent_id AS parent_task_id, diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-follow-up.test.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-follow-up.test.ts index 67f3668fe4b..5fcc1b9f42a 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-follow-up.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-follow-up.test.ts @@ -408,6 +408,66 @@ describe('a follow-up to a finished terminal worker', () => { }) }) + it('lists the finished worker without the terminal its follow-up took over', async () => { + h.setup() + const { dispatchId } = await h.startSettledWorker() + await send(dispatchId, 'and the docs') + const followUpId = getFollowUpDispatch(h.db, dispatchId)!.id + expect(h.db.getWorkerTerminalResourceByOwner(followUpId)?.terminal_handle).toBe('term_worker') + + await expect(h.call('orchestration.workerList', { run: h.activeRunId })).resolves.toMatchObject( + { + workers: expect.arrayContaining([ + expect.objectContaining({ + dispatchId, + agentTerminalHandle: null, + terminalState: null, + projection: expect.objectContaining({ + // Nothing proves the original's process ended, so liveness stays unverifiable. + liveness: { verdict: 'unverifiable', reason: 'missing_status' }, + nextAction: expect.objectContaining({ kind: 'none' }) + }) + }), + expect.objectContaining({ + dispatchId: followUpId, + agentTerminalHandle: 'term_worker', + terminalState: 'active' + }) + ]), + counts: { active: 1 } + } + ) + + // A -> B -> C, then C's release: neither earlier row claims the terminal. + const b = getFollowUpDispatch(h.db, dispatchId)! + h.settle(b.task_id, b.id, 'succeeded') + await send(b.id, 'and the tests') + const c = getFollowUpDispatch(h.db, b.id)! + h.settle(c.task_id, c.id, 'succeeded') + await expect(h.call('orchestration.workerRelease', { dispatch: c.id })).resolves.toMatchObject({ + state: 'released' + }) + const handedOn = (id: string) => + expect.objectContaining({ dispatchId: id, agentTerminalHandle: null, terminalState: null }) + await expect(h.call('orchestration.workerList', { run: h.activeRunId })).resolves.toMatchObject( + { + workers: [ + expect.objectContaining({ + dispatchId: c.id, + agentTerminalHandle: 'term_worker', + terminalState: 'released' + }), + handedOn(b.id), + handedOn(dispatchId) + ], + counts: { released: 1 } + } + ) + await expect( + h.call('orchestration.workerList', { run: h.activeRunId, terminalState: 'retained' }) + ).resolves.toMatchObject({ workers: [] }) + }) + it('takes mail for a working worker even when its report settles the Dispatch in the same turn', async () => { h.setup() const { taskId, dispatchId } = await h.startWorker() diff --git a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-pagination.test.ts b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-pagination.test.ts index 5be08bde82d..2fa2d87da4f 100644 --- a/src/main/runtime/rpc/methods/orchestration/worker/worker-list-pagination.test.ts +++ b/src/main/runtime/rpc/methods/orchestration/worker/worker-list-pagination.test.ts @@ -417,6 +417,138 @@ describe('orchestration worker-list pagination', () => { expect(prepare).toHaveBeenCalledTimes(6) }) + it('fetches the same rows for one worker however many unrelated terminals were handed on', async () => { + const fetchedBeside = async (unrelated: number): Promise => { + db = new OrchestrationDb(':memory:') + const runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + const run = db.createRun({ + objective: 'Handed-on history read cost', + coordinatorHandle: 'term-coordinator', + coordinatorPaneKey: 'tab-coordinator:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + }) + insertWorkerInventory(db, run.id, 'dispatch-listed', 'succeeded', 'not_requested') + const sqlite = sqliteFor(db) + const insert = sqlite.prepare( + `INSERT INTO worker_terminal_resources ( + id, origin_dispatch_id, owner_dispatch_id, prior_owner_dispatch_ids, terminal_handle + ) VALUES (?, ?, ?, ?, ?)` + ) + for (let index = 0; index < unrelated; index += 1) { + insert.run( + `resource-${index}`, + `a-${index}`, + `b-${index}`, + `["a-${index}"]`, + `term-${index}` + ) + } + const prepare = sqlite.prepare.bind(sqlite) + let fetched = 0 + vi.spyOn(sqlite, 'prepare').mockImplementation((sql) => { + const statement = prepare(sql) + const all = statement.all.bind(statement) + statement.all = (...params) => { + const rows = all(...params) + fetched += rows.length + return rows + } + return statement + }) + await callWorkerList(runtime, { run: run.id, limit: 100 }) + // worker-show reads the same listing for its one Dispatch. + db.listWorkerTerminalResources({ dispatchIds: ['dispatch-listed'], limit: 1 }) + vi.mocked(sqlite.prepare).mockRestore() + return fetched + } + + const alone = await fetchedBeside(0) + db?.close() + expect(await fetchedBeside(10_000)).toBe(alone) + }) + + it('keeps a resource-less worker whose follow-up started fresh in another terminal', async () => { + db = new OrchestrationDb(':memory:') + const runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + const run = db.createRun({ + objective: 'Fresh follow-up of a legacy worker', + coordinatorHandle: 'term-coordinator', + coordinatorPaneKey: 'tab-coordinator:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + }) + const sqlite = sqliteFor(db) + const link = (followUp: string, of: string, incarnation: string) => { + sqlite + .prepare( + 'UPDATE dispatch_contexts SET follow_up_of_dispatch_id = ?, process_incarnation = ? WHERE id = ?' + ) + .run(of, incarnation, followUp) + } + // Legacy Dispatches with no resource: one continued in a fresh terminal, one in its own process. + for (const id of ['dispatch-fresh', 'dispatch-same']) { + insertDispatch(db, run.id, id) + sqlite + .prepare('UPDATE dispatch_contexts SET process_incarnation = ? WHERE id = ?') + .run(`incarnation-${id}`, id) + } + insertWorkerInventory(db, run.id, 'dispatch-fresh-next', 'ready', 'not_requested') + link('dispatch-fresh-next', 'dispatch-fresh', 'incarnation-other') + insertWorkerInventory(db, run.id, 'dispatch-same-next', 'ready', 'not_requested') + link('dispatch-same-next', 'dispatch-same', 'incarnation-dispatch-same') + + const page = await callWorkerList(runtime, { run: run.id, limit: 100 }) + + expect(page.workers).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + dispatchId: 'dispatch-fresh', + agentTerminalHandle: 'term-dispatch-fresh', + terminalState: 'retained' + }), + expect.objectContaining({ + dispatchId: 'dispatch-same', + agentTerminalHandle: null, + terminalState: null + }) + ]) + ) + expect(page.counts).toEqual({ active: 2, retained: 1 }) + }) + + it('finds a handed-on terminal through the follow-up index without a table scan', async () => { + db = new OrchestrationDb(':memory:') + const runtime = new OrcaRuntimeService() + runtime.setOrchestrationDb(db) + const run = db.createRun({ + objective: 'Handed-on lookup plan', + coordinatorHandle: 'term-coordinator', + coordinatorPaneKey: 'tab-coordinator:aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa' + }) + insertWorkerInventory(db, run.id, 'dispatch-listed', 'succeeded', 'not_requested') + const sqlite = sqliteFor(db) + const prepare = sqlite.prepare.bind(sqlite) + const lookups: string[] = [] + vi.spyOn(sqlite, 'prepare').mockImplementation((sql) => { + if (sql.includes('follow_up_of_dispatch_id')) { + lookups.push(sql) + } + return prepare(sql) + }) + await callWorkerList(runtime, { run: run.id, limit: 100 }) + vi.mocked(sqlite.prepare).mockRestore() + + // The listing and the counts both ask. + expect(lookups).toHaveLength(2) + for (const sql of lookups) { + const plan = sqlite + .prepare(`EXPLAIN QUERY PLAN ${sql}`) + .all() + .map((step) => String(step.detail)) + expect(plan.filter((detail) => detail.startsWith('SCAN '))).toEqual([]) + expect(plan).toContainEqual(expect.stringContaining('idx_dispatch_contexts_follow_up_of')) + } + }) + it('aggregates exact inventory counts while preserving filtered totals', async () => { db = new OrchestrationDb(':memory:') const runtime = new OrcaRuntimeService()