mirror of
https://github.com/stablyai/orca.git
synced 2026-11-14 18:07:45 +00:00
perf(native-chat): give the old agent 1 s to exit on /clear (#40103)
A process in Claude's group that ignored SIGTERM held /clear for the full 3 s grace; on /clear the supervisor now reaps the group after 1 s. Linear: STA-14361
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
import { describe, expect, it, vi } from 'vitest'
|
||||
import type { ClaudeStreamJsonConnection } from './claude-stream-json-connection'
|
||||
import { createClaudeNamingGeneration } from './claude-naming-generation'
|
||||
|
||||
describe('createClaudeNamingGeneration', () => {
|
||||
it("passes a close's options through to the connection it watches", async () => {
|
||||
const close = vi.fn(async () => true)
|
||||
// oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: attach and the close wrapper read only close and exitVerdict.
|
||||
const connection = {
|
||||
close,
|
||||
exitVerdict: { root: 'exited' }
|
||||
} as unknown as ClaudeStreamJsonConnection
|
||||
createClaudeNamingGeneration('chat', undefined, 'native', () => {}).attach(connection)
|
||||
|
||||
await connection.close({ grace: 'brief' })
|
||||
|
||||
expect(close).toHaveBeenCalledWith({ grace: 'brief' })
|
||||
})
|
||||
})
|
||||
@@ -55,9 +55,9 @@ export function createClaudeNamingGeneration(
|
||||
generation.phase = 'initializing'
|
||||
}
|
||||
const close = connection.close
|
||||
connection.close = () => {
|
||||
connection.close = (...args) => {
|
||||
generation.phase = 'closing'
|
||||
const done = close.call(connection)
|
||||
const done = close.apply(connection, args)
|
||||
generation.closing = done
|
||||
notify()
|
||||
return done.finally(() => {
|
||||
|
||||
@@ -24,6 +24,7 @@ import type { ClaudeStructuredSdkOptions } from './structured/launch/claude-stru
|
||||
import { withProviderSpawnAbsence } from '../provider-process/provider-executable-missing'
|
||||
import { claudeStreamJsonExitError } from './claude-stream-json-exit-error'
|
||||
import { withObservedProviderExit } from '../native-chat/agent-session-wire/startup/structured-agent-session-failure-text'
|
||||
import type { ProviderProcessCloseOptions } from '../provider-process/provider-process-close'
|
||||
|
||||
export { ClaudeControlRequestError }
|
||||
|
||||
@@ -95,7 +96,7 @@ export type ClaudeStreamJsonConnection = ClaudeControlSurface & {
|
||||
resumeReading?: () => void
|
||||
send: (message: Record<string, unknown>, beforeDispatch?: () => Promise<void>) => Promise<void>
|
||||
/** Resolves true after processless settlement, or root exit plus observed tree exit. */
|
||||
close: () => Promise<boolean>
|
||||
close: (options?: ProviderProcessCloseOptions) => Promise<boolean>
|
||||
}
|
||||
|
||||
export async function openClaudeStreamJsonConnection(
|
||||
@@ -270,14 +271,14 @@ export async function openClaudeStreamJsonConnection(
|
||||
return inbox.push(message as unknown as SDKUserMessage, beforeDispatch)
|
||||
}
|
||||
|
||||
const close = (): Promise<boolean> => {
|
||||
const close = (options?: ProviderProcessCloseOptions): Promise<boolean> => {
|
||||
closePromise ??= (async () => {
|
||||
closing = true
|
||||
resumeReading()
|
||||
// Capture before stopping: exited roots can leave reparented descendants undiscoverable.
|
||||
await (tree.refresh?.() ?? tree.capture())
|
||||
inbox.end()
|
||||
const result = await managed.closeWriter(tree)
|
||||
const result = await managed.closeWriter(tree, options)
|
||||
const proven = claudeChildCloseProven(result) || result.killDelivered === true
|
||||
inbox.fail(new Error('claude stream-json connection closed'))
|
||||
if (!proven) {
|
||||
|
||||
@@ -11,7 +11,7 @@ import {
|
||||
PROVIDER_STDIN_END_GRACE_MS,
|
||||
PROVIDER_SUPERVISOR_MAX_STOP_MS
|
||||
} from '../provider-process/provider-process-supervisor'
|
||||
import { proveClaudeChildExit } from './claude-agent-sdk-exit-proof'
|
||||
import { createClaudeChildTreeReaper, proveClaudeChildExit } from './claude-agent-sdk-exit-proof'
|
||||
import { createClaudeCodeProcessSpawn } from './claude-agent-sdk-process-spawn'
|
||||
|
||||
// EOF leaves the turn running; SIGTERM reaps a detached tool but preserves a double-forked daemon.
|
||||
@@ -195,6 +195,26 @@ describe.runIf(process.platform !== 'win32')('Claude under the POSIX provider su
|
||||
expect(alive(pids.claude)).toBe(false)
|
||||
})
|
||||
|
||||
it('forces a Claude that ignores SIGTERM soon after a brief close, with its exit proven', async () => {
|
||||
const { managed, exit, pids } = await spawnClaude({ ORCA_TEST_CLAUDE_IGNORES_SIGTERM: '1' })
|
||||
const tree = createClaudeChildTreeReaper(managed.child, {
|
||||
exited: () => managed.rootVerdict === 'exited'
|
||||
})
|
||||
|
||||
// ast-grep-ignore: test-no-wall-clock -- the brief grace is the behaviour under test
|
||||
const startedAt = Date.now()
|
||||
await expect(managed.close(tree, { grace: 'brief' })).resolves.toEqual({
|
||||
root: 'exited',
|
||||
tree: 'exited'
|
||||
})
|
||||
|
||||
// ast-grep-ignore: test-no-wall-clock -- the full SIGTERM grace is 3 s; a brief close must not wait it out
|
||||
expect(Date.now() - startedAt).toBeLessThan(2_500)
|
||||
await expect(exit).resolves.toEqual({ code: null, signal: 'SIGTERM', processless: false })
|
||||
expect(alive(pids.claude)).toBe(false)
|
||||
expect(alive(pids.tool)).toBe(false)
|
||||
})
|
||||
|
||||
it("stops Claude and its tool when Orca's main dies, and leaves a daemon that left its tree alone", async () => {
|
||||
const specs: ProcessSpec[] = []
|
||||
createClaudeCodeProcessSpawn((spec) => {
|
||||
|
||||
@@ -10,19 +10,23 @@ import type {
|
||||
ClaudeSessionExit,
|
||||
ClaudeStructuredSessionAdapterDeps
|
||||
} from './claude-structured-session-state'
|
||||
import type { ProviderProcessCloseOptions } from '../../provider-process/provider-process-close'
|
||||
|
||||
export type ClaudeReleaseRequest = { sessionId: string; closeOptions?: ProviderProcessCloseOptions }
|
||||
|
||||
/** An observed exit still requires connection cleanup; absence alone cannot prove release. */
|
||||
export async function releaseClaudeAcquisition(input: {
|
||||
sessionId: string
|
||||
sessions: Map<string, ClaudeSession>
|
||||
acquisitions: ClaudeAcquisitionRegistry
|
||||
exits: Map<string, ClaudeSessionExit>
|
||||
onExitProven?: (sessionId: string, exit: ClaudeSessionExit) => Promise<void>
|
||||
persistHandle?: ClaudeStructuredSessionAdapterDeps['persistHandle']
|
||||
onEvent?: ClaudeStructuredSessionAdapterDeps['onEvent']
|
||||
logger?: ClaudeStructuredSessionAdapterDeps['logger']
|
||||
settledExitErrors?: Map<string, Error>
|
||||
}): Promise<boolean> {
|
||||
export async function releaseClaudeAcquisition(
|
||||
input: ClaudeReleaseRequest & {
|
||||
sessions: Map<string, ClaudeSession>
|
||||
acquisitions: ClaudeAcquisitionRegistry
|
||||
exits: Map<string, ClaudeSessionExit>
|
||||
onExitProven?: (sessionId: string, exit: ClaudeSessionExit) => Promise<void>
|
||||
persistHandle?: ClaudeStructuredSessionAdapterDeps['persistHandle']
|
||||
onEvent?: ClaudeStructuredSessionAdapterDeps['onEvent']
|
||||
logger?: ClaudeStructuredSessionAdapterDeps['logger']
|
||||
settledExitErrors?: Map<string, Error>
|
||||
}
|
||||
): Promise<boolean> {
|
||||
const exit = input.exits.get(input.sessionId)
|
||||
if (!exit || input.sessions.has(input.sessionId) || input.acquisitions.get(input.sessionId)) {
|
||||
return closeClaudeSession(input)
|
||||
|
||||
@@ -9,7 +9,10 @@ import type {
|
||||
} from '../../native-chat/agent-session-wire/providers/structured-agent-session-adapter'
|
||||
import { stopCurrentClaudeBackgroundTasks } from './claude-structured-control-actions'
|
||||
import { claudeHeldSends, claudeHoldsDispatch } from '../claude-command-lifecycle'
|
||||
import { releaseClaudeAcquisition } from './claude-structured-acquisition-release'
|
||||
import {
|
||||
releaseClaudeAcquisition,
|
||||
type ClaudeReleaseRequest
|
||||
} from './claude-structured-acquisition-release'
|
||||
import { acquireClaudeSession } from './claude-structured-session-acquisition'
|
||||
import { supportsClaudeStructuredLocation } from './launch/claude-structured-location-support'
|
||||
import * as claudeOptions from './launch/claude-structured-options'
|
||||
@@ -97,7 +100,7 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
|
||||
finishClaudeCloseAfterExit({
|
||||
sessions: this.sessions,
|
||||
exits: this.exits,
|
||||
closeExit: this.releaseProviderSession,
|
||||
closeExit: (sessionId) => this.releaseProviderSession({ sessionId }),
|
||||
sessionId,
|
||||
connection: attempt.connection,
|
||||
deps: { ...this.deps, settledExitErrors: this.settledExitErrors },
|
||||
@@ -227,9 +230,9 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
|
||||
})
|
||||
readOptions = (input: { sessionId: string; fence: number }) =>
|
||||
readClaudeStructuredSessionOptions(this.session(input.sessionId))
|
||||
releaseAcquisition = (input: { sessionId: string }): Promise<boolean> => {
|
||||
releaseAcquisition = (input: ClaudeReleaseRequest): Promise<boolean> => {
|
||||
this.endings.retire(input.sessionId)
|
||||
return this.afterClose(input.sessionId, () => this.releaseProviderSession(input.sessionId))
|
||||
return this.afterClose(input.sessionId, () => this.releaseProviderSession(input))
|
||||
}
|
||||
acknowledgeSessionRelease = (sessionId: string): void => {
|
||||
this.settledExitErrors.delete(sessionId)
|
||||
@@ -260,9 +263,9 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
|
||||
}
|
||||
}
|
||||
|
||||
private releaseProviderSession = (sessionId: string): Promise<boolean> =>
|
||||
private releaseProviderSession = (request: ClaudeReleaseRequest): Promise<boolean> =>
|
||||
releaseClaudeAcquisition({
|
||||
sessionId,
|
||||
...request,
|
||||
sessions: this.sessions,
|
||||
acquisitions: this.acquisitions,
|
||||
exits: this.exits,
|
||||
@@ -272,8 +275,12 @@ export class ClaudeStructuredSessionAdapter implements StructuredAgentSessionAda
|
||||
...this.deps
|
||||
})
|
||||
|
||||
closeSession = (sessionId: string): Promise<boolean> =>
|
||||
this.releaseAcquisition({ sessionId }).finally(() => {
|
||||
closeSession: NonNullable<StructuredAgentSessionAdapter['closeSession']> = (
|
||||
sessionId,
|
||||
_,
|
||||
options
|
||||
) =>
|
||||
this.releaseAcquisition({ sessionId, closeOptions: options }).finally(() => {
|
||||
const receipt = this.settledExitErrors.get(sessionId)
|
||||
if (
|
||||
!(
|
||||
|
||||
@@ -21,6 +21,7 @@ import { claudeHeldSends } from '../claude-command-lifecycle'
|
||||
import { settledClaudeTurnEndLeaf } from './claude-structured-resume-point'
|
||||
import { settleClaudeTurnEndWaiters } from '../claude-request-end-wait'
|
||||
import type { StructuredAgentSessionLogger } from '../../native-chat/agent-session-wire/host/structured-agent-session-logger'
|
||||
import type { ProviderProcessCloseOptions } from '../../provider-process/provider-process-close'
|
||||
|
||||
/** Root exit releases the lease even with unverified descendants; a failed spawn proves no exit. */
|
||||
export function claudeRootExitObserved(
|
||||
@@ -105,6 +106,7 @@ type CloseClaudePublishedSessionInput = {
|
||||
onEvent?: (event: ClaudeStructuredSessionEvent) => void
|
||||
logger?: StructuredAgentSessionLogger
|
||||
settledExitErrors?: Map<string, Error>
|
||||
closeOptions?: ProviderProcessCloseOptions
|
||||
}
|
||||
|
||||
async function finalizeClaudePublishedSession(
|
||||
@@ -116,7 +118,7 @@ async function finalizeClaudePublishedSession(
|
||||
for (const prompt of session.prompts.clear()) {
|
||||
prompt.settle(null)
|
||||
}
|
||||
const connectionClosed = await session.connection.close()
|
||||
const connectionClosed = await session.connection.close(input.closeOptions)
|
||||
session.unbindReadingControl?.()
|
||||
const rootExitVerdict = claudeCloseReport(
|
||||
session.connection,
|
||||
|
||||
@@ -330,8 +330,8 @@ export class CodexStructuredSessionAdapter implements SessionAdapter {
|
||||
projectOptions = (sessionId: string, options: Readonly<Record<string, string>>) =>
|
||||
projectCodexSessionOptions(this.sessions.get(sessionId), options)
|
||||
|
||||
closeSession: NonNullable<SessionAdapter['closeSession']> = (...args) =>
|
||||
this.teardown.close(...args)
|
||||
closeSession: NonNullable<SessionAdapter['closeSession']> = (sessionId, observeClose) =>
|
||||
this.teardown.close(sessionId, observeClose)
|
||||
forceCloseSession = (sessionId: string): Promise<boolean> => this.teardown.forceClose(sessionId)
|
||||
disposeSession: NonNullable<SessionAdapter['disposeSession']> = this.closeSession
|
||||
closeAll = (): Promise<void> => this.teardown.closeAll()
|
||||
|
||||
+7
@@ -175,4 +175,11 @@ describe('same-conversation clear', () => {
|
||||
store.getOperationRow(caller.callerKey, params.envelope.clientOperationId)?.outcome.status
|
||||
).toBe('succeeded')
|
||||
})
|
||||
|
||||
it('stops the old agent with a brief grace, since the divider waits on that stop', async () => {
|
||||
await clearCommits()
|
||||
expect(adapter.closeSession).toHaveBeenCalledWith(HOST_TEST_SESSION, undefined, {
|
||||
grace: 'brief'
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
+5
-5
@@ -231,17 +231,17 @@ export class StructuredAgentSessionAdapterRouter implements Adapter {
|
||||
return adapter.copyConversation(input)
|
||||
}
|
||||
|
||||
closeSession: NonNullable<Adapter['closeSession']> = (sessionId, observeClose) =>
|
||||
this.stopSession(sessionId, (adapter) => adapter.closeSession?.(sessionId, observeClose))
|
||||
closeSession: NonNullable<Adapter['closeSession']> = (...args) =>
|
||||
this.stopSession(args[0], (adapter) => adapter.closeSession?.(...args))
|
||||
|
||||
forceCloseSession = (sessionId: string): Promise<boolean> =>
|
||||
this.stopSession(sessionId, (adapter) =>
|
||||
(adapter.forceCloseSession ?? adapter.closeSession)?.call(adapter, sessionId)
|
||||
)
|
||||
|
||||
disposeSession: NonNullable<Adapter['disposeSession']> = (sessionId, observeClose) =>
|
||||
this.stopSession(sessionId, (adapter) =>
|
||||
(adapter.disposeSession ?? adapter.closeSession)?.call(adapter, sessionId, observeClose)
|
||||
disposeSession: NonNullable<Adapter['disposeSession']> = (...args) =>
|
||||
this.stopSession(args[0], (adapter) =>
|
||||
(adapter.disposeSession ?? adapter.closeSession)?.call(adapter, ...args)
|
||||
)
|
||||
|
||||
private async stopSession(
|
||||
|
||||
+11
-2
@@ -1,4 +1,5 @@
|
||||
import type { PrepareStructuredRewind } from './structured-agent-session-rewind-contract'
|
||||
import type { ProviderProcessCloseOptions } from '../../../provider-process/provider-process-close'
|
||||
import type { NativeCopyCoverage } from '../../agent-session-journal/context-epochs/journal-history-origin'
|
||||
import type { StructuredAgentSessionConversationCopyInput } from './structured-agent-session-conversation-copy-input'
|
||||
import type { CodexAccountUsageRead } from '../../../rate-limits/codex-rpc-rate-limit-mapper'
|
||||
@@ -347,11 +348,19 @@ export type StructuredAgentSessionAdapter = StructuredAgentSessionAdapterStop &
|
||||
}): Promise<void> | undefined
|
||||
stopProviderNaming?(): void
|
||||
/** True proves child exit; root/processless verdicts throw only after finalization (stopAgentSessionProviderRoot). */
|
||||
closeSession?(sessionId: string, observeClose?: AgentSessionCloseObserver): Promise<boolean>
|
||||
closeSession?(
|
||||
sessionId: string,
|
||||
observeClose?: AgentSessionCloseObserver,
|
||||
options?: ProviderProcessCloseOptions
|
||||
): Promise<boolean>
|
||||
/** Sink-failure shutdown leaves exit recovery unexpected. */
|
||||
forceCloseSession?(sessionId: string): Promise<boolean>
|
||||
/** Teardown without requiring a future-resume cursor. */
|
||||
disposeSession?(sessionId: string, observeClose?: AgentSessionCloseObserver): Promise<boolean>
|
||||
disposeSession?(
|
||||
sessionId: string,
|
||||
observeClose?: AgentSessionCloseObserver,
|
||||
options?: ProviderProcessCloseOptions
|
||||
): Promise<boolean>
|
||||
/** Acknowledge only after dead-child, lease and journal ownership release. */
|
||||
acknowledgeSessionRelease?(sessionId: string, acquisition?: AcquisitionReceipt): void
|
||||
}
|
||||
|
||||
+9
-3
@@ -24,7 +24,7 @@ export async function joinStructuredAgentSessionChildClose(
|
||||
): Promise<StructuredAgentSessionChildCloseVerdict> {
|
||||
// Capture startup state before provider release loses that evidence.
|
||||
const unanswered = child.phase === 'starting'
|
||||
const closed = await closeProviderRoot(context, sessionId, observeClose)
|
||||
const closed = await closeProviderRoot(context, sessionId, child, observeClose)
|
||||
if (!closed) {
|
||||
return 'unverifiable'
|
||||
}
|
||||
@@ -71,7 +71,7 @@ export async function joinClosingStructuredAgentSessionChild(
|
||||
return { ok: true }
|
||||
}
|
||||
const unanswered = child.phase === 'starting'
|
||||
const closing = closeProviderRoot(context, sessionId)
|
||||
const closing = closeProviderRoot(context, sessionId, child)
|
||||
const closed = await (lock ? lock.wait(closing) : closing)
|
||||
if (closed) {
|
||||
const end = () =>
|
||||
@@ -96,6 +96,7 @@ export async function joinClosingStructuredAgentSessionChild(
|
||||
function closeProviderRoot(
|
||||
context: StructuredAgentSessionLifetimeContext,
|
||||
sessionId: string,
|
||||
child: StructuredAgentSessionProviderChild,
|
||||
observeClose?: AgentSessionCloseObserver
|
||||
): Promise<boolean | 'kill-delivered'> {
|
||||
const { adapter, logger } = context.deps
|
||||
@@ -107,7 +108,12 @@ function closeProviderRoot(
|
||||
return stopAgentSessionProviderRoot(
|
||||
async () => {
|
||||
const close = adapter.disposeSession ?? adapter.closeSession
|
||||
return (await close?.call(adapter, sessionId, observeClose)) === true
|
||||
// The clear's divider waits on this stop, so the old agent gets only a brief grace.
|
||||
const closing =
|
||||
child.close?.cause === 'context-clear'
|
||||
? close?.call(adapter, sessionId, observeClose, { grace: 'brief' })
|
||||
: close?.call(adapter, sessionId, observeClose)
|
||||
return (await closing) === true
|
||||
},
|
||||
// Proven exit stands; descendant/follow-up failures are log-only.
|
||||
(error) =>
|
||||
|
||||
@@ -3,6 +3,7 @@ import { spawnProcess } from '@orca/process-host'
|
||||
import { RetryableProcessExitProof } from '@orca/process-host/retryable-process-exit-proof'
|
||||
import type { ProviderProcessLaunch } from './provider-process-launch'
|
||||
import {
|
||||
PROVIDER_BRIEF_SIGTERM_GRACE_MS,
|
||||
PROVIDER_SUPERVISOR_MAX_STOP_MS,
|
||||
createProviderSpawnSpec,
|
||||
type ProviderSupervisorLifetime
|
||||
@@ -19,6 +20,7 @@ import {
|
||||
acceptProviderWriterClose,
|
||||
closeProviderProcess,
|
||||
rootOnlyProviderClosePolicy,
|
||||
type ProviderProcessCloseOptions,
|
||||
type ProviderProcessClosePolicy,
|
||||
type ProviderProcessCloseResult,
|
||||
type ProviderProcessTree
|
||||
@@ -72,9 +74,16 @@ export type ManagedProviderProcess = {
|
||||
stderrTail(): string
|
||||
onExit(listener: (exit: ProviderProcessExit) => void): void
|
||||
terminateTree(): Promise<ProviderProcessTeardownVerdict>
|
||||
close(tree?: ProviderProcessTree): Promise<ProviderProcessCloseResult>
|
||||
/** A close joining one already in flight shares that close's grace. */
|
||||
close(
|
||||
tree?: ProviderProcessTree,
|
||||
options?: ProviderProcessCloseOptions
|
||||
): Promise<ProviderProcessCloseResult>
|
||||
/** Writer retirement can finish while explicitly stricter tree cleanup remains retryable. */
|
||||
closeWriter(tree: ProviderProcessTree): Promise<ProviderProcessCloseResult>
|
||||
closeWriter(
|
||||
tree: ProviderProcessTree,
|
||||
options?: ProviderProcessCloseOptions
|
||||
): Promise<ProviderProcessCloseResult>
|
||||
}
|
||||
|
||||
export function spawnManagedProviderProcess(
|
||||
@@ -199,7 +208,10 @@ export function spawnManagedProviderProcess(
|
||||
return lastCloseResult?.killDelivered ? lastCloseResult : null
|
||||
}
|
||||
|
||||
const close = (tree?: ProviderProcessTree): Promise<ProviderProcessCloseResult> =>
|
||||
const close = (
|
||||
tree?: ProviderProcessTree,
|
||||
closeOptions?: ProviderProcessCloseOptions
|
||||
): Promise<ProviderProcessCloseResult> =>
|
||||
exitProof.run(async () => {
|
||||
const known = knownWriterClose(tree)
|
||||
if (known && !options.acceptClose) {
|
||||
@@ -217,7 +229,10 @@ export function spawnManagedProviderProcess(
|
||||
supervised: spec.supervised,
|
||||
policy,
|
||||
tree,
|
||||
terminateTree
|
||||
terminateTree,
|
||||
...(closeOptions?.grace === 'brief'
|
||||
? { briefGraceMs: PROVIDER_BRIEF_SIGTERM_GRACE_MS }
|
||||
: {})
|
||||
})
|
||||
lastCloseResult = {
|
||||
...result,
|
||||
@@ -261,9 +276,9 @@ export function spawnManagedProviderProcess(
|
||||
},
|
||||
terminateTree,
|
||||
close,
|
||||
closeWriter(tree) {
|
||||
closeWriter(tree, closeOptions) {
|
||||
const known = knownWriterClose(tree)
|
||||
return known ? Promise.resolve(known) : close(tree)
|
||||
return known ? Promise.resolve(known) : close(tree, closeOptions)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,7 +1,10 @@
|
||||
import type { SpawnedProcess } from '@orca/process-host/process-spec'
|
||||
import type { DescendantTreeVerdict } from '../pty/process-retirement/pty-descendant-exit-verification'
|
||||
import { waitForProcessExitUntil } from './provider-process-exit-deadline'
|
||||
import { PROVIDER_SUPERVISOR_MAX_STOP_MS } from './provider-process-supervisor'
|
||||
import {
|
||||
PROVIDER_SUPERVISOR_MAX_STOP_MS,
|
||||
PROVIDER_SUPERVISOR_REAP_NOW_SIGNAL
|
||||
} from './provider-process-supervisor'
|
||||
import type { ProviderProcessTeardownVerdict } from './provider-process-teardown'
|
||||
|
||||
export type ProviderProcessTree = {
|
||||
@@ -30,8 +33,13 @@ export type ProviderProcessCloseInput = {
|
||||
policy: ProviderProcessClosePolicy
|
||||
tree?: ProviderProcessTree
|
||||
terminateTree: () => Promise<ProviderProcessTeardownVerdict>
|
||||
/** Time after stdin end/SIGTERM before a supervisor is asked to reap now, or an unsupervised root is forced. */
|
||||
briefGraceMs?: number
|
||||
}
|
||||
|
||||
/** 'brief' is for a stop the user is waiting on; providers without a SIGTERM ladder may ignore it. */
|
||||
export type ProviderProcessCloseOptions = { grace?: 'brief' }
|
||||
|
||||
export type ProviderProcessCloseResult = {
|
||||
root: DescendantTreeVerdict
|
||||
/** Null when this close made no observation of the descendants. */
|
||||
@@ -56,6 +64,25 @@ export function acceptProviderWriterClose(result: ProviderProcessCloseResult): b
|
||||
return result.root === 'exited' || result.killDelivered === true
|
||||
}
|
||||
|
||||
// A supervised close without its SIGTERM has no ladder to shorten, so it keeps the full wait.
|
||||
async function waitForGracefulExit(
|
||||
input: ProviderProcessCloseInput,
|
||||
signalled: boolean
|
||||
): Promise<void> {
|
||||
const { gracefulExitMs } = input.policy
|
||||
if (input.briefGraceMs === undefined || (input.supervised && !signalled)) {
|
||||
return waitForProcessExitUntil(input.exitPromise, gracefulExitMs)
|
||||
}
|
||||
const briefMs = Math.min(input.briefGraceMs, gracefulExitMs)
|
||||
await waitForProcessExitUntil(input.exitPromise, briefMs)
|
||||
if (!signalled || input.rootVerdict() === 'exited') {
|
||||
return
|
||||
}
|
||||
// The supervisor still owns the reap: asked now, it SIGKILLs its group instead of waiting out the grace.
|
||||
input.child.kill(PROVIDER_SUPERVISOR_REAP_NOW_SIGNAL)
|
||||
await waitForProcessExitUntil(input.exitPromise, gracefulExitMs - briefMs)
|
||||
}
|
||||
|
||||
/** The supervisor owns the POSIX signal ladder; its wrapper must outlive that ladder. */
|
||||
export async function closeProviderProcess(
|
||||
input: ProviderProcessCloseInput
|
||||
@@ -69,14 +96,18 @@ export async function closeProviderProcess(
|
||||
} catch {
|
||||
// A broken pipe still owes the reap.
|
||||
}
|
||||
if (input.supervised && policy.signalSupervisorOnClose && input.rootVerdict() !== 'exited') {
|
||||
const signalled =
|
||||
input.supervised === true &&
|
||||
policy.signalSupervisorOnClose === true &&
|
||||
input.rootVerdict() !== 'exited'
|
||||
if (signalled) {
|
||||
child.kill('SIGTERM')
|
||||
}
|
||||
let reaped = false
|
||||
let fallbackTree: DescendantTreeVerdict | null = null
|
||||
let killDelivered = false
|
||||
if (input.rootVerdict() !== 'exited') {
|
||||
await waitForProcessExitUntil(input.exitPromise, policy.gracefulExitMs)
|
||||
await waitForGracefulExit(input, signalled)
|
||||
if (input.rootVerdict() !== 'exited') {
|
||||
reaped = true
|
||||
await tree?.refresh?.()
|
||||
|
||||
@@ -6,6 +6,7 @@ import { afterEach, describe, expect, it } from 'vitest'
|
||||
import {
|
||||
PROVIDER_STDIN_END_GRACE_MS,
|
||||
PROVIDER_SUPERVISOR_MAX_STOP_MS,
|
||||
PROVIDER_SUPERVISOR_REAP_NOW_SIGNAL,
|
||||
supervisedPosixLaunch,
|
||||
type ProviderSupervisorOptions
|
||||
} from './provider-process-supervisor'
|
||||
@@ -14,7 +15,9 @@ import { supervisedProviderSpawnFailure } from './provider-spawn-failure-report'
|
||||
// The provider leads its own group; its grandchild shares that group and ignores SIGTERM.
|
||||
const PROVIDER = String.raw`
|
||||
const { spawn } = require('node:child_process')
|
||||
if (process.env.ORCA_TEST_PROVIDER_IGNORES_SIGTERM) process.on('SIGTERM', () => {})
|
||||
if (process.env.ORCA_TEST_PROVIDER_IGNORES_SIGTERM) {
|
||||
process.on('SIGTERM', () => console.log(JSON.stringify({ sigterm: process.pid })))
|
||||
}
|
||||
if (process.env.ORCA_TEST_PROVIDER_SIGNAL_FILE) {
|
||||
process.on('SIGTERM', () => {
|
||||
require('node:fs').writeFileSync(process.env.ORCA_TEST_PROVIDER_SIGNAL_FILE, 'SIGTERM')
|
||||
@@ -267,6 +270,25 @@ describe.runIf(process.platform !== 'win32')('POSIX provider supervisor processe
|
||||
expect(alive(grandchild)).toBe(false)
|
||||
})
|
||||
|
||||
it('reaps the group at once when a stopping supervisor is asked to', async () => {
|
||||
const { supervisor, exit } = launchSupervisor({}, { ORCA_TEST_PROVIDER_IGNORES_SIGTERM: '1' })
|
||||
const { provider, grandchild } = await readPids(supervisor, ['provider', 'grandchild'])
|
||||
|
||||
// ast-grep-ignore: test-no-wall-clock -- the cut SIGTERM grace is the behaviour under test
|
||||
const startedAt = Date.now()
|
||||
supervisor.kill('SIGTERM')
|
||||
// Repeat only once the first stop reached the group, so the two signals cannot coalesce.
|
||||
await readPids(supervisor, ['sigterm'])
|
||||
expect(alive(-provider)).toBe(true)
|
||||
supervisor.kill(PROVIDER_SUPERVISOR_REAP_NOW_SIGNAL)
|
||||
|
||||
await expect(exit).resolves.toEqual({ code: null, signal: 'SIGTERM' })
|
||||
// ast-grep-ignore: test-no-wall-clock -- the full grace is 3 s; a reap-now request must not wait it out
|
||||
expect(Date.now() - startedAt).toBeLessThan(1_500)
|
||||
expect(alive(provider)).toBe(false)
|
||||
expect(alive(grandchild)).toBe(false)
|
||||
})
|
||||
|
||||
it('lets a session provider group finish its SIGTERM cleanup after the provider exits', async () => {
|
||||
const cleanupFile = join(tempDir(), 'grandchild-cleaned-up')
|
||||
const { supervisor, exit } = launchSupervisor(
|
||||
|
||||
@@ -5,6 +5,11 @@ import { PROVIDER_SPAWN_FAILURE_MARKER } from './provider-spawn-failure-report'
|
||||
export const PROVIDER_STDIN_END_GRACE_MS = 1_000
|
||||
/** Time the provider group gets to flush and exit after SIGTERM, before SIGKILL. */
|
||||
export const PROVIDER_SIGTERM_GRACE_MS = 3_000
|
||||
/** A stop the user waits on (context clear) asks the supervisor to reap its group after this;
|
||||
* Claude itself usually exits within it, so only a straggler in its group gets killed. */
|
||||
export const PROVIDER_BRIEF_SIGTERM_GRACE_MS = 1_000
|
||||
/** Asks a stopping supervisor to reap its group now; a repeated SIGTERM stays a plain stop. */
|
||||
export const PROVIDER_SUPERVISOR_REAP_NOW_SIGNAL = 'SIGUSR2'
|
||||
/** Bound kernel kill completion; a process stuck on a hung drive must not hang close/recovery forever. */
|
||||
export const PROVIDER_GROUP_REAP_TIMEOUT_MS = 1_500
|
||||
/** Longest a supervisor waits, after its provider exits, to relay output still in the pipes. */
|
||||
@@ -24,6 +29,10 @@ const [providerCommand, ...providerArgs] = process.argv.slice(1)
|
||||
const ownerGone = () => process.ppid !== spec.ownerPid
|
||||
// Registered before the spawn, so a stop that lands while the provider starts still reaps it.
|
||||
for (const signal of ['SIGTERM', 'SIGINT', 'SIGHUP']) process.on(signal, () => stopProviderGroup(signal))
|
||||
process.on(${JSON.stringify(PROVIDER_SUPERVISOR_REAP_NOW_SIGNAL)}, () => {
|
||||
sigtermGraceCut = true
|
||||
stopProviderGroup('SIGTERM')
|
||||
})
|
||||
// Orca can die before this runs; spawning then would start a provider nothing watches.
|
||||
if (ownerGone()) process.exit(1)
|
||||
const childEnv = { ...process.env, ...spec.nodeEnv }
|
||||
@@ -63,6 +72,7 @@ if (reports) {
|
||||
let timer
|
||||
let ownerShutdownTimer
|
||||
let settling = false
|
||||
let sigtermGraceCut = false
|
||||
let providerExited = false
|
||||
const providerGroupExists = () => {
|
||||
if (!child.pid) return false
|
||||
@@ -73,10 +83,10 @@ const providerGroupExists = () => {
|
||||
return Boolean(error && error.code !== 'ESRCH')
|
||||
}
|
||||
}
|
||||
const waitForProviderGroupExit = async (timeoutMs, untilProviderExits = false) => {
|
||||
const waitForProviderGroupExit = async (timeoutMs, stopWaiting = () => false) => {
|
||||
const deadline = Date.now() + timeoutMs
|
||||
while (providerGroupExists()) {
|
||||
if (Date.now() >= deadline || (untilProviderExits && providerExited)) return false
|
||||
if (Date.now() >= deadline || stopWaiting()) return false
|
||||
await new Promise((resolve) => setTimeout(resolve, 25))
|
||||
}
|
||||
return true
|
||||
@@ -94,8 +104,8 @@ const finishWithProviderOutcome = (code, signal) => {
|
||||
process.removeAllListeners(signal)
|
||||
process.kill(process.pid, signal)
|
||||
}
|
||||
// Every stop is the same: SIGTERM the group, SIGKILL it after its grace, and exit only once it is
|
||||
// gone. Whoever stops this pid judges the provider by it, so a dead supervisor means a dead group.
|
||||
// Every stop is the same: SIGTERM the group, SIGKILL it after its grace (or once asked to reap now), and exit
|
||||
// only once it is gone. Whoever stops this pid judges the provider by it, so a dead supervisor means a dead group.
|
||||
const stopProviderGroup = (receivedSignal) => {
|
||||
if (settling) return
|
||||
settling = true
|
||||
@@ -103,7 +113,8 @@ const stopProviderGroup = (receivedSignal) => {
|
||||
if (ownerShutdownTimer) clearTimeout(ownerShutdownTimer)
|
||||
try { process.kill(-child.pid, 'SIGTERM') } catch {}
|
||||
// A one-shot's helpers die with it once it has exited; a session's keep the grace to clean up.
|
||||
void waitForProviderGroupExit(spec.sigtermGraceMs, spec.lifetime === 'one-shot')
|
||||
const graceOver = () => sigtermGraceCut || (spec.lifetime === 'one-shot' && providerExited)
|
||||
void waitForProviderGroupExit(spec.sigtermGraceMs, graceOver)
|
||||
.then((exited) => exited || reapOwnedProviderGroup())
|
||||
.then((reaped) => {
|
||||
if (!reaped) return process.exit(1)
|
||||
|
||||
Reference in New Issue
Block a user