mirror of
https://github.com/stablyai/orca.git
synced 2026-10-31 00:57:45 +00:00
Brennan Benson
b1e12d7bb4
* fix(native-chat): a new structured chat's model list comes from the host that runs it agentSession.modelCatalog builds the structured-session host like agentSession.options, so a host with no saved chats since it started answers instead of refusing. When the host answers unknown because its first listing runs in the background, the picker re-reads on a bounded schedule until that listing lands. Local terminal-backed chat skips the structured catalog when a custom launch command is configured. * fix(native-chat): wait on the host's first model listing instead of re-reading on a timer A new structured chat's picker read the host catalog once; on an account the host had never listed, the answer was "unknown" while a background listing ran, and the client re-read on a 1-30 s schedule. Replace the schedule with the host's own completion signal: - The host answers a cold read with `listingInProgress: true` (new optional field) once it has started or joined that listing. A read that passes the new optional `waitForListing` param awaits the same joined listing and answers with it, or a plain "unknown" if it failed. The at-rest options read never waits. A host that predates the field never sends it, so the client never sends the param to a host that would refuse it. - The picker reads once per open and attach; after the host's report it sends one waiting read, with a 90 s client timeout above the slowest listing. While that read is out, the model pill keeps its label but cannot open or be set (typed /model included). An answer, failure, timeout, hide, attach or the provider's own list releases it. - `agentSession.modelCatalog` builds the structured-session host only for a read that names a session (a structured chat). Terminal-backed chat's session-less read keeps the non-building gate, so a desktop that never runs structured chat never opens the session journal. * fix(native-chat): one waiting model-list read per chat, and no late menu open The waiting catalog read was owned by one run of the picker's effect. Attach (a new fence), hide/show or a send re-ran the effect: the cleanup released the model picker onto the built-in list for a round trip, and the new run sent a second waiting read while the first, which cannot be withdrawn, kept a remote call slot until the listing ended. The waiting read now belongs to the chat (runtime target + agent + session): a small registry keeps one in flight per chat, every re-run or remount joins it, and the entry is deleted when the read settles. The picker hold is derived from that entry being in flight, so it lasts across attach and hide/show and ends when the read settles, the provider reports its own list, or the pane switches to another session. Answers still pass the stale and record checks. A bare /model typed while the list loads no longer opens the model menu by itself when the list lands: the menu stays keyed on the request, and only its initial open is suppressed while pending, so the request is spent shut and the end of the pending period never remounts it. * fix(native-chat): release the model picker in the same commit as the host list When the waiting catalog read settled in the chat that started it, the registry dropped its entry and told subscribers first, and the host list was applied a few microtasks later. React committed once with the picker enabled on the built-in list, then again with the host's list. Joiners now hand the registry their apply callback, and the registry runs every joiner (with the answer, or nothing when the read failed or timed out) before it deletes the entry and notifies. The release and the list land in one commit. An effect cleanup leaves the wait instead of flagging itself stale. * fix(runtime): queue model catalog reads in the long-wait lane A model catalog read that waits on a host's first listing replies only when that listing ends, yet it took one of the 8 foreground call slots for its server. Enough chats opened during one cold listing would stall that server's sends and interrupts until a wait settled. agentSession.modelCatalog now joins worktree.rm in the long-wait lane: same concurrency, counted apart from the foreground calls. The queue classifies by method only, and a warm catalog read answers at once, so the whole method moves.
237 lines
8.9 KiB
TypeScript
237 lines
8.9 KiB
TypeScript
import { describe, expect, it, vi } from 'vitest'
|
|
import {
|
|
isBackgroundRuntimeMethod,
|
|
RuntimeRpcCallQueueOverloadError,
|
|
RuntimeRpcCallQueuePool
|
|
} from './runtime-rpc-call-queue'
|
|
|
|
describe('runtime RPC call queue', () => {
|
|
it('classifies per-worktree decoration lookups as background work', () => {
|
|
expect(isBackgroundRuntimeMethod('github.prForBranch')).toBe(true)
|
|
expect(isBackgroundRuntimeMethod('hostedReview.forBranch')).toBe(true)
|
|
expect(isBackgroundRuntimeMethod('worktree.prefetchCreateBase')).toBe(true)
|
|
expect(isBackgroundRuntimeMethod('terminal.send')).toBe(false)
|
|
expect(isBackgroundRuntimeMethod('terminal.agentStatus')).toBe(false)
|
|
expect(isBackgroundRuntimeMethod('worktree.create')).toBe(false)
|
|
})
|
|
|
|
it('limits background calls while allowing foreground runtime work through', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(2, 1)
|
|
const started: string[] = []
|
|
const pending: ((value: string) => void)[] = []
|
|
const enqueuePending = (method: string, label: string): Promise<string> =>
|
|
queue.enqueue('web-runtime', method, async () => {
|
|
started.push(label)
|
|
return await new Promise<string>((resolve) => pending.push(resolve))
|
|
})
|
|
|
|
const background1 = enqueuePending('github.prForBranch', 'background-1')
|
|
const background2 = enqueuePending('hostedReview.forBranch', 'background-2')
|
|
await vi.waitFor(() => expect(started).toEqual(['background-1']))
|
|
|
|
const foreground = queue.enqueue('web-runtime', 'terminal.send', async () => {
|
|
started.push('foreground')
|
|
return 'foreground'
|
|
})
|
|
await expect(foreground).resolves.toBe('foreground')
|
|
expect(started).toEqual(['background-1', 'foreground'])
|
|
|
|
pending.shift()?.('background-1')
|
|
await expect(background1).resolves.toBe('background-1')
|
|
await vi.waitFor(() => expect(started).toEqual(['background-1', 'foreground', 'background-2']))
|
|
|
|
pending.shift()?.('background-2')
|
|
await expect(background2).resolves.toBe('background-2')
|
|
})
|
|
|
|
it('runs foreground calls while worktree deletes wait, still bounding the deletes', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(2, 1)
|
|
const started: string[] = []
|
|
const pending: (() => void)[] = []
|
|
const deletes = ['rm-1', 'rm-2', 'rm-3'].map((label) =>
|
|
queue.enqueue('web-runtime', 'worktree.rm', async () => {
|
|
started.push(label)
|
|
await new Promise<void>((resolve) => pending.push(resolve))
|
|
return label
|
|
})
|
|
)
|
|
await vi.waitFor(() => expect(started).toEqual(['rm-1', 'rm-2']))
|
|
|
|
const listing = queue.enqueue('web-runtime', 'worktree.list', async () => 'listed')
|
|
await expect(listing).resolves.toBe('listed')
|
|
expect(started).toEqual(['rm-1', 'rm-2'])
|
|
|
|
pending.shift()?.()
|
|
await expect(deletes[0]).resolves.toBe('rm-1')
|
|
await vi.waitFor(() => expect(started).toEqual(['rm-1', 'rm-2', 'rm-3']))
|
|
pending.shift()?.()
|
|
pending.shift()?.()
|
|
await expect(Promise.all(deletes)).resolves.toEqual(['rm-1', 'rm-2', 'rm-3'])
|
|
})
|
|
|
|
it('runs foreground calls while model catalog reads wait on a host listing', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(8, 2)
|
|
const release: (() => void)[] = []
|
|
const waits = Array.from({ length: 8 }, () =>
|
|
queue.enqueue('web-runtime', 'agentSession.modelCatalog', async () => {
|
|
await new Promise<void>((resolve) => release.push(resolve))
|
|
return 'listed'
|
|
})
|
|
)
|
|
await vi.waitFor(() => expect(release).toHaveLength(8))
|
|
|
|
const send = queue.enqueue('web-runtime', 'agentSession.send', async () => 'sent')
|
|
await expect(send).resolves.toBe('sent')
|
|
|
|
for (const resolve of release) {
|
|
resolve()
|
|
}
|
|
await expect(Promise.all(waits)).resolves.toEqual(Array(8).fill('listed'))
|
|
})
|
|
|
|
it('frees the queue slot when a runtime call throws synchronously', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(1, 1)
|
|
const first = queue.enqueue('web-runtime', 'status.get', () => {
|
|
throw new Error('invalid stored runtime pairing')
|
|
})
|
|
|
|
await expect(first).rejects.toThrow('invalid stored runtime pairing')
|
|
|
|
const second = queue.enqueue('web-runtime', 'status.get', async () => 'second')
|
|
await expect(second).resolves.toBe('second')
|
|
})
|
|
|
|
it('removes an aborted call before it starts', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(1, 1)
|
|
let releaseFirst: () => void = () => {}
|
|
const first = queue.enqueue('runtime-a', 'status.get', async () => {
|
|
await new Promise<void>((resolve) => {
|
|
releaseFirst = resolve
|
|
})
|
|
})
|
|
const controller = new AbortController()
|
|
const run = vi.fn(async () => 'cancelled')
|
|
const cancelled = queue.enqueue('runtime-a', 'status.get', run, 1, controller.signal)
|
|
|
|
controller.abort()
|
|
await expect(cancelled).rejects.toMatchObject({ name: 'AbortError' })
|
|
expect(run).not.toHaveBeenCalled()
|
|
|
|
releaseFirst()
|
|
await expect(first).resolves.toBeUndefined()
|
|
await expect(
|
|
queue.enqueue('runtime-a', 'status.get', async () => 'recovered', 1)
|
|
).resolves.toBe('recovered')
|
|
})
|
|
|
|
it('preserves queued background ordering across large bursts', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(1, 1)
|
|
const started: number[] = []
|
|
let releaseFirst: () => void = () => {}
|
|
const first = queue.enqueue('web-runtime', 'github.prForBranch', async () => {
|
|
started.push(0)
|
|
await new Promise<void>((resolve) => {
|
|
releaseFirst = resolve
|
|
})
|
|
return 0
|
|
})
|
|
const rest = Array.from({ length: 70 }, (_, index) =>
|
|
queue.enqueue('web-runtime', 'github.prForBranch', async () => {
|
|
const value = index + 1
|
|
started.push(value)
|
|
return value
|
|
})
|
|
)
|
|
|
|
await vi.waitFor(() => expect(started).toEqual([0]))
|
|
releaseFirst()
|
|
|
|
await expect(Promise.all([first, ...rest])).resolves.toEqual(
|
|
Array.from({ length: 71 }, (_, index) => index)
|
|
)
|
|
expect(started).toEqual(Array.from({ length: 71 }, (_, index) => index))
|
|
})
|
|
|
|
it('rejects per-selector overload and accepts work after the queue drains', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(1, 1, 2, 10)
|
|
let releaseFirst: () => void = () => {}
|
|
const first = queue.enqueue('runtime-a', 'status.get', async () => {
|
|
await new Promise<void>((resolve) => {
|
|
releaseFirst = resolve
|
|
})
|
|
return 'first'
|
|
})
|
|
const second = queue.enqueue('runtime-a', 'status.get', async () => 'second')
|
|
const third = queue.enqueue('runtime-a', 'status.get', async () => 'third')
|
|
|
|
await expect(queue.enqueue('runtime-a', 'status.get', async () => 'overflow')).rejects.toEqual(
|
|
expect.objectContaining({
|
|
code: 'runtime_rpc_queue_overloaded',
|
|
scope: 'selector'
|
|
})
|
|
)
|
|
|
|
releaseFirst()
|
|
await expect(Promise.all([first, second, third])).resolves.toEqual(['first', 'second', 'third'])
|
|
await expect(queue.enqueue('runtime-a', 'status.get', async () => 'recovered')).resolves.toBe(
|
|
'recovered'
|
|
)
|
|
})
|
|
|
|
it('caps queued calls across selectors and recovers after draining', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(1, 1, 10, 2)
|
|
const releases: (() => void)[] = []
|
|
const blockers = ['runtime-a', 'runtime-b'].map((selector) =>
|
|
queue.enqueue(selector, 'status.get', async () => {
|
|
await new Promise<void>((resolve) => releases.push(resolve))
|
|
})
|
|
)
|
|
const queuedA = queue.enqueue('runtime-a', 'status.get', async () => 'queued-a')
|
|
const queuedB = queue.enqueue('runtime-b', 'status.get', async () => 'queued-b')
|
|
|
|
const overload = queue.enqueue('runtime-c', 'status.get', async () => 'overflow')
|
|
await expect(overload).rejects.toBeInstanceOf(RuntimeRpcCallQueueOverloadError)
|
|
await expect(overload).rejects.toMatchObject({ scope: 'global' })
|
|
|
|
releases.splice(0).forEach((release) => release())
|
|
await expect(Promise.all([...blockers, queuedA, queuedB])).resolves.toEqual([
|
|
undefined,
|
|
undefined,
|
|
'queued-a',
|
|
'queued-b'
|
|
])
|
|
await expect(queue.enqueue('runtime-c', 'status.get', async () => 'recovered')).resolves.toBe(
|
|
'recovered'
|
|
)
|
|
})
|
|
|
|
it('caps retained call bytes across active and queued work, then recovers', async () => {
|
|
const queue = new RuntimeRpcCallQueuePool(1, 1, 10, 10, 10)
|
|
let releaseFirst: () => void = () => {}
|
|
let firstStarted = false
|
|
const first = queue.enqueue(
|
|
'runtime-a',
|
|
'status.get',
|
|
async () => {
|
|
firstStarted = true
|
|
await new Promise<void>((resolve) => {
|
|
releaseFirst = resolve
|
|
})
|
|
return 'first'
|
|
},
|
|
10
|
|
)
|
|
|
|
await vi.waitFor(() => expect(firstStarted).toBe(true))
|
|
await expect(
|
|
queue.enqueue('runtime-b', 'status.get', async () => 'overflow', 1)
|
|
).rejects.toMatchObject({ scope: 'memory' })
|
|
|
|
releaseFirst()
|
|
await expect(first).resolves.toBe('first')
|
|
await expect(
|
|
queue.enqueue('runtime-b', 'status.get', async () => 'recovered', 10)
|
|
).resolves.toBe('recovered')
|
|
})
|
|
})
|