mirror of
https://github.com/stablyai/orca.git
synced 2026-11-13 17:37:40 +00:00
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
This commit is contained in:
Jinwoo Hong
+8
-1
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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<number> => {
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user