mirror of
https://github.com/stablyai/orca.git
synced 2026-09-27 16:02:35 +00:00
fix(pi): stop pi-subagents workflows from pinning a pane on working (#22533)
Awaited workflow children announce subagent:async-started but never get a subagent:async-complete, so the #21882 roster kept their ids forever and suppressed every later agent_end. Treat subagent:process-terminal (the runner exiting) as the end signal for tracked runs, with a short grace so runs that report their own completion keep settling through that path. Also accept completion events that carry only runId. Fixes #22527
This commit is contained in:
mmarabel
@@ -0,0 +1,179 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
|
||||
|
||||
import {
|
||||
AGENT_STATUS_EXTENSION_SELF_PID,
|
||||
createAgentStatusExtensionHarness,
|
||||
type AgentStatusExtensionHarness
|
||||
} from './agent-status-extension-test-harness'
|
||||
|
||||
// Event shapes and orderings mirror traces recorded from pi-subagents 0.71.0.
|
||||
const WORKFLOW = 'workflow-1'
|
||||
const idle = { isIdle: () => true }
|
||||
|
||||
function postedHookNames(harness: AgentStatusExtensionHarness): string[] {
|
||||
return harness.fetchMock.mock.calls.map((call) => {
|
||||
const body: { payload?: { hook_event_name?: unknown } } = JSON.parse(String(call[1]?.body))
|
||||
return String(body.payload?.hook_event_name)
|
||||
})
|
||||
}
|
||||
|
||||
function agentEndCount(harness: AgentStatusExtensionHarness): number {
|
||||
return postedHookNames(harness).filter((name) => name === 'agent_end').length
|
||||
}
|
||||
|
||||
function startWorkflow(harness: AgentStatusExtensionHarness): void {
|
||||
harness.emitPiEvent('subagent:async-started', {
|
||||
id: WORKFLOW,
|
||||
mode: 'workflow',
|
||||
pid: AGENT_STATUS_EXTENSION_SELF_PID
|
||||
})
|
||||
}
|
||||
|
||||
function startChild(harness: AgentStatusExtensionHarness, id: string, parent = WORKFLOW): void {
|
||||
harness.emitPiEvent('subagent:async-started', {
|
||||
id,
|
||||
mode: 'single',
|
||||
pid: 4000,
|
||||
parentWorkflowRunId: parent
|
||||
})
|
||||
}
|
||||
|
||||
function exitRunner(harness: AgentStatusExtensionHarness, runId: string): void {
|
||||
harness.emitPiEvent('subagent:process-terminal', { runId, state: 'observed' })
|
||||
}
|
||||
|
||||
function complete(harness: AgentStatusExtensionHarness, id: string): void {
|
||||
harness.emitPiEvent('subagent:async-complete', { id, runId: id, state: 'complete' })
|
||||
}
|
||||
|
||||
async function endTurn(harness: AgentStatusExtensionHarness): Promise<void> {
|
||||
await harness.callHook('agent_end', {}, idle)
|
||||
await harness.callHook('agent_settled', undefined, idle)
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
}
|
||||
|
||||
describe('Pi async subagent roster', () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers()
|
||||
})
|
||||
afterEach(() => {
|
||||
vi.useRealTimers()
|
||||
})
|
||||
|
||||
it('settles after an async workflow whose awaited children never report completion', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
startWorkflow(harness)
|
||||
startChild(harness, 'child-a')
|
||||
startChild(harness, 'child-b')
|
||||
await endTurn(harness)
|
||||
exitRunner(harness, 'child-a')
|
||||
exitRunner(harness, 'child-b')
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
expect(agentEndCount(harness)).toBe(0)
|
||||
|
||||
// pi-subagents wakes the lead just before announcing only the workflow's completion.
|
||||
await harness.callHook('agent_start')
|
||||
complete(harness, WORKFLOW)
|
||||
await endTurn(harness)
|
||||
|
||||
expect(postedHookNames(harness).at(-1)).toBe('agent_end')
|
||||
})
|
||||
|
||||
it('settles as soon as the workflow completes when its children already exited', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
startWorkflow(harness)
|
||||
startChild(harness, 'child-a')
|
||||
await endTurn(harness)
|
||||
exitRunner(harness, 'child-a')
|
||||
complete(harness, WORKFLOW)
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
|
||||
expect(agentEndCount(harness)).toBe(1)
|
||||
})
|
||||
|
||||
it('settles after a foreground workflow whose awaited children never report completion', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
startChild(harness, 'child-a', 'tool-call-1')
|
||||
startChild(harness, 'child-b', 'tool-call-1')
|
||||
exitRunner(harness, 'child-a')
|
||||
exitRunner(harness, 'child-b')
|
||||
await endTurn(harness)
|
||||
|
||||
expect(agentEndCount(harness)).toBe(1)
|
||||
})
|
||||
|
||||
it('settles once a child runner exits after its workflow already completed', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
startWorkflow(harness)
|
||||
startChild(harness, 'child-a')
|
||||
await endTurn(harness)
|
||||
complete(harness, WORKFLOW)
|
||||
exitRunner(harness, 'child-a')
|
||||
await vi.advanceTimersByTimeAsync(500)
|
||||
expect(agentEndCount(harness)).toBe(0)
|
||||
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
expect(agentEndCount(harness)).toBe(1)
|
||||
})
|
||||
|
||||
it('keeps working while an explicit async child outlives its workflow', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
startWorkflow(harness)
|
||||
startChild(harness, 'child-a')
|
||||
complete(harness, WORKFLOW)
|
||||
await endTurn(harness)
|
||||
await vi.advanceTimersByTimeAsync(60_000)
|
||||
expect(agentEndCount(harness)).toBe(0)
|
||||
|
||||
// The child's own completion and the wake turn land right after its runner exits.
|
||||
exitRunner(harness, 'child-a')
|
||||
await vi.advanceTimersByTimeAsync(150)
|
||||
await harness.callHook('agent_start')
|
||||
complete(harness, 'child-a')
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
expect(agentEndCount(harness)).toBe(0)
|
||||
|
||||
await endTurn(harness)
|
||||
expect(agentEndCount(harness)).toBe(1)
|
||||
})
|
||||
|
||||
it('ignores runner exits for runs it is not tracking', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
harness.emitPiEvent('subagent:async-started', { id: 'run-1', mode: 'single', pid: 4000 })
|
||||
await endTurn(harness)
|
||||
exitRunner(harness, 'other-run')
|
||||
harness.emitPiEvent('subagent:process-terminal', {})
|
||||
await vi.advanceTimersByTimeAsync(5_000)
|
||||
|
||||
expect(agentEndCount(harness)).toBe(0)
|
||||
})
|
||||
|
||||
it('accepts completion events that identify the run only by runId', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
harness.emitPiEvent('subagent:async-started', { id: 'run-1', mode: 'single', pid: 4000 })
|
||||
await endTurn(harness)
|
||||
harness.emitPiEvent('subagent:async-complete', { runId: 'run-1' })
|
||||
await vi.advanceTimersByTimeAsync(0)
|
||||
|
||||
expect(agentEndCount(harness)).toBe(1)
|
||||
})
|
||||
|
||||
it('keeps one runner-exit subscription and the roster across reloads', async () => {
|
||||
const harness = createAgentStatusExtensionHarness({ kind: 'pi' })
|
||||
await harness.callHook('agent_start')
|
||||
startChild(harness, 'child-a', 'tool-call-1')
|
||||
harness.reload()
|
||||
expect(harness.piEventListenerCount('subagent:process-terminal')).toBe(1)
|
||||
|
||||
exitRunner(harness, 'child-a')
|
||||
await endTurn(harness)
|
||||
expect(agentEndCount(harness)).toBe(1)
|
||||
})
|
||||
})
|
||||
@@ -4,6 +4,10 @@ import { getAgentStatusInputRedactionSourceLines } from './agent-status-input-re
|
||||
import type { PiAgentKind } from '../../shared/pi-agent-kind'
|
||||
import { getOmpSessionOwnerHandlerSourceLines } from './omp-session-status-owner-source'
|
||||
import { getPiAgentStatusUiPromptHandlerSourceLines } from './agent-status-ui-prompt-source'
|
||||
import {
|
||||
getPiSubagentRosterEventSourceLines,
|
||||
getPiSubagentRosterSetupSourceLines
|
||||
} from './agent-status-subagent-roster-source'
|
||||
|
||||
// Why: keep the generated handler registrations separate from hook transport;
|
||||
// both are independently sizeable and the installed extension concatenates them.
|
||||
@@ -129,19 +133,10 @@ export function getPiAgentStatusHandlerSourceLines(kind: PiAgentKind): string[]
|
||||
' if (ownerPid && ownerPid !== selfPid && isStatusOwnerAlive(ownerPid)) return',
|
||||
` process.env.${ownerEnv} = selfPid`,
|
||||
' resetPostQueue()',
|
||||
' const piEventBus = (pi as { events?: { on?: (name: string, handler: (event: unknown) => void) => void } }).events',
|
||||
' const lifecycleState = (piEventBus as { __orcaPiSubagents?: { active: Set<string>; waiting: boolean; onEvent?: (event: unknown, forcedStatus?: string) => void; listener?: (event: unknown) => void } } | undefined)?.__orcaPiSubagents ?? { active: new Set<string>(), waiting: false }',
|
||||
' if (piEventBus) (piEventBus as { __orcaPiSubagents?: unknown }).__orcaPiSubagents = lifecycleState',
|
||||
' if (piEventBus?.on && !(lifecycleState as { listener?: unknown }).listener) {',
|
||||
' const listener = (event: unknown) => lifecycleState.onEvent?.(event)',
|
||||
' lifecycleState.listener = listener',
|
||||
" piEventBus.on('task:subagent:lifecycle', listener)",
|
||||
" piEventBus.on('subagent:async-started', (event: unknown) => lifecycleState.onEvent?.(event, 'started'))",
|
||||
" piEventBus.on('subagent:async-complete', (event: unknown) => lifecycleState.onEvent?.(event, 'completed'))",
|
||||
' }',
|
||||
...getPiSubagentRosterSetupSourceLines(),
|
||||
...(kind !== 'pi'
|
||||
? [
|
||||
" pi.on('session_shutdown', () => { lifecycleState.active.clear(); lifecycleState.waiting = false; resetPostQueue(); clearPendingAgentEndCheck() })"
|
||||
" pi.on('session_shutdown', () => { lifecycleState.active.clear(); lifecycleState.exited?.clear(); lifecycleState.waiting = false; resetPostQueue(); clearPendingAgentEndCheck() })"
|
||||
]
|
||||
: []),
|
||||
...(kind !== 'prime-agent'
|
||||
@@ -149,6 +144,7 @@ export function getPiAgentStatusHandlerSourceLines(kind: PiAgentKind): string[]
|
||||
" pi.on('session_switch', (_event, ctx) => {",
|
||||
' if (!isOmpRuntime()) return',
|
||||
' lifecycleState.active.clear()',
|
||||
' lifecycleState.exited?.clear()',
|
||||
' lifecycleState.waiting = false',
|
||||
' resetPostQueue()',
|
||||
' clearPendingAgentEndCheck()',
|
||||
@@ -239,25 +235,15 @@ export function getPiAgentStatusHandlerSourceLines(kind: PiAgentKind): string[]
|
||||
' pendingAgentEndCheck = null',
|
||||
' pendingAgentEndContext = null',
|
||||
' }',
|
||||
' // Defer completion while live child work remains.',
|
||||
' lifecycleState.onEvent = (event: unknown, forcedStatus?: string): void => {',
|
||||
" if (!event || typeof event !== 'object') return",
|
||||
" const id = typeof (event as { id?: unknown }).id === 'string' ? (event as { id: string }).id : ''",
|
||||
' const status = forcedStatus ?? (event as { status?: unknown }).status',
|
||||
' if (!id) return',
|
||||
" if (status === 'started') { lifecycleState.active.add(id); post('agent_start'); return }",
|
||||
" if (status !== 'completed' && status !== 'failed' && status !== 'aborted') return",
|
||||
' lifecycleState.active.delete(id)',
|
||||
' if (lifecycleState.active.size === 0 && lifecycleState.waiting) {',
|
||||
' lifecycleState.waiting = false',
|
||||
' postAgentEndOnce()',
|
||||
' }',
|
||||
' }',
|
||||
...getPiSubagentRosterEventSourceLines(),
|
||||
' function postAgentEndOnce(): void {',
|
||||
' for (const id of lifecycleState.exited ?? []) lifecycleState.active.delete(id)',
|
||||
' lifecycleState.exited?.clear()',
|
||||
' if (lifecycleState.active.size > 0) {',
|
||||
' lifecycleState.waiting = true',
|
||||
' return',
|
||||
' }',
|
||||
' lifecycleState.waiting = false',
|
||||
' if (completionPostedGeneration === endedRunGeneration) return',
|
||||
' completionPostedGeneration = endedRunGeneration',
|
||||
// Why: distinct from the completion guard, which holds the generation of the posted run
|
||||
|
||||
@@ -0,0 +1,62 @@
|
||||
// Why: Pi settles its own turn while pi-subagents children keep running, so the
|
||||
// generated extension holds the pane's completion until every child it saw start is gone.
|
||||
|
||||
// The roster lives on pi.events so an in-process /reload keeps children and listeners.
|
||||
export function getPiSubagentRosterSetupSourceLines(): string[] {
|
||||
return [
|
||||
' const piEventBus = (pi as { events?: { on?: (name: string, handler: (event: unknown) => void) => void } }).events',
|
||||
' const lifecycleState = (piEventBus as { __orcaPiSubagents?: { active: Set<string>; exited?: Set<string>; waiting: boolean; onEvent?: (event: unknown, forcedStatus?: string) => void; listener?: (event: unknown) => void; onRunnerExit?: (event: unknown) => void; runnerExitListener?: (event: unknown) => void } } | undefined)?.__orcaPiSubagents ?? { active: new Set<string>(), waiting: false }',
|
||||
' if (piEventBus) (piEventBus as { __orcaPiSubagents?: unknown }).__orcaPiSubagents = lifecycleState',
|
||||
' if (piEventBus?.on && !(lifecycleState as { listener?: unknown }).listener) {',
|
||||
' const listener = (event: unknown) => lifecycleState.onEvent?.(event)',
|
||||
' lifecycleState.listener = listener',
|
||||
" piEventBus.on('task:subagent:lifecycle', listener)",
|
||||
" piEventBus.on('subagent:async-started', (event: unknown) => lifecycleState.onEvent?.(event, 'started'))",
|
||||
" piEventBus.on('subagent:async-complete', (event: unknown) => lifecycleState.onEvent?.(event, 'completed'))",
|
||||
' }',
|
||||
// Why: separate guard so a roster created by an older in-process build still subscribes.
|
||||
' if (piEventBus?.on && !lifecycleState.runnerExitListener) {',
|
||||
' const runnerExitListener = (event: unknown) => lifecycleState.onRunnerExit?.(event)',
|
||||
' lifecycleState.runnerExitListener = runnerExitListener',
|
||||
" piEventBus.on('subagent:process-terminal', runnerExitListener)",
|
||||
' }'
|
||||
]
|
||||
}
|
||||
|
||||
// Expects post() and postAgentEndOnce() from the handler scope; the latter prunes
|
||||
// exited runners before deciding whether children still hold the pane.
|
||||
export function getPiSubagentRosterEventSourceLines(): string[] {
|
||||
return [
|
||||
// Why: a run that reports its own completion does so ~150ms after its runner exits;
|
||||
// the grace lets that path (and the wake turn it triggers) settle the pane first.
|
||||
' const RUNNER_EXIT_GRACE_MS = 2000',
|
||||
' let runnerExitCheck: ReturnType<typeof setTimeout> | null = null',
|
||||
' lifecycleState.onEvent = (event: unknown, forcedStatus?: string): void => {',
|
||||
" if (!event || typeof event !== 'object') return",
|
||||
' const record = event as { id?: unknown; runId?: unknown }',
|
||||
" const id = typeof record.id === 'string' && record.id ? record.id : typeof record.runId === 'string' ? record.runId : ''",
|
||||
' const status = forcedStatus ?? (event as { status?: unknown }).status',
|
||||
' if (!id) return',
|
||||
" if (status === 'started') { lifecycleState.active.add(id); post('agent_start'); return }",
|
||||
" if (status !== 'completed' && status !== 'failed' && status !== 'aborted') return",
|
||||
' lifecycleState.active.delete(id)',
|
||||
' lifecycleState.exited?.delete(id)',
|
||||
' if (lifecycleState.waiting) postAgentEndOnce()',
|
||||
' }',
|
||||
// Why: awaited workflow children never get subagent:async-complete; their runner
|
||||
// exiting is the only end signal pi-subagents publishes for them.
|
||||
' lifecycleState.onRunnerExit = (event: unknown): void => {',
|
||||
" const runId = event && typeof event === 'object' ? (event as { runId?: unknown }).runId : undefined",
|
||||
" if (typeof runId !== 'string' || !lifecycleState.active.has(runId)) return",
|
||||
' if (!lifecycleState.exited) lifecycleState.exited = new Set<string>()',
|
||||
' lifecycleState.exited.add(runId)',
|
||||
' if (!lifecycleState.waiting) return',
|
||||
' if (runnerExitCheck !== null) clearTimeout(runnerExitCheck)',
|
||||
' runnerExitCheck = setTimeout(() => {',
|
||||
' runnerExitCheck = null',
|
||||
' if (lifecycleState.waiting) postAgentEndOnce()',
|
||||
' }, RUNNER_EXIT_GRACE_MS)',
|
||||
" if (typeof runnerExitCheck.unref === 'function') runnerExitCheck.unref()",
|
||||
' }'
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user